Xのタイムラインを自動要約！FastAPI + LLMで作るトピックダイジェストシステムの設計

ユーザー（ブラウザ）
    ↓
React + Vite（フロントエンド: port 3000）
    ↓ REST API
FastAPI（バックエンド: port 8000）
    ↓
PostgreSQL（データベース: port 5432）

[非同期タスク - 実装予定]
FastAPI BackgroundTasks / Celery
    ├── X API v2 クライアント（ツイート収集）
    └── OpenAI API クライアント（LLM 要約）

# backend/main.py
from pydantic_settings import BaseSettings, SettingsConfigDict
from functools import lru_cache

class Settings(BaseSettings):
    x_bearer_token: str = ""
    openai_api_key: str = ""
    database_url: str = "sqlite:///./test.db"
    secret_key: str = ""
    llm_model_name: str = "gpt-3.5-turbo"

    model_config = SettingsConfigDict(
        env_file="../.env",
        env_file_encoding="utf-8",
        extra="ignore",  # .env に未定義のキーがあっても無視
    )

@lru_cache()
def get_settings() -> Settings:
    return Settings()

# backend/model.py
from sqlalchemy import Column, Integer, String, Text, DateTime, func
from database import Base

class Topic(Base):
    __tablename__ = "topics"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True, nullable=False)       # トピック名（例：「LLM」）
    keywords = Column(Text, nullable=True)                   # 検索キーワード（カンマ区切り）
    created_at = Column(DateTime, server_default=func.now())
    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())

# backend/main.py
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session

app = FastAPI(title="TopicDigestX API")

@app.get("/health")
async def health_check():
    return {"status": "ok"}

@app.get("/topics")
async def list_topics(
    db: Session = Depends(get_db),
    settings: Settings = Depends(get_settings),
):
    topics = db.query(Topic).all()
    return {"topics": topics}

@app.post("/topics")
async def create_topic(
    topic_data: TopicCreate,
    db: Session = Depends(get_db),
):
    topic = Topic(**topic_data.dict())
    db.add(topic)
    db.commit()
    db.refresh(topic)
    return topic

# docker-compose.yml
services:
  backend:
    build: ./backend
    ports: ["8000:8000"]
    volumes: ["./backend:/app"]
    env_file: [.env]
    depends_on: [db]
    command: uvicorn main:app --host 0.0.0.0 --port 8000 --reload

  db:
    image: postgres:15-alpine
    volumes: ["postgres_data:/var/lib/postgresql/data"]
    env_file: [.env]
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER}"]
      interval: 5s

  frontend:
    build: ./frontend
    ports: ["3000:5173"]  # Vite の dev server
    volumes:
      - ./frontend:/app
      - /app/node_modules  # コンテナ内の node_modules を優先
    environment:
      - CHOKIDAR_USEPOLLING=true  # ホットリロード対応

volumes:
  postgres_data:

# backend/clients/x_client.py（実装予定）
import httpx

class XClient:
    BASE_URL = "https://api.twitter.com/2"

    def __init__(self, bearer_token: str):
        self.headers = {"Authorization": f"Bearer {bearer_token}"}

    async def search_recent(self, query: str, max_results: int = 10) -> list[dict]:
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.BASE_URL}/tweets/search/recent",
                headers=self.headers,
                params={
                    "query": f"{query} lang:ja",
                    "max_results": max_results,
                    "tweet.fields": "created_at,author_id,public_metrics",
                },
            )
            response.raise_for_status()
            return response.json()["data"]

# backend/clients/llm_client.py（実装予定）
from openai import AsyncOpenAI

async def summarize_tweets(tweets: list[str], topic: str) -> str:
    client = AsyncOpenAI()
    
    tweets_text = "\n".join(f"- {t}" for t in tweets)
    
    response = await client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[
            {
                "role": "system",
                "content": "あなたはSNSの投稿を分析してダイジェストを作成するアシスタントです。"
            },
            {
                "role": "user",
                "content": f"以下は「{topic}」に関するXの投稿です。\n\n{tweets_text}\n\n"
                           "主要な論点・トレンド・重要な情報を箇条書きで要約してください。"
            },
        ],
    )
    return response.choices[0].message.content

# FastAPI BackgroundTasks を使う例
from fastapi import BackgroundTasks

@app.post("/topics/{topic_id}/collect")
async def trigger_collection(
    topic_id: int,
    background_tasks: BackgroundTasks,
    db: Session = Depends(get_db),
):
    topic = db.query(Topic).filter(Topic.id == topic_id).first()
    background_tasks.add_task(collect_and_summarize, topic)
    return {"message": "収集を開始しました"}