Merge pull request #208 from codeforjapan/feat/add-skip-logic

refactor: add `skip_topic_detect` and `skip_tweet_lookup` flags acros…

Author: yu23ki14

etl/seed/settings.json

@@ -2,8 +2,8 @@
   "filter": {
     "languages": ["ja"],
     "keywords": [],
-    "start_millis": 1768834800000,
+    "start_millis": 1762009200000,
     "end_millis": 1771858740000
   },
-  "description": "Note filtering configuration. start_millis is required. end_millis is optional (null = no upper limit). Timestamps in milliseconds UTC. Current: 2025-01-20 00:00:00 JST ~ 2025-02-23 23:59:59 JST"
+  "description": "Note filtering configuration. start_millis is required. end_millis is optional (null = no upper limit). Timestamps in milliseconds UTC. Current: 2025-11-01 00:00:00 JST ~ 2026-02-23 23:59:00 JST"
 }

etl/src/birdxplorer_etl/extract_ecs.py

@@ -173,7 +173,7 @@ def extract_data(postgresql: Session):
 
                     # バッチ処理後にSQSキューイング（新規追加のみ）
                     for note in rows_to_add.values():
-                        enqueue_notes(note.note_id, note.summary, note.tweet_id)
+                        enqueue_notes(note.note_id, note.summary, note.tweet_id, note.language)
 
                     rows_to_add = {}
                     rows_to_update = []
@@ -185,7 +185,7 @@ def extract_data(postgresql: Session):
 
             # 最後のバッチのSQSキューイング（新規追加のみ）
             for note in rows_to_add.values():
