140 lines
5.3 KiB
Python
140 lines
5.3 KiB
Python
"""与中心服务端同步经验：上报本校草稿、拉取中心审核发布的结果。
|
||
|
||
两条原则：
|
||
1. 永远不影响办事。同步全部在后台跑，失败只记一行日志，任务该怎么跑还怎么跑。
|
||
2. 不配 HUB_URL / HUB_TOKEN 就整个关掉，桌面端单机照常工作。
|
||
|
||
拉回来的条目直接落成本地 verified（来源记 hub），下次执行同类业务时注入提示词；
|
||
中心驳回的条目在本地也置为 rejected，避免本地自动晋升过的错知识继续生效。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
|
||
import httpx
|
||
|
||
from backend.config import settings
|
||
from backend.db import database as db
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
VERSION_KEY = "hub_pulled_version"
|
||
_LAST_ERROR_KEY = "hub_last_error"
|
||
|
||
|
||
def enabled() -> bool:
|
||
return bool(settings.hub_url and settings.hub_token)
|
||
|
||
|
||
def _client() -> httpx.AsyncClient:
|
||
return httpx.AsyncClient(
|
||
base_url=settings.hub_url.rstrip("/"),
|
||
headers={"Authorization": f"Bearer {settings.hub_token}"},
|
||
timeout=settings.hub_timeout_seconds,
|
||
)
|
||
|
||
|
||
def _note_error(action: str, error: Exception) -> None:
|
||
text = f"{action} 失败：{type(error).__name__}: {str(error)[:160]}"
|
||
log.warning("[hub] %s", text)
|
||
try:
|
||
db.set_setting(_LAST_ERROR_KEY, text)
|
||
except Exception: # noqa: BLE001 连本地库都挂了就别再添乱
|
||
pass
|
||
|
||
|
||
async def report_drafts(knowledge_id: str | None = None, business_id: str | None = None) -> dict:
|
||
"""把本校还没被中心处理过的草稿报上去。中心按内容判重，重复上报是安全的。"""
|
||
if not enabled():
|
||
return {"skipped": "未配置中心服务端"}
|
||
# 草稿要报（让中心看到队列），本地经试用升级的也要报——那是最硬的证据，中心据此可以推给别的学校
|
||
drafts = db.list_learned(knowledge_id=knowledge_id, business_id=business_id, status="draft")
|
||
proven = [
|
||
r for r in db.list_learned(knowledge_id=knowledge_id, business_id=business_id, status="verified")
|
||
if r.get("verified_by") == "trial"
|
||
]
|
||
rows = drafts + proven
|
||
if not rows:
|
||
return {"reported": 0}
|
||
payload = [
|
||
{
|
||
"knowledge_id": r["knowledge_id"],
|
||
"business_id": r["business_id"] or "",
|
||
"key": r["key"],
|
||
"value": r["value"],
|
||
"category": r["category"],
|
||
"observations": int(r["observations"] or 1),
|
||
"source_task_id": r.get("source_task_id") or "",
|
||
# 试用证据：Agent 在真实页面上验过几次。比观察次数硬得多，中心据此判断可信度
|
||
"confirmed": int(r.get("confirmed") or 0),
|
||
"contradicted": int(r.get("contradicted") or 0),
|
||
"status": r.get("status") or "draft",
|
||
}
|
||
for r in rows
|
||
]
|
||
try:
|
||
async with _client() as client:
|
||
resp = await client.post("/sync/report", json={"school_id": settings.school_id, "entries": payload})
|
||
resp.raise_for_status()
|
||
result = resp.json()
|
||
log.info("[hub] 上报 %d 条草稿：%s", len(payload), result)
|
||
return {"reported": len(payload), **result}
|
||
except Exception as e: # noqa: BLE001
|
||
_note_error("上报草稿", e)
|
||
return {"error": str(e)[:200]}
|
||
|
||
|
||
async def pull_published() -> dict:
|
||
"""拉取中心审核后的结果，落成本地 verified / rejected。按版本号增量。"""
|
||
if not enabled():
|
||
return {"skipped": "未配置中心服务端"}
|
||
since = db.get_setting(VERSION_KEY, "0")
|
||
since_version = int(since) if str(since).isdigit() else 0
|
||
try:
|
||
async with _client() as client:
|
||
resp = await client.get("/sync/published", params={"school_id": settings.school_id, "since": since_version})
|
||
resp.raise_for_status()
|
||
data = resp.json()
|
||
except Exception as e: # noqa: BLE001
|
||
_note_error("拉取已发布", e)
|
||
return {"error": str(e)[:200]}
|
||
|
||
entries = data.get("entries") or []
|
||
applied = db.apply_hub_entries(entries)
|
||
if entries:
|
||
db.set_setting(VERSION_KEY, str(max(int(e.get("version") or 0) for e in entries)))
|
||
log.info("[hub] 拉取 %d 条，应用 %s", len(entries), applied)
|
||
return {"pulled": len(entries), **applied, "hub_version": data.get("version")}
|
||
|
||
|
||
async def sync_once() -> dict:
|
||
"""先报后拉：刚沉淀的草稿也能在这一轮被中心看到。"""
|
||
report = await report_drafts()
|
||
pull = await pull_published()
|
||
return {"report": report, "pull": pull}
|
||
|
||
|
||
async def sync_loop(interval_seconds: int) -> None:
|
||
if not enabled():
|
||
log.info("[hub] 未配置 HUB_URL / HUB_TOKEN，中心同步关闭")
|
||
return
|
||
log.info("[hub] 中心同步已开启：%s，每 %d 秒一次", settings.hub_url, interval_seconds)
|
||
while True:
|
||
try:
|
||
await sync_once()
|
||
except Exception as e: # noqa: BLE001 循环不能被任何异常打断
|
||
_note_error("定时同步", e)
|
||
await asyncio.sleep(max(60, interval_seconds))
|
||
|
||
|
||
def status() -> dict:
|
||
return {
|
||
"enabled": enabled(),
|
||
"url": settings.hub_url,
|
||
"school_id": settings.school_id,
|
||
"pulled_version": db.get_setting(VERSION_KEY, "0") if enabled() else "0",
|
||
"last_error": db.get_setting(_LAST_ERROR_KEY, "") if enabled() else "",
|
||
}
|