"""阶段 2 智能受理案例导入、逐案评审和质量指标计算。""" import json from pathlib import Path from django.utils import timezone from apps.tickets.models import ( IntakeEvaluationCase, IntakeEvaluationResult, TicketDraft, ) SOURCE_KINDS = {"anonymized_real", "synthetic"} class IntakeEvaluationError(ValueError): pass def _normalize_value(value): if value is None: return "" if isinstance(value, str): return " ".join(value.casefold().split()) if isinstance(value, (dict, list)): return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) return str(value).casefold() def import_evaluation_cases(path: str | Path) -> tuple[int, int]: """导入版本化案例清单；真实案例必须提供完整期望与材料清单。""" source_path = Path(path) payload = json.loads(source_path.read_text(encoding="utf-8")) if not isinstance(payload, dict) or payload.get("schema_version") != 1: raise IntakeEvaluationError("案例文件 schema_version 必须为 1") cases = payload.get("cases") if isinstance(payload, dict) else None if not isinstance(cases, list): raise IntakeEvaluationError("案例文件必须包含 cases 数组") created_count = 0 updated_count = 0 for raw_case in cases: if not isinstance(raw_case, dict): raise IntakeEvaluationError("每个案例必须是对象") key = str(raw_case.get("key") or "").strip() source_kind = str(raw_case.get("source_kind") or "").strip() if not key or source_kind not in SOURCE_KINDS: raise IntakeEvaluationError("案例必须提供 key 和合法 source_kind") boolean_fields = { "valid_input", "expected_auto_created", "expected_similar_ticket", "enabled", } invalid_boolean_fields = sorted( field_name for field_name in boolean_fields if field_name in raw_case and not isinstance(raw_case[field_name], bool) ) if invalid_boolean_fields: raise IntakeEvaluationError( f"案例 {key} 的布尔字段类型错误：{', '.join(invalid_boolean_fields)}" ) input_manifest = raw_case.get("input_manifest") or {} expected_fields = raw_case.get("expected_fields") or {} expected_timeline = raw_case.get("expected_timeline") or [] if not isinstance(input_manifest, dict): raise IntakeEvaluationError(f"案例 {key} 的 input_manifest 必须是对象") if not isinstance(expected_fields, dict): raise IntakeEvaluationError(f"案例 {key} 的 expected_fields 必须是对象") if not isinstance(expected_timeline, list) or not all( isinstance(item, dict) for item in expected_timeline ): raise IntakeEvaluationError(f"案例 {key} 的 expected_timeline 必须是对象数组") if source_kind == "anonymized_real": required_keys = { "valid_input", "input_manifest", "expected_fields", "expected_timeline", "expected_auto_created", "expected_similar_ticket", } missing_keys = sorted(required_keys - set(raw_case)) if missing_keys: raise IntakeEvaluationError( f"真实案例 {key} 缺少字段：{', '.join(missing_keys)}" ) source_text = str(input_manifest.get("source_text") or "").strip() materials = input_manifest.get("materials") or [] if not isinstance(materials, list) or not all( isinstance(item, dict) for item in materials ): raise IntakeEvaluationError(f"真实案例 {key} 的 materials 必须是对象数组") if not source_text and not materials: raise IntakeEvaluationError(f"真实案例 {key} 必须包含文本或材料") _, created = IntakeEvaluationCase.objects.update_or_create( key=key, defaults={ "name": str(raw_case.get("name") or key), "source_kind": source_kind, "valid_input": bool(raw_case.get("valid_input", True)), "input_manifest": input_manifest, "expected_fields": expected_fields, "expected_timeline": expected_timeline, "expected_auto_created": bool(raw_case.get("expected_auto_created", True)), "expected_similar_ticket": bool( raw_case.get("expected_similar_ticket", False) ), "notes": str(raw_case.get("notes") or ""), "enabled": bool(raw_case.get("enabled", True)), }, ) created_count += int(created) updated_count += int(not created) return created_count, updated_count def _evaluate_timeline(expected_timeline: list, actual_timeline: list) -> tuple[list, int]: results = [] correct_count = 0 for index, expected in enumerate(expected_timeline): actual = actual_timeline[index] if index < len(actual_timeline) else None field_results = {} correct = isinstance(actual, dict) for field_name, expected_value in expected.items(): actual_value = actual.get(field_name) if isinstance(actual, dict) else None field_correct = _normalize_value(actual_value) == _normalize_value(expected_value) correct = correct and field_correct field_results[field_name] = { "expected": expected_value, "actual": actual_value, "correct": field_correct, } correct_count += int(correct) results.append( { "sequence": index + 1, "correct": correct, "field_results": field_results, } ) return results, correct_count def record_evaluation_result( case: IntakeEvaluationCase, draft: TicketDraft, *, reviewer=None, ) -> IntakeEvaluationResult: """把一次受理结果与案例期望逐字段比对，并保留人工评审证据。""" expected_fields = case.expected_fields if isinstance(case.expected_fields, dict) else {} actual_fields = draft.current_fields if isinstance(draft.current_fields, dict) else {} field_results = {} correct_count = 0 for field_name, expected in expected_fields.items(): actual = actual_fields.get(field_name) correct = _normalize_value(actual) == _normalize_value(expected) correct_count += int(correct) field_results[field_name] = { "expected": expected, "actual": actual, "correct": correct, } evaluated_count = len(field_results) field_accuracy = correct_count / evaluated_count if evaluated_count else None field_evidence = draft.field_evidence if isinstance(draft.field_evidence, dict) else {} evidence_results = {} evidence_count = 0 evidence_covered_count = 0 for field_name, expected in expected_fields.items(): if not _normalize_value(expected): continue evidence_count += 1 evidence = field_evidence.get(field_name) covered = isinstance(evidence, dict) and any( bool(evidence.get(key)) for key in ( "source_refs", "excerpt", "location", "generated_from", "raw_value", ) ) evidence_covered_count += int(covered) evidence_results[field_name] = { "covered": covered, "evidence": evidence if isinstance(evidence, dict) else {}, } expected_timeline = ( case.expected_timeline if isinstance(case.expected_timeline, list) else [] ) actual_timeline = ( draft.message_timeline if isinstance(draft.message_timeline, list) else [] ) timeline_results, correct_timeline_count = _evaluate_timeline( expected_timeline, actual_timeline, ) timeline_correct = correct_timeline_count == len(expected_timeline) actual_auto_created = draft.decision == "auto_created" and bool(draft.confirmed_ticket_id) actual_similar_ticket = bool( draft.confirmed_ticket_id and draft.confirmed_ticket.similarity_suggestions.exists() ) passed = ( draft.problem_valid is case.valid_input and actual_auto_created == case.expected_auto_created and actual_similar_ticket == case.expected_similar_ticket and correct_count == evaluated_count and timeline_correct and evidence_covered_count == evidence_count ) result, _ = IntakeEvaluationResult.objects.update_or_create( case=case, draft=draft, defaults={ "ticket": draft.confirmed_ticket, "actual_fields": actual_fields, "actual_auto_created": actual_auto_created, "correct_field_count": correct_count, "evaluated_field_count": evaluated_count, "field_accuracy": field_accuracy, "passed": passed, "details": { "field_results": field_results, "evidence_results": evidence_results, "evidence_covered_count": evidence_covered_count, "evidence_field_count": evidence_count, "timeline_results": timeline_results, "correct_timeline_count": correct_timeline_count, "evaluated_timeline_count": len(expected_timeline), "timeline_correct": timeline_correct, "expected_problem_valid": case.valid_input, "actual_problem_valid": draft.problem_valid, "expected_similar_ticket": case.expected_similar_ticket, "actual_similar_ticket": actual_similar_ticket, }, "reviewed_by": reviewer, }, ) return result def _latest_evaluation_scope(*, include_synthetic: bool) -> tuple[list, list]: cases = IntakeEvaluationCase.objects.filter(enabled=True) if not include_synthetic: cases = cases.filter(source_kind="anonymized_real") case_list = list(cases.order_by("key")) queryset = ( IntakeEvaluationResult.objects.select_related( "case", "draft", "ticket", "reviewed_by", ) .filter(case__in=case_list) .order_by("case_id", "-updated_at", "-id") ) latest_by_case = {} for result in queryset: latest_by_case.setdefault(result.case_id, result) return case_list, list(latest_by_case.values()) def _build_modality_coverage(cases: list[IntakeEvaluationCase]) -> dict: image_types = {"jpg", "jpeg", "png", "webp", "image"} document_types = { "pdf", "txt", "log", "docx", "xlsx", "pptx", "document", "text", "spreadsheet", "presentation", } coverage = { "text_case_count": 0, "image_case_count": 0, "multi_image_case_count": 0, "document_case_count": 0, "mixed_text_image_case_count": 0, "valid_input_case_count": 0, "invalid_input_case_count": 0, } for case in cases: manifest = case.input_manifest if isinstance(case.input_manifest, dict) else {} has_text = bool(str(manifest.get("source_text") or "").strip()) materials = manifest.get("materials") materials = materials if isinstance(materials, list) else [] material_types = [] for material in materials: if not isinstance(material, dict): continue raw_type = str( material.get("type") or material.get("extension") or Path(str(material.get("file_name") or "")).suffix.lstrip(".") ) material_types.append(raw_type.casefold()) image_count = sum(item in image_types for item in material_types) has_document = any(item in document_types for item in material_types) coverage["text_case_count"] += int(has_text) coverage["image_case_count"] += int(image_count > 0) coverage["multi_image_case_count"] += int(image_count > 1) coverage["document_case_count"] += int(has_document) coverage["mixed_text_image_case_count"] += int(has_text and image_count > 0) coverage["valid_input_case_count"] += int(case.valid_input) coverage["invalid_input_case_count"] += int(not case.valid_input) return coverage def build_intake_quality_metrics(*, include_synthetic: bool = False) -> dict: """按每个案例最新评审结果计算覆盖率、自动建单率和字段准确率。""" operational = TicketDraft.objects.filter(problem_valid=True).exclude( status__in={"rejected", "expired"} ) operational_valid_count = operational.count() operational_auto_count = operational.filter( decision="auto_created", confirmed_ticket__isnull=False, ).count() operational_auto_rate = ( operational_auto_count / operational_valid_count if operational_valid_count else None ) cases, results = _latest_evaluation_scope(include_synthetic=include_synthetic) valid_results = [result for result in results if result.case.valid_input] evaluated_valid_count = len(valid_results) evaluated_auto_count = sum(result.actual_auto_created for result in valid_results) evaluated_auto_rate = ( evaluated_auto_count / evaluated_valid_count if evaluated_valid_count else None ) correct_fields = sum(result.correct_field_count for result in results) evaluated_fields = sum(result.evaluated_field_count for result in results) field_accuracy = correct_fields / evaluated_fields if evaluated_fields else None evidence_covered_fields = 0 evidence_fields = 0 correct_timeline_items = 0 evaluated_timeline_items = 0 for result in results: details = result.details if not isinstance(details, dict): continue evidence_covered_fields += int(details.get("evidence_covered_count") or 0) evidence_fields += int(details.get("evidence_field_count") or 0) correct_timeline_items += int(details.get("correct_timeline_count") or 0) evaluated_timeline_items += int(details.get("evaluated_timeline_count") or 0) reviewed_case_count = len(results) enabled_case_count = len(cases) coverage_complete = ( enabled_case_count > 0 and reviewed_case_count == enabled_case_count ) return { "include_synthetic": include_synthetic, "operational": { "valid_input_count": operational_valid_count, "auto_created_count": operational_auto_count, "auto_creation_rate": operational_auto_rate, }, "evaluation": { "enabled_case_count": enabled_case_count, "reviewed_case_count": reviewed_case_count, "coverage_rate": ( reviewed_case_count / enabled_case_count if enabled_case_count else None ), "valid_input_count": evaluated_valid_count, "auto_created_count": evaluated_auto_count, "auto_creation_rate": evaluated_auto_rate, "correct_field_count": correct_fields, "evaluated_field_count": evaluated_fields, "field_accuracy": field_accuracy, "evidence_covered_field_count": evidence_covered_fields, "evidence_field_count": evidence_fields, "evidence_coverage_rate": ( evidence_covered_fields / evidence_fields if evidence_fields else None ), "correct_timeline_item_count": correct_timeline_items, "evaluated_timeline_item_count": evaluated_timeline_items, "timeline_accuracy": ( correct_timeline_items / evaluated_timeline_items if evaluated_timeline_items else None ), "result_selection": "latest_per_case", }, "targets": { "auto_creation_rate": 0.70, "field_accuracy": 0.90, "coverage_complete": coverage_complete, "auto_creation_rate_met": ( evaluated_auto_rate >= 0.70 if coverage_complete and evaluated_auto_rate is not None else None ), "field_accuracy_met": ( field_accuracy >= 0.90 if coverage_complete and field_accuracy is not None else None ), }, } def build_stage_2_acceptance_report(*, include_synthetic: bool = False) -> dict: """生成阶段 2 退出报告；默认只允许脱敏真实案例作为验收依据。""" cases, results = _latest_evaluation_scope(include_synthetic=include_synthetic) metrics = build_intake_quality_metrics(include_synthetic=include_synthetic) modality_coverage = _build_modality_coverage(cases) result_by_case = {result.case_id: result for result in results} case_results = [] for case in cases: result = result_by_case.get(case.id) details = result.details if result is not None and isinstance(result.details, dict) else {} case_results.append( { "key": case.key, "name": case.name, "source_kind": case.source_kind, "valid_input": case.valid_input, "reviewed": result is not None, "passed": result.passed if result is not None else None, "draft_public_id": str(result.draft.public_id) if result is not None else None, "ticket_id": result.ticket_id if result is not None else None, "field_accuracy": ( float(result.field_accuracy) if result is not None and result.field_accuracy is not None else None ), "evidence_coverage_rate": ( details.get("evidence_covered_count", 0) / details.get("evidence_field_count", 0) if details.get("evidence_field_count", 0) else None ), "timeline_accuracy": ( details.get("correct_timeline_count", 0) / details.get("evaluated_timeline_count", 0) if details.get("evaluated_timeline_count", 0) else None ), "reviewed_at": result.updated_at.isoformat() if result is not None else None, } ) reasons = [] readiness_state = "failed" if include_synthetic: readiness_state = "diagnostic" reasons.append( { "code": "synthetic_scope", "message": "合成案例只验证评估框架，不能作为阶段 2 业务退出证据。", } ) elif not cases: readiness_state = "not_ready" reasons.append( { "code": "missing_real_cases", "message": "尚未导入启用的脱敏真实案例。", } ) elif not metrics["targets"]["coverage_complete"]: readiness_state = "not_ready" reasons.append( { "code": "incomplete_review_coverage", "message": "仍有启用案例没有评审结果。", } ) if not include_synthetic and cases: required_modalities = { "image_case_count": "真实案例集未覆盖图片输入。", "mixed_text_image_case_count": "真实案例集未覆盖文本加图片混合输入。", "document_case_count": "真实案例集未覆盖文档或日志输入。", "invalid_input_case_count": "真实案例集未覆盖无效输入与人工追问路径。", } for key, message in required_modalities.items(): if modality_coverage[key] == 0: if readiness_state == "failed": readiness_state = "not_ready" reasons.append({"code": f"missing_{key}", "message": message}) if metrics["targets"]["coverage_complete"] and not include_synthetic: if metrics["targets"]["auto_creation_rate_met"] is not True: reasons.append( { "code": "auto_creation_rate_below_target", "message": "有效输入自动建单率未达到 70%。", } ) if metrics["targets"]["field_accuracy_met"] is not True: reasons.append( { "code": "field_accuracy_below_target", "message": "自动建单字段准确率未达到 90%。", } ) evidence_count = metrics["evaluation"]["evidence_field_count"] evidence_rate = metrics["evaluation"]["evidence_coverage_rate"] if not evidence_count or evidence_rate != 1.0: reasons.append( { "code": "incomplete_field_evidence", "message": "非空评估字段尚未全部具备可追溯证据。", } ) if metrics["evaluation"]["evaluated_timeline_item_count"] == 0: reasons.append( { "code": "missing_timeline_baseline", "message": "真实案例集没有可比对的期望消息时间线。", } ) exit_ready = not reasons and not include_synthetic if exit_ready: readiness_state = "passed" elif readiness_state == "failed" and any( reason["code"] in { "missing_real_cases", "incomplete_review_coverage", "missing_image_case_count", "missing_mixed_text_image_case_count", "missing_document_case_count", "missing_invalid_input_case_count", } for reason in reasons ): readiness_state = "not_ready" return { "schema_version": 1, "generated_at": timezone.now().isoformat(), "scope": "diagnostic_with_synthetic" if include_synthetic else "anonymized_real", "status": readiness_state, "exit_ready": exit_ready, "reasons": reasons, "modality_coverage": modality_coverage, "metrics": metrics, "case_results": case_results, }