"""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)