#!/usr/bin/env python3 """REPAIR005: close the remaining B1 company-self / target-bucket role defects. This replay deliberately reuses the 729 CNINFO PDFs and the 858 page-search receipts accepted by the independent REPAIR004 review. It does not download, expand, or replace research sources. The only substantive change is a stricter semantic role adjudication of the 50 page-search positives, followed by a full 858-row state replay, non-eligible field cleanup, mechanical reranking, and deterministic downstream rebuild. """ from __future__ import annotations import argparse import json import sys from collections import Counter, defaultdict from pathlib import Path from typing import Any import newenergy_batch001_repair as base import newenergy_batch001_repair004 as r4 TOOL_VERSION = "REPAIR-005" ROLE_RULE_VERSION = "REPAIR005_COMPANY_SELF_TARGET_BUCKET_ROLE_GATE_V2_HUMAN_ADJUDICATED_DIRECT_SET" REVIEW_STATUS = "DRAFT_FOR_REVIEW" TASK_ID = r4.TASK_ID CASE_ID = r4.CASE_ID BATCH_ID = r4.BATCH_ID RUN_ID = r4.RUN_ID # The independent reviewer expressly allowed necessary human reading in # REPAIR005. Every one of the 50 prior regex-positive rows is therefore frozen # here with an explicit semantic outcome. Accepted rows prove a direct business # of the listed company or a controlled subsidiary in the target bucket. The # remaining 808 rows are replayed from the already accepted full-page negative # or context-only search receipt. APPROVED_ROLE_DECISIONS: dict[str, str] = { "QUAL-00019": "CONTROLLED_SUBSIDIARY_TARGET_PRODUCT_MANUFACTURE", "QUAL-00061": "DIRECT_TARGET_PRODUCT_RESEARCH_PRODUCTION_AND_SALE", "QUAL-00072": "DIRECT_TARGET_PRODUCT_RESEARCH_PRODUCTION_AND_SALE", "QUAL-00103": "CONTROLLED_SUBSIDIARY_TARGET_PRODUCT_MANUFACTURE", "QUAL-00128": "DIRECT_TARGET_PRODUCT_PRODUCTION_AND_SALE", "QUAL-00129": "DIRECT_TARGET_PRODUCT_BUSINESS", "QUAL-00136": "DIRECT_TARGET_PRODUCT_PRODUCTION_AND_SALE", "QUAL-00141": "DIRECT_TARGET_PRODUCT_SALE", "QUAL-00148": "DIRECT_TARGET_PRODUCT_PRODUCTION", "QUAL-00149": "DIRECT_TARGET_RESOURCE_DEVELOPMENT_PRODUCTION_AND_SALE", "QUAL-00213": "DIRECT_TARGET_PRODUCT_RESEARCH_PRODUCTION_AND_SALE", "QUAL-00226": "DIRECT_TARGET_PRODUCT_RESEARCH_PRODUCTION_AND_SALE", "QUAL-00227": "DIRECT_TARGET_PRODUCT_RESEARCH_PRODUCTION_AND_SALE", "QUAL-00348": "DIRECT_TARGET_CELL_RESEARCH_PRODUCTION_AND_SALE", "QUAL-01029": "DIRECT_TARGET_SYSTEM_PRODUCT", "QUAL-01137": "DIRECT_TARGET_SYSTEM_PRODUCT", "QUAL-01625": "DIRECT_TARGET_WAFER_PRODUCTION", "QUAL-01639": "DIRECT_TARGET_SILICON_INGOT_AND_WAFER_PRODUCTION_AND_SALE", "QUAL-01732": "DIRECT_TARGET_WAFER_PRODUCT_BUSINESS", "QUAL-01846": "CONTROLLED_SUBSIDIARY_TARGET_COMPONENT_MANUFACTURE", "QUAL-01904": "DIRECT_TARGET_SPECIALTY_COMPONENT_PRODUCTION_AND_SALE", "QUAL-02192": "DIRECT_TARGET_COMPONENT_PRODUCT_SALE", "QUAL-02265": "DIRECT_TARGET_COMPONENT_PRODUCT_SALE", "QUAL-02299": "CONTROLLED_SUBSIDIARY_TARGET_COMPONENT_PRODUCTION_AND_SALE", "QUAL-02364": "CONTROLLED_SUBSIDIARY_TARGET_COMPONENT_RESEARCH_PRODUCTION_AND_SALE", "QUAL-02648": "DIRECT_TARGET_INVERTER_PRODUCT", "QUAL-02692": "CONTROLLED_SUBSIDIARY_TARGET_STATION_INVESTMENT_AND_OPERATION", "QUAL-02760": "CONTROLLED_SUBSIDIARY_TARGET_STATION_INVESTMENT_AND_OPERATION", "QUAL-02761": "DIRECT_TARGET_STATION_PROJECT_CONSTRUCTION_AND_OPERATION", "QUAL-02875": "CONTROLLED_SUBSIDIARY_TARGET_STATION_CONSTRUCTION_AND_OPERATION", "QUAL-02888": "DIRECT_TARGET_STATION_CONSTRUCTION_AND_OPERATION_WITH_REVENUE", "QUAL-02900": "CONTROLLED_SUBSIDIARY_TARGET_STATION_OPERATION", "QUAL-02956": "DIRECT_TARGET_STATION_EPC_AND_OPERATION_SERVICE", "QUAL-02982": "DIRECT_TARGET_STATION_CONSTRUCTION_OPERATION_AND_EXTERNAL_EXPANSION", "QUAL-02991": "DIRECT_TARGET_STATION_INVESTMENT_AND_OPERATION", "QUAL-03042": "DIRECT_TARGET_STATION_CONSTRUCTION_AND_OPERATION_WITH_REVENUE", "QUAL-03055": "DIRECT_TARGET_STATION_INVESTMENT_AND_OPERATION_WITH_REVENUE", "QUAL-03067": "DIRECT_TARGET_STATION_EPC_SERVICE", "QUAL-03199": "DIRECT_TARGET_WIND_COMPONENT_MANUFACTURE", "QUAL-03750": "DIRECT_TARGET_WIND_PROJECT_CONSTRUCTION_AND_OPERATION", } REJECTED_ROLE_DECISIONS: dict[str, str] = { "QUAL-00063": "ADJACENT_EQUIPMENT_AND_EPC_SERVICE_NOT_BATTERY_RESOURCE_OR_MAIN_MATERIAL", "QUAL-00223": "ADJACENT_INDUSTRIAL_GAS_SUPPLY_TO_BATTERY_MATERIAL_PROJECT", "QUAL-01640": "CUTTING_WIRE_AUXILIARY_NOT_SILICON_OR_WAFER_MANUFACTURE", "QUAL-01889": "COMPONENT_BACKSHEET_FILM_AND_EPOXY_AUXILIARY_NOT_CELL_OR_COMPONENT_MANUFACTURE", "QUAL-01901": "PHOTOVOLTAIC_GLASS_AUXILIARY_OR_POWER_GENERATION_NOT_CELL_OR_COMPONENT_MANUFACTURE", "QUAL-01962": "COMPONENT_ENCAPSULATION_MATERIAL_NOT_CELL_OR_COMPONENT_MANUFACTURE", "QUAL-02159": "COMPANY_PROCUREMENT_OF_COMPONENTS_AND_INVERTERS_NOT_TARGET_PRODUCT_BUSINESS", "QUAL-02499": "COMPONENTS_LISTED_AS_CONSUMED_RAW_MATERIAL_NOT_TARGET_PRODUCT_BUSINESS", "QUAL-02828": "INTERNAL_ROOFTOP_AND_STATION_SUPPORT_SERVICE_ONLY_WITHOUT_EXTERNAL_TARGET_BUSINESS", "QUAL-02852": "SELF_USE_TEXTILE_PROJECT_ONLY_WITHOUT_STATION_BUSINESS", } ROLE_HEADERS = [ "role_adjudication_id", "qualification_row_id", "task_id", "case_id", "batch_id", "run_id", "company_id", "security_code", "security_name", "track_code", "selection_bucket", "prior_page_search_result", "prior_qualification_result", "prior_keyword_hit_count", "prior_direct_context_hit_count", "full_page_search_reused", "human_semantic_review", "company_self_gate", "target_bucket_role_gate", "adjacent_or_self_use_exclusion_gate", "final_role_decision", "decision_basis_code", "selected_attachment_id", "selected_page", "selected_exact_sentence", "rule_version", "adjudicated_at", "tool_version", "review_status", "receipt_sha256", ] PRESTATE_HEADERS = [ "snapshot_id", "blocker_id", "record_type", "company_id", "security_code", "track_code", "state", "subject_type", "direct_business_source_id", "direct_business_locator", "evidence_grade", "exposure_specificity", "evidence_fact_id", "chain_node_id", "classification_reason", "snapshot_status", ] def role_receipt_path(industry_root: Path) -> Path: return industry_root / "extracted" / "candidate_role_adjudication_receipt.csv" def prestate_path(industry_root: Path) -> Path: return industry_root / "manifest" / "repair005_pollution_prestate_snapshot.csv" def validation_path(industry_root: Path) -> Path: return industry_root / "manifest" / "repair005_validation_receipt.json" def stable_adjudicated_at(industry_root: Path) -> str: path = role_receipt_path(industry_root) if path.exists(): rows = r4.read_csv(path) values = sorted({row.get("adjudicated_at", "") for row in rows if row.get("adjudicated_at")}) if len(values) == 1: return values[0] return r4.now_iso() def build_role_adjudication_receipt( industry_root: Path, pair_rows: list[dict[str, str]], candidates: list[dict[str, str]], ) -> list[dict[str, str]]: direct_ids = {row["qualification_row_id"] for row in pair_rows if int(row["direct_context_hit_count"]) > 0} frozen_decisions = set(APPROVED_ROLE_DECISIONS) | set(REJECTED_ROLE_DECISIONS) if len(pair_rows) != 858 or len({row["qualification_row_id"] for row in pair_rows}) != 858: raise RuntimeError("REPAIR005 requires the frozen 858 unique pair receipts") if len(direct_ids) != 50 or direct_ids != frozen_decisions: raise RuntimeError( f"the frozen 50 direct-context rows changed: direct={len(direct_ids)} " f"missing={sorted(direct_ids - frozen_decisions)} extra={sorted(frozen_decisions - direct_ids)}" ) if set(APPROVED_ROLE_DECISIONS) & set(REJECTED_ROLE_DECISIONS): raise RuntimeError("approved and rejected semantic decision sets overlap") candidate_by_pair = {(row["company_id"], row["track_code"]): row for row in candidates} adjudicated_at = stable_adjudicated_at(industry_root) results: list[dict[str, str]] = [] for source in sorted(pair_rows, key=lambda row: row["qualification_row_id"]): qid = source["qualification_row_id"] candidate = candidate_by_pair[(source["company_id"], source["track_code"])] attachment_id = "" page = "" sentence = "" if qid in APPROVED_ROLE_DECISIONS: attachment_id, page_number, sentence = r4.parse_direct_context(source) page = str(page_number) company_gate = "PASS" target_gate = "PASS" exclusion_gate = "PASS_NO_EXCLUSION_TRIGGERED" decision = "ELIGIBLE_DIRECT_COMPANY_SELF_TARGET_BUCKET_ROLE" basis = APPROVED_ROLE_DECISIONS[qid] human_review = "YES_ALL_PRIOR_DIRECT_CONTEXTS_REVIEWED" elif qid in REJECTED_ROLE_DECISIONS: attachment_id, page_number, sentence = r4.parse_direct_context(source) page = str(page_number) company_gate = "FAIL_OR_NOT_SAME_TARGET_ROLE" target_gate = "FAIL" exclusion_gate = "FAIL_EXCLUSION_TRIGGERED" decision = "HELD_ROLE_MISMATCH_ADJACENT_PROCUREMENT_OR_SELF_USE" basis = REJECTED_ROLE_DECISIONS[qid] human_review = "YES_ALL_PRIOR_DIRECT_CONTEXTS_REVIEWED" else: company_gate = "FAIL_NO_PRIOR_DIRECT_SELF_CONTEXT" target_gate = "FAIL_NO_PRIOR_DIRECT_TARGET_ROLE_CONTEXT" exclusion_gate = "NOT_APPLICABLE_NO_DIRECT_CONTEXT" if source["page_search_result"] == "ALL_REFERENCED_READABLE_PDFS_SEARCHED_NO_FROZEN_KEYWORD_HIT": decision = "HELD_TRUE_NEGATIVE_AFTER_FULL_PAGE_SEARCH" basis = "NO_FROZEN_KEYWORD_HIT_IN_REFERENCED_READABLE_PDFS" elif source["page_search_result"] == "FROZEN_KEYWORD_FOUND_BUT_NO_DIRECT_SELF_BUSINESS_PAGE_CONTEXT": decision = "HELD_KEYWORD_CONTEXT_ONLY_AFTER_FULL_PAGE_SEARCH" basis = "KEYWORD_CONTEXT_DID_NOT_PASS_COMPANY_SELF_AND_TARGET_ROLE_GATE" else: decision = "HELD_RETRIEVAL_OR_PARSE_GAP" basis = "REPAIR004_REVIEWABLE_RETRIEVAL_OR_PARSE_FAILURE" human_review = "NOT_REQUIRED_PRIOR_FULL_PAGE_GATE_FOUND_NO_DIRECT_CONTEXT" core: dict[str, str] = { "role_adjudication_id": f"ROLE-R005-{qid}", "qualification_row_id": qid, "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "company_id": source["company_id"], "security_code": source["security_code"], "security_name": candidate["security_name"], "track_code": source["track_code"], "selection_bucket": source["selection_bucket"], "prior_page_search_result": source["page_search_result"], "prior_qualification_result": source["qualification_result"], "prior_keyword_hit_count": source["keyword_hit_count"], "prior_direct_context_hit_count": source["direct_context_hit_count"], "full_page_search_reused": "YES_REPAIR004_ACCEPTED_729_PDFS_858_PAIR_SEARCH", "human_semantic_review": human_review, "company_self_gate": company_gate, "target_bucket_role_gate": target_gate, "adjacent_or_self_use_exclusion_gate": exclusion_gate, "final_role_decision": decision, "decision_basis_code": basis, "selected_attachment_id": attachment_id, "selected_page": page, "selected_exact_sentence": sentence, "rule_version": ROLE_RULE_VERSION, "adjudicated_at": adjudicated_at, "tool_version": TOOL_VERSION, "review_status": REVIEW_STATUS, } payload = json.dumps(core, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") core["receipt_sha256"] = r4.sha256_bytes(payload) results.append({key: core.get(key, "") for key in ROLE_HEADERS}) r4.write_csv(role_receipt_path(industry_root), ROLE_HEADERS, results) return results def write_prestate_snapshot_once( industry_root: Path, candidates: list[dict[str, str]], classifications: list[dict[str, str]], ) -> list[dict[str, str]]: path = prestate_path(industry_root) if path.exists(): prior = r4.read_csv(path) rows = [ row for row in prior if ( row["record_type"] == "CANDIDATE_LEDGER" and bool(row["direct_business_source_id"]) and row["evidence_grade"] in {"S", "A"} ) or ( row["record_type"] == "CLASSIFICATION_SUMMARY" and ( row["subject_type"] == "COMPANY_DIRECT_BUSINESS" or bool(row["evidence_fact_id"]) ) ) ] if len(rows) != len(prior): r4.write_csv(path, PRESTATE_HEADERS, rows) return rows rows: list[dict[str, str]] = [] for candidate in candidates: held = candidate["candidate_state"] == "HELD_BY_EVIDENCE_GAP" polluted = ( held and bool(candidate.get("direct_business_source_id", "")) and candidate.get("evidence_grade", "") in {"S", "A"} ) if not polluted: continue rows.append({ "snapshot_id": f"PRE-R005-LEDGER-{candidate['track_code']}-{candidate['security_code']}", "blocker_id": "B1-R004-B", "record_type": "CANDIDATE_LEDGER", "company_id": candidate["company_id"], "security_code": candidate["security_code"], "track_code": candidate["track_code"], "state": candidate["candidate_state"], "subject_type": "", "direct_business_source_id": candidate["direct_business_source_id"], "direct_business_locator": candidate["direct_business_locator"], "evidence_grade": candidate["evidence_grade"], "exposure_specificity": candidate["exposure_specificity"], "evidence_fact_id": "", "chain_node_id": candidate["chain_nodes"], "classification_reason": candidate["include_or_exclude_reason"], "snapshot_status": "FROZEN_PRE_REPAIR005_POLLUTION_EVIDENCE", }) for row in classifications: held = row["data_status"] == "HELD_BY_EVIDENCE_GAP" polluted = held and ( row.get("subject_type", "") == "COMPANY_DIRECT_BUSINESS" or bool(row.get("evidence_fact_id", "")) ) if not polluted: continue rows.append({ "snapshot_id": f"PRE-R005-CLASS-{row['track_code']}-{row['company_id']}", "blocker_id": "B1-R004-B", "record_type": "CLASSIFICATION_SUMMARY", "company_id": row["company_id"], "security_code": row["company_id"].split(":")[-1], "track_code": row["track_code"], "state": row["data_status"], "subject_type": row["subject_type"], "direct_business_source_id": row["source_doc_id"], "direct_business_locator": "", "evidence_grade": "", "exposure_specificity": "", "evidence_fact_id": row["evidence_fact_id"], "chain_node_id": row["chain_node_id"], "classification_reason": row["classification_reason"], "snapshot_status": "FROZEN_PRE_REPAIR005_POLLUTION_EVIDENCE", }) if not rows: raise RuntimeError("expected REPAIR004 polluted prestate was not found before REPAIR005") r4.write_csv(path, PRESTATE_HEADERS, rows) return rows def clear_noneligible_candidate(candidate: dict[str, str], reason: str) -> None: candidate["chain_nodes"] = "" candidate["direct_business_source_id"] = "" candidate["direct_business_locator"] = "" candidate["evidence_grade"] = "" candidate["exposure_specificity"] = "" candidate["latest_disclosed_period"] = "" candidate["candidate_state"] = "HELD_BY_EVIDENCE_GAP" candidate["selection_rank"] = "" candidate["tier"] = "" candidate["tie_break_rule"] = "" candidate["include_or_exclude_reason"] = reason def build_evidence_and_candidates( industry_root: Path, role_rows: list[dict[str, str]], source_rows: list[dict[str, str]], ) -> tuple[list[dict[str, str]], list[dict[str, str]]]: evidence_path = industry_root / "evidence" / "evidence_fact_table.csv" prior_evidence = r4.read_csv(evidence_path) evidence_headers = list(prior_evidence[0]) evidence_rows = [row for row in prior_evidence if not row["evidence_fact_id"].startswith("EVF-QUAL-")] candidate_path = industry_root / "extracted" / "company_track_candidate_ledger.csv" candidates = r4.read_csv(candidate_path) source_by_id = {row["doc_id"]: row for row in source_rows} candidate_by_pair = {(row["company_id"], row["track_code"]): row for row in candidates} role_by_pair = {(row["company_id"], row["track_code"]): row for row in role_rows} for role in role_rows: if role["final_role_decision"] != "ELIGIBLE_DIRECT_COMPANY_SELF_TARGET_BUCKET_ROLE": continue pair = (role["company_id"], role["track_code"]) candidate = candidate_by_pair[pair] attachment_id = role["selected_attachment_id"] page = role["selected_page"] doc_id = f"S-QUAL-AR-{attachment_id}" source = source_by_id[doc_id] evidence_id = f"EVF-QUAL-{role['track_code']}-{role['security_code']}-R005" evidence = { "evidence_fact_id": evidence_id, "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "doc_id": doc_id, "industry_case": "新能源案例", "industry_id": "IND-NEWENERGY", "subindustry_id": role["track_code"], "company_id": role["company_id"], "track_code": role["track_code"], "chain_node_id": r4.CHAIN_NODES[(role["track_code"], role["selection_bucket"])], "subject_type": "COMPANY", "subject_id": role["company_id"], "source_text_path": source["converted_text_path"], "raw_pool_path": source["raw_pool_path"], "raw_file_sha256": source["file_sha256"], "source_page": page, "source_table_id": "", "source_sentence_index": "", "locator_type": "PDF_PAGE_AND_EXACT_CONTEXT", "locator_value": f"attachment_id={attachment_id};page={page};receipt={role['qualification_row_id']};role_receipt={role['role_adjudication_id']}", "evidence_text": role["selected_exact_sentence"], "evidence_type": "OFFICIAL_ANNUAL_REPORT", "statement_type": "FACT", "business_dimension": "COMPANY_EXPOSURE", "research_dimension": "COMPANY_TRACK_QUALIFICATION", "numeric_value_raw": "", "metric_candidate_name": "", "metric_candidate_unit": "", "metric_period": "2025", "metric_date": "2025-12-31", "geography": "CN", "original_qualifier": ( "REPAIR005复用已审核的CNINFO年度报告与全页检索回执，并对全部50条先前direct-context做公司自身/" "受控子公司与目标桶角色人读裁决；仅支持直接业务事实，不作份额、质量或投资判断。" ), "related_company_id": "", "viewpoint_id": "", "darkline_signal_flag": "NO", "confidence_level": "HIGH", "conclusion_strength": "DIRECT_FACT", "sensitivity_screen": "LEGAL_PUBLIC_SCREENED_HIGH_LEVEL_ONLY", "contradicts_evidence_fact_id": "", "normalization_status": "NORMALIZED", "processing_status": "READY", "data_status": "VERIFIED_PUBLIC", "review_status": REVIEW_STATUS, } evidence_rows.append({key: evidence.get(key, "") for key in evidence_headers}) candidate.update({ "chain_nodes": r4.CHAIN_NODES[(role["track_code"], role["selection_bucket"])], "direct_business_source_id": doc_id, "direct_business_locator": f"2025年报第{page}页;attachment_id={attachment_id};role_receipt={role['role_adjudication_id']}", "evidence_grade": "S", "exposure_specificity": "NAMED_PRODUCT_PROJECT_AND_OPERATING_FACT", "latest_disclosed_period": "2025-12-31", }) evidence_by_pair: dict[tuple[str, str], list[dict[str, str]]] = defaultdict(list) for evidence in evidence_rows: if evidence["subject_type"] == "COMPANY" and evidence["company_id"] and evidence["track_code"]: evidence_by_pair[(evidence["company_id"], evidence["track_code"])].append(evidence) eligible_groups: dict[tuple[str, str], list[dict[str, str]]] = defaultdict(list) valid_specificity = { "SEGMENT_REVENUE_OR_ASSET_DISCLOSED", "NAMED_PRODUCT_PROJECT_AND_OPERATING_FACT", "GENERAL_DIRECT_BUSINESS_DESCRIPTION", } for candidate in candidates: pair = (candidate["company_id"], candidate["track_code"]) matching = [ evidence for evidence in evidence_by_pair.get(pair, []) if candidate["direct_business_source_id"] and evidence["doc_id"] == candidate["direct_business_source_id"] ] eligible = bool(matching) and candidate["evidence_grade"] in {"S", "A"} and candidate["exposure_specificity"] in valid_specificity if eligible: candidate["candidate_state"] = "ELIGIBLE" candidate["selection_rank"] = "" candidate["tier"] = "" candidate["tie_break_rule"] = "source_grade>exposure_specificity>latest_period>publish_date>exchange_code>security_code" if pair in role_by_pair: candidate["include_or_exclude_reason"] = ( "REPAIR005复用年度报告PDF与全页检索，对先前direct-context执行公司自身/受控子公司和目标桶角色语义裁决；" "S级主源、页码原句与全部资格gate通过，进入同桶ELIGIBLE机械排序。" ) else: candidate["include_or_exclude_reason"] = ( "设计前序已登记直接业务主源与证据事实在REPAIR005重放中保持有效；全部资格gate通过，进入同桶ELIGIBLE机械排序。" ) eligible_groups[(candidate["track_code"], candidate["selection_bucket"])].append(candidate) else: role = role_by_pair.get(pair) if role: reason = ( f"REPAIR005全量重放结论={role['final_role_decision']};" f"依据={role['decision_basis_code']}；未通过公司自身及目标桶直接业务证据闸门，保持HELD。" ) else: reason = "REPAIR005重放未找到可与候选直接业务字段闭合的S/A级公司证据，清除全部资格派生字段并保持HELD。" clear_noneligible_candidate(candidate, reason) grade_order = {"S": 0, "A": 1} specificity_order = { "SEGMENT_REVENUE_OR_ASSET_DISCLOSED": 0, "NAMED_PRODUCT_PROJECT_AND_OPERATING_FACT": 1, "GENERAL_DIRECT_BUSINESS_DESCRIPTION": 2, } for group_rows in eligible_groups.values(): group_rows.sort(key=lambda row: ( grade_order.get(row["evidence_grade"], 9), specificity_order.get(row["exposure_specificity"], 9), r4.descending_date(row["latest_disclosed_period"]), r4.descending_date(source_by_id[row["direct_business_source_id"]]["publish_date"]), row["exchange_code"], row["security_code"], )) for rank, candidate in enumerate(group_rows, 1): candidate["selection_rank"] = str(rank) if rank == 1: candidate["candidate_state"] = "INCLUDED_T1" candidate["tier"] = "T1_PRIMARY" elif rank == 2: candidate["candidate_state"] = "INCLUDED_T2" candidate["tier"] = "T2_CANDIDATE" else: candidate["candidate_state"] = "ELIGIBLE_NOT_SELECTED_BATCH001" candidate["tier"] = "" candidate["include_or_exclude_reason"] += " 同桶排名超过2，保留为ELIGIBLE_NOT_SELECTED_BATCH001。" track_order = {"BATTERY": 0, "SOLAR": 1, "WIND": 2, "NUCLEAR": 3} bucket_order = {key: index for index, key in enumerate(r4.TRACK_KEYWORDS)} state_order = { "INCLUDED_T1": 0, "INCLUDED_T2": 1, "ELIGIBLE_NOT_SELECTED_BATCH001": 2, "HELD_BY_EVIDENCE_GAP": 3, } candidates.sort(key=lambda row: ( track_order[row["track_code"]], bucket_order[(row["track_code"], row["selection_bucket"])], state_order.get(row["candidate_state"], 9), int(row["selection_rank"] or 999999), row["exchange_code"], row["security_code"], )) r4.write_csv(evidence_path, evidence_headers, evidence_rows) r4.write_csv(candidate_path, list(candidates[0]), candidates) return evidence_rows, candidates def update_funnel( industry_root: Path, role_rows: list[dict[str, str]], candidates: list[dict[str, str]], evidence_rows: list[dict[str, str]], source_rows: list[dict[str, str]], ) -> list[dict[str, str]]: path = industry_root / "extracted" / "candidate_qualification_funnel.csv" funnel = r4.read_csv(path) role_by_pair = {(row["company_id"], row["track_code"]): row for row in role_rows} candidate_by_pair = {(row["company_id"], row["track_code"]): row for row in candidates} source_by_id = {row["doc_id"]: row for row in source_rows} evidence_by_pair: dict[tuple[str, str], list[dict[str, str]]] = defaultdict(list) for evidence in evidence_rows: if evidence["subject_type"] == "COMPANY": evidence_by_pair[(evidence["company_id"], evidence["track_code"])].append(evidence) eligible_states = {"INCLUDED_T1", "INCLUDED_T2", "ELIGIBLE_NOT_SELECTED_BATCH001"} for row in funnel: pair = (row["company_id"], row["track_code"]) candidate = candidate_by_pair[pair] role = role_by_pair.get(pair) eligible = candidate["candidate_state"] in eligible_states matching = [ evidence for evidence in evidence_by_pair.get(pair, []) if candidate["direct_business_source_id"] and evidence["doc_id"] == candidate["direct_business_source_id"] ] if role: row["page_level_verification_attempt"] = "REUSE_REPAIR004_FULL_PAGE_SEARCH_PLUS_REPAIR005_COMPANY_SELF_TARGET_BUCKET_ROLE_ADJUDICATION" row["page_level_verification_result"] = role["final_role_decision"] row["business_context_rule_result"] = ( "REPAIR005_COMPANY_SELF_TARGET_BUCKET_ROLE_CONFIRMED" if role["final_role_decision"] == "ELIGIBLE_DIRECT_COMPANY_SELF_TARGET_BUCKET_ROLE" else "REPAIR005_HELD_BY_ROLE_CONTEXT_OR_SEARCH_RESULT" ) row["replay_status"] = "REPAIR005_FULL_858_ROLE_STATE_REPLAY_COMPLETED" row["direct_business_source_id"] = candidate["direct_business_source_id"] if eligible else "" for gate in ["direct_source_gate", "locator_gate", "company_evidence_fact_gate", "source_grade_gate", "exposure_specificity_gate"]: row[gate] = "PASS" if eligible else "FAIL" row["evidence_fact_ids"] = ";".join(sorted(e["evidence_fact_id"] for e in matching)) if eligible else "" row["failed_gates"] = "" if eligible else "DIRECT_BUSINESS_SOURCE;PAGE_OR_TEXT_LOCATOR;COMPANY_EVIDENCE_FACT;SOURCE_GRADE_S_OR_A;EXPOSURE_SPECIFICITY" row["eligibility_result"] = "ELIGIBLE" if eligible else "HELD_BY_EVIDENCE_GAP" row["eligible_rank_in_bucket"] = candidate["selection_rank"] if eligible else "" row["final_candidate_state"] = candidate["candidate_state"] if eligible: source = source_by_id[candidate["direct_business_source_id"]] row["mechanical_sort_key"] = "|".join([ candidate["evidence_grade"], candidate["exposure_specificity"], candidate["latest_disclosed_period"], source["publish_date"], candidate["exchange_code"], candidate["security_code"], ]) else: row["mechanical_sort_key"] = "" row["funnel_rule_version"] = ROLE_RULE_VERSION row["review_status"] = REVIEW_STATUS r4.write_csv(path, list(funnel[0]), funnel) return funnel def update_classification_and_matrix( industry_root: Path, candidates: list[dict[str, str]], evidence_rows: list[dict[str, str]], source_rows: list[dict[str, str]], ) -> tuple[list[dict[str, str]], list[dict[str, str]]]: candidate_by_pair = {(row["company_id"], row["track_code"]): row for row in candidates} source_by_id = {row["doc_id"]: row for row in source_rows} evidence_by_pair: dict[tuple[str, str], list[dict[str, str]]] = defaultdict(list) for evidence in evidence_rows: if evidence["subject_type"] == "COMPANY": evidence_by_pair[(evidence["company_id"], evidence["track_code"])].append(evidence) classification_path = industry_root / "extracted" / "classification_summary.csv" classifications = r4.read_csv(classification_path) eligible_states = {"INCLUDED_T1", "INCLUDED_T2", "ELIGIBLE_NOT_SELECTED_BATCH001"} for row in classifications: pair = (row["company_id"], row["track_code"]) candidate = candidate_by_pair[pair] eligible = candidate["candidate_state"] in eligible_states matching = [ evidence for evidence in evidence_by_pair.get(pair, []) if candidate["direct_business_source_id"] and evidence["doc_id"] == candidate["direct_business_source_id"] ] row["classification_reason"] = candidate["include_or_exclude_reason"] row["data_status"] = candidate["candidate_state"] row["review_status"] = REVIEW_STATUS if eligible and matching: evidence = matching[0] source = source_by_id[evidence["doc_id"]] row["subject_type"] = "COMPANY_DIRECT_BUSINESS" row["source_doc_id"] = evidence["doc_id"] row["evidence_fact_id"] = evidence["evidence_fact_id"] row["chain_node_id"] = candidate["chain_nodes"] row["raw_pool_path"] = source["raw_pool_path"] row["raw_file_sha256"] = source["file_sha256"] else: discovery_source = source_by_id.get(candidate["candidate_source_id"], {}) row["subject_type"] = "COMPANY_QUERY_HIT" row["source_doc_id"] = discovery_source.get("doc_id", "") row["evidence_fact_id"] = "" row["chain_node_id"] = "" row["raw_pool_path"] = discovery_source.get("raw_pool_path", "") row["raw_file_sha256"] = discovery_source.get("file_sha256", "") r4.write_csv(classification_path, list(classifications[0]), classifications) selected = [row for row in candidates if row["candidate_state"] in {"INCLUDED_T1", "INCLUDED_T2"}] matrix_path = industry_root / "extracted" / "newenergy_company_exposure_matrix.csv" matrix_headers = list(r4.read_csv(matrix_path)[0]) matrix_rows: list[dict[str, str]] = [] for candidate in selected: pair = (candidate["company_id"], candidate["track_code"]) evidence = next( evidence for evidence in evidence_by_pair[pair] if evidence["doc_id"] == candidate["direct_business_source_id"] ) row = { "mapping_id": f"MAP-{candidate['track_code']}-{candidate['security_code']}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "company_id": candidate["company_id"], "security_code": candidate["security_code"], "security_name": candidate["security_name"], "legal_name": candidate["legal_name"], "exchange_code": candidate["exchange_code"], "track_code": candidate["track_code"], "chain_nodes": candidate["chain_nodes"], "selection_bucket": candidate["selection_bucket"], "tier": candidate["tier"], "candidate_state": candidate["candidate_state"], "direct_business_source_id": candidate["direct_business_source_id"], "direct_business_locator": candidate["direct_business_locator"], "evidence_fact_id": evidence["evidence_fact_id"], "evidence_grade": candidate["evidence_grade"], "exposure_specificity": candidate["exposure_specificity"], "latest_disclosed_period": candidate["latest_disclosed_period"], "primary_region": "MAINLAND_CHINA", "scope_status": "CORE_SCOPE_DIRECT_BUSINESS", "coverage_claim": "NONE_INITIAL_CANDIDATE_POOL_ONLY", "data_status": "VERIFIED_PUBLIC", "review_status": REVIEW_STATUS, } matrix_rows.append({key: row.get(key, "") for key in matrix_headers}) r4.write_csv(matrix_path, matrix_headers, matrix_rows) return classifications, matrix_rows def build_manifests( industry_root: Path, source_rows: list[dict[str, str]], conversion_rows: list[dict[str, str]], evidence_rows: list[dict[str, str]], candidates: list[dict[str, str]], case_map: list[dict[str, str]], ) -> tuple[list[dict[str, str]], list[dict[str, str]]]: base.COLLECTED_AT = max( (row["collected_at"] for row in source_rows if row.get("collected_at")), default="2026-08-06T00:00:00+08:00", ) base.ARTIFACT_TOOL_VERSION = TOOL_VERSION base.ARTIFACT_PARAMETERS_SUMMARY = ( "B1 REPAIR005; reuse 729 acquired PDFs; replay 858 pairs through the company-self and " "target-bucket semantic role gate; clear non-eligible derived fields and dangling evidence FKs; " "mechanical rerank; exact-set hash coverage" ) base.ARTIFACT_STATUS = "READY_FOR_REPAIR005_FOCUSED_REREVIEW" base.build_input_manifests(source_rows) base.build_source_gap_audit(source_rows, evidence_rows) universe_rows = r4.read_csv(industry_root / "extracted" / "a_share_universe.csv") base.rewrite_batch_summary(candidates, len(source_rows), len(conversion_rows), universe_rows) base.build_human_receipt(case_map, len(source_rows)) output_rows = base.rebuild_output_manifest() artifact_rows = base.build_artifact_manifest(source_rows, output_rows, Path(__file__)) return output_rows, artifact_rows def split_evidence_ids(value: str) -> list[str]: return [item.strip() for item in (value or "").split(";") if item.strip()] def validate_and_write_receipt( industry_root: Path, role_rows: list[dict[str, str]], prestate_rows: list[dict[str, str]], source_rows: list[dict[str, str]], conversion_rows: list[dict[str, str]], evidence_rows: list[dict[str, str]], candidates: list[dict[str, str]], funnel: list[dict[str, str]], classifications: list[dict[str, str]], matrix_rows: list[dict[str, str]], case_map: list[dict[str, str]], ) -> dict[str, Any]: errors: list[str] = [] eligible_states = {"INCLUDED_T1", "INCLUDED_T2", "ELIGIBLE_NOT_SELECTED_BATCH001"} evidence_ids = {row["evidence_fact_id"] for row in evidence_rows} source_ids = {row["doc_id"] for row in source_rows} conversion_ids = {row["source_doc_id"] for row in conversion_rows} candidate_by_pair = {(row["company_id"], row["track_code"]): row for row in candidates} role_counts = Counter(row["final_role_decision"] for row in role_rows) expected_role_counts = { "ELIGIBLE_DIRECT_COMPANY_SELF_TARGET_BUCKET_ROLE": 40, "HELD_ROLE_MISMATCH_ADJACENT_PROCUREMENT_OR_SELF_USE": 10, "HELD_KEYWORD_CONTEXT_ONLY_AFTER_FULL_PAGE_SEARCH": 221, "HELD_TRUE_NEGATIVE_AFTER_FULL_PAGE_SEARCH": 587, } if dict(role_counts) != expected_role_counts: errors.append(f"role distribution mismatch: {dict(role_counts)}") if len(role_rows) != 858 or len({row["qualification_row_id"] for row in role_rows}) != 858: errors.append("role adjudication receipt is not exact 858") if any(row["rule_version"] != ROLE_RULE_VERSION or row["tool_version"] != TOOL_VERSION for row in role_rows): errors.append("role receipt rule/tool version mismatch") candidate_counts = Counter(row["candidate_state"] for row in candidates) expected_candidate_counts = { "INCLUDED_T1": 16, "INCLUDED_T2": 16, "ELIGIBLE_NOT_SELECTED_BATCH001": 40, "HELD_BY_EVIDENCE_GAP": 4479, } if dict(candidate_counts) != expected_candidate_counts: errors.append(f"candidate distribution mismatch: {dict(candidate_counts)}") if len(candidates) != 4551 or len({(row["company_id"], row["track_code"]) for row in candidates}) != 4551: errors.append("candidate ledger pair set is not exact 4551") noneligible_derived_fields = [ "chain_nodes", "direct_business_source_id", "direct_business_locator", "evidence_grade", "exposure_specificity", "latest_disclosed_period", "selection_rank", "tier", "tie_break_rule", ] polluted_candidates = [ row for row in candidates if row["candidate_state"] not in eligible_states and any(row.get(field, "") for field in noneligible_derived_fields) ] if polluted_candidates: errors.append(f"noneligible candidate derived-field pollution={len(polluted_candidates)}") polluted_classifications = [ row for row in classifications if row["data_status"] not in eligible_states and ( row["subject_type"] == "COMPANY_DIRECT_BUSINESS" or bool(row["evidence_fact_id"]) or row["source_doc_id"].startswith("S-QUAL-") or bool(row["chain_node_id"]) ) ] if polluted_classifications: errors.append(f"noneligible classification pollution={len(polluted_classifications)}") for qid in REJECTED_ROLE_DECISIONS: role = next(row for row in role_rows if row["qualification_row_id"] == qid) candidate = candidate_by_pair[(role["company_id"], role["track_code"])] if candidate["candidate_state"] != "HELD_BY_EVIDENCE_GAP" or any( candidate.get(field, "") for field in noneligible_derived_fields ): errors.append(f"rejected role candidate not clean: {qid}") selected_groups: dict[tuple[str, str], list[dict[str, str]]] = defaultdict(list) for candidate in candidates: if candidate["candidate_state"] in {"INCLUDED_T1", "INCLUDED_T2"}: selected_groups[(candidate["track_code"], candidate["selection_bucket"])].append(candidate) if len(selected_groups) != 16 or any( sorted(row["candidate_state"] for row in rows) != ["INCLUDED_T1", "INCLUDED_T2"] for rows in selected_groups.values() ): errors.append("16-bucket T1/T2 coverage mismatch") if source_ids != conversion_ids or len(source_rows) != 786 or len(conversion_rows) != 786: errors.append(f"source/conversion exact set mismatch {len(source_rows)}/{len(conversion_rows)}") if any(row["direct_business_source_id"] not in source_ids for row in candidates if row["candidate_state"] in eligible_states): errors.append("eligible candidate source FK dangling") if any(evidence_id not in evidence_ids for row in funnel for evidence_id in split_evidence_ids(row["evidence_fact_ids"])): errors.append("funnel evidence FK dangling") if any(row["evidence_fact_id"] and row["evidence_fact_id"] not in evidence_ids for row in classifications): errors.append("classification evidence FK dangling") if any(row["evidence_fact_id"] not in evidence_ids for row in matrix_rows): errors.append("exposure matrix evidence FK dangling") if any(row["evidence_fact_id"] not in evidence_ids for row in case_map): errors.append("case map evidence FK dangling") new_qual_ids = {row["evidence_fact_id"] for row in evidence_rows if row["evidence_fact_id"].startswith("EVF-QUAL-")} if len(new_qual_ids) != 40 or any(not evidence_id.endswith("-R005") for evidence_id in new_qual_ids): errors.append(f"REPAIR005 qualification evidence exact set mismatch={len(new_qual_ids)}") if len(matrix_rows) != 32 or len(case_map) != 110: errors.append(f"matrix/map count mismatch {len(matrix_rows)}/{len(case_map)}") if len(funnel) != 4551: errors.append(f"funnel count mismatch={len(funnel)}") if len(prestate_rows) != 8: errors.append(f"direct-claim pollution prestate snapshot mismatch={len(prestate_rows)}") core_hashes = { "role_adjudication_receipt": r4.sha256_file(role_receipt_path(industry_root)), "candidate_ledger": r4.sha256_file(industry_root / "extracted" / "company_track_candidate_ledger.csv"), "qualification_funnel": r4.sha256_file(industry_root / "extracted" / "candidate_qualification_funnel.csv"), "classification_summary": r4.sha256_file(industry_root / "extracted" / "classification_summary.csv"), "evidence_fact_table": r4.sha256_file(industry_root / "evidence" / "evidence_fact_table.csv"), "exposure_matrix": r4.sha256_file(industry_root / "extracted" / "newenergy_company_exposure_matrix.csv"), "case_evidence_map": r4.sha256_file(industry_root / CASE_ID / "evidence" / "case_evidence_map.csv"), "pollution_prestate_snapshot": r4.sha256_file(prestate_path(industry_root)), } receipt = { "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "tool_version": TOOL_VERSION, "rule_version": ROLE_RULE_VERSION, "generated_at": role_rows[0]["adjudicated_at"], "review_status": REVIEW_STATUS, "scope": "B1_REPAIR005_ONLY_REUSE_729_PDFS_FULL_858_ROLE_STATE_REPLAY", "role_status_distribution": dict(sorted(role_counts.items())), "candidate_state_distribution": dict(sorted(candidate_counts.items())), "prestate_direct_claim_pollution_rows": len(prestate_rows), "poststate_noneligible_candidate_pollution_rows": len(polluted_candidates), "poststate_noneligible_classification_pollution_rows": len(polluted_classifications), "dangling_fk_count": sum("FK dangling" in error for error in errors), "selected_bucket_count": len(selected_groups), "selected_mapping_count": len(matrix_rows), "source_count": len(source_rows), "conversion_count": len(conversion_rows), "evidence_fact_count": len(evidence_rows), "case_map_count": len(case_map), "core_hashes": core_hashes, "validation_errors": errors, "validation_status": "PASS" if not errors else "FAIL", } path = validation_path(industry_root) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(receipt, ensure_ascii=False, sort_keys=True, indent=2) + "\n", encoding="utf-8") if errors: raise RuntimeError("REPAIR005 validation failed: " + "; ".join(errors)) return receipt def validate_artifacts( industry_root: Path, source_rows: list[dict[str, str]], output_rows: list[dict[str, str]], artifact_rows: list[dict[str, str]], ) -> dict[str, Any]: formal = base.formal_files(Path(__file__)) formal_by_rel = {base.rel(path): path for path in formal} artifact_by_rel = {row["relative_path"]: row for row in artifact_rows} errors: list[str] = [] if set(formal_by_rel) != set(artifact_by_rel): errors.append( f"artifact exact set diff missing={sorted(set(formal_by_rel) - set(artifact_by_rel))[:5]} " f"extra={sorted(set(artifact_by_rel) - set(formal_by_rel))[:5]}" ) for relative_path, path in formal_by_rel.items(): artifact = artifact_by_rel.get(relative_path) if not artifact: continue if artifact["file_size"] != str(path.stat().st_size) or artifact["sha256"] != r4.sha256_file(path): errors.append(f"artifact hash mismatch {relative_path}") break if any(row["tool_version"] != TOOL_VERSION for row in artifact_rows): errors.append("artifact tool version is not REPAIR-005") for row in output_rows: path = base.PROJECT_ROOT / row["output_path"] if not path.exists() or row["output_sha256"] != r4.sha256_file(path): errors.append(f"output hash mismatch {row['output_path']}") break if len(source_rows) != 786 or len(output_rows) != 24: errors.append(f"source/output count mismatch {len(source_rows)}/{len(output_rows)}") if errors: raise RuntimeError("REPAIR005 artifact validation failed: " + "; ".join(errors)) return { "formal_exact_set": len(formal), "artifact_rows": len(artifact_rows), "output_rows": len(output_rows), "artifact_manifest_sha256": r4.sha256_file(industry_root / "manifest" / "artifact_manifest.csv"), "output_manifest_sha256": r4.sha256_file( industry_root / CASE_ID / "manifest" / "output_manifest.csv" ), } def run() -> dict[str, Any]: project_root = Path.cwd().resolve() industry_root = r4.find_industry_root(project_root) source_rows = r4.read_csv(industry_root / "manifest" / "source_document.csv") conversion_rows = r4.read_csv(industry_root / "manifest" / "conversion_status.csv") pair_rows = r4.read_csv(industry_root / "extracted" / "candidate_page_qualification_receipt.csv") candidates_before = r4.read_csv(industry_root / "extracted" / "company_track_candidate_ledger.csv") classifications_before = r4.read_csv(industry_root / "extracted" / "classification_summary.csv") prestate_rows = write_prestate_snapshot_once(industry_root, candidates_before, classifications_before) role_rows = build_role_adjudication_receipt(industry_root, pair_rows, candidates_before) evidence_rows, candidates = build_evidence_and_candidates(industry_root, role_rows, source_rows) funnel = update_funnel(industry_root, role_rows, candidates, evidence_rows, source_rows) classifications, matrix_rows = update_classification_and_matrix( industry_root, candidates, evidence_rows, source_rows, ) case_map = r4.write_company_outputs_and_map(industry_root, candidates, evidence_rows, source_rows) validation = validate_and_write_receipt( industry_root, role_rows, prestate_rows, source_rows, conversion_rows, evidence_rows, candidates, funnel, classifications, matrix_rows, case_map, ) output_rows, artifact_rows = build_manifests( industry_root, source_rows, conversion_rows, evidence_rows, candidates, case_map, ) artifact_validation = validate_artifacts(industry_root, source_rows, output_rows, artifact_rows) return { "status": "REPAIR005_FORMAL_PACKAGE_REBUILT_DRAFT_FOR_REVIEW", "role_status_distribution": validation["role_status_distribution"], "candidate_state_distribution": validation["candidate_state_distribution"], "dangling_fk_count": validation["dangling_fk_count"], "prestate_pollution_rows": validation["prestate_direct_claim_pollution_rows"], "poststate_pollution_rows": ( validation["poststate_noneligible_candidate_pollution_rows"] + validation["poststate_noneligible_classification_pollution_rows"] ), "sources": len(source_rows), "conversions": len(conversion_rows), "evidence_facts": len(evidence_rows), "candidates": len(candidates), "selected": len(matrix_rows), "case_map": len(case_map), **artifact_validation, "role_receipt_sha256": r4.sha256_file(role_receipt_path(industry_root)), "validation_receipt_sha256": r4.sha256_file(validation_path(industry_root)), } def main() -> None: parser = argparse.ArgumentParser() parser.parse_args() result = run() print(json.dumps(result, ensure_ascii=False, sort_keys=True, indent=2)) if __name__ == "__main__": main()