-                enqueue_notes(note.note_id, note.summary, note.tweet_id)
+                enqueue_notes(note.note_id, note.summary, note.tweet_id, note.language)
 
             status_url = (
                 f"https://ton.twimg.com/birdwatch-public-data/{dateString}/"
@@ -280,17 +280,18 @@ def extract_data(postgresql: Session):
     return
 
 
-def enqueue_notes(note_id: str, summary: str, post_id: str = None):
+def enqueue_notes(note_id: str, summary: str, post_id: str = None, language: str = None):
     """
     ノート処理用のSQSキューにメッセージを送信
     lang-detect-queueに送信（summaryとpost_idも含める）
     """
     sqs_client = boto3.client("sqs", region_name=os.environ.get("AWS_REGION", "ap-northeast-1"))
 
     # lang-detect-queue用のメッセージ（summaryとpost_idを含める）
-    lang_detect_message = json.dumps(
-        {"note_id": note_id, "summary": summary, "post_id": post_id, "processing_type": "language_detect"}
-    )
+    message = {"note_id": note_id, "summary": summary, "post_id": post_id, "processing_type": "language_detect"}
+    if language:
+        message["language"] = language
+    lang_detect_message = json.dumps(message)
 
     # lang-detect-queueに送信
     try:
@@ -324,4 +325,3 @@ def enqueue_note_status_update(note_id: str):
         logging.info(f"Enqueued note {note_id} to note-status-update queue, messageId={response.get('MessageId')}")
     except Exception as e:
         logging.error(f"Failed to enqueue note {note_id} to note-status-update queue: {e}")
-

etl/src/birdxplorer_etl/lib/lambda_handler/language_detect_lambda.py

@@ -4,7 +4,9 @@
 
 from birdxplorer_etl import settings
 from birdxplorer_etl.lib.ai_model.ai_model_interface import get_ai_service
-from birdxplorer_etl.lib.lambda_handler.common.retry_handler import call_ai_api_with_retry
+from birdxplorer_etl.lib.lambda_handler.common.retry_handler import (
+    call_ai_api_with_retry,
+)
 from birdxplorer_etl.lib.lambda_handler.common.sqs_handler import SQSHandler
 
 # Lambda用のロガー設定
@@ -37,6 +39,7 @@ def lambda_handler(event, context):
     try:
         note_id = None
         summary = None
+        existing_language = None
 
         # SQSイベントの場合
         if "Records" in event:
@@ -53,6 +56,7 @@ def lambda_handler(event, context):
                     if processing_type == "language_detect":
                         note_id = message_body.get("note_id")
                         summary = message_body.get("summary")
+                        existing_language = message_body.get("language")
                         logger.info(f"Found language_detect message for note_id: {note_id}")
                         break
                     else:
@@ -71,16 +75,20 @@ def lambda_handler(event, context):
         if note_id and summary:
             logger.info(f"[START] Detecting language for note: {note_id}")
 
-            ai_service = get_ai_service()
-
-            # 言語判定を実行（リトライ付き）
-            logger.info(f"[PROCESSING] Calling AI service for language detection...")
-            detected_language = call_ai_api_with_retry(
-                ai_service.detect_language,
-                summary,
-                max_retries=3,
-                initial_delay=1.0,
-            )
+            if existing_language:
+                logger.info(f"[SKIP] Language already known for note {note_id}: {existing_language}")
+                detected_language = existing_language
+            else:
+                ai_service = get_ai_service()
+
+                # 言語判定を実行（リトライ付き）
+                logger.info(f"[PROCESSING] Calling AI service for language detection...")
+                detected_language = call_ai_api_with_retry(
+                    ai_service.detect_language,
+                    summary,
+                    max_retries=3,
+                    initial_delay=1.0,
+                )
 
             logger.info(f"[SUCCESS] Language detected for note {note_id}: {detected_language}")
 

etl/src/birdxplorer_etl/lib/lambda_handler/note_transform_lambda.py

@@ -9,8 +9,10 @@
 
 from birdxplorer_common.storage import (
     NoteRecord,
+    NoteTopicAssociation,
     RowNoteRecord,
     RowNoteStatusRecord,
+    RowPostRecord,
     TopicRecord,
 )
 from birdxplorer_etl import settings
@@ -204,8 +206,31 @@ def process_single_message(message: dict, postgresql, sqs_handler, topics_cache:
     existing_note = postgresql.query(NoteRecord).filter(NoteRecord.note_id == note_id).first()
 
     if existing_note:
-        logger.info(f"Note already exists in notes table: {note_id}")
-        return {"note_id": note_id, "status": "skipped", "message": "Note already exists"}
+        logger.info(f"Note already exists in notes table: {note_id}, checking downstream status")
+
+        # note_topicテーブルでトピック設定済みかチェック
+        has_topics = (
+            postgresql.query(NoteTopicAssociation).filter(NoteTopicAssociation.note_id == note_id).first() is not None
+        )
+
+        # row_postsテーブルで投稿取得済みかチェック
+        post_id = existing_note.post_id
+        has_post = False
+        if post_id:
+            has_post = postgresql.query(RowPostRecord).filter(RowPostRecord.post_id == post_id).first() is not None
+
+        logger.info(f"Note {note_id}: has_topics={has_topics}, has_post={has_post}")
+
+        return {
+            "note_id": note_id,
+            "status": "existing",
+            "detected_language": str(existing_note.language) if existing_note.language else "",
+            "summary": existing_note.summary,
+            "post_id": str(post_id) if post_id else None,
+            "created_at_millis": int(existing_note.created_at) if existing_note.created_at else None,
+            "skip_topic_detect": has_topics,
+            "skip_tweet_lookup": has_post,
+        }
 
     # 言語を取得（優先順位: メッセージ > DB）
     detected_language = message_language or note_row.language
@@ -233,8 +258,7 @@ def process_single_message(message: dict, postgresql, sqs_handler, topics_cache:
         )
         if sent_message_id:
             logger.info(
-                f"Language not available for note {note_id}, "
-                f"requeued with retry_count={retry_count + 1}, delay=30s"
+                f"Language not available for note {note_id}, " f"requeued with retry_count={retry_count + 1}, delay=30s"
             )
             return {"note_id": note_id, "status": "requeued", "retry_count": retry_count + 1}
         else:
@@ -311,11 +335,7 @@ def lambda_handler(event, context):
                 logger.error(f"Commit error: {e}")
                 postgresql.rollback()
                 # コミット失敗時は全メッセージを失敗扱い
-                return {
-                    "batchItemFailures": [
-                        {"itemIdentifier": record.get("messageId")} for record in records
-                    ]
-                }
+                return {"batchItemFailures": [{"itemIdentifier": record.get("messageId")} for record in records]}
 
         # 成功したノートに対してtopic-detect-queueに送信
         settings_config = load_settings()
@@ -327,7 +347,7 @@ def lambda_handler(event, context):
 
         topic_detect_queued = 0
         for result in results:
-            if result.get("status") != "success":
+            if result.get("status") not in ("success", "existing"):
                 continue
 
             note_id = result["note_id"]
@@ -356,6 +376,8 @@ def lambda_handler(event, context):
                 "post_id": post_id,
                 "topics": lambda_handler._topics_cache,
                 "processing_type": "topic_detect",
+                "skip_topic_detect": result.get("skip_topic_detect", False),
+                "skip_tweet_lookup": result.get("skip_tweet_lookup", False),
             }
 
             if sqs_handler.send_message(queue_url=settings.TOPIC_DETECT_QUEUE_URL, message_body=topic_detect_message):
@@ -376,9 +398,7 @@ def lambda_handler(event, context):
         logger.error(f"Lambda execution error: {str(e)}")
         # 全体エラー時は全メッセージを失敗扱い
         return {
-            "batchItemFailures": [
-                {"itemIdentifier": record.get("messageId")} for record in event.get("Records", [])
-            ]
+            "batchItemFailures": [{"itemIdentifier": record.get("messageId")} for record in event.get("Records", [])]
         }
     finally:
         postgresql.close()

etl/src/birdxplorer_etl/lib/lambda_handler/postlookup_lambda.py

@@ -150,13 +150,16 @@ def lambda_handler(event: dict, context: Any) -> dict:
     try:
         tweet_id = None
 
+        skip_tweet_lookup = False
+
         # SQSイベントの場合
         if "Records" in event:
             for record in event["Records"]:
                 try:
                     message_body = json.loads(record["body"])
                     if message_body.get("processing_type") == "tweet_lookup":
                         tweet_id = message_body.get("tweet_id")
+                        skip_tweet_lookup = message_body.get("skip_tweet_lookup", False)
                         break
                 except json.JSONDecodeError as e:
                     logger.error(f"Failed to parse SQS message body: {e}")
@@ -166,6 +169,24 @@ def lambda_handler(event: dict, context: Any) -> dict:
         elif "tweet_id" in event:
             tweet_id = event["tweet_id"]
 
+        if tweet_id and skip_tweet_lookup:
+            logger.info(f"[SKIP] Post already fetched for tweet: {tweet_id}")
+            # POST_TRANSFORM_QUEUEへの転送のみ実行
+            post_transform_queue_url = os.environ.get("POST_TRANSFORM_QUEUE_URL")
+            if post_transform_queue_url:
+                transform_message = {
+                    "operation": "transform_post",
+                    "post_id": tweet_id,
+                    "retry_count": 0,
+                }
+                sqs_handler.send_message(
+                    queue_url=post_transform_queue_url,
+                    message_body=transform_message,
+                    delay_seconds=5,
+                )
+                logger.info(f"[SQS_SUCCESS] Sent transform request for skipped tweet {tweet_id}")
+            return {"statusCode": 200, "body": json.dumps({"skipped": True, "tweet_id": tweet_id})}
+
         if tweet_id:
             logger.info(f"Looking up tweet: {tweet_id}")
 
@@ -217,11 +238,13 @@ def lambda_handler(event: dict, context: Any) -> dict:
 
                     return {
                         "statusCode": 200,
-                        "body": json.dumps({
-                            "rate_limited": True,
-                            "tweet_id": tweet_id,
-                            "delay_seconds": delay_seconds,
-                        }),
+                        "body": json.dumps(
+                            {
+                                "rate_limited": True,
+                                "tweet_id": tweet_id,
+                                "delay_seconds": delay_seconds,
+                            }
+                        ),
                     }
 
                 # 削除/非公開/その他エラーの場合はスキップ
@@ -236,12 +259,14 @@ def lambda_handler(event: dict, context: Any) -> dict:
                 # 正常終了として返す（DLQに送らない）
                 return {
                     "statusCode": 200,
-                    "body": json.dumps({
-                        "skipped": True,
-                        "tweet_id": tweet_id,
-                        "reason": status,
-                        "detail": detail,
-                    }),
+                    "body": json.dumps(
+                        {
+                            "skipped": True,
+                            "tweet_id": tweet_id,
+                            "reason": status,
+                            "detail": detail,
+                        }
+                    ),
                 }
 
             # dataがない場合は予期しないエラー