import os import sys import json import gzip import subprocess from pathlib import Path from contextlib import asynccontextmanager from fastapi import FastAPI, Response, Depends, HTTPException, Security, Query from fastapi.responses import RedirectResponse, JSONResponse from fastapi.security import APIKeyHeader from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from dotenv import load_dotenv from posthog import Posthog sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) sys.path.append(os.path.dirname(__file__)) from utils.logger import setup_logger from schemas.circles import Image, CircleSummary, CircleDetail, CircleIds from .repositories.circles_repository import CirclesRepository from .search.engine import SearchEngine from .services.circles_service import CirclesService log = setup_logger(__name__) load_dotenv() # --- Auth --- API_KEY_NAME = "X-API-KEY" APISECRETKEY = os.getenv("API_SECRET_KEY") api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=True) POSTHOG_PROJECT_API_KEY = os.getenv("POSTHOG_PROJECT_API_KEY") async def get_api_key(key: str = Security(api_key_header)): """APIキーを検証する依存関係""" if not APISECRETKEY or key != APISECRETKEY: raise HTTPException(status_code=403, detail="Could not validate credentials.") return key # --- App --- engine = SearchEngine() circles_repository = CirclesRepository() circles_cache_ttl = int(os.getenv("CIRCLES_CACHE_TTL", "300")) circles_service = CirclesService( circles_repository, cache_ttl_seconds=circles_cache_ttl ) def _select_primary_image(images: list[Image]) -> Image | None: primary: Image | None = None for image in images: if image.order == 0: primary = image if primary is not None: return primary if images: return images[0] return None @asynccontextmanager async def lifespan(app: FastAPI): """アプリケーションの起動時と終了時に実行されるコード""" # --- 起動時処理 --- log.info("Initializing search engine and loading assets...") engine.initialize() log.info("Initialization complete.") yield # --- 終了時処理 --- log.info("Shutting down search engine...") app = FastAPI(lifespan=lifespan) # CORS設定 app.add_middleware( CORSMiddleware, allow_origins=[ "https://circle-search-26.pages.dev", "http://localhost:3000", ], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # PostHog（ログ分析） if POSTHOG_PROJECT_API_KEY: posthog = Posthog( project_api_key=POSTHOG_PROJECT_API_KEY, host="https://us.i.posthog.com" ) else: log.warning("POSTHOG_PROJECT_API_KEY is not set. PostHog is disabled.") posthog = None @app.get("/", include_in_schema=False) def root(): return RedirectResponse("/docs") @app.get("/api/health") def health_check(): log.info("Health check OK") return {"status": "ok"} @app.get( "/api/circles", response_model=list[CircleSummary], dependencies=[Depends(get_api_key)], ) def get_summary_data(): # ここで必要なフィールドのみ抽出して返す summaries_payload: list[dict[str, object]] = [] for circle in circles_service.list_circles(): # 画像はcircle.mainImageを使用、なければimagesから選択 image = circle.mainImage if image is None and circle.images: image = _select_primary_image(circle.images) summary = CircleSummary( circleId=circle.circleId, circleName=circle.circleName, circleNameKana=circle.circleNameKana, projectId=circle.projectId, projectName=circle.projectName, genre=circle.genre, areaCode=circle.areaCode, pamphletNumber=circle.pamphletNumber, shortIntro=circle.shortIntro, detailDescription=circle.detailDescription, mainImage=image, memberCount=circle.memberCount, memberNote=circle.memberNote, annualFee=circle.annualFee, otherCosts=circle.otherCosts, activityFrequency=circle.activityFrequency, activityFrequencyNote=circle.activityFrequencyNote, activityLocation=circle.activityLocation, isArchived=circle.isArchived, ) summaries_payload.append(summary.model_dump(mode="json")) content = json.dumps(summaries_payload, ensure_ascii=False).encode("utf-8") log.info(f"Circle summaries fetched: {len(summaries_payload)} items") return Response( content=gzip.compress(content), headers={"Content-Encoding": "gzip", "Content-Type": "application/json"}, ) @app.get( "/api/details", response_model=CircleDetail, dependencies=[Depends(get_api_key)], ) def get_circle_detail(circleId: str = Query(..., description="取得したいサークルのID")): c = circles_service.get_circle(circleId) if not c: raise HTTPException(status_code=404, detail="Circle not found") log.info(f"Circle detail fetched: {circleId}") return CircleDetail(**c.model_dump(include=set(CircleDetail.model_fields.keys()))) class SearchRequest(BaseModel): query: str debug: bool = False class TaskUpdateResponse(BaseModel): message: str @app.post( "/api/search", response_model=CircleIds, dependencies=[Depends(get_api_key)], ) def search(request: SearchRequest): if not request.query: raise HTTPException(status_code=400, detail="Query cannot be empty") result = engine.search(request.query, debug=request.debug) if request.debug: pairs, diag = result ids = [cid for cid, _ in pairs] return JSONResponse( content={ "circleIds": ids, "scores": [ {"circleId": cid, "score": float(score)} for cid, score in pairs ], "details": diag.get("details", []), } ) pairs = result ids = [cid for cid, _ in pairs] # ログ送信 GET_LOGS = os.getenv("GET_LOGS", "false").lower() == "true" if GET_LOGS and posthog: try: posthog.capture( event="circles searched", properties={ "query": request.query, "result_count": len(ids), "$process_person_profile": False, }, ) except Exception: pass log.info(f'Search query="{request.query}" => {len(ids)} results') return CircleIds(circleIds=ids) @app.post( "/tasks/update", response_model=TaskUpdateResponse, dependencies=[Depends(get_api_key)], ) def update_tasks(): project_root = Path(__file__).resolve().parents[1] script_path = project_root / "scripts" / "build_all.py" if not script_path.exists(): log.error("Requested update script not found: %s", script_path) raise HTTPException(status_code=500, detail="Update script not found.") try: result = subprocess.run( [sys.executable, str(script_path)], check=True, capture_output=True, text=True, cwd=str(project_root), ) except subprocess.CalledProcessError as exc: stdout = exc.stdout.strip() if exc.stdout else "" stderr = exc.stderr.strip() if exc.stderr else "" if stdout: log.error("build_all.py stdout:\n%s", stdout) if stderr: log.error("build_all.py stderr:\n%s", stderr) raise HTTPException( status_code=500, detail="Data update failed while running build_all.py.", ) from exc stdout = result.stdout.strip() if result.stdout else "" stderr = result.stderr.strip() if result.stderr else "" if stdout: log.info("build_all.py stdout:\n%s", stdout) if stderr: log.warning("build_all.py stderr:\n%s", stderr) return TaskUpdateResponse(message="Data update completed.")