#!/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()
|
