284 lines
12 KiB
Python
284 lines
12 KiB
Python
"""管理员显式批准后发布一期客服公开知识；默认只验证清单，不写外部服务。"""
|
|
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
from collections.abc import Mapping
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any, cast
|
|
|
|
from sqlalchemy import select, text, update
|
|
|
|
# 直接执行 tools 脚本时优先解析当前工作树，避免误导入相邻 worktree 的 app 包。
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from app.service.knowledge_publication_service import ( # noqa: E402
|
|
KnowledgePublicationRecord,
|
|
KnowledgePublicationService,
|
|
)
|
|
|
|
# 外部服务依赖仅在显式 --apply 时载入，dry-run 不需要本地 .env 或服务可达。
|
|
SessionFactory: Any
|
|
InteractionAudit: Any
|
|
FinKnowledgeMeta: Any
|
|
DatabaseModelGateway: Any
|
|
get_settings: Any
|
|
|
|
|
|
class DatabaseKnowledgeEmbedder:
|
|
"""发布工具只走专用知识向量端点，禁止意外使用聊天端点。"""
|
|
|
|
async def embed(self, text_value: str) -> list[float]:
|
|
settings = get_settings()
|
|
endpoint_code = settings.knowledge_embedding_endpoint_code
|
|
if not endpoint_code:
|
|
raise RuntimeError("KNOWLEDGE_EMBEDDING_ENDPOINT_CODE 未配置")
|
|
return cast(
|
|
list[float],
|
|
await DatabaseModelGateway().embed(
|
|
endpoint_code=endpoint_code,
|
|
text=text_value,
|
|
timeout_ms=settings.knowledge_embedding_timeout_ms,
|
|
),
|
|
)
|
|
|
|
|
|
class SqlAlchemyKnowledgePublicationStore:
|
|
"""使用现有知识表暂存、发布和停用记录，不变更数据库表结构。"""
|
|
|
|
def __init__(self, reviewer_id: int) -> None:
|
|
self._reviewer_id = reviewer_id
|
|
|
|
async def stage(self, records: tuple[KnowledgePublicationRecord, ...]) -> dict[str, int]:
|
|
now = datetime.now(UTC).replace(tzinfo=None)
|
|
async with SessionFactory() as session, session.begin():
|
|
await self._assert_reviewer(session)
|
|
await self._reject_existing_qa_ids(session, records)
|
|
rows: list[Any] = []
|
|
for record in records:
|
|
row = FinKnowledgeMeta(
|
|
knowledge_type=str(record.metadata["knowledge_type"]),
|
|
title=record.title,
|
|
source_file=str(record.metadata["source_file"]),
|
|
minio_path=None,
|
|
milvus_collection=record.milvus_collection,
|
|
version=record.version,
|
|
effective_date=record.metadata.get("effective_date"),
|
|
expire_date=record.metadata.get("expire_date"),
|
|
content_text=str(record.metadata["content_text"]),
|
|
tags=list(record.tags),
|
|
reviewer_id=None,
|
|
review_status="pending",
|
|
status="disabled",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(row)
|
|
rows.append(row)
|
|
await session.flush()
|
|
session.add(InteractionAudit(
|
|
actor_type="admin",
|
|
actor_id=self._reviewer_id,
|
|
portal="admin",
|
|
action_type="knowledge.publication_staged",
|
|
detail={"qa_ids": [record.qa_id for record in records]},
|
|
created_at=now,
|
|
))
|
|
return {record.qa_id: int(row.id) for record, row in zip(records, rows, strict=True)}
|
|
|
|
async def publish(self, knowledge_ids: tuple[int, ...], reviewer_id: int) -> None:
|
|
now = datetime.now(UTC).replace(tzinfo=None)
|
|
async with SessionFactory() as session, session.begin():
|
|
await session.execute(
|
|
update(FinKnowledgeMeta)
|
|
.where(FinKnowledgeMeta.id.in_(knowledge_ids))
|
|
.values(
|
|
reviewer_id=reviewer_id,
|
|
review_status="published",
|
|
status="active",
|
|
updated_at=now,
|
|
)
|
|
)
|
|
session.add(InteractionAudit(
|
|
actor_type="admin",
|
|
actor_id=reviewer_id,
|
|
portal="admin",
|
|
action_type="knowledge.publication_completed",
|
|
detail={"knowledge_ids": list(knowledge_ids)},
|
|
created_at=now,
|
|
))
|
|
|
|
async def disable(self, knowledge_ids: tuple[int, ...]) -> None:
|
|
now = datetime.now(UTC).replace(tzinfo=None)
|
|
async with SessionFactory() as session, session.begin():
|
|
await session.execute(
|
|
update(FinKnowledgeMeta)
|
|
.where(FinKnowledgeMeta.id.in_(knowledge_ids))
|
|
.values(status="disabled", updated_at=now)
|
|
)
|
|
session.add(InteractionAudit(
|
|
actor_type="system",
|
|
actor_id=None,
|
|
portal="admin",
|
|
action_type="knowledge.publication_failed",
|
|
detail={"knowledge_ids": list(knowledge_ids)},
|
|
created_at=now,
|
|
))
|
|
|
|
async def _assert_reviewer(self, session: Any) -> None:
|
|
row = await session.execute(
|
|
text(
|
|
"SELECT id FROM sys_user "
|
|
"WHERE id = :reviewer_id AND status IN ('正常', 'active') "
|
|
"AND user_type IN ('employee', 'admin')"
|
|
),
|
|
{"reviewer_id": self._reviewer_id},
|
|
)
|
|
if row.scalar_one_or_none() is None:
|
|
raise RuntimeError("reviewer_id 不是有效的在职管理员或员工账号")
|
|
|
|
@staticmethod
|
|
async def _reject_existing_qa_ids(
|
|
session: Any, records: tuple[KnowledgePublicationRecord, ...]
|
|
) -> None:
|
|
collections = tuple({record.milvus_collection for record in records})
|
|
rows = await session.scalars(
|
|
select(FinKnowledgeMeta)
|
|
.where(FinKnowledgeMeta.milvus_collection.in_(collections))
|
|
.with_for_update()
|
|
)
|
|
existing_ids: set[str] = set()
|
|
for row in rows:
|
|
try:
|
|
content = json.loads(row.content_text)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
qa_id = content.get("qa_id") if isinstance(content, dict) else None
|
|
if isinstance(qa_id, str):
|
|
existing_ids.add(qa_id)
|
|
duplicates = sorted(existing_ids & {record.qa_id for record in records})
|
|
if duplicates:
|
|
raise RuntimeError(f"qa_id 已存在，拒绝重复发布: {', '.join(duplicates)}")
|
|
|
|
|
|
class MilvusKnowledgePublicationStore:
|
|
"""管理员发布期的最小 Milvus 写适配器；客服 Agent 运行期仍只能检索。"""
|
|
|
|
def __init__(self) -> None:
|
|
settings = get_settings()
|
|
self._uri = settings.resolved_milvus_uri
|
|
self._token = settings.milvus_token or None
|
|
self._client: Any | None = None
|
|
|
|
async def _client_instance(self) -> Any:
|
|
if self._client is None:
|
|
from pymilvus import AsyncMilvusClient # type: ignore[import-untyped]
|
|
|
|
self._client = AsyncMilvusClient(uri=self._uri, token=self._token)
|
|
return self._client
|
|
|
|
async def upsert(self, collection: str, records: tuple[dict[str, object], ...]) -> None:
|
|
client = await self._client_instance()
|
|
payload = [
|
|
{**record, "tags": json.dumps(record["tags"], ensure_ascii=False)}
|
|
for record in records
|
|
]
|
|
await client.upsert(collection_name=collection, data=payload)
|
|
|
|
async def delete(self, collection: str, knowledge_ids: tuple[str, ...]) -> None:
|
|
if not all(knowledge_id.isdecimal() for knowledge_id in knowledge_ids):
|
|
raise ValueError("knowledge_id 必须为十进制主键")
|
|
client = await self._client_instance()
|
|
values = ", ".join(json.dumps(knowledge_id) for knowledge_id in knowledge_ids)
|
|
await client.delete(collection_name=collection, filter=f"knowledge_id in [{values}]")
|
|
|
|
|
|
def load_pending_manifest(value: Mapping[str, object]) -> tuple[KnowledgePublicationRecord, ...]:
|
|
"""只接受预检工具输出的 pending_review 清单，拒绝手工伪造已发布状态。"""
|
|
summary = value.get("summary")
|
|
records = value.get("records")
|
|
if not isinstance(summary, dict) or summary.get("publication_state") != "pending_review":
|
|
raise ValueError("发布清单必须处于 pending_review 状态")
|
|
if not isinstance(records, list) or summary.get("eligible_records") != len(records):
|
|
raise ValueError("发布清单记录数与汇总不一致")
|
|
result: list[KnowledgePublicationRecord] = []
|
|
for raw in records:
|
|
if not isinstance(raw, dict):
|
|
raise ValueError("发布清单记录必须是对象")
|
|
required = ("qa_id", "milvus_collection", "retrieval_text", "title", "snippet", "version")
|
|
if any(not isinstance(raw.get(field), str) or not raw[field].strip() for field in required):
|
|
raise ValueError("发布清单缺少字符串字段")
|
|
tags = raw.get("tags")
|
|
if not isinstance(tags, list) or not all(isinstance(tag, str) and tag for tag in tags):
|
|
raise ValueError("发布清单标签无效")
|
|
if raw.get("review_status") != "pending_review" or raw.get("status") != "active":
|
|
raise ValueError("只有预检待审核记录可以发布")
|
|
result.append(KnowledgePublicationRecord(
|
|
qa_id=str(raw["qa_id"]),
|
|
milvus_collection=str(raw["milvus_collection"]),
|
|
retrieval_text=str(raw["retrieval_text"]),
|
|
title=str(raw["title"]),
|
|
snippet=str(raw["snippet"]),
|
|
tags=tuple(tags),
|
|
version=str(raw["version"]),
|
|
metadata=dict(raw),
|
|
))
|
|
return tuple(result)
|
|
|
|
|
|
def _read_manifest(path: Path) -> tuple[KnowledgePublicationRecord, ...]:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
if not isinstance(value, dict):
|
|
raise ValueError("发布清单根节点必须是对象")
|
|
return load_pending_manifest(value)
|
|
|
|
|
|
async def _apply(records: tuple[KnowledgePublicationRecord, ...], reviewer_id: int) -> None:
|
|
global DatabaseModelGateway, FinKnowledgeMeta, InteractionAudit, SessionFactory, get_settings
|
|
from app.core.config import get_settings
|
|
from app.infrastructure.db import SessionFactory
|
|
from app.model.audit import InteractionAudit
|
|
from app.model.knowledge import FinKnowledgeMeta
|
|
from app.service.model_gateway import DatabaseModelGateway
|
|
|
|
service = KnowledgePublicationService(
|
|
DatabaseKnowledgeEmbedder(),
|
|
SqlAlchemyKnowledgePublicationStore(reviewer_id),
|
|
MilvusKnowledgePublicationStore(),
|
|
)
|
|
result = await service.publish(records, reviewer_id=reviewer_id)
|
|
print(
|
|
json.dumps(
|
|
{"published_records": len(result.knowledge_ids), "collections": result.collections},
|
|
ensure_ascii=False,
|
|
)
|
|
)
|
|
|
|
|
|
def main() -> int:
|
|
"""默认 dry-run，且 apply 必须三重确认，防止候选资料被误发布。"""
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--input", type=Path, required=True)
|
|
parser.add_argument("--reviewer-id", type=int)
|
|
parser.add_argument("--apply", action="store_true")
|
|
parser.add_argument("--confirm-count", type=int)
|
|
arguments = parser.parse_args()
|
|
records = _read_manifest(arguments.input)
|
|
if not arguments.apply:
|
|
print(f"DRY RUN: {len(records)} records are pending administrator review")
|
|
return 0
|
|
if arguments.reviewer_id is None or arguments.reviewer_id <= 0:
|
|
raise SystemExit("--apply requires a positive --reviewer-id")
|
|
if arguments.confirm_count != len(records):
|
|
raise SystemExit("--apply requires --confirm-count equal to the manifest record count")
|
|
asyncio.run(_apply(records, arguments.reviewer_id))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|