195 lines
12 KiB
Python
195 lines
12 KiB
Python
"""Immutable maintainer judgements over existing multi-axis analysis sources."""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import sqlite3
|
||
from typing import Annotated, Literal
|
||
|
||
from fastapi import HTTPException
|
||
from pydantic import Field, ValidationError
|
||
|
||
from .analysis import RESULT_VERSION, location_key, quotes_valid
|
||
from .business_briefs import safe_text
|
||
from .consultation import validate_consultation
|
||
from .db import StoreConflict, digest, encoded, identifier, now
|
||
from .learning_contracts import ClosedInput
|
||
from .model_client import ModelCallError
|
||
|
||
LEGACY_RESULT_VERSION = "demand-result-v2"
|
||
LEGACY_CONTRACT_VERSION = "analysis-review-v1"
|
||
CONTRACT_VERSION = "analysis-review-v2"
|
||
AXES = ("role", "demand", "expressions", "business_relevance", "match")
|
||
CONTEXT_AXES = (*AXES, "consultation")
|
||
DECISIONS = ("confirmed", "corrected", "uncertain", "unreviewable")
|
||
METRIC_LABEL = "已判定标注的一致率（非准确率或召回率）"
|
||
|
||
|
||
def reject(code, message, status=422):
|
||
raise HTTPException(status, {"code": code, "message": message})
|
||
|
||
|
||
class ReviewLabels(ClosedInput):
|
||
role: Literal["suspected_demand_side", "suspected_service_provider", "participant", "unknown"]
|
||
demand: Literal["explicit", "possible", "discussion", "insufficient"]
|
||
expressions: list[Literal["self_need", "question", "seek_advice", "service_offer", "advice", "case_share", "other", "unknown"]] = Field(min_length=1, max_length=6)
|
||
business_relevance: Literal["relevant", "irrelevant", "unclear"]
|
||
match: Literal["matched", "possible", "insufficient", "not_matched"]
|
||
consultation: Literal["yes", "no", "unclear"] | None = None
|
||
|
||
|
||
class ReviewContextEvidence(ClosedInput):
|
||
source_ref: Literal["note"]
|
||
evidence: list[Annotated[str, Field(min_length=1, max_length=1000, strict=True)]] = Field(min_length=1, max_length=3)
|
||
|
||
|
||
class ReviewInput(ClosedInput):
|
||
source_ref: str = Field(min_length=1, max_length=100, pattern=r"^[A-Za-z0-9_]+$", strict=True)
|
||
previous_review_id: str | None = Field(min_length=1, max_length=100, pattern=r"^[A-Za-z0-9_-]+$", strict=True)
|
||
decision: Literal["confirmed", "corrected", "uncertain", "unreviewable"]
|
||
labels: ReviewLabels | None
|
||
evidence: list[Annotated[str, Field(min_length=1, max_length=1000, strict=True)]] = Field(default_factory=list, max_length=3)
|
||
context_evidence: list[ReviewContextEvidence] = Field(default_factory=list, max_length=1)
|
||
note: str = Field(default="", max_length=1000, strict=True)
|
||
|
||
|
||
def model_labels(result):
|
||
labels = {"role": result["role"], "demand": result["demand"], "expressions": sorted(entry["type"] for entry in result["expressions"]),
|
||
"business_relevance": result["business_relevance"]["value"], "match": result["match"]}
|
||
if "consultation" in result:
|
||
labels["consultation"] = result["consultation"]["value"]
|
||
return labels
|
||
|
||
|
||
def eligible(analysis):
|
||
return analysis.get("status") == "completed" and analysis.get("result_version") in {LEGACY_RESULT_VERSION, RESULT_VERSION}
|
||
|
||
|
||
def review_contract(analysis):
|
||
return CONTRACT_VERSION if analysis.get("result_version") == RESULT_VERSION else LEGACY_CONTRACT_VERSION
|
||
|
||
|
||
def review_axes(analysis):
|
||
return CONTEXT_AXES if analysis.get("result_version") == RESULT_VERSION else AXES
|
||
|
||
|
||
def validate_labels(labels, result, source, conditions):
|
||
expressions = labels["expressions"]
|
||
seen = set(expressions)
|
||
own_need = labels["demand"] in {"explicit", "possible"}
|
||
if (len(seen) != len(expressions) or ("unknown" in seen and len(seen) != 1) or own_need != ("self_need" in seen)
|
||
or (labels["role"] == "suspected_demand_side" and not own_need)
|
||
or (labels["role"] == "suspected_service_provider" and "service_offer" not in seen)):
|
||
reject("REVIEW_LABELS_INVALID", "角色、自身需求与表达标签不一致，请重新核对")
|
||
if labels["match"] in {"matched", "possible"}:
|
||
scope = conditions.get("region_scope")
|
||
fact = result.get("region")
|
||
region_ok = scope == "unrestricted" or (scope == "specified" and isinstance(conditions.get("region"), str) and isinstance(fact, dict)
|
||
and isinstance(fact.get("value"), str) and location_key(fact["value"]) == location_key(conditions["region"]))
|
||
if not own_need or labels["business_relevance"] != "relevant" or not region_ok or not source["text"].strip():
|
||
reject("REVIEW_LABELS_INVALID", "匹配须有自身需求、业务相关且原条件下地区已满足；人工标签不能补造地点")
|
||
|
||
|
||
class AnalysisReviewService:
|
||
def __init__(self, analysis):
|
||
self.analysis, self.store = analysis, analysis.store
|
||
|
||
@staticmethod
|
||
def dto(row):
|
||
return {"id": row["id"], "created_at": row["created_at"], **json.loads(row["data_json"])}
|
||
|
||
def get(self, review_id, owner):
|
||
row = self.store.one("SELECT * FROM analysis_reviews WHERE id=$1 AND owner_id=$2", (review_id, owner))
|
||
if not row:
|
||
raise HTTPException(404, "人工核对记录不存在")
|
||
self.analysis.get(row["analysis_run_id"], owner)
|
||
return self.dto(row)
|
||
|
||
def list(self, analysis_id, owner):
|
||
self.analysis.get(analysis_id, owner)
|
||
return [self.dto(row) for row in self.store.rows(f"SELECT * FROM analysis_reviews WHERE analysis_run_id=$1 AND owner_id=$2 ORDER BY {self.store.order_column()} DESC", (analysis_id, owner))]
|
||
|
||
def summary(self, analysis_id, owner):
|
||
analysis = self.analysis.get(analysis_id, owner)
|
||
can_review = eligible(analysis)
|
||
axes = review_axes(analysis)
|
||
sources = analysis.get("sources", []) if can_review else []
|
||
latest = {}
|
||
for review in self.list(analysis_id, owner):
|
||
latest.setdefault(review["source_ref"], review)
|
||
current = [latest[source["source_ref"]] for source in sources if source["source_ref"] in latest]
|
||
counts = {decision: sum(review["decision"] == decision for review in current) for decision in DECISIONS}
|
||
judged = counts["confirmed"] + counts["corrected"]
|
||
return {"analysis_id": analysis_id, "eligible": can_review, "contract_version": review_contract(analysis), "metric_label": METRIC_LABEL,
|
||
"total": len(sources), "reviewed": len(current), "pending": len(sources) - len(current), "judged": judged, "decision_counts": counts,
|
||
"agreement_count": counts["confirmed"], "disagreement_count": counts["corrected"], "agreement_rate": counts["confirmed"] / judged if judged else None,
|
||
"axis_diff_counts": {axis: sum(bool(review.get("axis_diff", {}).get(axis)) for review in current if review["decision"] in {"confirmed", "corrected"}) for axis in axes},
|
||
"latest_reviews": current}
|
||
|
||
def create(self, analysis_id, body, owner):
|
||
analysis = self.analysis.get(analysis_id, owner)
|
||
if not eligible(analysis):
|
||
reject("REVIEW_ANALYSIS_INELIGIBLE", "仅已完成的多维需求分析支持人工核对；旧分析保持只读")
|
||
try:
|
||
body = body if isinstance(body, ReviewInput) else ReviewInput.model_validate(body)
|
||
except ValidationError:
|
||
reject("REVIEW_LABELS_INVALID", "人工核对输入不符合固定合同")
|
||
contextual = analysis.get("result_version") == RESULT_VERSION
|
||
axes = review_axes(analysis)
|
||
if not contextual and ("context_evidence" in body.model_fields_set or
|
||
(body.labels is not None and "consultation" in body.labels.model_fields_set)):
|
||
reject("REVIEW_LABELS_INVALID", "旧五维核对不接受咨询标签或笔记背景依据，请按原合同提交")
|
||
source = next((source for source in analysis.get("sources", []) if source.get("source_ref") == body.source_ref), None)
|
||
result = next((result for result in analysis.get("results", []) if result.get("source_ref") == body.source_ref), None)
|
||
if source is None or result is None:
|
||
reject("REVIEW_SOURCE_NOT_FOUND", "人工核对来源不属于当前分析", 404)
|
||
note = body.note.strip()
|
||
try:
|
||
safe_text(note)
|
||
except HTTPException:
|
||
reject("REVIEW_SENSITIVE_INPUT", "核对备注不应包含凭据、联系方式、链接或控制字符")
|
||
if not quotes_valid(body.evidence, source["text"]):
|
||
reject("REVIEW_EVIDENCE_INVALID", "核对依据须为本来源1至3条逐字原文，不能引用其他来源")
|
||
original = model_labels(result)
|
||
labels = body.labels.model_dump(exclude_none=True) if body.labels is not None else None
|
||
context_evidence = [entry.model_dump() for entry in body.context_evidence]
|
||
difference = {axis: False for axis in axes}
|
||
if body.decision in {"confirmed", "corrected"}:
|
||
if not source["text"].strip():
|
||
reject("REVIEW_MEDIA_UNREVIEWABLE", "无可用文字的媒体来源仅可标为不确定或无法核对")
|
||
if labels is None:
|
||
reject("REVIEW_LABELS_INVALID", "确认或纠错须提交当前版本的完整标签")
|
||
if contextual and "consultation" not in labels:
|
||
reject("REVIEW_LABELS_INVALID", "新版确认或纠错须提交独立咨询意向标签")
|
||
validate_labels(labels, result, source, (analysis.get("context") or {}).get("conditions") or {})
|
||
if contextual:
|
||
try:
|
||
validate_consultation({"value": labels["consultation"], "evidence": body.evidence, "context_evidence": context_evidence},
|
||
source, analysis.get("sources", []), labels["expressions"], labels["business_relevance"])
|
||
except ModelCallError:
|
||
reject("REVIEW_CONSULTATION_INVALID", "咨询意向须符合表达与相关性，并分别引用本来源和已关联笔记的逐字依据")
|
||
labels["expressions"] = sorted(labels["expressions"])
|
||
difference = {axis: labels[axis] != original[axis] for axis in axes}
|
||
if body.decision == "confirmed" and any(difference.values()):
|
||
reject("REVIEW_DECISION_INVALID", "确认须与原分析各维一致；改变标签请选择纠错")
|
||
if body.decision == "corrected" and (not any(difference.values()) or not note or not body.evidence):
|
||
reject("REVIEW_DECISION_INVALID", "纠错须至少改变一维，并填写备注和本来源逐字依据")
|
||
elif labels is not None or not note or context_evidence:
|
||
reject("REVIEW_DECISION_INVALID", "不确定或无法核对须清空判断标签并填写原因")
|
||
run_id, stamp = identifier("review"), now()
|
||
content = {"analysis_id": analysis_id, "source_ref": body.source_ref, "previous_review_id": body.previous_review_id, "decision": body.decision,
|
||
"labels": labels, "evidence": body.evidence, "note": note, "analysis_digest": digest(analysis), "source_digest": digest(source),
|
||
"contract_version": review_contract(analysis), "reviewer_id": owner, "model_labels": original, "axis_diff": difference}
|
||
if contextual:
|
||
content["context_evidence"] = context_evidence
|
||
try:
|
||
with self.store.connect(immediate=True) as connection:
|
||
latest = connection.execute(f"SELECT id FROM analysis_reviews WHERE analysis_run_id=$1 AND owner_id=$2 AND source_ref=$3 ORDER BY {self.store.order_column()} DESC LIMIT 1", (analysis_id, owner, body.source_ref)).fetchone()
|
||
latest_id = latest["id"] if latest else None
|
||
if body.previous_review_id != latest_id:
|
||
reject("REVIEW_STALE", "本来源已有更新的核对记录，请刷新后重新核对；没有自动重试", 409)
|
||
connection.execute("INSERT INTO analysis_reviews(id,owner_id,analysis_run_id,source_ref,previous_id,data_json,created_at) VALUES($1,$2,$3,$4,$5,$6,$7)",
|
||
(run_id, owner, analysis_id, body.source_ref, latest_id, encoded(content), stamp))
|
||
except (StoreConflict, sqlite3.IntegrityError):
|
||
reject("REVIEW_STALE", "核对版本已变化，请刷新后重新核对；没有自动重试", 409)
|
||
return self.get(run_id, owner)
|