222 lines
8.9 KiB
Python
222 lines
8.9 KiB
Python
"""将一期公开 QA 候选转换为不含外部写入的待审核导入清单。"""
|
||
|
||
import argparse
|
||
import json
|
||
from collections.abc import Mapping, Sequence
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from app.core.knowledge_contracts import ALLOWED_KNOWLEDGE_COLLECTIONS
|
||
from app.service.knowledge_config import KnowledgeRuntimeConfig
|
||
|
||
|
||
# 仅允许一期三类公开知识按既定路由进入后续发布流程。
|
||
EXPECTED_COLLECTIONS = {
|
||
intent: collection
|
||
for intent, (collection, _top_k) in KnowledgeRuntimeConfig.DEFAULT_ROUTES.items()
|
||
}
|
||
# 公开候选必须具备的字段，缺失时不能生成不完整的发布清单。
|
||
REQUIRED_PUBLIC_FIELDS = frozenset({
|
||
"qa_id",
|
||
"title",
|
||
"question",
|
||
"paraphrases",
|
||
"answer",
|
||
"scope",
|
||
"intent",
|
||
"collection",
|
||
"execution_mode",
|
||
"retrieval_status",
|
||
"audience",
|
||
"agent_data_access",
|
||
"tags",
|
||
"source_type",
|
||
"source_file",
|
||
"source_version",
|
||
"review_status",
|
||
"status",
|
||
})
|
||
|
||
|
||
def _required_string(record: Mapping[str, object], field: str, qa_id: str) -> str:
|
||
"""读取非空字符串字段，拒绝将不完整资料带入后续发布阶段。"""
|
||
value = record.get(field)
|
||
if not isinstance(value, str) or not value.strip():
|
||
raise ValueError(f"{qa_id}: {field} must be a non-empty string")
|
||
return value.strip()
|
||
|
||
|
||
def _required_strings(record: Mapping[str, object], field: str, qa_id: str) -> list[str]:
|
||
"""读取非空字符串数组，并保持来源中已经审核的条目顺序。"""
|
||
value = record.get(field)
|
||
if not isinstance(value, list) or not value:
|
||
raise ValueError(f"{qa_id}: {field} must be a non-empty string list")
|
||
strings = [item.strip() for item in value if isinstance(item, str) and item.strip()]
|
||
if len(strings) != len(value):
|
||
raise ValueError(f"{qa_id}: {field} must contain only non-empty strings")
|
||
return strings
|
||
|
||
|
||
def _is_rule_only(record: Mapping[str, object]) -> bool:
|
||
"""控制类记录必须由应用层固定路由处理，绝不能进入向量导入清单。"""
|
||
return (
|
||
record.get("collection") is None
|
||
and record.get("execution_mode") == "fixed_route"
|
||
and record.get("retrieval_status") == "rule_only"
|
||
)
|
||
|
||
|
||
def _validate_public_record(record: Mapping[str, object]) -> None:
|
||
"""验证候选是否符合一期公开知识与最小权限边界。"""
|
||
qa_id = _required_string(record, "qa_id", "<unknown>")
|
||
missing = sorted(field for field in REQUIRED_PUBLIC_FIELDS if field not in record)
|
||
if missing:
|
||
raise ValueError(f"{qa_id}: missing required fields: {', '.join(missing)}")
|
||
intent = _required_string(record, "intent", qa_id)
|
||
collection = _required_string(record, "collection", qa_id)
|
||
if intent not in EXPECTED_COLLECTIONS:
|
||
raise ValueError(f"{qa_id}: unsupported public intent: {intent}")
|
||
if collection not in ALLOWED_KNOWLEDGE_COLLECTIONS:
|
||
raise ValueError(f"{qa_id}: collection is not allowlisted: {collection}")
|
||
if collection != EXPECTED_COLLECTIONS[intent]:
|
||
raise ValueError(f"{qa_id}: collection does not match intent route")
|
||
if record.get("scope") != "public":
|
||
raise ValueError(f"{qa_id}: public vector record must have scope=public")
|
||
if record.get("execution_mode") != "vector_search":
|
||
raise ValueError(f"{qa_id}: public record must use vector_search")
|
||
if record.get("retrieval_status") != "approved_candidate":
|
||
raise ValueError(f"{qa_id}: record is not an approved candidate")
|
||
if record.get("review_status") != "approved_candidate":
|
||
raise ValueError(f"{qa_id}: review state cannot enter preflight")
|
||
if record.get("status") != "active":
|
||
raise ValueError(f"{qa_id}: inactive record cannot enter preflight")
|
||
if record.get("agent_data_access") != "none":
|
||
raise ValueError(f"{qa_id}: agent data access must remain none")
|
||
audience = _required_strings(record, "audience", qa_id)
|
||
if set(audience) != {"visitor", "authenticated_user"}:
|
||
raise ValueError(f"{qa_id}: audience must be visitor and authenticated_user")
|
||
_required_string(record, "title", qa_id)
|
||
_required_string(record, "question", qa_id)
|
||
_required_strings(record, "paraphrases", qa_id)
|
||
_required_string(record, "answer", qa_id)
|
||
_required_strings(record, "tags", qa_id)
|
||
_required_string(record, "source_type", qa_id)
|
||
_required_string(record, "source_file", qa_id)
|
||
_required_string(record, "source_version", qa_id)
|
||
|
||
|
||
def _public_entry(record: Mapping[str, object]) -> dict[str, object]:
|
||
"""构造供管理员审核的确定性条目，不生成数据库主键或向量。"""
|
||
qa_id = _required_string(record, "qa_id", "<unknown>")
|
||
question = _required_string(record, "question", qa_id)
|
||
paraphrases = _required_strings(record, "paraphrases", qa_id)
|
||
tags = _required_strings(record, "tags", qa_id)
|
||
answer = _required_string(record, "answer", qa_id)
|
||
content = {
|
||
"qa_id": qa_id,
|
||
"question": question,
|
||
"paraphrases": paraphrases,
|
||
"answer": answer,
|
||
"audience": _required_strings(record, "audience", qa_id),
|
||
"agent_data_access": "none",
|
||
"source_version": _required_string(record, "source_version", qa_id),
|
||
}
|
||
return {
|
||
"qa_id": qa_id,
|
||
"knowledge_type": _required_string(record, "intent", qa_id),
|
||
"title": _required_string(record, "title", qa_id),
|
||
"milvus_collection": _required_string(record, "collection", qa_id),
|
||
"version": _required_string(record, "source_version", qa_id),
|
||
"source_file": _required_string(record, "source_file", qa_id),
|
||
"source_type": _required_string(record, "source_type", qa_id),
|
||
"source_url": record.get("source_url"),
|
||
"effective_date": record.get("effective_date"),
|
||
"expire_date": record.get("expire_date"),
|
||
"tags": tags,
|
||
"content_text": json.dumps(content, ensure_ascii=False, separators=(",", ":")),
|
||
"retrieval_text": (
|
||
f"标准问题：{question}\n"
|
||
f"相似问法：{'；'.join(paraphrases)}\n"
|
||
f"标签：{'、'.join(tags)}"
|
||
),
|
||
"snippet": question[:300],
|
||
# 管理员填入真实审核人并批准前，预检清单绝不伪装成已发布数据。
|
||
"review_status": "pending_review",
|
||
"status": "active",
|
||
}
|
||
|
||
|
||
def build_import_manifest(
|
||
records: Sequence[Mapping[str, object]], *, source_name: str
|
||
) -> dict[str, object]:
|
||
"""构建可复查的导入清单，并拒绝任何不满足公开边界的非控制类记录。"""
|
||
entries: list[dict[str, object]] = []
|
||
seen_ids: set[str] = set()
|
||
excluded_rule_records = 0
|
||
for record in records:
|
||
if _is_rule_only(record):
|
||
excluded_rule_records += 1
|
||
continue
|
||
_validate_public_record(record)
|
||
entry = _public_entry(record)
|
||
qa_id = str(entry["qa_id"])
|
||
if qa_id in seen_ids:
|
||
raise ValueError(f"duplicate qa_id: {qa_id}")
|
||
seen_ids.add(qa_id)
|
||
entries.append(entry)
|
||
return {
|
||
"source_name": source_name,
|
||
"summary": {
|
||
"total_records": len(records),
|
||
"eligible_records": len(entries),
|
||
"excluded_rule_records": excluded_rule_records,
|
||
"publication_state": "pending_review",
|
||
},
|
||
"records": entries,
|
||
}
|
||
|
||
|
||
def load_jsonl(source: Path) -> list[dict[str, object]]:
|
||
"""读取 UTF-8 JSONL，并为每个无效 JSON 行返回带行号的明确错误。"""
|
||
records: list[dict[str, object]] = []
|
||
for line_number, line in enumerate(source.read_text(encoding="utf-8").splitlines(), start=1):
|
||
if not line.strip():
|
||
continue
|
||
try:
|
||
value: Any = json.loads(line)
|
||
except json.JSONDecodeError as exc:
|
||
raise ValueError(f"line {line_number}: invalid JSON") from exc
|
||
if not isinstance(value, dict):
|
||
raise ValueError(f"line {line_number}: record must be an object")
|
||
records.append(value)
|
||
return records
|
||
|
||
|
||
def write_manifest(target: Path, manifest: Mapping[str, object]) -> None:
|
||
"""仅在调用者显式传入输出路径时，写入本地待审核 JSON 清单。"""
|
||
target.parent.mkdir(parents=True, exist_ok=True)
|
||
target.write_text(
|
||
json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
|
||
)
|
||
|
||
|
||
def main() -> int:
|
||
"""执行本地预检；该入口不读取配置也不连接任何外部服务。"""
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument("--input", type=Path, required=True)
|
||
parser.add_argument("--output", type=Path)
|
||
arguments = parser.parse_args()
|
||
manifest = build_import_manifest(
|
||
load_jsonl(arguments.input), source_name=arguments.input.name
|
||
)
|
||
if arguments.output is None:
|
||
print(json.dumps(manifest, ensure_ascii=False, indent=2))
|
||
else:
|
||
write_manifest(arguments.output, manifest)
|
||
print(f"WROTE: {arguments.output}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|