162 lines
5.6 KiB
Python
162 lines
5.6 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
import hmac
|
|
from contextlib import asynccontextmanager
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi import Request
|
|
from fastapi.responses import JSONResponse
|
|
from fastapi.responses import FileResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
from app.api.routes import router as api_router
|
|
from app.core.config import get_settings
|
|
from app.engine.detector import PlagiarismDetector
|
|
from app.jobs.store import JobStore
|
|
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
|
|
|
|
|
|
class AuthConfigurationError(RuntimeError):
|
|
"""운영 인증 설정이 모순될 때 기동을 막는다."""
|
|
|
|
|
|
def validate_auth_settings(settings) -> None:
|
|
"""REQUIRE_API_KEY=true 인데 키가 없으면 기동 실패 (fail-closed).
|
|
|
|
'키를 깜빡해서 무인증으로 떠 있었다'가 가능한 구성을 없애는 것이 목적이다.
|
|
미출간 원고를 다루는 서버라 조용한 무인증이 가장 위험하다.
|
|
"""
|
|
if settings.require_api_key and not settings.api_key.strip():
|
|
raise AuthConfigurationError(
|
|
"REQUIRE_API_KEY=true 인데 API_KEY 가 비어 있습니다. "
|
|
"키를 설정하거나 REQUIRE_API_KEY=false 로 두십시오."
|
|
)
|
|
|
|
|
|
def public_paths(settings) -> set[str]:
|
|
"""인증 없이 접근 가능한 경로. 설정으로 좁힐 수 있다."""
|
|
paths = {"/"}
|
|
if settings.public_health:
|
|
paths.add("/v1/health")
|
|
if settings.public_docs:
|
|
paths.update({"/docs", "/openapi.json", "/redoc"})
|
|
return paths
|
|
|
|
|
|
def is_authorized(settings, path: str, supplied: str) -> bool:
|
|
"""요청 허용 여부. 순수 함수라 앱 기동 없이 테스트할 수 있다.
|
|
|
|
api_key 가 비어 있으면(개발 기본값) 인증을 걸지 않는다. 이 경우 기동 시
|
|
critical 경고가 남는다.
|
|
"""
|
|
configured = settings.api_key.strip()
|
|
if not configured:
|
|
return True
|
|
if path in public_paths(settings):
|
|
return True
|
|
protected_docs = {"/docs", "/openapi.json", "/redoc"}
|
|
if not path.startswith("/v1") and path not in protected_docs:
|
|
return True
|
|
# compare_digest 는 비ASCII str 에서 TypeError 를 던진다. 한글/이모지 키를
|
|
# 넣으면 전 요청이 500 이 되므로 반드시 bytes 로 비교한다.
|
|
return hmac.compare_digest(configured.encode("utf-8"), (supplied or "").encode("utf-8"))
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
settings = get_settings()
|
|
validate_auth_settings(settings)
|
|
if not settings.api_key.strip():
|
|
logging.critical(
|
|
"API_KEY is empty: all /v1 endpoints are unauthenticated. "
|
|
"Set REQUIRE_API_KEY=true with a key before exposing this server. "
|
|
"Do not expose unpublished manuscripts publicly until client key rollout is complete."
|
|
)
|
|
app.state.settings = settings
|
|
app.state.detector = PlagiarismDetector(settings=settings)
|
|
app.state.job_store = JobStore()
|
|
import threading
|
|
app.state.detector_lock = threading.Lock()
|
|
logging.info(
|
|
"Engine ready: version=%s, corpus_size=%d",
|
|
settings.engine_version,
|
|
app.state.detector.corpus_size,
|
|
)
|
|
yield
|
|
|
|
|
|
def rebuild_detector(app: FastAPI) -> int:
|
|
"""코퍼스 변경 후 인덱스 재빌드. 호출 시점에 lock으로 보호."""
|
|
settings = app.state.settings
|
|
with app.state.detector_lock:
|
|
app.state.detector = PlagiarismDetector(settings=settings)
|
|
return app.state.detector.corpus_size
|
|
|
|
|
|
_settings = get_settings()
|
|
|
|
app = FastAPI(
|
|
title="O2O 저작권 침해 여부 탐지 API",
|
|
description=(
|
|
"오투오 1단계 산출물 - 콘텐츠 표절 여부 AI 탐지 모듈. "
|
|
"본 응답 스키마는 커뮤니케이션북스(아카이빙) 및 바이칼AI(분석 보고서) 통합 기준."
|
|
),
|
|
version="1.0.0",
|
|
lifespan=lifespan,
|
|
root_path=_settings.root_path,
|
|
)
|
|
|
|
app.include_router(api_router)
|
|
|
|
|
|
@app.middleware("http")
|
|
async def optional_api_key_auth(request: Request, call_next):
|
|
"""API_KEY가 설정된 운영 환경에서만 API 인증을 강제한다."""
|
|
settings = getattr(request.app.state, "settings", None) or _settings
|
|
if not is_authorized(settings, request.url.path, request.headers.get("x-api-key", "")):
|
|
return JSONResponse(status_code=401, content={"detail": "Invalid or missing API key"})
|
|
return await call_next(request)
|
|
|
|
_STATIC_DIR = Path(__file__).resolve().parent / "static"
|
|
if _STATIC_DIR.exists():
|
|
app.mount("/static", StaticFiles(directory=str(_STATIC_DIR)), name="static")
|
|
|
|
|
|
@app.get("/", include_in_schema=False)
|
|
async def root() -> FileResponse:
|
|
"""검토 콘솔 페이지 — 컴북스 측이 브라우저에서 직접 엔진 성능 확인."""
|
|
index = _STATIC_DIR / "index.html"
|
|
if index.exists():
|
|
return FileResponse(str(index))
|
|
from fastapi.responses import JSONResponse
|
|
return JSONResponse({"service": "o2o-plagiarism-api", "docs": "/docs"})
|
|
|
|
|
|
def main() -> None:
|
|
"""`.env` 의 HOST/PORT/LOG_LEVEL/RELOAD 를 읽어 서버 기동.
|
|
|
|
사용법:
|
|
python -m app.main # .env 따라 실행
|
|
PORT=9000 python -m app.main # 환경변수 override
|
|
"""
|
|
import uvicorn
|
|
|
|
settings = get_settings()
|
|
uvicorn.run(
|
|
"app.main:app",
|
|
host=settings.host,
|
|
port=settings.port,
|
|
log_level=settings.log_level,
|
|
reload=settings.reload,
|
|
root_path=settings.root_path,
|
|
forwarded_allow_ips="*", # 리버스 프록시(Apache) 헤더 신뢰
|
|
proxy_headers=True,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|