# kb_diag4.dspy - 临时:验证 PM 建议入库全链路(落表+建待办),用后即删 import json as _j uid = await get_user() if not uid: return _j.dumps({"ok": False, "error": "未登录"}, ensure_ascii=False) project_id = (params_kw or {}).get('project_id', '').strip() file_path = (params_kw or {}).get('file_path', '').strip() kb_id = (params_kw or {}).get('kb_id', '').strip() act = (params_kw or {}).get('act', 'suggest').strip() out = {"act": act} try: dbname = get_module_dbname('pipeline-sdlc') if act == 'suggest': from pipeline_service.kb_ingest_capability import suggest_kb_ingest ok, msg = await suggest_kb_ingest( project_id, kb_id, file_path, reason="自测:该文档包含产线接入知识库的标准流程,建议沉淀供后续项目复用", who="agent.pm", agent_id="ag.testpm", task_id="") out["suggest_ok"] = ok out["suggest_msg"] = msg # 查落库 async with DBPools().sqlorContext(dbname) as sor: recs = await sor.sqlExe( "SELECT id, kb_id, file_name, status FROM pipeline_kb_suggestions " "ORDER BY created_at DESC LIMIT 1") await sor.sqlExe("COMMIT", {}) if recs: out["suggestion"] = {"id": recs[0].id, "kb_id": recs[0].kb_id, "file_name": recs[0].file_name, "status": recs[0].status} ht = await sor.sqlExe( "SELECT id, task_type, assignee_id, status, form_schema FROM pipeline_human_tasks " "WHERE task_type='kb_ingest_confirm' ORDER BY created_at DESC LIMIT 1") await sor.sqlExe("COMMIT", {}) if ht: out["human_task"] = {"id": ht[0].id, "task_type": ht[0].task_type, "assignee_id": ht[0].assignee_id, "status": ht[0].status, "form_schema_head": (ht[0].form_schema or "")[:120]} elif act == 'decide': sid = (params_kw or {}).get('suggestion_id', '').strip() approve = (params_kw or {}).get('approve', '1') in ('1', 'true') from pipeline_service.kb_ingest_capability import decide_kb_ingest # operator 须是项目 owner ok, msg = await decide_kb_ingest(sid, approve, uid, comment="自测批准") out["decide_ok"] = ok out["decide_msg"] = msg except Exception as e: import traceback out["traceback"] = traceback.format_exc()[-800:] return _j.dumps(out, ensure_ascii=False, default=str)