imfact-summarizer-sequential.py

import json
import boto3
import time
from botocore.exceptions import ClientError

s3 = boto3.client("s3")
bedrock = boto3.client(service_name="bedrock-runtime", region_name="ap-northeast-2")

MODEL_ID = "apac.amazon.nova-micro-v1:0"
RETRY_LIMIT = 3       # 요청 실패 시 재시도 횟수
DELAY_BETWEEN_CALLS = 0.1  # 요청 간 대기 시간 (초)

def summarize_text(text: str) -> str:
    """단일 뉴스 기사 요약 (Bedrock Nova Micro 모델 호출)"""
    instruction = (
        "다음은 뉴스 기사 원문입니다.\n"
        "이 기사의 핵심 내용을 정확히 3문장(서론·본론·결론 구조)으로 요약해 주세요.\n\n"
        "지침:\n"
        "1. 평어체(신문 기사문체)를 사용합니다.\n"
        "2. 객관적·보도문형 어투로 작성합니다. (예: '~다', '~했다', '~에 따르면')\n"
        "3. 반드시 3문장으로 구성합니다.\n"
        "   - 1문장: 기사에서 다루는 가장 중요한 문제를 명확히 제시합니다.\n"
        "   - 2문장: 그 문제의 근거 또는 원인을 구체적 사실이나 데이터로 설명합니다.\n"
        "   - 3문장: 기사에서 제시하는 결론, 주장, 또는 해결 방안을 요약합니다.\n"
        "4. 쉼표(,)를 사용하지 않습니다. 각 문장은 짧고 명확하게 완결된 형태로 작성합니다.\n"
        "5. 줄바꿈, 주석, 설명, 질문 없이 요약문만 출력합니다.\n\n"
        "---\n\n"
        f"기사 원문:\n{text}\n\n---"
    )

    for attempt in range(RETRY_LIMIT):
        try:
            response = bedrock.converse(
                modelId=MODEL_ID,
                messages=[
                    {
                        "role": "user",
                        "content": [{"text": instruction}]
                    }
                ],
                inferenceConfig={
                    "maxTokens": 512,
                    "temperature": 0.7,
                },
            )

            summary = response["output"]["message"]["content"][0]["text"].strip()
            return summary

        except (ClientError, Exception) as e:
            print(f"ERROR: Can't invoke '{MODEL_ID}' ({attempt+1}/{RETRY_LIMIT}). Reason: {e}")
            time.sleep(1.0 * (attempt + 1))  # 재시도 간 대기
    return ""


def summarize_articles(articles):
    """모든 기사를 순차적으로 요약"""
    results = []
    success, fail = 0, 0

    for idx, article in enumerate(articles, 1):
        text = article.get("body", "")
        print(f"[{idx}/{len(articles)}] 요약 중...")

        summary = summarize_text(text)
        if summary:
            success += 1
        else:
            fail += 1

        article["summary"] = summary
        results.append(article)

        # Bedrock 호출 간 대기 (Throttling 방지)
        time.sleep(DELAY_BETWEEN_CALLS)

    print(f"[요약 완료] 성공: {success}, 실패: {fail}")
    return results


def lambda_handler(event, context):
    print("[ImFact] S3 → Bedrock Summarizer Triggered")

    start = time.time()

    for record in event["Records"]:
        bucket = record["s3"]["bucket"]["name"]
        key = record["s3"]["object"]["key"]

        if not key.startswith("curated/"):
            print(f"Skip: {key} (not under curated/)")
            continue

        print(f"Input File: s3://{bucket}/{key}")

        try:
            # S3에서 파일 로드
            obj = s3.get_object(Bucket=bucket, Key=key)
            lines = obj["Body"].read().decode("utf-8").strip().split("\n")
            articles = [json.loads(line) for line in lines if line.strip()]
            print(f"총 {len(articles)}건 기사 처리 시작")

            # 요약 수행
            summarized_articles = summarize_articles(articles)

            # S3에 저장
            output_key = key.replace(
                "curated/news_curated.jsonl", "summarized/news_summarized.jsonl"
            )
            output_data = "\n".join(
                json.dumps(a, ensure_ascii=False) for a in summarized_articles
            ) + "\n"

            s3.put_object(
                Bucket=bucket,
                Key=output_key,
                Body=output_data.encode("utf-8"),
                ContentType="application/json",
            )

            print(f"[저장 완료] s3://{bucket}/{output_key}")

        except Exception as e:
            print(f"Lambda Error: {e}")

    end = time.time()
    print(f"총 처리 시간: {end - start:.2f}초")

    return {"status": "OK"}