fix: added finfast collectors

app/collectors/finfast/article.py

@@ -0,0 +1,114 @@
+"""Module for collecting and managing article data from DynamoDB to MongoDB."""
+from venv import logger
+from datetime import datetime, timedelta
+from database import FinFastMongoClient as MongodbClient
+from pymongo.errors import PyMongoError
+from .utils import scan_dynamodb_table, delete_old_documents
+
+collection = MongodbClient["FinFAST_China"]["Article"]
+
+
+def _process_article_item(item):
+    """
+    Process a single article item by converting data types and upserting to MongoDB.
+    
+    Args:
+        item (dict): The article item to process
+        
+    Raises:
+        ValueError: If data type conversion fails
+        KeyError: If required keys are missing
+        TypeError: If type conversion fails
+        PyMongoError: If MongoDB operation fails
+    """
+    try:
+        # Convert sentimentScore to float
+        item["sentimentScore"] = float(item.get("sentimentScore", 0.0))
+        # Set _id and remove id
+        item["_id"] = item.pop("id", None)
+
+        # Convert entityList inner values to float (for MongoDB compatibility)
+        if "entityList" in item and isinstance(item["entityList"], list):
+            for entity in item["entityList"]:
+                if isinstance(entity, dict):
+                    if "sentimentScore" in entity:
+                        try:
+                            entity["sentimentScore"] = float(entity["sentimentScore"])
+                        except (ValueError, TypeError):
+                            entity["sentimentScore"] = 0.0
+                    if "occurrence" in entity:
+                        try:
+                            entity["occurrence"] = float(entity["occurrence"])
+                        except (ValueError, TypeError):
+                            entity["occurrence"] = 0.0
+
+        # Upsert into MongoDB
+        collection.update_one(
+            {'_id': item['_id']},
+            {'$set': item},
+            upsert=True
+        )
+        print(f"Successfully processed item: {item['_id']}")
+
+    except (ValueError, KeyError, TypeError, PyMongoError) as e:
+        logger.error("Error processing item with _id %s: %s",
+                     item.get('_id', 'unknown'), e)
+
+
+def upsert_documents(filter_date):
+    """
+    Scan and upsert documents from DynamoDB Article_China table to MongoDB collection.
+
+    This function scans the DynamoDB table for articles published on or after the specified
+    filter date, processes each item by converting data types and field names, then upserts
+    them into the MongoDB collection. The operation handles pagination automatically using
+    LastEvaluatedKey from DynamoDB scan responses.
+
+    Args:
+        filter_date (str): The minimum publish date in 'YYYY-MM-DD' format to filter articles.
+
+    Raises:
+        ClientError: If there is an error during DynamoDB scan operation
+        BotoCoreError: If there is a boto3 core error
+
+    Note:
+        - Converts sentimentScore to float type
+        - Renames 'id' field to '_id' for MongoDB compatibility
+        - Uses upsert operation to avoid duplicates
+    """
+    scan_dynamodb_table("Article_China", filter_date, _process_article_item)
+
+
+def collect():
+    """
+    Main collection function that orchestrates the article data collection process.
+
+    This function performs a complete data collection cycle by:
+    1. Calculating a filter date (30 days ago from current date)
+    2. Scanning and upserting recent articles from DynamoDB to MongoDB
+    3. Deleting articles older than 30 days from MongoDB to maintain data freshness
+
+    The function maintains a rolling 30-day window of article data, ensuring that
+    only recent and relevant articles are kept in the MongoDB collection.
+
+    Args:
+        None
+
+    Returns:
+        None
+
+    Raises:
+        Exception: Propagates any exceptions from upsert_documents or delete_old_documents.
+
+    Note:
+        This function is typically called as part of a scheduled data collection process.
+    """
+    # Calculate date 60 days ago
+    sixty_days_ago = datetime.now() - timedelta(days=60)
+    filter_date = sixty_days_ago.strftime('%Y-%m-%d')
+
+    # Scan and process items
+    upsert_documents(filter_date)
+
+    # Delete documents older than 60 days
+    delete_old_documents(collection, filter_date, logger)

app/collectors/finfast/entity.py

