494 lines
21 KiB
Python
494 lines
21 KiB
Python
import json
|
||
import re
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
from wiki_engine import service
|
||
|
||
|
||
CONFIRM_WORDS = ("确认", "执行", "开始", "确定", "同意", "dry_run=false", "不是预览")
|
||
CANCEL_WORDS = ("取消", "先不", "不要执行", "作废")
|
||
|
||
|
||
async def run_wiki_admin_workflow(
|
||
extracted_text: str,
|
||
combined_query: str,
|
||
history_message: str = "",
|
||
route_flag: str = "",
|
||
chat_id: str = "",
|
||
message_id: str = "",
|
||
**_: Any,
|
||
) -> Dict[str, Any]:
|
||
query = (combined_query or extracted_text or "").strip()
|
||
if not query:
|
||
return _response("请告诉我要管理哪个知识库或文件，例如：检查知识库 163舰 的 Wiki 同步情况。")
|
||
|
||
try:
|
||
result = await _dispatch(query=query, chat_id=chat_id)
|
||
return _response(_format_result(result), result=result)
|
||
except Exception as exc:
|
||
return _response(f"Wiki 管理操作失败：{exc}", error=str(exc))
|
||
|
||
|
||
async def _dispatch(query: str, chat_id: str = "") -> Dict[str, Any]:
|
||
pending_plan = await service.get_pending_admin_plan(chat_id) if chat_id else None
|
||
confirmed = _is_confirmed(query)
|
||
selected_indexes = _extract_selection_indexes(query)
|
||
job_id = _extract_job_id(query)
|
||
|
||
if job_id and any(word in query for word in ("进度", "状态", "查看", "job", "任务", "同步")):
|
||
job = await service.get_sync_job(job_id)
|
||
if not job:
|
||
return {"operation": "job_progress", "data": {"success": False, "job_id": job_id, "message": "未找到这个同步任务。"}}
|
||
return {"operation": "job_progress", "data": {"success": True, "job": job}}
|
||
|
||
if pending_plan and _is_cancel(query):
|
||
cancelled = await service.cancel_admin_plan(str(pending_plan.get("id") or ""))
|
||
return {"operation": "plan_cancelled", "data": {"success": True, "plan": cancelled or pending_plan}}
|
||
|
||
if pending_plan and confirmed and not _contains_new_target(query):
|
||
data = await service.execute_admin_plan(pending_plan, selected_indexes=selected_indexes)
|
||
data["plan"] = pending_plan
|
||
return {"operation": "plan_executed", "data": data}
|
||
|
||
kb_id = _extract_kb_id(query)
|
||
file_id = _extract_id(query, ("文件", "file", "file_id"))
|
||
include_children = any(word in query for word in ("子知识库", "子库", "包含子", "包括子"))
|
||
use_llm = any(word.lower() in query.lower() for word in ("大模型", "llm", "模型摘要", "智能摘要"))
|
||
limit = _extract_limit(query) or 20
|
||
|
||
if any(word in query for word in ("统计", "状态", "概览", "数量")) and not kb_id and not file_id:
|
||
return {"operation": "stats", "data": {"success": True, "stats": await service.get_stats()}}
|
||
|
||
if any(word in query for word in ("检查", "对比", "差异", "巡检", "同步情况")):
|
||
target = await _resolve_kb_target(query, kb_id)
|
||
if not target.get("success"):
|
||
return {"operation": "need_kb_name", "data": target}
|
||
diff = await service.diff_kb(kb_id=target["kb_id"], include_children=include_children, limit=limit)
|
||
diff.update({"kb_name": target.get("kb_name", ""), "kb_path": target.get("kb_path", "")})
|
||
actions = _sync_actions_from_diff(diff, use_llm=use_llm)
|
||
plan = await _save_plan(
|
||
chat_id=chat_id,
|
||
intent="sync_after_check",
|
||
scope="kb",
|
||
target=target,
|
||
summary=diff.get("summary", {}),
|
||
actions=actions,
|
||
metadata={"source_operation": "diff_kb", "include_children": include_children, "limit": limit},
|
||
)
|
||
diff["plan"] = plan
|
||
diff["planned_actions"] = actions
|
||
return {"operation": "diff_kb", "data": diff}
|
||
|
||
if any(word in query for word in ("清理", "修复", "对齐", "同步当前", "reconcile")):
|
||
target = await _resolve_kb_target(query, kb_id)
|
||
if not target.get("success"):
|
||
return {"operation": "need_kb_name", "data": target}
|
||
delete_orphan = any(word in query for word in ("删除孤儿", "删除脏数据", "清理脏数据", "已删除", "不存在", "孤儿"))
|
||
preview = await service.reconcile_kb(
|
||
kb_id=target["kb_id"],
|
||
include_children=include_children,
|
||
limit=limit,
|
||
sync_missing=True,
|
||
rebuild_stale=True,
|
||
delete_orphan=delete_orphan,
|
||
use_llm=use_llm,
|
||
dry_run=True,
|
||
)
|
||
preview.update({"kb_name": target.get("kb_name", ""), "kb_path": target.get("kb_path", "")})
|
||
return await _plan_or_execute(
|
||
chat_id=chat_id,
|
||
confirmed=confirmed,
|
||
intent="reconcile_kb",
|
||
scope="kb",
|
||
target=target,
|
||
summary=preview.get("summary", {}),
|
||
actions=preview.get("planned_actions", []),
|
||
metadata={"delete_orphan": delete_orphan, "include_children": include_children, "limit": limit},
|
||
preview=preview,
|
||
)
|
||
|
||
if any(word in query for word in ("删除", "移除")):
|
||
if file_id:
|
||
action = {"action": "delete_wiki_file", "file_id": file_id}
|
||
return await _plan_or_execute(
|
||
chat_id=chat_id,
|
||
confirmed=confirmed,
|
||
intent="delete_wiki_file",
|
||
scope="file",
|
||
target={"kb_id": kb_id or "", "kb_name": "", "kb_path": ""},
|
||
summary={"delete_files": 1},
|
||
actions=[action],
|
||
metadata={"destructive": True},
|
||
preview={"success": True, "planned_actions": [action]},
|
||
)
|
||
target = await _resolve_kb_target(query, kb_id)
|
||
if target.get("success"):
|
||
action = {"action": "delete_wiki_kb", "kb_id": target["kb_id"], "kb_name": target.get("kb_name", "")}
|
||
return await _plan_or_execute(
|
||
chat_id=chat_id,
|
||
confirmed=confirmed,
|
||
intent="delete_wiki_kb",
|
||
scope="kb",
|
||
target=target,
|
||
summary={"delete_kb": 1},
|
||
actions=[action],
|
||
metadata={"destructive": True},
|
||
preview={"success": True, "planned_actions": [action], **target},
|
||
)
|
||
return {"operation": "need_target", "data": {"success": False, "message": "删除 Wiki 数据需要提供文件 id，或说清楚知识库名称。", **target}}
|
||
|
||
if any(word in query for word in ("同步", "构建", "建立", "重建", "生成")):
|
||
if file_id:
|
||
action = {"action": "sync_file", "file_id": file_id, "kb_id": kb_id or "", "use_llm": use_llm}
|
||
return await _plan_or_execute(
|
||
chat_id=chat_id,
|
||
confirmed=confirmed,
|
||
intent="sync_file",
|
||
scope="file",
|
||
target={"kb_id": kb_id or "", "kb_name": "", "kb_path": ""},
|
||
summary={"sync_files": 1},
|
||
actions=[action],
|
||
metadata={"use_llm": use_llm},
|
||
preview={"success": True, "planned_actions": [action]},
|
||
)
|
||
target = await _resolve_kb_target(query, kb_id)
|
||
if target.get("success"):
|
||
preview = await service.reconcile_kb(
|
||
kb_id=target["kb_id"],
|
||
include_children=include_children,
|
||
limit=limit,
|
||
sync_missing=True,
|
||
rebuild_stale=True,
|
||
delete_orphan=False,
|
||
use_llm=use_llm,
|
||
dry_run=True,
|
||
)
|
||
preview.update({"kb_name": target.get("kb_name", ""), "kb_path": target.get("kb_path", "")})
|
||
return await _plan_or_execute(
|
||
chat_id=chat_id,
|
||
confirmed=confirmed,
|
||
intent="sync_kb",
|
||
scope="kb",
|
||
target=target,
|
||
summary=preview.get("summary", {}),
|
||
actions=preview.get("planned_actions", []),
|
||
metadata={"include_children": include_children, "limit": limit, "use_llm": use_llm},
|
||
preview=preview,
|
||
)
|
||
return {"operation": "need_target", "data": {"success": False, "message": "同步 Wiki 需要提供文件 id，或说清楚知识库名称。", **target}}
|
||
|
||
if pending_plan:
|
||
return {"operation": "pending_plan", "data": {"success": True, "plan": pending_plan}}
|
||
|
||
return {
|
||
"operation": "help",
|
||
"data": {
|
||
"success": True,
|
||
"examples": [
|
||
"检查知识库 163舰 的 Wiki 同步情况",
|
||
"同步知识库 163舰 的前 50 个文件",
|
||
"清理知识库 163舰 中 11000 已删除但 Wiki 还存在的数据",
|
||
"确认执行",
|
||
"只执行第 1、3 个",
|
||
],
|
||
},
|
||
}
|
||
|
||
|
||
async def _plan_or_execute(
|
||
chat_id: str,
|
||
confirmed: bool,
|
||
intent: str,
|
||
scope: str,
|
||
target: Dict[str, Any],
|
||
summary: Dict[str, Any],
|
||
actions: List[Dict[str, Any]],
|
||
metadata: Dict[str, Any],
|
||
preview: Dict[str, Any],
|
||
) -> Dict[str, Any]:
|
||
plan = await _save_plan(chat_id, intent, scope, target, summary, actions, metadata)
|
||
preview["plan"] = plan
|
||
preview["planned_actions"] = actions
|
||
if confirmed and actions:
|
||
if not plan:
|
||
return {"operation": "plan_preview", "data": preview}
|
||
executed = await service.execute_admin_plan(plan)
|
||
executed["plan"] = plan
|
||
return {"operation": "plan_executed", "data": executed}
|
||
return {"operation": "plan_preview", "data": preview}
|
||
|
||
|
||
async def _save_plan(
|
||
chat_id: str,
|
||
intent: str,
|
||
scope: str,
|
||
target: Dict[str, Any],
|
||
summary: Dict[str, Any],
|
||
actions: List[Dict[str, Any]],
|
||
metadata: Dict[str, Any],
|
||
) -> Dict[str, Any]:
|
||
if not chat_id or not actions:
|
||
return {}
|
||
return await service.create_admin_plan(
|
||
chat_id=chat_id,
|
||
intent=intent,
|
||
scope=scope,
|
||
kb_id=str(target.get("kb_id") or ""),
|
||
kb_name=str(target.get("kb_name") or target.get("kb_path") or ""),
|
||
summary=summary,
|
||
actions=actions,
|
||
metadata=metadata,
|
||
)
|
||
|
||
|
||
def _sync_actions_from_diff(diff: Dict[str, Any], use_llm: bool = False) -> List[Dict[str, Any]]:
|
||
actions: List[Dict[str, Any]] = []
|
||
for item in diff.get("missing_in_wiki", []):
|
||
actions.append({
|
||
"action": "sync_file",
|
||
"file_id": item.get("file_id", ""),
|
||
"kb_id": item.get("kb_id") or diff.get("kb_id", ""),
|
||
"filename": item.get("filename", ""),
|
||
"use_llm": use_llm,
|
||
})
|
||
for item in diff.get("stale_in_wiki", []):
|
||
actions.append({
|
||
"action": "rebuild_file",
|
||
"file_id": item.get("file_id", ""),
|
||
"kb_id": item.get("kb_id") or diff.get("kb_id", ""),
|
||
"filename": item.get("filename", ""),
|
||
"use_llm": use_llm,
|
||
})
|
||
return actions
|
||
|
||
|
||
def _format_result(result: Dict[str, Any]) -> str:
|
||
op = result.get("operation", "")
|
||
data = result.get("data", {})
|
||
if op == "help":
|
||
return "我是 Wiki 管理助手，可以检查、生成计划、同步、清理和删除 Wiki 指针数据。\n\n示例：\n" + "\n".join(f"- {x}" for x in data.get("examples", []))
|
||
if op in {"need_kb_id", "need_kb_name", "need_target"}:
|
||
matches = data.get("matches") or data.get("candidates") or []
|
||
if matches:
|
||
options = "\n".join(
|
||
f"- {item.get('path') or item.get('name')}（id: {item.get('kb_id')}）"
|
||
for item in matches[:5]
|
||
)
|
||
return f"{data.get('message') or data.get('error') or '请说清楚知识库名称。'}\n可选知识库：\n{options}"
|
||
return data.get("message", "请补充必要参数。")
|
||
if op == "stats":
|
||
stats = data.get("stats", {})
|
||
return "当前 Wiki 数据统计：\n" + "\n".join(f"- {k}: {v}" for k, v in stats.items())
|
||
if op == "job_progress":
|
||
if not data.get("success"):
|
||
return data.get("message") or "未找到同步任务。"
|
||
return _format_job_progress(data.get("job") or {})
|
||
if op == "diff_kb":
|
||
summary = data.get("summary", {})
|
||
target = data.get("kb_path") or data.get("kb_name") or data.get("kb_id")
|
||
lines = [
|
||
f"知识库 {target} 的 Wiki 同步检查完成。",
|
||
f"- 11000 可用文件数：{summary.get('htknow_files', 0)}",
|
||
f"- Wiki 已同步文件数：{summary.get('wiki_files', 0)}",
|
||
f"- 11000 有但 Wiki 缺失：{summary.get('missing_in_wiki', 0)}",
|
||
f"- Wiki 已过期需重建：{summary.get('stale_in_wiki', 0)}",
|
||
f"- Wiki 有但 11000 不存在：{summary.get('orphan_in_wiki', 0)}",
|
||
f"- 跳过未解析文件：{summary.get('skipped_unparsed', 0)}",
|
||
]
|
||
lines.extend(_format_plan_tail(data.get("plan") or {}, data.get("planned_actions") or []))
|
||
return "\n".join(lines)
|
||
if op in {"plan_preview", "pending_plan"}:
|
||
plan = data.get("plan") or {}
|
||
actions = data.get("planned_actions") or plan.get("actions") or []
|
||
target = data.get("kb_path") or data.get("kb_name") or plan.get("kb_name") or plan.get("kb_id") or "当前目标"
|
||
lines = [f"已生成 {target} 的操作计划。", f"- 计划编号：{plan.get('id') or '未保存'}", f"- 待执行动作：{len(actions)}"]
|
||
lines.extend(_format_actions(actions))
|
||
if plan:
|
||
lines.append("确认无误后，请回复“确认执行”；也可以说“只执行第 1、3 个”或“取消”。")
|
||
else:
|
||
lines.append("当前请求没有 chat_id，计划未落库；需要前端带 chat_id 才能多轮确认执行。")
|
||
return "\n".join(lines)
|
||
if op == "plan_executed":
|
||
plan = data.get("plan") or {}
|
||
lines = [
|
||
f"已提交 Wiki 同步计划 {plan.get('id') or ''}。",
|
||
f"- 已提交任务：{data.get('executed_count', 0)}",
|
||
f"- 提交失败：{data.get('failed_count', 0)}",
|
||
]
|
||
jobs = _extract_executed_jobs(data.get("executed") or [])
|
||
if jobs:
|
||
lines.append("同步正在后台执行，可继续使用系统。当前任务：")
|
||
for job in jobs[:8]:
|
||
lines.append(f"- job_id: {job.get('id') or job.get('job_id')}，进度：{_job_progress_text(job)}")
|
||
if len(jobs) > 8:
|
||
lines.append(f"- 还有 {len(jobs) - 8} 个后台任务")
|
||
lines.append("想看进度时，可以问：查看同步进度 <job_id>")
|
||
return "\n".join(lines)
|
||
if op == "plan_cancelled":
|
||
plan = data.get("plan") or {}
|
||
return f"已取消待执行计划 {plan.get('id') or ''}。"
|
||
return json.dumps(data, ensure_ascii=False, indent=2)
|
||
|
||
|
||
def _format_plan_tail(plan: Dict[str, Any], actions: List[Dict[str, Any]]) -> List[str]:
|
||
if not actions:
|
||
return ["当前没有需要同步或重建的 Wiki 动作。"]
|
||
lines = [f"已为缺失/过期数据生成待确认计划：{plan.get('id') or '未保存'}", f"- 待同步/重建动作：{len(actions)}"]
|
||
lines.extend(_format_actions(actions))
|
||
if plan:
|
||
lines.append("如果要执行，请回复“确认执行”；如果要清理孤儿数据，请明确说“清理孤儿数据”。")
|
||
return lines
|
||
|
||
|
||
def _format_actions(actions: List[Dict[str, Any]], max_items: int = 8) -> List[str]:
|
||
lines = []
|
||
for idx, action in enumerate(actions[:max_items], start=1):
|
||
lines.append(f" {idx}. {_action_label(action)}")
|
||
if len(actions) > max_items:
|
||
lines.append(f" ... 还有 {len(actions) - max_items} 个动作")
|
||
return lines
|
||
|
||
|
||
def _action_label(action: Dict[str, Any]) -> str:
|
||
action_type = action.get("action", "")
|
||
if action_type == "sync_file":
|
||
return f"同步文件 {action.get('filename') or action.get('file_id')}"
|
||
if action_type == "rebuild_file":
|
||
return f"重建文件 {action.get('filename') or action.get('file_id')}"
|
||
if action_type == "delete_wiki_file":
|
||
return f"删除 Wiki 中的文件数据 {action.get('file_id')}"
|
||
if action_type == "delete_wiki_kb":
|
||
return f"删除 Wiki 中的知识库数据 {action.get('kb_name') or action.get('kb_id')}"
|
||
return json.dumps(action, ensure_ascii=False)
|
||
|
||
|
||
def _extract_executed_jobs(executed: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
||
jobs: List[Dict[str, Any]] = []
|
||
for item in executed:
|
||
result = item.get("result") or {}
|
||
job = result.get("job") if isinstance(result.get("job"), dict) else {}
|
||
if job:
|
||
jobs.append(job)
|
||
elif result.get("job_id"):
|
||
jobs.append({"id": result.get("job_id"), "status": result.get("status") or "running"})
|
||
return jobs
|
||
|
||
|
||
def _format_job_progress(job: Dict[str, Any]) -> str:
|
||
current = str(job.get("current_filename") or job.get("current_file_id") or "").strip()
|
||
lines = [
|
||
f"同步任务 {job.get('id') or ''}",
|
||
f"- 状态：{job.get('status') or ''}",
|
||
f"- 目标：{job.get('target_type') or ''} {job.get('target_id') or ''}",
|
||
f"- 当前阶段：{job.get('stage') or '处理中'}",
|
||
]
|
||
if current:
|
||
lines.append(f"- 当前文件：{current}")
|
||
lines.extend([
|
||
f"- 进度：{_job_progress_text(job)}",
|
||
f"- 已处理文件：{job.get('processed_files', 0)} / {job.get('total_files', 0)}",
|
||
f"- 已跳过未变化文件：{job.get('skipped_files', 0)}",
|
||
f"- 已同步切片：{job.get('total_slices', 0)}",
|
||
f"- 已生成/更新页面：{job.get('page_count', 0)}",
|
||
f"- 更新时间：{job.get('updated_at') or ''}",
|
||
f"- 错误：{job.get('error') or '无'}",
|
||
])
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _job_progress_text(job: Dict[str, Any]) -> str:
|
||
total = int(job.get("total_files") or 0)
|
||
processed = int(job.get("processed_files") or 0)
|
||
status = str(job.get("status") or "")
|
||
if total > 0:
|
||
percent = min(100.0, max(0.0, processed * 100.0 / total))
|
||
return f"{processed}/{total}（{percent:.0f}%），状态：{status}"
|
||
return f"等待统计文件数，状态：{status}"
|
||
|
||
|
||
def _response(text: str, result: Optional[Dict[str, Any]] = None, error: str = "") -> Dict[str, Any]:
|
||
return {
|
||
"response": text,
|
||
"actions": [],
|
||
"result_tag": "wiki_admin",
|
||
"suggestedReplies": [
|
||
{"title": "检查同步情况", "content": "检查知识库 163舰 的 Wiki 同步情况"},
|
||
{"title": "确认执行", "content": "确认执行"},
|
||
],
|
||
"route_flag": "wiki_admin",
|
||
"wiki_admin_result": result or {},
|
||
"error_message": error,
|
||
}
|
||
|
||
|
||
def _is_confirmed(query: str) -> bool:
|
||
return any(word in query for word in CONFIRM_WORDS)
|
||
|
||
|
||
def _is_cancel(query: str) -> bool:
|
||
return any(word in query for word in CANCEL_WORDS)
|
||
|
||
|
||
def _contains_new_target(query: str) -> bool:
|
||
return any(word in query for word in ("知识库", "文件", "file_id", "kb_id"))
|
||
|
||
|
||
def _extract_selection_indexes(query: str) -> Optional[List[int]]:
|
||
if not any(word in query for word in ("第", "只", "选择", "操作", "执行")):
|
||
return None
|
||
nums = [int(x) for x in re.findall(r"\d+", query)]
|
||
return nums or None
|
||
|
||
|
||
def _extract_job_id(query: str) -> Optional[str]:
|
||
match = re.search(r"\b[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\b", query or "")
|
||
return match.group(0) if match else None
|
||
|
||
|
||
def _extract_id(query: str, labels: tuple[str, ...]) -> Optional[str]:
|
||
for label in labels:
|
||
patterns = [
|
||
rf"{re.escape(label)}\s*[:：]?\s*([A-Za-z0-9_\-]+)",
|
||
rf"([A-Za-z0-9_\-]+)\s*号?{re.escape(label)}",
|
||
]
|
||
for pattern in patterns:
|
||
match = re.search(pattern, query, flags=re.IGNORECASE)
|
||
if match:
|
||
return match.group(1)
|
||
return None
|
||
|
||
|
||
def _extract_kb_id(query: str) -> Optional[str]:
|
||
patterns = [
|
||
r"(?:kb_id|kb|知识库\s*(?:id|ID|编号))\s*[:：=]?\s*([A-Za-z0-9_\-]+)",
|
||
r"知识库\s*[:：]?\s*(\d+)(?![A-Za-z0-9_\-])",
|
||
r"(\d+)\s*号?\s*知识库",
|
||
]
|
||
for pattern in patterns:
|
||
match = re.search(pattern, query, flags=re.IGNORECASE)
|
||
if match:
|
||
return match.group(1)
|
||
return None
|
||
|
||
|
||
async def _resolve_kb_target(query: str, kb_id: Optional[str]) -> Dict[str, Any]:
|
||
if kb_id:
|
||
return {"success": True, "kb_id": kb_id, "kb_name": "", "kb_path": ""}
|
||
resolved = await service.resolve_kb_reference(query)
|
||
if resolved.get("success"):
|
||
return resolved
|
||
return {
|
||
"success": False,
|
||
"message": resolved.get("error") or "请说清楚知识库名称。",
|
||
"matches": resolved.get("matches") or resolved.get("candidates") or [],
|
||
}
|
||
|
||
|
||
def _extract_limit(query: str) -> Optional[int]:
|
||
patterns = [r"前\s*(\d+)\s*个", r"limit\s*[:=]\s*(\d+)", r"最多\s*(\d+)\s*个"]
|
||
for pattern in patterns:
|
||
match = re.search(pattern, query, flags=re.IGNORECASE)
|
||
if match:
|
||
return max(1, min(int(match.group(1)), 1000))
|
||
return None
|