49 lines
2.8 KiB
Python
49 lines
2.8 KiB
Python
"""Wiki package flow orchestration; all remote writes use WeKnora APIs."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
from starlette.requests import Request
|
|
|
|
|
|
async def sync_to_global(app, mapping):
|
|
mapping_id = str(mapping["id"])
|
|
# Multiple local-index triggers (manual, mutation debounce and periodic
|
|
# reconciliation) may finish close together. Coalesce the check/create/
|
|
# import sequence per mapping inside this Gateway process; the repository's
|
|
# database base-row lock remains the cross-process authority.
|
|
locks = getattr(app.state, "knowledge_transfer_locks", None)
|
|
if locks is None:
|
|
locks = {}
|
|
app.state.knowledge_transfer_locks = locks
|
|
lock = locks.get(mapping_id)
|
|
if lock is None:
|
|
lock = asyncio.Lock()
|
|
locks[mapping_id] = lock
|
|
async with lock:
|
|
return await _sync_to_global_locked(app, mapping, mapping_id=mapping_id)
|
|
|
|
|
|
async def _sync_to_global_locked(app, mapping, *, mapping_id: str):
|
|
from app.gateway.routers.assistant_knowledge import _export_weknora_then_import_for_job
|
|
|
|
if mapping.get("is_conversation_deposit") or mapping.get("owner_user_id") == "system":
|
|
return {"status": "excluded"}
|
|
store = app.state.assistant_knowledge_store
|
|
revision, fingerprint, _ = await app.state.llmwiki_index_store.get_index_revision(mapping_id)
|
|
base = await store.initialize(created_by=None)
|
|
for source in await store.list_sources(base_id=base["id"]):
|
|
meta = source.get("metadata") or {}
|
|
if source.get("source_key") == mapping_id and meta.get("index_revision") == revision:
|
|
job = await store.get_job(source.get("last_import_job_id")) if source.get("last_import_job_id") else None
|
|
if job and job.get("status") == "completed":
|
|
return {"status": "unchanged", "job_id": job["id"], "base_id": base["id"]}
|
|
source = {"provider": "weknora", "type": "weknora", "knowledge_base_id": mapping_id, "knowledge_base_name": mapping["name"], "index_revision": revision, "embedding_fingerprint": fingerprint}
|
|
job = await store.create_queued_import_job(source_type="weknora", source_key=mapping_id, source_name=mapping["name"], trigger="vectorization_completed", created_by=None, metadata=source, mapping_id=mapping_id, base_id=base["id"])
|
|
await _export_weknora_then_import_for_job(request=Request({"type": "http", "app": app}), job_id=job["id"], mapping=mapping, remote_id=mapping["weknora_id"], source=source, actor=None, index_fresh=True)
|
|
result = await store.get_job(job["id"])
|
|
if not result or result.get("status") != "completed":
|
|
raise RuntimeError(f"向量化已完成，但知识梳理同步失败：{(result or {}).get('last_error_message') or job['id']}")
|
|
return {"status": "completed", "job_id": job["id"], "base_id": base["id"], "counts": result.get("counts")}
|