fengxiangdebaoziandClaude Opus 4.8
42f55251f2
- 后端新增 PUT /api/ai-writing/sessions/{id}/draft（属主校验，draft_title 可选）
- 前端历史模式放开只读，工具栏新增「保存」按钮持久化修改
- 实时写作完成态保持原样（仅导出，不落库）
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
628 lines
25 KiB
Python
628 lines
25 KiB
Python
"""AI 写作路由 —— 业务表 CRUD + transcript 持久化 + admin cleanup。
|
||
|
||
写作主流程（``/start /stream /resume /state``）已迁到 LangGraph Server 的
|
||
``ai_writing`` assistant（前端走 ``useAIWritingStream`` hook + SDK），路由层
|
||
不再承载 graph 推进，只保留以下「不依赖 graph 状态」的辅助接口：
|
||
|
||
GET /api/ai-writing/sessions 列出用户的会话历史
|
||
GET /api/ai-writing/sessions/{id} 获取单个历史会话
|
||
PATCH /api/ai-writing/sessions/{id} 重命名会话标题
|
||
DELETE /api/ai-writing/sessions/{id} 删除会话
|
||
PUT /api/ai-writing/sessions/{id}/transcript 保存完整对话时间线
|
||
GET /api/ai-writing/article-types 列出所有文章类型
|
||
POST /api/ai-writing/article-types 新增文章类型
|
||
PATCH /api/ai-writing/article-types/{id} 更新文章类型
|
||
DELETE /api/ai-writing/article-types/{id} 删除文章类型
|
||
POST /api/ai-writing/admin/cleanup （admin）手动清理会话
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from typing import Optional
|
||
|
||
from fastapi import APIRouter, HTTPException, Request
|
||
from pydantic import BaseModel, Field
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
router = APIRouter(prefix="/api/ai-writing", tags=["ai-writing"])
|
||
|
||
# ── 请求/响应模型 ─────────────────────────────────────────────────────────────
|
||
|
||
class RenameSessionRequest(BaseModel):
|
||
title: str
|
||
|
||
|
||
class UpdateDraftRequest(BaseModel):
|
||
# 历史回看时用户在右侧编辑器里修改后的正文 Markdown
|
||
draft_markdown: str
|
||
# 可选：标题一并更新（用户改了文章标题时）
|
||
draft_title: Optional[str] = None
|
||
|
||
|
||
class AdminCleanupRequest(BaseModel):
|
||
"""手动清理接口的可选覆盖参数。
|
||
|
||
全部留空 → 用 config.yaml 的 ai_writing.cleanup 段默认值；显式传值
|
||
→ 覆盖配置（一般用于运维想验证一个临时阈值，例如清掉「最近 1 小时
|
||
以前的」会话来模拟整轮调度）。
|
||
"""
|
||
retention_days: Optional[int] = Field(default=None, ge=0, description="保留多少天以内的会话；0 = 不保留")
|
||
only_finished: Optional[bool] = Field(default=None, description="True 时只清 done/error 状态")
|
||
batch_size: Optional[int] = Field(default=None, ge=1, le=10000, description="单批最多删多少行")
|
||
|
||
|
||
class AdminCleanupResponse(BaseModel):
|
||
deleted_db: int
|
||
deleted_checkpoints: int
|
||
# 因「仍在跑 / 等用户干预」被保护掉的会话数（threads_meta.status in
|
||
# running/interrupted）。即使 retention_days=0 也不会秒杀用户正在写的稿。
|
||
skipped_active: int = 0
|
||
elapsed_seconds: float
|
||
params: dict
|
||
|
||
|
||
class SaveTranscriptRequest(BaseModel):
|
||
# 前端组装的完整对话时间线（progressEvents + completedInterventions）
|
||
transcript: dict
|
||
|
||
|
||
class ArticleTypeRequest(BaseModel):
|
||
key: str
|
||
label: str
|
||
description: Optional[str] = None
|
||
prompt_hint: Optional[str] = None
|
||
default_word_count: int = 800
|
||
sort_order: int = 0
|
||
|
||
|
||
class UpdateArticleTypeRequest(BaseModel):
|
||
key: Optional[str] = None
|
||
label: Optional[str] = None
|
||
description: Optional[str] = None
|
||
prompt_hint: Optional[str] = None
|
||
default_word_count: Optional[int] = None
|
||
sort_order: Optional[int] = None
|
||
|
||
|
||
class ArticleTypeResponse(BaseModel):
|
||
id: str
|
||
key: str
|
||
label: str
|
||
description: Optional[str] = None
|
||
prompt_hint: Optional[str] = None
|
||
default_word_count: int
|
||
sort_order: int
|
||
created_at: str
|
||
updated_at: str
|
||
|
||
|
||
class ResearchTemplateRequest(BaseModel):
|
||
name: str
|
||
description: Optional[str] = None
|
||
article_type_key: Optional[str] = None
|
||
outline: Optional[str] = None
|
||
material_source: str = "general"
|
||
word_count: int = 800
|
||
audience: Optional[str] = None
|
||
strict_mode: bool = False
|
||
topic: Optional[str] = None
|
||
sort_order: int = 0
|
||
|
||
|
||
class UpdateResearchTemplateRequest(BaseModel):
|
||
name: Optional[str] = None
|
||
description: Optional[str] = None
|
||
article_type_key: Optional[str] = None
|
||
outline: Optional[str] = None
|
||
material_source: Optional[str] = None
|
||
word_count: Optional[int] = None
|
||
audience: Optional[str] = None
|
||
strict_mode: Optional[bool] = None
|
||
topic: Optional[str] = None
|
||
sort_order: Optional[int] = None
|
||
|
||
|
||
class ResearchTemplateResponse(BaseModel):
|
||
id: str
|
||
name: str
|
||
description: Optional[str] = None
|
||
article_type_key: Optional[str] = None
|
||
outline: Optional[str] = None
|
||
material_source: str
|
||
word_count: int
|
||
audience: Optional[str] = None
|
||
strict_mode: bool
|
||
topic: Optional[str] = None
|
||
sort_order: int
|
||
created_at: str
|
||
updated_at: str
|
||
|
||
|
||
class SessionResponse(BaseModel):
|
||
id: str
|
||
user_id: Optional[str] = None
|
||
title: str
|
||
user_intent: Optional[str] = None
|
||
status: str
|
||
draft_title: Optional[str] = None
|
||
draft_markdown: Optional[str] = None
|
||
completed_interventions: Optional[list] = None
|
||
review_result: Optional[dict] = None
|
||
transcript: Optional[dict] = None
|
||
created_at: str
|
||
updated_at: str
|
||
|
||
|
||
# ── 路由 ──────────────────────────────────────────────────────────────────────
|
||
@router.get("/sessions", response_model=list[SessionResponse])
|
||
async def list_sessions(request: Request):
|
||
"""列出当前用户的 AI 写作历史会话。"""
|
||
from app.gateway.deps import get_ai_writing_session_store, get_current_user
|
||
from app.gateway.routers._ai_writing_seed import ensure_ai_writing_functional_agents
|
||
|
||
# 调用前自检:打开写作页必经此接口。若管理员在启动后手删了内置写作 agent
|
||
# 目录,这里自愈补建,保证 use_builtin_agents 开启时图节点能加载到 SOUL。
|
||
ensure_ai_writing_functional_agents()
|
||
|
||
repo = get_ai_writing_session_store(request)
|
||
if repo is None:
|
||
return []
|
||
user_id = await get_current_user(request)
|
||
return await repo.list(user_id=user_id)
|
||
|
||
|
||
@router.get("/sessions/{session_id}", response_model=SessionResponse)
|
||
async def get_session(session_id: str, request: Request):
|
||
"""获取单个历史会话。"""
|
||
from app.gateway.deps import get_ai_writing_session_store, get_current_user
|
||
|
||
repo = get_ai_writing_session_store(request)
|
||
if repo is None:
|
||
raise HTTPException(status_code=503, detail="Session store not available")
|
||
user_id = await get_current_user(request)
|
||
record = await repo.get(session_id, user_id=user_id)
|
||
if record is None:
|
||
raise HTTPException(status_code=404, detail="Session not found")
|
||
return record
|
||
|
||
|
||
@router.patch("/sessions/{session_id}", response_model=SessionResponse)
|
||
async def rename_session(session_id: str, body: RenameSessionRequest, request: Request):
|
||
"""重命名会话标题。"""
|
||
from app.gateway.deps import get_ai_writing_session_store, get_current_user
|
||
|
||
repo = get_ai_writing_session_store(request)
|
||
if repo is None:
|
||
raise HTTPException(status_code=503, detail="Session store not available")
|
||
user_id = await get_current_user(request)
|
||
updated = await repo.update(session_id, user_id=user_id, title=body.title.strip())
|
||
if updated is None:
|
||
raise HTTPException(status_code=404, detail="Session not found")
|
||
return updated
|
||
|
||
|
||
@router.put("/sessions/{session_id}/draft", response_model=SessionResponse)
|
||
async def update_session_draft(session_id: str, body: UpdateDraftRequest, request: Request):
|
||
"""更新会话的成稿正文（历史回看时用户在右侧编辑器里直接修改并保存）。"""
|
||
from app.gateway.deps import get_ai_writing_session_store, get_current_user
|
||
|
||
repo = get_ai_writing_session_store(request)
|
||
if repo is None:
|
||
raise HTTPException(status_code=503, detail="Session store not available")
|
||
user_id = await get_current_user(request)
|
||
# draft_title 缺省（None）时不动原标题，传了才更新
|
||
extra = {} if body.draft_title is None else {"draft_title": body.draft_title}
|
||
updated = await repo.update(
|
||
session_id,
|
||
user_id=user_id,
|
||
draft_markdown=body.draft_markdown,
|
||
**extra,
|
||
)
|
||
if updated is None:
|
||
raise HTTPException(status_code=404, detail="Session not found")
|
||
return updated
|
||
|
||
|
||
@router.delete("/sessions/{session_id}")
|
||
async def delete_session(session_id: str, request: Request):
|
||
"""删除历史会话。"""
|
||
from app.gateway.deps import get_ai_writing_session_store, get_current_user
|
||
|
||
repo = get_ai_writing_session_store(request)
|
||
if repo is None:
|
||
raise HTTPException(status_code=503, detail="Session store not available")
|
||
user_id = await get_current_user(request)
|
||
deleted = await repo.delete(session_id, user_id=user_id)
|
||
if not deleted:
|
||
raise HTTPException(status_code=404, detail="Session not found")
|
||
return {"success": True}
|
||
|
||
|
||
@router.put("/sessions/{session_id}/transcript")
|
||
async def save_transcript(session_id: str, body: SaveTranscriptRequest, request: Request):
|
||
"""保存会话的完整对话时间线，供历史会话回显出与实时对话一致的富时间线。
|
||
|
||
时间线持久化属于「锦上添花」功能：即便保存失败（如内网部署缺少 transcript
|
||
列、或 MySQL TEXT 列超长），也绝不能阻断写作流程。任何异常都吞掉并返回
|
||
success=False，由前端忽略即可。
|
||
"""
|
||
from app.gateway.deps import get_ai_writing_session_store, get_current_user
|
||
|
||
repo = get_ai_writing_session_store(request)
|
||
if repo is None:
|
||
return {"success": False, "reason": "store_unavailable"}
|
||
try:
|
||
user_id = await get_current_user(request)
|
||
updated = await repo.update(session_id, user_id=user_id, transcript=body.transcript)
|
||
except Exception:
|
||
logger.exception("ai_writing: failed to save transcript: session=%s", session_id)
|
||
return {"success": False, "reason": "save_failed"}
|
||
if updated is None:
|
||
# 会话不存在或不属于当前用户：同样不抛 500，返回 success=False 即可
|
||
return {"success": False, "reason": "session_not_found"}
|
||
return {"success": True}
|
||
|
||
|
||
class BackgroundResponse(BaseModel):
|
||
# 是否新拉起了后台任务（False = 已在后台跑，幂等返回）
|
||
started: bool
|
||
# 会话当前状态（background = 挂起中）
|
||
status: str
|
||
|
||
|
||
@router.post("/sessions/{session_id}/background", response_model=BackgroundResponse)
|
||
async def suspend_to_background(session_id: str, request: Request):
|
||
"""把这篇写作转入「后台挂起」：后端自动驱动剩余流程到定稿。
|
||
|
||
前端在用户点「后台挂起」或离开页面时调用。幂等：同一会话已在后台跑则直接返回。
|
||
驱动逻辑见 ``app.gateway.ai_writing_job_executor``（素材/大纲自动确认、缺素材
|
||
通用知识续写、草稿送审、审核不过自动重写直到通过或用完次数再定稿）。
|
||
"""
|
||
from app.gateway.deps import (
|
||
get_ai_writing_job_executor,
|
||
get_ai_writing_session_store,
|
||
get_current_user,
|
||
)
|
||
|
||
executor = get_ai_writing_job_executor(request)
|
||
if executor is None:
|
||
raise HTTPException(status_code=503, detail="Background executor not available")
|
||
|
||
repo = get_ai_writing_session_store(request)
|
||
if repo is None:
|
||
raise HTTPException(status_code=503, detail="Session store not available")
|
||
|
||
user_id = await get_current_user(request)
|
||
# 鉴权 + 存在性：会话必须属于当前用户（get 已带 user_id 过滤）。
|
||
record = await repo.get(session_id, user_id=user_id)
|
||
if record is None:
|
||
raise HTTPException(status_code=404, detail="Session not found")
|
||
if record.get("status") == "done":
|
||
# 已定稿，无需后台驱动。
|
||
return BackgroundResponse(started=False, status="done")
|
||
|
||
started = executor.start(thread_id=session_id, user_id=user_id)
|
||
return BackgroundResponse(started=started, status="background")
|
||
|
||
|
||
class ProgressResponse(BaseModel):
|
||
# 业务表 status（background = 挂起中 / done / error / in_progress）
|
||
sessionStatus: str
|
||
# 规范化阶段 key（intent/research/material_confirm/outline/outline_confirm/draft/draft_confirm/review/done），
|
||
# 从 graph checkpoint 推导；None = 还没建 checkpoint。
|
||
stage: Optional[str] = None
|
||
# 阶段细化说明（素材不足求助 / 编辑打回）
|
||
note: Optional[str] = None
|
||
revisionCount: int = 0
|
||
maxRevisions: int = 3
|
||
reviewVerdict: Optional[str] = None
|
||
# 后台驱动任务是否在本进程在跑（重启后可能为 False 但 status 仍 background，由 reconcile 续起）
|
||
isRunning: bool = False
|
||
|
||
|
||
@router.get("/sessions/{session_id}/progress", response_model=ProgressResponse)
|
||
async def get_session_progress(session_id: str, request: Request):
|
||
"""读取一篇写作的实际进度（当前阶段），供「查看进度」流程图弹窗轮询。
|
||
|
||
阶段从 graph checkpoint 推导，反映后端真实推进位置；鉴权要求会话属于当前用户。
|
||
"""
|
||
from app.gateway.deps import (
|
||
get_ai_writing_job_executor,
|
||
get_ai_writing_session_store,
|
||
get_current_user,
|
||
)
|
||
|
||
repo = get_ai_writing_session_store(request)
|
||
if repo is None:
|
||
raise HTTPException(status_code=503, detail="Session store not available")
|
||
user_id = await get_current_user(request)
|
||
record = await repo.get(session_id, user_id=user_id)
|
||
if record is None:
|
||
raise HTTPException(status_code=404, detail="Session not found")
|
||
|
||
executor = get_ai_writing_job_executor(request)
|
||
progress = await executor.get_progress(session_id) if executor is not None else {}
|
||
return ProgressResponse(
|
||
sessionStatus=str(record.get("status") or "in_progress"),
|
||
stage=progress.get("stage") or None,
|
||
note=progress.get("note"),
|
||
revisionCount=int(progress.get("revisionCount") or 0),
|
||
maxRevisions=int(progress.get("maxRevisions") or 3),
|
||
reviewVerdict=progress.get("reviewVerdict"),
|
||
isRunning=bool(executor.is_running(session_id)) if executor is not None else False,
|
||
)
|
||
|
||
|
||
@router.get("/article-types", response_model=list[ArticleTypeResponse])
|
||
async def list_article_types(request: Request):
|
||
"""列出所有文章类型，按 sort_order 升序排列。"""
|
||
from app.gateway.deps import get_article_type_store
|
||
from app.gateway.routers._ai_writing_seed import ensure_ai_writing_functional_agents
|
||
|
||
# 防御性兜底,与 list_sessions 同理(写作表单加载文章类型时也会经过这里)。
|
||
ensure_ai_writing_functional_agents()
|
||
|
||
store = get_article_type_store(request)
|
||
if store is None:
|
||
return []
|
||
return await store.list()
|
||
|
||
|
||
@router.post("/article-types", response_model=ArticleTypeResponse, status_code=201)
|
||
async def create_article_type(body: ArticleTypeRequest, request: Request):
|
||
"""新增文章类型。key 必须全局唯一。"""
|
||
from app.gateway.deps import get_article_type_store
|
||
|
||
store = get_article_type_store(request)
|
||
if store is None:
|
||
raise HTTPException(status_code=503, detail="Article type store not available")
|
||
existing = await store.get_by_key(body.key)
|
||
if existing is not None:
|
||
raise HTTPException(status_code=409, detail=f"Article type key '{body.key}' already exists")
|
||
return await store.create(
|
||
key=body.key,
|
||
label=body.label,
|
||
description=body.description,
|
||
prompt_hint=body.prompt_hint,
|
||
default_word_count=body.default_word_count,
|
||
sort_order=body.sort_order,
|
||
)
|
||
|
||
|
||
@router.patch("/article-types/{article_type_id}", response_model=ArticleTypeResponse)
|
||
async def update_article_type(article_type_id: str, body: UpdateArticleTypeRequest, request: Request):
|
||
"""更新文章类型字段（只传需要修改的字段）。"""
|
||
from app.gateway.deps import get_article_type_store
|
||
|
||
store = get_article_type_store(request)
|
||
if store is None:
|
||
raise HTTPException(status_code=503, detail="Article type store not available")
|
||
# 如果修改了 key，检查唯一性
|
||
if body.key is not None:
|
||
existing = await store.get_by_key(body.key)
|
||
if existing is not None and existing["id"] != article_type_id:
|
||
raise HTTPException(status_code=409, detail=f"Article type key '{body.key}' already exists")
|
||
|
||
_UNSET = object.__new__(object)
|
||
kwargs = {}
|
||
for field in ("key", "label", "description", "prompt_hint", "default_word_count", "sort_order"):
|
||
val = getattr(body, field)
|
||
if val is not None:
|
||
kwargs[field] = val
|
||
|
||
updated = await store.update(article_type_id, **kwargs)
|
||
if updated is None:
|
||
raise HTTPException(status_code=404, detail="Article type not found")
|
||
return updated
|
||
|
||
|
||
@router.delete("/article-types/{article_type_id}")
|
||
async def delete_article_type(article_type_id: str, request: Request):
|
||
"""删除文章类型。"""
|
||
from app.gateway.deps import get_article_type_store
|
||
|
||
store = get_article_type_store(request)
|
||
if store is None:
|
||
raise HTTPException(status_code=503, detail="Article type store not available")
|
||
deleted = await store.delete(article_type_id)
|
||
if not deleted:
|
||
raise HTTPException(status_code=404, detail="Article type not found")
|
||
return {"success": True}
|
||
|
||
|
||
# ── 研究模板（课题研究左栏一键预填表单，需求 0616 #5）────────────────────────────
|
||
|
||
|
||
@router.get("/research-templates", response_model=list[ResearchTemplateResponse])
|
||
async def list_research_templates(request: Request):
|
||
"""列出所有研究模板，按 sort_order 升序。
|
||
|
||
新机器/空表时自动写入内置范例模板（网摘/周报/要讯），幂等仅在表为空时执行。
|
||
"""
|
||
from app.gateway.deps import get_research_template_store
|
||
|
||
store = get_research_template_store(request)
|
||
if store is None:
|
||
return []
|
||
try:
|
||
await store.seed_defaults()
|
||
except Exception:
|
||
logger.exception("ai_writing: failed to seed default research templates")
|
||
return await store.list()
|
||
|
||
|
||
@router.post("/research-templates", response_model=ResearchTemplateResponse, status_code=201)
|
||
async def create_research_template(body: ResearchTemplateRequest, request: Request):
|
||
"""新增研究模板。"""
|
||
from app.gateway.deps import get_research_template_store
|
||
|
||
store = get_research_template_store(request)
|
||
if store is None:
|
||
raise HTTPException(status_code=503, detail="Research template store not available")
|
||
return await store.create(
|
||
name=body.name,
|
||
description=body.description,
|
||
article_type_key=body.article_type_key,
|
||
outline=body.outline,
|
||
material_source=body.material_source,
|
||
word_count=body.word_count,
|
||
audience=body.audience,
|
||
strict_mode=body.strict_mode,
|
||
topic=body.topic,
|
||
sort_order=body.sort_order,
|
||
)
|
||
|
||
|
||
@router.patch("/research-templates/{template_id}", response_model=ResearchTemplateResponse)
|
||
async def update_research_template(template_id: str, body: UpdateResearchTemplateRequest, request: Request):
|
||
"""更新研究模板字段（只传需要修改的字段）。"""
|
||
from app.gateway.deps import get_research_template_store
|
||
|
||
store = get_research_template_store(request)
|
||
if store is None:
|
||
raise HTTPException(status_code=503, detail="Research template store not available")
|
||
|
||
kwargs = {}
|
||
for field in (
|
||
"name",
|
||
"description",
|
||
"article_type_key",
|
||
"outline",
|
||
"material_source",
|
||
"word_count",
|
||
"audience",
|
||
"strict_mode",
|
||
"topic",
|
||
"sort_order",
|
||
):
|
||
val = getattr(body, field)
|
||
if val is not None:
|
||
kwargs[field] = val
|
||
|
||
updated = await store.update(template_id, **kwargs)
|
||
if updated is None:
|
||
raise HTTPException(status_code=404, detail="Research template not found")
|
||
return updated
|
||
|
||
|
||
@router.delete("/research-templates/{template_id}")
|
||
async def delete_research_template(template_id: str, request: Request):
|
||
"""删除研究模板。"""
|
||
from app.gateway.deps import get_research_template_store
|
||
|
||
store = get_research_template_store(request)
|
||
if store is None:
|
||
raise HTTPException(status_code=503, detail="Research template store not available")
|
||
deleted = await store.delete(template_id)
|
||
if not deleted:
|
||
raise HTTPException(status_code=404, detail="Research template not found")
|
||
return {"success": True}
|
||
|
||
|
||
# ── 管理接口 ──────────────────────────────────────────────────────────────────
|
||
|
||
async def _require_admin(request: Request) -> None:
|
||
"""放行 admin / auth-disabled，其余 403。
|
||
|
||
与 scheduled_tasks._require_admin 同套语义：auth_disabled 模式下
|
||
get_optional_user_from_request 返回 None，此时放行（开发环境一致体验）。
|
||
"""
|
||
from app.gateway.deps import get_optional_user_from_request
|
||
|
||
user = await get_optional_user_from_request(request)
|
||
if user is None:
|
||
return
|
||
if getattr(user, "system_role", None) != "admin":
|
||
raise HTTPException(status_code=403, detail="AI 写作清理管理接口仅限管理员")
|
||
|
||
|
||
def _resolve_cleanup_params(override: AdminCleanupRequest) -> dict:
|
||
"""合并 config.yaml 默认值 + 请求显式覆盖。
|
||
|
||
显式传 0 / False 也算覆盖（用 ``is not None`` 判定），方便测「保留 0 天 =
|
||
全删」这种极端场景。
|
||
"""
|
||
from deerflow.config import get_app_config
|
||
|
||
raw = getattr(get_app_config(), "ai_writing", None) or {}
|
||
cfg = (raw.get("cleanup") if isinstance(raw, dict) else None) or {}
|
||
|
||
return {
|
||
"retention_days": (
|
||
override.retention_days if override.retention_days is not None
|
||
else int(cfg.get("retention_days", 7))
|
||
),
|
||
"only_finished": (
|
||
override.only_finished if override.only_finished is not None
|
||
else bool(cfg.get("delete_only_finished", False))
|
||
),
|
||
"batch_size": (
|
||
override.batch_size if override.batch_size is not None
|
||
else int(cfg.get("batch_size", 500))
|
||
),
|
||
}
|
||
|
||
|
||
@router.post("/admin/cleanup", response_model=AdminCleanupResponse)
|
||
async def admin_cleanup(
|
||
body: AdminCleanupRequest,
|
||
request: Request,
|
||
) -> AdminCleanupResponse:
|
||
"""**仅管理员** 手动触发一次会话清理（不等凌晨 3 点定时调度）。
|
||
|
||
用途：
|
||
- 验证 cleanup 配置改动是否生效
|
||
- 内网紧急清理（磁盘满 / MySQL 行数过多）
|
||
- 调试：传 ``retention_days: 0`` 一刷拉清空，方便重新跑回归
|
||
|
||
所有参数留空 → 沿用 config.yaml 的 ``ai_writing.cleanup`` 段默认。显式
|
||
传值会覆盖配置但**不修改 config.yaml**（一次性参数，下次定时调度还是用
|
||
配置文件的值）。
|
||
"""
|
||
import time
|
||
|
||
from app.gateway.ai_writing_cleanup import run_cleanup_once
|
||
|
||
await _require_admin(request)
|
||
params = _resolve_cleanup_params(body)
|
||
|
||
# 透传主 LangGraph checkpointer / thread_store —— 新路径下 ai_writing 的
|
||
# thread 状态在主 checkpointer 里，thread_store 用来按 status 过滤活跃 thread。
|
||
# getattr 嵌套防御：测试用 stub Request 时 ``request.app`` 可能不存在。
|
||
_app_state = getattr(getattr(request, "app", None), "state", None)
|
||
main_checkpointer = getattr(_app_state, "checkpointer", None)
|
||
main_thread_store = getattr(_app_state, "thread_store", None)
|
||
|
||
started = time.monotonic()
|
||
try:
|
||
result = await run_cleanup_once(
|
||
**params,
|
||
checkpointer=main_checkpointer,
|
||
thread_store=main_thread_store,
|
||
)
|
||
except Exception as e:
|
||
logger.exception("ai_writing admin cleanup failed: params=%s", params)
|
||
# 抛 500 是合理的 —— 管理员触发清理失败需要看到失败原因，跟「业务
|
||
# 接口被失败拖累」是两回事
|
||
raise HTTPException(
|
||
status_code=500,
|
||
detail=f"清理执行失败: {type(e).__name__}: {e}",
|
||
) from e
|
||
|
||
elapsed = time.monotonic() - started
|
||
logger.info(
|
||
"ai_writing admin cleanup done: params=%s deleted_db=%d deleted_cp=%d elapsed=%.2fs",
|
||
params,
|
||
result["deleted_db"],
|
||
result["deleted_checkpoints"],
|
||
elapsed,
|
||
)
|
||
return AdminCleanupResponse(
|
||
deleted_db=result["deleted_db"],
|
||
deleted_checkpoints=result["deleted_checkpoints"],
|
||
skipped_active=result.get("skipped_active", 0),
|
||
elapsed_seconds=round(elapsed, 3),
|
||
params=params,
|
||
)
|