@@ -0,0 +1,103 @@
+"""Module for collecting and managing entity data from DynamoDB to MongoDB."""
+from datetime import datetime, timedelta
+from database import FinFastMongoClient as MongodbClient
+from pymongo.errors import PyMongoError
+from .utils import scan_dynamodb_table, delete_old_documents
+
+
+entity_collection = MongodbClient["FinFAST_China"]["Entity"]
+
+
+def _process_entity_item(item):
+    """
+    Process a single entity item by converting data types and upserting to MongoDB.
+    
+    Args:
+        item (dict): The entity item to process
+        
+    Raises:
+        ValueError: If data type conversion fails
+        KeyError: If required keys are missing
+        TypeError: If type conversion fails
+        PyMongoError: If MongoDB operation fails
+    """
+    try:
+        # Convert fields to appropriate types
+        item["sentimentScore"] = float(item.get("sentimentScore", 0.0))
+        item["confidenceLevel"] = float(item.get("confidenceLevel", 0.0))
+        item["occurrence"] = int(item.get("occurrence", 0))
+        # Create composite _id
+        item["_id"] = f"{item.get('entity', '')}-{item.get('articleID', '')}"
+
+        # Upsert into MongoDB
+        entity_collection.update_one(
+            {'_id': item['_id']},
+            {'$set': item},
+            upsert=True
+        )
+        print(f"Successfully processed item: {item['_id']}")
+
+    except (ValueError, KeyError, TypeError, PyMongoError) as e:
+        print(f"Error processing item with _id {item.get('_id', 'unknown')}: {e}")
+
+
+def upsert_documents(filter_date):
+    """
+    Scan and upsert entity documents from DynamoDB article_entity_china table to MongoDB collection.
+
+    This function scans the DynamoDB table for entity records published on or after the specified
+    filter date, processes each item by converting data types and creating composite IDs, then 
+    upserts them into the MongoDB collection. The operation handles pagination automatically using
+    LastEvaluatedKey from DynamoDB scan responses.
+
+    Args:
+        filter_date (str): The minimum publish date in 'YYYY-MM-DD' format to filter entity records.
+
+    Raises:
+        ClientError: If there is an error during DynamoDB scan operation
+        BotoCoreError: If there is a boto3 core error
+
+    Note:
+        - Converts sentimentScore and confidenceLevel to float type
+        - Converts occurrence to int type
+        - Creates composite _id using format: '{entity}-{articleID}'
+        - Uses upsert operation to avoid duplicates
+    """
+    scan_dynamodb_table('article_entity_china', filter_date, _process_entity_item)
+
+
+def collect():
+    """
+    Main collection function that orchestrates the entity data collection process.
+
+    This function performs a complete entity data collection cycle by:
+    1. Calculating a filter date (30 days ago from current date)
+    2. Scanning and upserting recent entity records from DynamoDB to MongoDB
+    3. Deleting entity records older than 30 days from MongoDB to maintain data freshness
+
+    The function maintains a rolling 30-day window of entity data, ensuring that
+    only recent and relevant entity associations are kept in the MongoDB collection.
+
+    Args:
+        None
+
+    Returns:
+        None
+
+    Raises:
+        Exception: Propagates any exceptions from upsert_documents or delete_old_documents.
+
+    Note:
+        This function is typically called as part of a scheduled data collection process.
+        The timezone consideration (HKT) is noted in the original comment but uses system time.
+    """
+    # Calculate date 30 days ago (accounting for HKT timezone)
+    thirty_days_ago = datetime.now() - timedelta(days=30)
+    filter_date = thirty_days_ago.strftime('%Y-%m-%d')
+
+    # Scan and process items
+    upsert_documents(filter_date)
+
+    # Delete documents older than 30 days
+    delete_old_documents(entity_collection, filter_date)
+    

app/collectors/finfast/utils.py

@@ -0,0 +1,107 @@
+"""Utility functions for FinFast data collection from DynamoDB to MongoDB."""
+
+import os
+import boto3
+from boto3.dynamodb.types import TypeDeserializer
+from pymongo.errors import PyMongoError
+from botocore.exceptions import ClientError, BotoCoreError
+
+
+def get_dynamodb_client():
+    """
+    Create and return a DynamoDB client using AWS credentials from environment variables.
+    
+    Returns:
+        boto3.client: Configured DynamoDB client
+    """
+    aws_access_key_id = os.getenv('AWS_ACCESS_KEY_ID')
+    aws_secret_access_key = os.getenv('AWS_SECRET_ACCESS_KEY')
+
+    return boto3.client(
+        service_name='dynamodb',
+        region_name='us-east-1',
+        aws_access_key_id=aws_access_key_id,
+        aws_secret_access_key=aws_secret_access_key
+    )
+
+
+def scan_dynamodb_table(table_name, filter_date, item_processor):
+    """
+    Scan a DynamoDB table and process items using the provided processor function.
+    
+    Args:
+        table_name (str): Name of the DynamoDB table to scan
+        filter_date (str): Filter date in 'YYYY-MM-DD' format
+        item_processor (callable): Function to process each item
+        
+    Raises:
+        ClientError: If there is an error during DynamoDB scan operation
+        BotoCoreError: If there is a boto3 core error
+    """
+    dynamodb = get_dynamodb_client()
+    last_evaluated_key = None
+    deserializer = TypeDeserializer()
+
+    try:
+        while True:
+            # Prepare scan parameters
+            scan_params = {
+                'TableName': table_name,
+                'FilterExpression': 'publishDate >= :date',
+                'ExpressionAttributeValues': {
+                    ':date': {'S': filter_date}
+                }
+            }
+
+            # Add ExclusiveStartKey if it exists
+            if last_evaluated_key:
+                scan_params['ExclusiveStartKey'] = last_evaluated_key
+
+            # Scan the table
+            response = dynamodb.scan(**scan_params)
+            items = [
+                {k: deserializer.deserialize(v) for k, v in item.items()}
+                for item in response.get('Items', [])
+            ]
+
+            # Process items using the provided processor
+            for item in items:
+                item_processor(item)
+
+            last_evaluated_key = response.get('LastEvaluatedKey')
+            if not last_evaluated_key:
+                break
+
+    except (ClientError, BotoCoreError) as e:
+        print(f"Error in scan operation: {e}")
+        raise
+
+
+def delete_old_documents(collection, cutoff_date, use_logger=None):
+    """
+    Delete documents from MongoDB collection that are older than the specified cutoff date.
+    
+    Args:
+        collection: MongoDB collection object
+        cutoff_date (str): The cutoff date in 'YYYY-MM-DD' format
+        use_logger: Optional logger object. If provided, will use logger, otherwise use print
+        
+    Raises:
+        PyMongoError: If there is an error during the delete operation
+    """
+    try:
+        result = collection.delete_many({
+            'publishDate': {'$lt': cutoff_date}
+        })
+        message = f"Deleted {result.deleted_count} documents older than {cutoff_date}"
+        if use_logger:
+            use_logger.info("Deleted %d documents older than %s", result.deleted_count, cutoff_date)
+        else:
+            print(message)
+    except PyMongoError as e:
+        error_message = f"Error deleting old documents: {e}"
+        if use_logger:
+            use_logger.error("Error deleting old documents: %s", e)
+        else:
+            print(error_message)
+        raise