Create-knowledge-find-decode/creation_knowledge/api.py

"""FastAPI：把 ck_post / ck_knowledge_item 暴露给 web 可视化，并托管单页前端。

启动：
    CK_ENV_FILE=/path/.env uvicorn creation_knowledge.api:app --host 0.0.0.0 --port 8900
API 在 /api/*，前端单页在 /。
"""
from __future__ import annotations

import os
from functools import lru_cache
from pathlib import Path

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles

from creation_knowledge.config import PgConfig, Settings
from creation_knowledge.integrations import extractor as extractor_mod
from creation_knowledge.integrations.db import CkStore
from creation_knowledge.prompts import PROMPT_VERSION, load_prompt
from creation_knowledge.stages import deconstruct as decon_stage
from creation_knowledge.stages import screen as screen_stage
from creation_knowledge.stages import split as split_stage

WEB_DIR = Path(__file__).resolve().parent.parent / "web"


def _env_file() -> str:
    return os.getenv("CK_ENV_FILE", ".env")


@lru_cache(maxsize=1)
def _store() -> CkStore:
    return CkStore(PgConfig.from_env(_env_file()))


@lru_cache(maxsize=1)
def _settings() -> Settings:
    return Settings.from_env(_env_file())


app = FastAPI(title="创作知识")
app.add_middleware(
    CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]
)


@app.get("/api/posts")
def list_posts() -> list[dict]:
    return _store().posts_overview()


@app.get("/api/posts/{post_id}")
def get_post(post_id: str) -> dict:
    post = _store().read_post(post_id)
    if not post:
        raise HTTPException(status_code=404, detail="post not found")
    return post


@app.get("/api/posts/{post_id}/items")
def get_items(post_id: str) -> list[dict]:
    return _store().read_items(post_id)


@app.get("/api/prompts")
def get_prompts() -> dict:
    """各环节用到的提示词（system + user 模板），供前端弹窗查看。"""
    s = _settings()
    return {
        "version": PROMPT_VERSION,
        "items": [
            {"key": "extract", "label": "图文提取", "model": s.video_model,
             "system": extractor_mod._SYSTEM_PROMPT, "user": load_prompt("extract")},
            {"key": "extract_video", "label": "视频提炼（原生整段）", "model": s.video_model,
             "system": "（无独立 system，提示词自含）", "user": load_prompt("extract_video")},
            {"key": "screen", "label": "筛选", "model": s.llm_model,
             "system": screen_stage.SYSTEM, "user": load_prompt("screen")},
            {"key": "split", "label": "拆分", "model": s.llm_model,
             "system": split_stage.SYSTEM, "user": load_prompt("split")},
            {"key": "deconstruct", "label": "解构", "model": s.llm_model,
             "system": decon_stage.SYSTEM, "user": load_prompt("deconstruct")},
        ],
    }


# 视频帧静态服务（card.url = /frames/<post_id>/<file>）
_frames_dir = Path(_settings().frames_dir)
_frames_dir.mkdir(parents=True, exist_ok=True)
app.mount("/frames", StaticFiles(directory=str(_frames_dir)), name="frames")

# 媒体静态服务（card.url = /data/<run_id>/<platform>/<post_id>/<file>）
_data_dir = Path(_settings().data_dir or "data")
_data_dir.mkdir(parents=True, exist_ok=True)
app.mount("/data", StaticFiles(directory=str(_data_dir)), name="data")

# 单页前端挂在最后（catch-all），不影响上面的 /api、/frames、/data
if WEB_DIR.exists():
    app.mount("/", StaticFiles(directory=str(WEB_DIR), html=True), name="web")