from __future__ import annotations import csv import hashlib import json import re from collections import Counter, defaultdict from datetime import datetime from pathlib import Path from typing import Any, Iterable ROOT = Path(__file__).resolve().parents[2] INDUSTRY = ROOT / "ana-data" / "cases" / "新能源案例" CASE_ID = "ANA-NEWENERGY-FOUR-TRACK-ATLAS-20260806-002" TASK_ID = "TASK-NEWENERGY-FOUR-TRACK-ATLAS-BATCH002-20260806-001" BATCH_ID = "BATCH-002" RUN_ID = "RUN-ANA-NEWENERGY-FOUR-TRACK-ATLAS-20260806-002-BATCH-002-001" CASE = INDUSTRY / CASE_ID RESULT = ROOT / "ana-data" / "result" / "新能源案例" / CASE_ID IMG = ROOT / "ana-data" / "img" / "新能源案例" / CASE_ID TMP = ROOT / "ana-data" / "tmp" / "新能源案例" / CASE_ID / RUN_ID SCHEMA = "NEWENERGY_EXTENSION_V1" REVIEW = "DRAFT_FOR_REVIEW" SOURCE_CUTOFF = "2026-08-06T20:27:54+08:00" EXECUTED_AT = "2026-08-07T00:30:00+08:00" REPAIR_ID = "EXECUTION_OUTPUT_REPAIR001" LATEST_REPAIR_ID = "EXECUTION_OUTPUT_REPAIR004" QUERY_PROBE_PATH = INDUSTRY / "supplement" / "NEB2_external_public_query_probe_REPAIR001.json" QUERY_RECEIPT_PATH = INDUSTRY / "supplement" / "NEB2_external_public_query_receipt_BATCH002.csv" def rel(path: Path) -> str: return path.relative_to(ROOT).as_posix() def sha256(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as fh: for chunk in iter(lambda: fh.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest().upper() def read_csv(path: Path) -> list[dict[str, str]]: with path.open("r", encoding="utf-8-sig", newline="") as fh: return list(csv.DictReader(fh)) def write_csv(path: Path, rows: list[dict[str, Any]], fields: list[str] | None = None) -> None: path.parent.mkdir(parents=True, exist_ok=True) if fields is None: if not rows: raise ValueError(f"refuse empty csv without schema: {path}") fields = list(rows[0].keys()) with path.open("w", encoding="utf-8-sig", newline="") as fh: writer = csv.DictWriter(fh, fieldnames=fields, extrasaction="ignore", lineterminator="\n") writer.writeheader() writer.writerows(rows) def write_text(path: Path, text: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(text.rstrip() + "\n", encoding="utf-8", newline="\n") def write_json(path: Path, obj: Any) -> None: write_text(path, json.dumps(obj, ensure_ascii=False, indent=2, sort_keys=True)) def index_unique(rows: Iterable[dict[str, str]], key: str) -> dict[str, dict[str, str]]: out: dict[str, dict[str, str]] = {} for row in rows: value = row[key] if value in out: raise ValueError(f"duplicate {key}: {value}") out[value] = row return out BASELINE_HASHES = { INDUSTRY / "extracted" / "company_track_candidate_ledger.csv": "ABD3D0BF2CEF2D9D0225F2D37A9542A0899B75B127E829747C767FAED03E54E5", INDUSTRY / "extracted" / "candidate_qualification_funnel.csv": "8FCC1C9563A3709E04EE747173A55DF76135B760E186529DC1F1DA8A07C0891A", INDUSTRY / "extracted" / "candidate_role_adjudication_receipt.csv": "7E47068068E47EDDDF9B9B6766D5A8FBDF40CEB62FFF33508DBF65302F42D824", INDUSTRY / "evidence" / "evidence_fact_table.csv": "B9FE3822781F776E93A57309E52794EF445B86820B96410B486E57A852540AC9", INDUSTRY / "ANA-NEWENERGY-FOUR-TRACK-ATLAS-20260805-001" / "evidence" / "case_evidence_map.csv": "CE6CFFB10BD3B11FD4E29B5381B814DAC66A8EE1E53BB9821805312D19AFD103", INDUSTRY / "ANA-NEWENERGY-FOUR-TRACK-ATLAS-20260805-001" / "manifest" / "output_manifest.csv": "D0DC978D558CA56C9D32B9DC0F76489EDB0F643DB1B482585EF2D96DEE3DF6C4", ROOT / "ana-data" / "result" / "新能源案例" / "ANA-NEWENERGY-FOUR-TRACK-ATLAS-20260805-001" / "acceptance_record.md": "448E67471C53E1E59011A1ADBB44D2ED0BDA0068084D843251E2CCCD4A407CFA", } for baseline_path, expected in BASELINE_HASHES.items(): if not baseline_path.exists(): raise FileNotFoundError(f"missing immutable baseline: {baseline_path}") actual = sha256(baseline_path) if actual != expected: raise RuntimeError(f"immutable baseline drift: {rel(baseline_path)} {actual} != {expected}") for directory in ( CASE / "outputs" / "核心文档" / "子行业深化", CASE / "manifest", CASE / "evidence", RESULT, IMG, TMP, INDUSTRY / "supplement", ): directory.mkdir(parents=True, exist_ok=True) for forbidden in (CASE / "raw", CASE / "converted", CASE / "extracted", CASE / "supplement"): if forbidden.exists(): raise RuntimeError(f"case storage split violated: {forbidden}") ledger = read_csv(INDUSTRY / "extracted" / "company_track_candidate_ledger.csv") funnel = read_csv(INDUSTRY / "extracted" / "candidate_qualification_funnel.csv") roles = read_csv(INDUSTRY / "extracted" / "candidate_role_adjudication_receipt.csv") evidence = read_csv(INDUSTRY / "evidence" / "evidence_fact_table.csv") source_docs = read_csv(INDUSTRY / "manifest" / "source_document.csv") conversions = read_csv(INDUSTRY / "manifest" / "conversion_status.csv") artifacts_b1 = read_csv(INDUSTRY / "manifest" / "artifact_manifest.csv") evidence_by_id = index_unique(evidence, "evidence_fact_id") source_by_id = index_unique(source_docs, "doc_id") conversion_by_source = {r["source_doc_id"]: r for r in conversions} artifact_by_rel = {r["relative_path"]: r for r in artifacts_b1} funnel_by_pair = {(r["company_id"], r["track_code"], r["selection_bucket"]): r for r in funnel} role_by_pair = {(r["company_id"], r["track_code"], r["selection_bucket"]): r for r in roles} ledger_by_pair = {(r["company_id"], r["track_code"], r["selection_bucket"]): r for r in ledger} NODES = [ { "n": 1, "track": "BATTERY", "code": "BAT_RESOURCE_MATERIAL", "name": "资源与主材", "bucket": "资源与主材", "folder": "01_资源与主材", "boundary": "纳入锂资源、锂盐、正极/负极、电解液和隔膜的直接生产经营；排除一般化工、采购方、客户和题材关联。", "role": "上游资源与锂盐连接电芯材料成本，四大主材分别承担离子载体、电极反应、离子传导与安全隔离功能。", "route": "锂盐—正负极—电解液—隔膜分层；不同材料、规格与应用场景不得合并为单一价格或产能口径。", "market_evf": "EVF-MKT-BAT-03", "risk": "资源品价格和材料路线变化可能改变成本传导；规格、期间或税口径不同会使价格比较失真。", }, { "n": 2, "track": "BATTERY", "code": "BAT_CELL", "name": "电芯制造", "bucket": "电芯制造", "folder": "02_电芯制造", "boundary": "纳入动力或储能电芯的研发、生产和销售；模组、Pack、整车和单纯电池材料分别分账。", "role": "电芯把材料体系转化为可用电化学单元，是动力与储能系统的核心制造环节。", "route": "动力与储能应用分账；磷酸铁锂、三元等路线按应用、安全、寿命与成本约束比较。", "market_evf": "EVF-MKT-BAT-01", "risk": "产量增长不等于终端消化，库存、利用率、质量与安全信息仍需单独验证。", }, { "n": 3, "track": "BATTERY", "code": "BAT_SYSTEM_COMPONENT", "name": "系统/部件/BMS-Pack", "bucket": "系统/部件/BMS-Pack", "folder": "03_系统部件BMS-Pack", "boundary": "纳入电芯模组、Pack、BMS、结构件和系统集成的直接业务；排除仅使用电池的终端产品。", "role": "系统环节完成电芯成组、状态监测、热安全和应用适配，动力与储能的设计目标不可混用。", "route": "从电芯到模组、Pack、BMS和系统集成逐层识别；分部不可拆时降低暴露结论强度。", "market_evf": "EVF-MKT-BAT-02", "risk": "价值量受集成边界、外购比例和应用结构影响，不能由电池总量直接推导公司收入。", }, { "n": 4, "track": "BATTERY", "code": "BAT_EQUIPMENT_RECYCLE", "name": "设备与回收循环", "bucket": "设备与回收循环", "folder": "04_设备与回收循环", "boundary": "纳入锂电前中后段专用设备、检测设备和电池回收处理的直接业务；通用施工、包装和自用项目不纳入。", "role": "设备支撑制造工序，回收环节处理退役电池与生产废料；两者的订单、产能和经济性口径必须分开。", "route": "设备按工序验证，回收按来源料、处理工艺与再生产品验证；仅有‘新能源’字样不足以确认角色。", "market_evf": "EVF-MKT-BAT-01", "risk": "本批冻结候选池未找到可升级的直接主源，相关企业增量保持缺口。", }, { "n": 5, "track": "SOLAR", "code": "SOL_SILICON_WAFER", "name": "硅料/硅片与材料", "bucket": "硅料/硅片与材料", "folder": "05_硅料硅片与材料", "boundary": "纳入多晶硅、拉晶、硅棒/硅片和直接关键材料；电池片、组件和电站业务另列。", "role": "上游材料决定电池片的基础晶硅供给，产能必须区分名义、有效、在产与不同尺寸/导电类型。", "route": "硅料—拉晶—切片分层，N/P 型和尺寸规格分账；不以总产能替代可售合格产出。", "market_evf": "EVF-MKT-SOL-01", "risk": "装机需求不能直接推导硅片盈利，供给、价格、库存与技术切换均可能造成背离。", }, { "n": 6, "track": "SOLAR", "code": "SOL_CELL_MODULE", "name": "电池片/组件", "bucket": "电池片/组件", "folder": "06_电池片组件", "boundary": "纳入太阳能电池片和组件的直接制造销售；玻璃等辅材、逆变器与电站运营分别归桶。", "role": "电池片完成光电转换，组件完成封装和终端产品交付；技术效率、良率和出货口径需区分。", "route": "电池技术与组件产品分账，产能、产量、出货、并网装机不能互相替代。", "market_evf": "EVF-MKT-SOL-01", "risk": "快速路线迭代和供给扩张可能造成存货、减值与旧产线利用率风险。", }, { "n": 7, "track": "SOLAR", "code": "SOL_BALANCE_EQUIPMENT", "name": "设备/辅材/逆变器", "bucket": "设备/辅材/逆变器", "folder": "07_光伏设备辅材逆变器", "boundary": "纳入光伏专用设备、辅材和逆变器直接业务；通用工业品潜在用途和项目采购不纳入。", "role": "设备决定制造能力，辅材影响封装可靠性，逆变器承担直交流转换与系统控制，三者分别核验。", "route": "按设备、玻璃/胶膜等辅材、逆变器三个子层记录，不用组件需求替代各自产品证据。", "market_evf": "EVF-MKT-SOL-01", "risk": "需求联动存在时滞，设备订单、辅材出货与逆变器销售不能由装机量机械外推。", }, { "n": 8, "track": "SOLAR", "code": "SOL_SYSTEM_STATION", "name": "系统集成/电站建设运营", "bucket": "系统集成/电站建设运营", "folder": "08_光伏系统电站", "boundary": "纳入光伏 EPC、开发、建设、持有运营与发电收入；设备采购和内部自用电站不等于对外系统业务。", "role": "项目从开发、建设、并网到持有运营分阶段，工程收入和发电收入必须分账。", "route": "项目阶段、所有权和收入确认口径是核心；规划、在建、并网和运营不得静默合并。", "market_evf": "EVF-MKT-SOL-02", "risk": "装机增长不代表单个开发商收益，消纳、电价、融资和资产处置均会改变经营结果。", }, { "n": 9, "track": "WIND", "code": "WND_COMPONENT", "name": "材料与关键零部件", "bucket": "材料与关键零部件", "folder": "09_风电材料关键零部件", "boundary": "纳入叶片、铸锻件、主轴、轴承、齿轮箱等直接制造；通用材料和相邻工业品不纳入。", "role": "关键零部件连接整机技术平台与上游制造，产品规格、尺寸和认证决定可交付范围。", "route": "按叶片、铸锻、传动和轴承等子层识别，海陆产品、尺寸与交付周期分账。", "market_evf": "EVF-MKT-WIND-01", "risk": "大型化与海上化会改变材料、工艺和验证要求，旧规格产能不可直接视为有效供给。", }, { "n": 10, "track": "WIND", "code": "WND_TURBINE", "name": "整机", "bucket": "整机", "folder": "10_风电整机", "boundary": "纳入风力发电整机研发、制造、销售与随整机形成的直接服务；运营商和通用电气设备另列。", "role": "整机集成叶轮、传动、电气与控制系统，并承担交付、质保和部分运维责任。", "route": "海陆机型、平台功率、订单、交付和质保口径分账；采购或使用风机不构成整机业务。", "market_evf": "EVF-MKT-WIND-01", "risk": "本批空桶候选均未获得直接整机主源，不能由风电项目或发电业务反推整机制造。", }, { "n": 11, "track": "WIND", "code": "WND_TOWER_CABLE_ENGINEERING", "name": "塔筒/海缆/工程配套", "bucket": "塔筒/海缆/工程配套", "folder": "11_塔筒海缆工程配套", "boundary": "纳入塔筒、基础、海缆、吊装和风电工程直接业务；一般施工、钢管或项目相邻供应不纳入。", "role": "工程配套把设备连接到场址和电网，塔筒、海缆、基础与吊装的制造/施工能力分别验证。", "route": "海陆项目、制造与施工、订单与收入确认分账；通用工程资质不能替代风电直接业务。", "market_evf": "EVF-MKT-WIND-01", "risk": "项目进度、海况、原材料与验收会影响交付，本批未找到可升级的增量企业。", }, { "n": 12, "track": "WIND", "code": "WND_OPERATION_OM", "name": "项目运营与运维服务", "bucket": "项目运营与运维服务", "folder": "12_风电运营运维", "boundary": "纳入风电场持有运营和对外运维服务；核准、建设、并网、商运及运维收入分别记录。", "role": "运营环节把装机转化为发电量，利用小时、可利用率、消纳和电价共同影响经营事实。", "route": "项目阶段与海陆风分账，发电量和运维服务收入不互相替代。", "market_evf": "EVF-MKT-WIND-02", "risk": "资源条件、限电、检修和电价机制变化会造成装机与发电表现背离。", }, { "n": 13, "track": "NUCLEAR", "code": "NUC_OPERATOR", "name": "运营商", "bucket": "运营商", "folder": "13_核电运营商", "boundary": "仅纳入民用核电机组持有运营及公开高层运营指标；核准、在建、并网和商运严格分账。", "role": "运营商承担机组建设转运营后的发电与安全运行责任，本研究只处理公开聚合指标。", "route": "机组台数、装机、发电量与上网电量分别记录；不处理厂址安防、控制系统或敏感参数。", "market_evf": "EVF-MKT-NUC-01", "risk": "本批没有新的运营商合格余额；候选中的其他发电或设备业务不能升级为核电运营。", }, { "n": 14, "track": "NUCLEAR", "code": "NUC_ENGINEERING_EPC", "name": "工程/EPC", "bucket": "工程/EPC", "folder": "14_核电工程EPC", "boundary": "仅纳入民用核电工程设计、建设或总承包的直接公开业务；一般建材、防水和相邻施工不纳入核电 EPC。", "role": "工程环节连接核准、设计、施工和投运，合同阶段与收入确认需用正式披露核验。", "route": "只保留公开项目阶段和公司角色，不记录关键基础设施敏感细节。", "market_evf": "EVF-MKT-NUC-2024", "risk": "东方雨虹公开案例仅支持防水材料及施工服务，不能据此认定核电工程总承包。", }, { "n": 15, "track": "NUCLEAR", "code": "NUC_NI_CI_EQUIPMENT", "name": "核岛/常规岛主设备", "bucket": "核岛/常规岛主设备", "folder": "15_核岛常规岛主设备", "boundary": "纳入公开披露的民用核岛/常规岛主设备直接制造交付；通用设备潜在用途和采购方不纳入。", "role": "主设备与机组系统直接相关，公开产品、合同、交付与收入证据是公司资格基础。", "route": "只到公开产品类别和项目阶段，不处理非公开性能、布置、控制或脆弱性细节。", "market_evf": "EVF-MKT-NUC-2024", "risk": "冻结账本在该桶没有可处理候选，本批明确保留企业增量缺口。", }, { "n": 16, "track": "NUCLEAR", "code": "NUC_COMPONENT_MATERIAL_ICT", "name": "核级部件/材料/仪控电气", "bucket": "核级部件/材料/仪控电气", "folder": "16_核级部件材料仪控电气", "boundary": "纳入公开核级部件、材料、仪控和电气直接业务；一般工业产品、资质推测和潜在用途不纳入。", "role": "该环节依赖核级产品、认证或订单的明确公开证据，不能由公司能力或客户关系推断。", "route": "按部件、材料、仪控、电气分层；仅记录民用公开高层信息。", "market_evf": "EVF-MKT-NUC-2024", "risk": "10 个冻结候选均未形成直接核级业务主源，增量映射保持 HELD。", }, ] node_by_bucket = {(n["track"], n["bucket"]): n for n in NODES} if len(node_by_bucket) != 16: raise RuntimeError("node bucket uniqueness failed") COMPANY_EVF = { ("BATTERY", "资源与主材"): "EVF-QUAL-BATTERY-002497-R005", ("BATTERY", "电芯制造"): "EVF-BATTERY-300750-BUSINESS", ("BATTERY", "系统/部件/BMS-Pack"): "EVF-BATTERY-300207-BUSINESS", ("SOLAR", "硅料/硅片与材料"): "EVF-QUAL-SOLAR-601908-R005", ("SOLAR", "电池片/组件"): "EVF-QUAL-SOLAR-002623-R005", ("SOLAR", "设备/辅材/逆变器"): "EVF-SOLAR-300274-BUSINESS", ("SOLAR", "系统集成/电站建设运营"): "EVF-QUAL-SOLAR-301070-R005", ("WIND", "材料与关键零部件"): "EVF-WIND-300443-BUSINESS", ("WIND", "项目运营与运维服务"): "EVF-WIND-001289-BUSINESS", } def company_evidence_summary(row: dict[str, str], ev: dict[str, str]) -> str: if row["security_code"] == "002497": return "年报披露公司锂产品面向汽车厂商、电池企业和正极材料企业销售。" return ev["evidence_text"] eligible_by_bucket: dict[tuple[str, str], list[dict[str, str]]] = defaultdict(list) for row in ledger: if row["candidate_state"] == "ELIGIBLE_NOT_SELECTED_BATCH001": eligible_by_bucket[(row["track_code"], row["selection_bucket"])].append(row) for rows in eligible_by_bucket.values(): rows.sort(key=lambda r: (int(r["selection_rank"]), r["exchange_code"], r["security_code"])) incremental: list[dict[str, str]] = [] for node in NODES: rows = eligible_by_bucket.get((node["track"], node["bucket"]), []) if not rows: continue prior = rows[0] ev_id = COMPANY_EVF[(node["track"], node["bucket"])] ev = evidence_by_id[ev_id] if ev["company_id"] != prior["company_id"]: raise RuntimeError(f"company evidence mismatch for {node['code']}") incremental.append({**prior, "node_code": node["code"], "node_name": node["name"], "evidence_fact_id": ev_id}) expected_incremental_codes = {"002497", "300750", "300207", "601908", "002623", "300274", "301070", "300443", "001289"} if {r["security_code"] for r in incremental} != expected_incremental_codes: raise RuntimeError("mechanical inherited incremental set changed") QUEUE_CODES = { ("BATTERY", "设备与回收循环"): ["002081", "300793", "600120", "600248", "600353", "603052", "603687", "688335", "688610", "000036"], ("WIND", "整机"): ["600089", "600863", "603969", "000155", "000690", "600011", "600248", "600548", "600642", "601985"], ("WIND", "塔筒/海缆/工程配套"): ["601618", "002443", "600268", "600985", "601669", "000862", "002307", "601069", "000933"], ("NUCLEAR", "运营商"): ["300198", "601121", "000958", "300165", "600312", "600550", "600651", "601609", "601618", "603308"], ("NUCLEAR", "工程/EPC"): ["002271"], ("NUCLEAR", "核岛/常规岛主设备"): [], ("NUCLEAR", "核级部件/材料/仪控电气"): ["600685", "601199", "601212", "603013", "603282", "603700", "603800", "688198", "000571", "000581"], } if set(QUEUE_CODES) != {key for key in node_by_bucket if key not in eligible_by_bucket}: raise RuntimeError("empty-bucket set changed") queue_rows: list[dict[str, Any]] = [] for (track, bucket), codes in QUEUE_CODES.items(): node = node_by_bucket[(track, bucket)] for rank, code in enumerate(codes, 1): candidates = [r for r in ledger if r["security_code"] == code and r["track_code"] == track and r["selection_bucket"] == bucket] if len(candidates) != 1: raise RuntimeError(f"queue pair unresolved: {track}/{bucket}/{code} count={len(candidates)}") prior = candidates[0] f = funnel_by_pair[(prior["company_id"], track, bucket)] rr = role_by_pair[(prior["company_id"], track, bucket)] prior_attachment_id = rr["selected_attachment_id"] if not prior_attachment_id: prior_attachment_id = (f["announcement_ids"].split(";")[0] if f["announcement_ids"] else "") prior_source_doc_id = f"S-QUAL-AR-{prior_attachment_id}" if prior_attachment_id else "" if not prior_source_doc_id or prior_source_doc_id not in source_by_id: raise RuntimeError(f"queue prior annual report unresolved: {track}/{bucket}/{code}/{prior_source_doc_id}") prior_doc = source_by_id[prior_source_doc_id] new_doc = "S-NEB2-YUHONG-SUSTAINABILITY-2023" if code == "002271" and track == "NUCLEAR" else "" new_evf = "EVF-NE-B002-ADJ-002271" if new_doc else "" result = "HELD_ADJACENT_WATERPROOFING_MATERIAL_AND_SERVICE_NOT_NUCLEAR_EPC" if new_doc else "HELD_NO_NEW_DIRECT_TARGET_BUCKET_PRIMARY_SOURCE" queue_rows.append({ "queue_item_id": f"NEB2-QUEUE-{node['n']:02d}-{rank:02d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "subindustry_node_code": node["code"], "track_code": track, "selection_bucket": bucket, "queue_rank": rank, "company_id": prior["company_id"], "security_code": code, "security_name": prior["security_name"], "prior_candidate_state": prior["candidate_state"], "prior_query_hit_count": f["query_hit_count"], "prior_business_context_result": f["business_context_rule_result"], "prior_source_doc_id": prior_source_doc_id, "prior_locator": rr["selected_page"] and f"page={rr['selected_page']}" or f["fulltext_context_locator"], "prior_raw_file_path": prior_doc["raw_file_path"], "prior_raw_sha256": prior_doc["file_sha256"].upper(), "prior_converted_text_path": prior_doc["converted_text_path"], "prior_retrieval_result": f["annual_report_retrieval_result"], "prior_full_page_search_result": f["page_level_verification_result"], "prior_context_sha256": f["fulltext_context_sha256"], "prior_context_excerpt": f["fulltext_context_excerpt_sanitized"], "prior_company_self_gate": rr["company_self_gate"], "prior_target_bucket_role_gate": rr["target_bucket_role_gate"], "prior_adjacent_or_self_use_exclusion_gate": rr["adjacent_or_self_use_exclusion_gate"], "prior_role_decision": rr["final_role_decision"], "prior_decision_basis_code": rr["decision_basis_code"], "completion_channel": "IMMUTABLE_B001_OFFICIAL_ANNUAL_REPORT_REVIEW_PLUS_BOUNDED_PUBLIC_PRIMARY_QUERY", "query_receipt_path": rel(INDUSTRY / "supplement" / f"NEB2_QRY_{node['code']}_evidence_completion.json"), "actual_query_receipt_path": rel(QUERY_RECEIPT_PATH), "new_source_doc_id": new_doc, "new_evidence_fact_id": new_evf, "company_self_gate": "PASS_NEW_ADJACENT_SERVICE_CONTEXT" if new_doc else rr["company_self_gate"], "target_bucket_direct_role_gate": "FAIL_ADJACENT_SERVICE" if new_doc else rr["target_bucket_role_gate"], "current_public_primary_source_gate": "PASS_ADJACENT_ONLY" if new_doc else "NO_NEW_DIRECT_SOURCE", "locator_gate": "PASS_PAGE_14" if new_doc else "NO_NEW_DIRECT_LOCATOR", "final_state": "HELD_BY_EVIDENCE_GAP", "verification_result": result, "stop_reason": "BOUNDED_ONE_ROUND_COMPLETED_NO_QUALIFIED_PAIR", "coverage_claim": "NONE_INCREMENTAL_EVIDENCED_POOL_ONLY", "schema_version": SCHEMA, "review_status": REVIEW, }) if len(queue_rows) != 50: raise RuntimeError(f"bounded queue expected 50 rows, got {len(queue_rows)}") query_receipts = { "BAT_EQUIPMENT_RECYCLE": { "query": "冻结候选名称 + 锂电设备/电池回收 + 年报/公告/官网", "result": "未定位到可把 10 个候选升级为锂电专用设备或回收直接业务的公开主源；复用前批 2025 年报全页检索与角色裁决。", }, "WND_TURBINE": { "query": "冻结候选名称 + 风电整机/风力发电机组 + 年报/公告/官网", "result": "未定位到可把 10 个候选升级为风电整机直接制造的公开主源；项目、发电或通用设备语境不升级。", }, "WND_TOWER_CABLE_ENGINEERING": { "query": "冻结候选名称 + 风电塔筒/海缆/吊装/工程 + 年报/公告/官网", "result": "未定位到可把 9 个候选升级为目标桶直接业务的公开主源；一般工程、钢管和项目关联不升级。", }, "NUC_OPERATOR": { "query": "冻结候选名称 + 民用核电运营/核电机组 + 年报/公告/官网", "result": "未定位到可把 10 个候选升级为核电运营商的公开主源；其他发电、设备或项目语境不升级。", }, "NUC_ENGINEERING_EPC": { "query": "东方雨虹 + 核电 + EPC/工程 + 年报/可持续发展报告/官网", "result": "东方雨虹官网 2023 可持续发展报告仅支持核电机组厂房地下防水工程材料及施工服务，属于相邻专业服务，不是核电工程 EPC。", "official_url": "https://www.yuhong.com.cn/uploads/soft/240422/1-240422101057.pdf", }, "NUC_NI_CI_EQUIPMENT": { "query": "冻结账本目标桶候选生成", "result": "前批冻结账本在完成排除后无候选，按设计不自由扩池。", }, "NUC_COMPONENT_MATERIAL_ICT": { "query": "冻结候选名称 + 核级部件/材料/仪控/电气 + 年报/公告/官网", "result": "未定位到可把 10 个候选升级为核级直接业务的公开主源；一般工业产品、客户或潜在用途不升级。", }, } for node in NODES: if (node["track"], node["bucket"]) not in QUEUE_CODES: continue receipt = { "query_id": f"NEB2-QRY-{node['code']}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "darkline_mode": "EXTERNAL_PUBLIC_EVIDENCE_COMPLETION", "subindustry_node_code": node["code"], "selection_bucket": node["bucket"], "candidate_codes": QUEUE_CODES[(node["track"], node["bucket"])], "search_window": "ONE_INITIAL_BROAD_PUBLIC_PRIMARY_QUERY_PLUS_PRIOR_ACCEPTED_FULL_PAGE_REPORT_REVIEW", "source_priority": ["exchange_or_statutory_disclosure", "company_official_report_or_IR", "official_project_or_regulator"], "search_engine_usage": "LOCATE_ORIGINAL_ONLY", "market_reverse_scan": "NOT_APPLICABLE", "access_control_bypass": "NO", "nuclear_sensitive_content_saved": "NO", "source_cutoff_at": SOURCE_CUTOFF, "executed_at": EXECUTED_AT, "actual_query_receipt_path": rel(QUERY_RECEIPT_PATH), **query_receipts[node["code"]], "review_status": REVIEW, } write_json(INDUSTRY / "supplement" / f"NEB2_QRY_{node['code']}_evidence_completion.json", receipt) yuhong_raw = INDUSTRY / "raw" / "official_filings" / "NEB2_S-NEB2-YUHONG-SUSTAINABILITY-2023_2024-04-22_yuhong-sustainability-2023.pdf" yuhong_txt = INDUSTRY / "converted" / "official_filings" / "NEB2_S-NEB2-YUHONG-SUSTAINABILITY-2023_yuhong-sustainability-2023.txt" if not yuhong_raw.exists() or not yuhong_txt.exists(): raise FileNotFoundError("Yuhong official source or conversion missing") if not QUERY_PROBE_PATH.exists(): raise FileNotFoundError("REPAIR001 official URL probe receipt missing; run newenergy_batch002_query_probe.py first") probe_payload = json.loads(QUERY_PROBE_PATH.read_text(encoding="utf-8")) if probe_payload.get("status") != "COMPLETED_WITH_ALL_RESPONSE_OR_FAILURE_STATES_RECORDED": raise RuntimeError("official URL probe did not close all response/failure states") probe_by_url = {row["url"]: row for row in probe_payload["url_results"]} QUERY_TERMS = { "BAT_EQUIPMENT_RECYCLE": "锂电设备|电池设备|电池回收|梯次利用|再生利用", "WND_TURBINE": "风电整机|风力发电机组|风机制造|整机销售", "WND_TOWER_CABLE_ENGINEERING": "风电塔筒|风电海缆|风电吊装|风电工程", "NUC_OPERATOR": "核电运营|核电机组|核电发电|商运核电", "NUC_ENGINEERING_EPC": "核电EPC|核电工程总承包|核电工程设计建设", "NUC_COMPONENT_MATERIAL_ICT": "核级部件|核级材料|核电仪控|核电电气", } actual_query_rows: list[dict[str, Any]] = [] for row in queue_rows: source_id = row["new_source_doc_id"] or row["prior_source_doc_id"] if row["new_source_doc_id"]: source_url = "https://www.yuhong.com.cn/uploads/soft/240422/1-240422101057.pdf" returned_object = "东方雨虹2023可持续发展报告；仅返回相邻防水材料及施工服务事实" returned_locator = "page=14;converted_lines=740-764" else: source_url = source_by_id[source_id]["source_url"] returned_object = f"{row['security_code']} {row['security_name']} 2025年度报告" returned_locator = row["prior_locator"] if source_url not in probe_by_url: raise RuntimeError(f"actual probe URL missing for queue pair: {row['queue_item_id']} {source_url}") probe = probe_by_url[source_url] actual_query_rows.append({ "actual_query_id": row["queue_item_id"].replace("QUEUE", "ACTUAL-QUERY"), "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "repair_id": REPAIR_ID, "queue_item_id": row["queue_item_id"], "subindustry_node_code": row["subindustry_node_code"], "track_code": row["track_code"], "selection_bucket": row["selection_bucket"], "company_id": row["company_id"], "security_code": row["security_code"], "security_name": row["security_name"], "query_purpose": "Verify whether the frozen company has current direct business in the target bucket using its official primary source; do not discover candidates.", "complete_query_conditions": f"company_id={row['company_id']};security_code={row['security_code']};track={row['track_code']};bucket={row['selection_bucket']};source_cutoff<={SOURCE_CUTOFF};source_grade=S/A;company_self+target_role+locator required", "query_terms": QUERY_TERMS[row["subindustry_node_code"]], "queried_at": probe["queried_at"], "source_doc_id": source_id, "request_url": source_url, "request_method": probe["request_method"], "response_status": probe["response_status"], "response_result": probe["response_result"], "final_url": probe["final_url"], "content_type": probe["content_type"], "content_length": probe["content_length"], "failure_type": probe["error_type"], "failure_detail": probe["error_detail"], "returned_object": returned_object, "returned_original_locator": returned_locator, "prior_raw_file_path": row["prior_raw_file_path"], "prior_raw_sha256": row["prior_raw_sha256"], "prior_full_page_search_result": row["prior_full_page_search_result"], "prior_context_excerpt": row["prior_context_excerpt"], "mapping_status": "MAPPED_ONE_TO_ONE_TO_QUEUE_PAIR", "qualification_result": row["verification_result"], "termination_reason": "TARGET_ROLE_NOT_ESTABLISHED_AFTER_OFFICIAL_SOURCE_RESPONSE_AND_ACCEPTED_FULL_PAGE_ROLE_REVIEW", "candidate_added": "NO", "schema_version": SCHEMA, "review_status": REVIEW, }) actual_query_rows.append({ "actual_query_id": "NEB2-ACTUAL-QUERY-15-NA", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "repair_id": REPAIR_ID, "queue_item_id": "NOT_APPLICABLE_ZERO_CANDIDATE", "subindustry_node_code": "NUC_NI_CI_EQUIPMENT", "track_code": "NUCLEAR", "selection_bucket": "核岛/常规岛主设备", "company_id": "", "security_code": "", "security_name": "", "query_purpose": "No query is executable because the frozen ledger contains zero candidates for this bucket.", "complete_query_conditions": "frozen_candidate_count=0;free_expansion=DENIED", "query_terms": "NOT_APPLICABLE", "queried_at": probe_payload["queried_at"], "source_doc_id": "", "request_url": "NOT_FOUND_NOT_APPLICABLE_ZERO_CANDIDATE", "request_method": "NOT_APPLICABLE", "response_status": "NOT_APPLICABLE", "response_result": "NOT_APPLICABLE_ZERO_CANDIDATE", "final_url": "", "content_type": "", "content_length": "", "failure_type": "", "failure_detail": "", "returned_object": "NONE_ZERO_CANDIDATE", "returned_original_locator": "FROZEN_LEDGER_EMPTY_BUCKET", "prior_raw_file_path": "", "prior_raw_sha256": "", "prior_full_page_search_result": "NOT_APPLICABLE", "prior_context_excerpt": "", "mapping_status": "MAPPED_TO_EMPTY_BUCKET_NOT_TO_PAIR", "qualification_result": "NO_CANDIDATE_NO_QUERY_EXECUTED", "termination_reason": "FROZEN_LEDGER_NO_CANDIDATE_DO_NOT_EXPAND", "candidate_added": "NO", "schema_version": SCHEMA, "review_status": REVIEW, }) if len(actual_query_rows) != 51: raise RuntimeError(f"actual query receipt expected 51 rows, got {len(actual_query_rows)}") write_csv(QUERY_RECEIPT_PATH, actual_query_rows) yuhong_evf = { "evidence_fact_id": "EVF-NE-B002-ADJ-002271", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "doc_id": "S-NEB2-YUHONG-SUSTAINABILITY-2023", "industry_case": "新能源案例", "industry_id": "IND-NEWENERGY", "subindustry_id": "NUC_ENGINEERING_EPC", "company_id": "CN_A:SZSE:002271", "track_code": "NUCLEAR", "chain_node_id": "工程/EPC", "subject_type": "COMPANY_BOUNDARY", "subject_id": "CN_A:SZSE:002271", "source_text_path": rel(yuhong_txt), "raw_pool_path": rel(yuhong_raw.parent) + "/", "raw_file_sha256": sha256(yuhong_raw), "source_page": "14", "source_table_id": "", "source_sentence_index": "", "locator_type": "PDF_PAGE", "locator_value": "page=14;converted_lines=740-764", "evidence_text": "东方雨虹报告披露其为核电机组厂房地下防水工程提供材料及施工服务。", "evidence_type": "COMPANY_OFFICIAL_SUSTAINABILITY_REPORT", "statement_type": "FACT", "business_dimension": "BOUNDARY_EXCLUSION", "research_dimension": "COMPANY_ROLE", "numeric_value_raw": "", "metric_candidate_name": "", "metric_candidate_unit": "", "metric_period": "2023", "metric_date": "2023-12-31", "geography": "CN", "original_qualifier": "仅支持防水材料及施工服务，不支持核电工程EPC资格", "related_company_id": "", "viewpoint_id": "", "darkline_signal_flag": "NO", "confidence_level": "HIGH", "conclusion_strength": "DIRECT_BOUNDARY_FACT", "sensitivity_screen": "LEGAL_PUBLIC_CIVIL_NUCLEAR_HIGH_LEVEL_ONLY", "contradicts_evidence_fact_id": "", "normalization_status": "NORMALIZED", "processing_status": "READY", "data_status": "VERIFIED_PUBLIC_BOUNDARY_ONLY", "schema_version": SCHEMA, "review_status": REVIEW, } write_csv(INDUSTRY / "evidence" / "evidence_fact_table_BATCH002.csv", [yuhong_evf]) source_document_rows = [{ "doc_id": "S-NEB2-YUHONG-SUSTAINABILITY-2023", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "doc_type": "COMPANY_OFFICIAL_SUSTAINABILITY_REPORT", "title": "东方雨虹2023可持续发展报告", "source_org": "北京东方雨虹防水技术股份有限公司", "author": "", "publish_date": "2024-04-22", "collected_at": EXECUTED_AT, "source_url": "https://www.yuhong.com.cn/uploads/soft/240422/1-240422101057.pdf", "industry_case": "新能源案例", "industry_id": "IND-NEWENERGY", "subindustry_id": "NUC_ENGINEERING_EPC", "company_id": "CN_A:SZSE:002271", "raw_pool_path": rel(yuhong_raw.parent) + "/", "raw_file_path": rel(yuhong_raw), "converted_text_path": rel(yuhong_txt), "converted_markdown_path": "", "file_sha256": sha256(yuhong_raw), "file_name": yuhong_raw.name, "file_size": yuhong_raw.stat().st_size, "detected_type": "PDF", "source_language": "zh-CN", "public_access_basis": "COMPANY_OFFICIAL_PUBLIC_URL", "access_status": "HTTP_200_PUBLIC", "source_level": "A", "sensitivity_screen": "CIVIL_NUCLEAR_HIGH_LEVEL_PUBLIC_BOUNDARY_ONLY", "legal_access_note": "公开公司报告；未绕过访问控制；只引用角色边界所需最小事实。", "doc_status": "ARCHIVED", "processing_status": "TEXT_CONVERTED_INDEXED_BOUNDARY_EVIDENCE", "schema_version": SCHEMA, "review_status": REVIEW, }] write_csv(INDUSTRY / "manifest" / "source_document_BATCH002.csv", source_document_rows) used_evidence_ids = {node["market_evf"] for node in NODES} | {r["evidence_fact_id"] for r in incremental} used_source_ids = {evidence_by_id[eid]["doc_id"] for eid in used_evidence_ids} for row in queue_rows: if row["prior_source_doc_id"]: used_source_ids.add(row["prior_source_doc_id"]) used_source_ids.add("S-NEB2-YUHONG-SUSTAINABILITY-2023") missing_sources = sorted(s for s in used_source_ids if s != "S-NEB2-YUHONG-SUSTAINABILITY-2023" and s not in source_by_id) if missing_sources: raise RuntimeError(f"missing baseline source docs: {missing_sources}") conversion_rows: list[dict[str, Any]] = [] input_rows: list[dict[str, Any]] = [] source_gap_audit_rows: list[dict[str, Any]] = [] for idx, source_id in enumerate(sorted(used_source_ids), 1): if source_id == "S-NEB2-YUHONG-SUSTAINABILITY-2023": doc = source_document_rows[0] conv = { "raw_pool_path": doc["raw_pool_path"], "raw_file_path": doc["raw_file_path"], "raw_file_sha256": doc["file_sha256"], "detected_type": "PDF", "conversion_method": "PYPDF_TEXT_EXTRACT", "parameters_summary": "all_pages;utf8;page_markers", "converted_text_path": doc["converted_text_path"], "converted_markdown_path": "", "converted_path": doc["converted_text_path"], "converted_sha256": sha256(yuhong_txt), "page_or_duration_count": "39", "status": "CONVERTED_INDEXED_BOUNDARY_EVIDENCE", "error_code": "", "error_summary": "", "created_at": EXECUTED_AT, } source_url = doc["source_url"] source_level = "A" public_basis = doc["public_access_basis"] sensitivity = doc["sensitivity_screen"] raw_path = doc["raw_file_path"] raw_hash = doc["file_sha256"] mode = "NEW_BATCH002_SOURCE" else: doc = source_by_id[source_id] prior_conv = conversion_by_source.get(source_id) if not prior_conv: raise RuntimeError(f"missing baseline conversion: {source_id}") conv = { "raw_pool_path": prior_conv["raw_pool_path"], "raw_file_path": prior_conv["raw_file_path"], "raw_file_sha256": prior_conv["raw_file_sha256"], "detected_type": prior_conv["detected_type"], "conversion_method": "REUSE_ACCEPTED_B001_" + prior_conv["conversion_method"], "parameters_summary": "immutable accepted conversion; no recopy", "converted_text_path": prior_conv["converted_text_path"], "converted_markdown_path": prior_conv["converted_markdown_path"], "converted_path": prior_conv["converted_path"], "converted_sha256": prior_conv["converted_sha256"], "page_or_duration_count": prior_conv["page_or_duration_count"], "status": "REUSED_IMMUTABLE_ACCEPTED_CONVERSION", "error_code": "", "error_summary": "", "created_at": EXECUTED_AT, } source_url = doc["source_url"] source_level = doc["source_level"] public_basis = doc["public_access_basis"] sensitivity = doc["sensitivity_screen"] raw_path = doc["raw_file_path"] raw_hash = doc["file_sha256"] mode = "IMMUTABLE_B001_REFERENCE" conversion_rows.append({ "conversion_id": f"NEB2-CONV-{idx:04d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "source_doc_id": source_id, **conv, "schema_version": SCHEMA, "review_status": REVIEW, }) input_rows.append({ "input_item_id": f"NEB2-INPUT-{idx:04d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "source_doc_id": source_id, "include_decision": "INCLUDE", "exclude_reason": "", "raw_pool_path": conv["raw_pool_path"], "raw_file_path": raw_path, "raw_file_sha256": raw_hash, "source_url": source_url, "source_level": source_level, "public_access_basis": public_basis, "sensitivity_screen": sensitivity, "processing_status": mode, "schema_version": SCHEMA, "review_status": REVIEW, }) source_gap_audit_rows.append({ "source_gap_audit_id": f"NEB2-SGA-{idx:04d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "source_doc_id": source_id, "source_received_flag": "YES", "raw_archived_flag": "YES_IMMUTABLE_REFERENCE" if mode.startswith("IMMUTABLE") else "YES_NEW_BATCH002", "converted_flag": "YES", "indexed_flag": "YES", "evidence_linked_flag": "YES_STRONG_FACT_LINKED" if source_id in {evidence_by_id[e]["doc_id"] for e in used_evidence_ids} or source_id.startswith("S-NEB2") else "YES_QUEUE_ADJUDICATION_LINKED", "raw_pool_path": conv["raw_pool_path"], "raw_file_sha256": raw_hash, "gap_type": "NONE", "impact": "NONE", "status": "COMPLETE", "schema_version": SCHEMA, "review_status": REVIEW, }) write_csv(INDUSTRY / "manifest" / "conversion_status_BATCH002.csv", conversion_rows) write_csv(INDUSTRY / "manifest" / "input_manifest_BATCH002.csv", input_rows) write_csv(INDUSTRY / "manifest" / "source_gap_audit_BATCH002.csv", source_gap_audit_rows) incremental_ledger_rows: list[dict[str, Any]] = [] for idx, row in enumerate(sorted(incremental, key=lambda r: (r["track_code"], r["selection_bucket"])), 1): incremental_ledger_rows.append({ "candidate_id": f"NEB2-CAND-INCR-{idx:02d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "company_id": row["company_id"], "security_code": row["security_code"], "security_name": row["security_name"], "exchange_code": row["exchange_code"], "track_code": row["track_code"], "subindustry_node_code": row["node_code"], "selection_bucket": row["selection_bucket"], "predecessor_candidate_id": row["candidate_id"], "predecessor_state": row["candidate_state"], "predecessor_selection_rank": row["selection_rank"], "batch002_state": "INCLUDED_INCREMENTAL", "batch002_rank_in_node": "1", "source_doc_id": row["direct_business_source_id"], "evidence_fact_id": row["evidence_fact_id"], "locator": row["direct_business_locator"], "company_self_gate": "PASS", "target_bucket_direct_role_gate": "PASS", "current_public_primary_source_gate": "PASS", "locator_gate": "PASS", "mechanical_reason": "FIRST_ELIGIBLE_NOT_SELECTED_BATCH001_BY_ACCEPTED_SELECTION_RANK", "coverage_claim": "NONE_INCREMENTAL_EVIDENCED_POOL_ONLY", "schema_version": SCHEMA, "review_status": REVIEW, }) for row in queue_rows: incremental_ledger_rows.append({ "candidate_id": row["queue_item_id"].replace("QUEUE", "CAND-HELD"), "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "company_id": row["company_id"], "security_code": row["security_code"], "security_name": row["security_name"], "exchange_code": "SSE" if row["security_code"].startswith("6") else "SZSE", "track_code": row["track_code"], "subindustry_node_code": row["subindustry_node_code"], "selection_bucket": row["selection_bucket"], "predecessor_candidate_id": ledger_by_pair[(row["company_id"], row["track_code"], row["selection_bucket"])]["candidate_id"], "predecessor_state": row["prior_candidate_state"], "predecessor_selection_rank": "", "batch002_state": row["final_state"], "batch002_rank_in_node": "", "source_doc_id": row["new_source_doc_id"] or row["prior_source_doc_id"], "evidence_fact_id": row["new_evidence_fact_id"], "locator": row["locator_gate"], "company_self_gate": row["company_self_gate"], "target_bucket_direct_role_gate": row["target_bucket_direct_role_gate"], "current_public_primary_source_gate": row["current_public_primary_source_gate"], "locator_gate": row["locator_gate"], "mechanical_reason": row["verification_result"], "coverage_claim": row["coverage_claim"], "schema_version": SCHEMA, "review_status": REVIEW, }) write_csv(INDUSTRY / "extracted" / "batch002_evidence_completion_queue.csv", queue_rows) write_csv(INDUSTRY / "extracted" / "company_incremental_candidate_ledger_BATCH002.csv", incremental_ledger_rows) market_evfs = {eid: evidence_by_id[eid] for eid in {n["market_evf"] for n in NODES}} ATOMIC_METRICS = { "EVF-MKT-BAT-01": [ ("全国锂电池总产量", ">473", "GWh", "2025-01-01", "2025-04-30", "工信部测算；期间产量"), ], "EVF-MKT-BAT-02": [ ("储能型锂电池产量", ">110", "GWh", "2025-01-01", "2025-04-30", "储能应用；期间产量"), ("动力电池装车量", "~184", "GWh", "2025-01-01", "2025-04-30", "动力应用；装车口径"), ], "EVF-MKT-BAT-03": [ ("电池级碳酸锂均价", "7.4", "万元/吨", "2025-01-01", "2025-04-30", "电池级；期间均价"), ("微粉级氢氧化锂均价", "7.6", "万元/吨", "2025-01-01", "2025-04-30", "微粉级；期间均价"), ], "EVF-MKT-SOL-01": [ ("全国新增光伏装机", "317", "GW", "2025-01-01", "2025-12-31", "并网新增装机"), ("全国累计光伏装机", "1200", "GW", "2025-01-01", "2025-12-31", "年末累计并网装机"), ], "EVF-MKT-SOL-02": [ ("全国光伏发电量", "1.17", "万亿千瓦时", "2025-01-01", "2025-12-31", "年度发电量"), ("全国光伏利用率", "95", "%", "2025-01-01", "2025-12-31", "年度利用率"), ], "EVF-MKT-WIND-01": [ ("全国新增风电装机", "120", "GW", "2025-01-01", "2025-12-31", "并网新增装机"), ("全国累计风电装机", "640", "GW", "2025-01-01", "2025-12-31", "年末累计并网装机"), ], "EVF-MKT-WIND-02": [ ("全国风电发电量", "1.13", "万亿千瓦时", "2025-01-01", "2025-12-31", "年度发电量"), ("全国风电利用率", "94", "%", "2025-01-01", "2025-12-31", "年度利用率"), ], "EVF-MKT-NUC-01": [ ("全国投入商业运营核电机组", "59", "台", "2025-01-01", "2025-12-31", "不含台湾省；商运机组"), ("全国商运核电机组上网电量", "4389.29", "亿千瓦时", "2025-01-01", "2025-12-31", "商运机组；年度上网电量"), ], "EVF-MKT-NUC-2024": [ ("全国运行核电机组", "57", "台", "2024-01-01", "2024-12-31", "年末运行机组"), ("全国运行核电装机", "59431.7", "MWe", "2024-01-01", "2024-12-31", "年末运行装机"), ("全国核电发电量", "4451.75", "亿千瓦时", "2024-01-01", "2024-12-31", "年度发电量"), ("全国核电上网电量", "4184", "亿千瓦时", "2024-01-01", "2024-12-31", "年度上网电量"), ], } selected_by_node = {r["node_code"]: r for r in incremental} baseline_selected_by_bucket: dict[tuple[str, str], list[dict[str, str]]] = defaultdict(list) for row in ledger: if row["candidate_state"] in {"INCLUDED_T1", "INCLUDED_T2"}: baseline_selected_by_bucket[(row["track_code"], row["selection_bucket"])].append(row) for rows in baseline_selected_by_bucket.values(): rows.sort(key=lambda r: int(r["selection_rank"])) subindustry_rows: list[dict[str, Any]] = [] scope_rows: list[dict[str, Any]] = [] technology_rows: list[dict[str, Any]] = [] metric_rows: list[dict[str, Any]] = [] risk_rows: list[dict[str, Any]] = [] classification_rows: list[dict[str, Any]] = [] gap_rows: list[dict[str, Any]] = [] next_action_rows: list[dict[str, Any]] = [] for node in NODES: ev = market_evfs[node["market_evf"]] selected = selected_by_node.get(node["code"]) queue_count = len(QUEUE_CODES.get((node["track"], node["bucket"]), [])) subindustry_rows.append({ "node_row_id": f"NEB2-NODE-{node['n']:02d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "node_order": node["n"], "track_code": node["track"], "subindustry_node_code": node["code"], "subindustry_name": node["name"], "selection_bucket": node["bucket"], "primary_region": "MAINLAND_CHINA", "jurisdiction": "CN", "global_comparator": "SEPARATE_CONTEXT_ONLY", "scope_boundary": node["boundary"], "value_chain_role": node["role"], "technology_framework": node["route"], "market_evidence_fact_id": node["market_evf"], "incremental_company_count": 1 if selected else 0, "bounded_queue_count": queue_count, "node_company_status": "INCREMENTAL_INCLUDED" if selected else "HELD_BY_EVIDENCE_GAP", "coverage_claim": "NONE_INCREMENTAL_EVIDENCED_POOL_ONLY", "schema_version": SCHEMA, "review_status": REVIEW, }) scope_rows.append({ "scope_row_id": f"NEB2-SCOPE-NODE-{node['n']:02d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "object_type": "SUBINDUSTRY_NODE", "object_id": node["code"], "track_code": node["track"], "chain_node": node["name"], "scope_type": "CORE_INDUSTRY", "inclusion_reason": node["boundary"], "exclusion_reason": "NOT_APPLICABLE_FOR_CORE_INDUSTRY_INCLUDED_SCOPE", "scope_as_of": SOURCE_CUTOFF, "primary_region": "MAINLAND_CHINA", "jurisdiction": "CN", "source_doc_id": ev["doc_id"], "evidence_fact_id": node["market_evf"], "artifact_id": "", "scope_version": "BATCH002", "coverage_claim": "NONE", "schema_version": SCHEMA, "review_status": REVIEW, }) route_metric = ATOMIC_METRICS[node["market_evf"]][0] technology_rows.append({ "technology_route_id": f"NEB2-ROUTE-{node['n']:02d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "track_code": node["track"], "subindustry_node_code": node["code"], "route_name": node["route"].split("；", 1)[0], "product_or_system": node["name"], "technical_stage": "PUBLIC_ROUTE_FRAMEWORK_NOT_PRODUCT_CERTIFICATION", "commercial_stage": "COMMERCIAL_INDUSTRY_CONTEXT_ONLY_NOT_COMPANY_STAGE", "performance_metric": route_metric[0], "metric_value": route_metric[1], "unit": route_metric[2], "cost_boundary": "GAP_NOT_ESTABLISHED_AT_ROUTE_LEVEL", "validation_or_certification": "GAP_NOT_ESTABLISHED", "alternative_route": "GAP_NOT_COMPARABLY_EVIDENCED", "limiting_factor": node["risk"], "route_description": node["route"], "value_chain_stage": node["name"], "region": "MAINLAND_CHINA", "metric_as_of": SOURCE_CUTOFF, "source_doc_id": ev["doc_id"], "evidence_fact_id": node["market_evf"], "artifact_id": "", "evidence_boundary": "行业需求/运营事实只作路线背景，不证明单一公司份额或盈利", "schema_version": SCHEMA, "review_status": REVIEW, }) for atomic_index, (metric_name, metric_value, unit, period_start, period_end, basis) in enumerate(ATOMIC_METRICS[node["market_evf"]], 1): is_price_metric = "均价" in metric_name if "装机" in metric_name or "机组" in metric_name: capacity_status = "GRID_CONNECTED_OR_COMMERCIAL_OPERATION_AS_STATED" elif "产量" in metric_name or "装车量" in metric_name: capacity_status = "ACTUAL_PERIOD_OUTPUT_NOT_CAPACITY" else: capacity_status = "NOT_APPLICABLE_NON_CAPACITY_METRIC" metric_rows.append({ "market_metric_id": f"NEB2-METRIC-{node['n']:02d}-{atomic_index:02d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "metric_name": metric_name, "object_id": node["code"], "track_code": node["track"], "value": metric_value, "unit": unit, "period_start": period_start, "period_end": period_end, "region": "MAINLAND_CHINA", "jurisdiction": "CN", "price_or_volume_basis": basis, "source_doc_id": ev["doc_id"], "evidence_fact_id": node["market_evf"], "nominal_or_real": "NOMINAL_PUBLIC_CURRENCY_VALUE" if is_price_metric else "ACTUAL_PUBLIC_STATISTIC", "spot_or_contract": "UNSPECIFIED_PUBLIC_AVERAGE_NOT_SPOT_OR_CONTRACT" if is_price_metric else "NOT_APPLICABLE_NON_PRICE", "tax_basis": "UNSPECIFIED_BY_SOURCE" if is_price_metric else "NOT_APPLICABLE_NON_PRICE", "capacity_status": capacity_status, "source_method": "OFFICIAL_GOVERNMENT_STATISTIC" if ev["doc_id"].startswith("S-MARKET-") else "STATUTORY_FILING_TRANSCRIBED_INDUSTRY_STATISTIC", "comparability_note": "仅限同地域、同期间、同单位、同阶段及原限定语；不得外推单一公司份额或盈利。", "artifact_id": "", "qualifier": "同一行业事实可作多个节点的共同需求/运营背景，不作跨节点因果外推", "schema_version": SCHEMA, "review_status": REVIEW, }) risk_rows.append({ "catalyst_risk_id": f"NEB2-RISK-{node['n']:02d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "object_id": node["code"], "track_code": node["track"], "event_or_risk_type": "RISK", "source_as_of": SOURCE_CUTOFF, "statement_type": "INFERENCE", "trigger_condition": "后续同口径官方统计、法定披露或项目状态显著偏离本批公开事实边界", "transmission_path": f"公开行业/运营事实变化 -> {node['name']}节点供需、技术或项目约束变化", "affected_metric_or_object": node["code"], "time_window": "AFTER_SOURCE_CUTOFF_FUTURE_MONITORING", "risk_statement": node["risk"], "observable_indicator": "后续同口径官方统计、法定披露或项目状态", "alternative_explanation": "行业总量变化可能由口径、库存、项目节奏或结构变化造成", "invalidation_condition": "后续主源证明角色、口径或阶段与本批判断不一致", "current_status": "OPEN_MONITORING_NOT_PREDICTION", "source_doc_id": ev["doc_id"], "evidence_fact_id": node["market_evf"], "artifact_id": "", "schema_version": SCHEMA, "review_status": REVIEW, }) classification_rows.append({ "classification_id": f"NEB2-CLASS-NODE-{node['n']:02d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "subject_type": "SUBINDUSTRY_NODE", "subject_id": node["code"], "industry_id": "IND-NEWENERGY", "subindustry_id": node["code"], "company_id": "", "track_code": node["track"], "chain_node_id": node["name"], "scope_type": "CORE_INDUSTRY", "classification_reason": node["boundary"], "source_doc_id": ev["doc_id"], "evidence_fact_id": node["market_evf"], "artifact_id": "", "raw_pool_path": ev["raw_pool_path"], "raw_file_sha256": ev["raw_file_sha256"], "data_status": "NODE_FROZEN", "schema_version": SCHEMA, "review_status": REVIEW, }) if not selected: gap_id = f"NEB2-GAP-COMPANY-{node['n']:02d}" gap_rows.append({ "gap_id": gap_id, "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "subject_id": node["code"], "track_code": node["track"], "research_dimension": "INCREMENTAL_COMPANY_DIRECT_ROLE", "missing_information": f"{node['name']}节点在冻结余额与最多10个补证对象中未找到通过全部资格gate的新增公司。", "attempted_scope": ( f"B001 eligible balance=0; bounded queue={queue_count}; one public-primary completion round" if queue_count else "B001 eligible balance=0; frozen bounded queue=0; no executable query; free expansion denied" ), "impact": "本节点不新增公司映射；不影响行业边界与前批已接受企业。", "status": "HELD", "source_doc_id": "S-NEB2-YUHONG-SUSTAINABILITY-2023" if node["code"] == "NUC_ENGINEERING_EPC" else "", "evidence_fact_id": "EVF-NE-B002-ADJ-002271" if node["code"] == "NUC_ENGINEERING_EPC" else "", "artifact_id": "ART-REPAIR-0015", "schema_version": SCHEMA, "review_status": REVIEW, }) next_action_rows.append({ "action_id": f"NEB2-ACTION-{node['n']:02d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "subject_id": node["code"], "action_type": "MONITOR_FUTURE_PRIMARY_DISCLOSURE", "trigger_gap_id": gap_id, "action_description": "仅在后续定期报告或正式公告出现公司自身、目标桶直接业务和可定位原句时重新评估；不得用行情或题材补数。", "priority": "NORMAL", "owner": "case_analysis.analyst.new_energy", "status": "OPEN_NOT_IN_CURRENT_BATCH", "schema_version": SCHEMA, "review_status": REVIEW, }) queue_by_pair_b2 = {(r["company_id"], r["track_code"], r["selection_bucket"]): r for r in queue_rows} for idx, row in enumerate(incremental_ledger_rows, 1): is_included = row["batch002_state"] == "INCLUDED_INCREMENTAL" if is_included: scope_type = "CORE_INDUSTRY" inclusion_reason = row["mechanical_reason"] exclusion_reason = "NOT_APPLICABLE_FOR_INCLUDED_SCOPE" trace_ev = evidence_by_id[row["evidence_fact_id"]] raw_pool_path = trace_ev["raw_pool_path"] raw_file_sha256 = trace_ev["raw_file_sha256"] else: queue_row = queue_by_pair_b2[(row["company_id"], row["track_code"], row["selection_bucket"])] scope_type = "ADJACENT_DOWNSTREAM" if queue_row["prior_role_decision"] == "HELD_KEYWORD_CONTEXT_ONLY_AFTER_FULL_PAGE_SEARCH" or row["evidence_fact_id"] == "EVF-NE-B002-ADJ-002271" else "FALSE_THEME_OR_NOISE" inclusion_reason = "NOT_APPLICABLE_FOR_EXCLUDED_OR_ADJACENT_SCOPE" exclusion_reason = row["mechanical_reason"] if row["source_doc_id"] == "S-NEB2-YUHONG-SUSTAINABILITY-2023": raw_pool_path = rel(yuhong_raw.parent) + "/" raw_file_sha256 = sha256(yuhong_raw) else: source_trace = source_by_id[row["source_doc_id"]] raw_pool_path = source_trace["raw_pool_path"] raw_file_sha256 = source_trace["file_sha256"].upper() scope_rows.append({ "scope_row_id": f"NEB2-SCOPE-CAND-{idx:03d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "object_type": "COMPANY_TRACK_PAIR", "object_id": f"{row['company_id']}|{row['track_code']}", "track_code": row["track_code"], "chain_node": row["selection_bucket"], "scope_type": scope_type, "inclusion_reason": inclusion_reason, "exclusion_reason": exclusion_reason, "scope_as_of": SOURCE_CUTOFF, "primary_region": "MAINLAND_CHINA", "jurisdiction": "CN", "source_doc_id": row["source_doc_id"], "evidence_fact_id": row["evidence_fact_id"], "artifact_id": "" if row["evidence_fact_id"] else "ART-REPAIR-0015", "scope_version": "BATCH002", "coverage_claim": "NONE", "schema_version": SCHEMA, "review_status": REVIEW, }) classification_rows.append({ "classification_id": f"NEB2-CLASS-CAND-{idx:03d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "subject_type": "COMPANY_DIRECT_BUSINESS" if is_included else "COMPANY_EVIDENCE_GAP", "subject_id": row["company_id"], "industry_id": "IND-NEWENERGY", "subindustry_id": row["subindustry_node_code"], "company_id": row["company_id"], "track_code": row["track_code"], "chain_node_id": row["selection_bucket"], "scope_type": scope_type, "classification_reason": row["mechanical_reason"], "source_doc_id": row["source_doc_id"], "evidence_fact_id": row["evidence_fact_id"], "artifact_id": "" if row["evidence_fact_id"] else "ART-REPAIR-0015", "raw_pool_path": raw_pool_path, "raw_file_sha256": raw_file_sha256, "data_status": row["batch002_state"], "schema_version": SCHEMA, "review_status": REVIEW, }) write_csv(INDUSTRY / "extracted" / "subindustry_node_matrix_BATCH002.csv", subindustry_rows) write_csv(INDUSTRY / "extracted" / "newenergy_scope_matrix_BATCH002.csv", scope_rows) write_csv(INDUSTRY / "extracted" / "newenergy_technology_route_matrix_BATCH002.csv", technology_rows) write_csv(INDUSTRY / "extracted" / "newenergy_supply_demand_price_metric_BATCH002.csv", metric_rows) write_csv(INDUSTRY / "extracted" / "newenergy_catalyst_risk_register_BATCH002.csv", risk_rows) write_csv(INDUSTRY / "extracted" / "classification_summary_BATCH002.csv", classification_rows) write_csv(INDUSTRY / "extracted" / "unresolved_data_gap_BATCH002.csv", gap_rows) write_csv(INDUSTRY / "extracted" / "next_action_list_BATCH002.csv", next_action_rows) exposure_rows: list[dict[str, Any]] = [] for idx, row in enumerate(sorted(incremental, key=lambda r: (r["track_code"], r["node_code"])), 1): ev = evidence_by_id[row["evidence_fact_id"]] exposure_rows.append({ "company_exposure_id": f"NEB2-EXPOSURE-{idx:02d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "company_id": row["company_id"], "security_code": row["security_code"], "security_name": row["security_name"], "track_code": row["track_code"], "chain_node": row["selection_bucket"], "business_segment": row["selection_bucket"], "product_or_service": ev["chain_node_id"], "revenue_exposure": "GAP_NOT_SEPARATELY_DISCLOSED_IN_CITED_FACT", "profit_exposure": "GAP_NOT_SEPARATELY_DISCLOSED_IN_CITED_FACT", "customer_or_project_stage": "CURRENT_DIRECT_BUSINESS_AS_DISCLOSED_IN_2025_ANNUAL_REPORT", "capacity_or_delivery_status": "GAP_NOT_STANDARDIZED_IN_CITED_FACT", "capex": "GAP_NOT_SEPARATELY_DISCLOSED_IN_CITED_FACT", "order_or_contract_status": "GAP_NOT_USED_FOR_QUALIFICATION", "competitive_position": "NOT_ASSESSED_NO_RANKING_CLAIM", "reporting_period": "2025", "exposure_description": company_evidence_summary(row, ev), "exposure_strength": "DIRECT_PUBLIC_PRIMARY_SOURCE", "selection_method": "FIRST_ELIGIBLE_NOT_SELECTED_BATCH001_BY_ACCEPTED_RANK", "source_doc_id": ev["doc_id"], "evidence_fact_id": row["evidence_fact_id"], "artifact_id": "", "coverage_claim": "NONE", "schema_version": SCHEMA, "review_status": REVIEW, }) write_csv(INDUSTRY / "extracted" / "newenergy_company_exposure_matrix_BATCH002.csv", exposure_rows) project_event_rows: list[dict[str, Any]] = [] event_source = read_csv(INDUSTRY / "extracted" / "newenergy_project_capacity_event.csv") for idx, row in enumerate(event_source, 1): is_national = row["subject_name"].startswith("全国") or row["subject_name"].startswith("储能型") owner_entity_id = "CN_NATIONAL_AGGREGATE" if is_national else "CN_A:SZSE:003816" construction_status = "NOT_APPLICABLE_AGGREGATE_OPERATING_FACT" commissioning_status = row["project_or_metric_stage"] project_event_rows.append({ "project_event_id": f"NEB2-PROJECT-{idx:02d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "project_or_line_id": f"B001-REF-{row['event_id']}", "owner_entity_id": owner_entity_id, "track_code": row["track_code"], "region": row["region"], "event_type": row["event_type"], "event_date": row["event_date"], "subject_name": row["subject_name"], "project_or_metric_stage": row["project_or_metric_stage"], "announced_capacity": "NOT_APPLICABLE_OBSERVED_EFFECTIVE_METRIC", "effective_capacity": row["capacity_or_count_value"], "unit": row["unit"], "investment_amount": "GAP_NOT_DISCLOSED_FOR_AGGREGATE_FACT", "construction_status": construction_status, "commissioning_status": commissioning_status, "expected_or_actual": "ACTUAL_PUBLIC_STATISTIC_OR_REPORTED_PORTFOLIO", "dependency": "IMMUTABLE_B001_ACCEPTED_EVENT_AND_ORIGINAL_SOURCE", "jurisdiction": row["jurisdiction"], "source_doc_id": row["source_id"], "evidence_fact_id": row["evidence_fact_id"], "artifact_id": "", "status_boundary": "IMMUTABLE_B001_ACCEPTED_EVENT_RECONTEXTUALIZED_PER_NODE; NO_NEW_EVENT_CLAIM", "schema_version": SCHEMA, "review_status": REVIEW, }) write_csv(INDUSTRY / "extracted" / "newenergy_project_capacity_event_BATCH002.csv", project_event_rows) source_gap_rows = [] for node in NODES: if node["code"] not in {n["code"] for n in NODES if (n["track"], n["bucket"]) in QUEUE_CODES}: continue source_gap_rows.append({ "source_gap_id": f"NEB2-SOURCE-GAP-{node['n']:02d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "subject_id": node["code"], "track_code": node["track"], "gap_type": "NO_QUALIFIED_INCREMENTAL_COMPANY_PRIMARY_SOURCE", "attempted_candidate_count": len(QUEUE_CODES[(node["track"], node["bucket"])]), "query_receipt_path": rel(INDUSTRY / "supplement" / f"NEB2_QRY_{node['code']}_evidence_completion.json"), "impact": "INCREMENTAL_COMPANY_COUNT_ZERO_FOR_NODE", "status": "HELD", "schema_version": SCHEMA, "review_status": REVIEW, }) write_csv(INDUSTRY / "extracted" / "source_gap_BATCH002.csv", source_gap_rows) case_maps: list[dict[str, Any]] = [] map_counter = 0 def mapped_line(output_path: Path, anchor: str, text: str, evidence_fact_id: str, limit: str = "") -> str: global map_counter map_counter += 1 case_maps.append({ "conclusion_evidence_map_id": f"NEB2-CEM-{map_counter:04d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "conclusion_id": f"NEB2-CONC-{map_counter:04d}", "output_path": rel(output_path), "section_anchor": anchor, "conclusion_text": text, "conclusion_strength": "DIRECT_FACT" if not limit else "DIRECT_FACT_WITH_LIMIT", "evidence_fact_id": evidence_fact_id, "support_type": "SUPPORT", "contradiction_or_limit": limit, "schema_version": SCHEMA, "review_status": REVIEW, }) return f'\n{text}' def md_meta(view_type: str, title: str) -> str: return ( f"# {title}\n\n" f"- task_id: `{TASK_ID}`\n- case_id: `{CASE_ID}`\n- batch_id: `{BATCH_ID}`\n- run_id: `{RUN_ID}`\n" f"- view_type: `{view_type}`\n- primary_region: `MAINLAND_CHINA`\n- source_cutoff_at: `{SOURCE_CUTOFF}`\n" f"- review_status: `{REVIEW}`\n- coverage_claim: `NONE_INCREMENTAL_EVIDENCED_POOL_ONLY`\n" ) for node in NODES: out_dir = CASE / "outputs" / "核心文档" / "子行业深化" / node["folder"] industry_doc = out_dir / "子行业研究_BATCH002.md" company_doc = out_dir / "相关企业_BATCH002.md" ev = market_evfs[node["market_evf"]] market_anchor = f"node-{node['n']:02d}-market-fact" market_text = ev["evidence_text"] industry_body = md_meta("SUBINDUSTRY_RESEARCH", f"{node['n']:02d} {node['name']}：子行业研究") industry_body += ( "\n## 范围定义\n\n" + node["boundary"] + "\n\n## 产业链角色\n\n" + node["role"] + "\n\n## 技术与产品路线\n\n" + node["route"] + "\n\n## 供需、项目或运营口径\n\n" + mapped_line(industry_doc, market_anchor, market_text, node["market_evf"], "行业级背景，不外推单一公司份额、订单或盈利。") + f"\n\n该事实仅按 `{ev['metric_period']}`、`{ev['geography']}` 与原限定语使用；跨阶段、跨应用或跨单位比较一律禁止。" "\n\n## 商业口径\n\n收入、订单、产能、产量、出货、装机、并网、发电量与上网电量分别记录；不能用一个口径替代另一个。" "\n\n## 风险、冲突与 GAP\n\n" + node["risk"] + "\n\n本节点不含估值、交易、收益、推荐或完整覆盖结论。核电节点只保留民用公开高层信息。\n" ) write_text(industry_doc, industry_body) selected = selected_by_node.get(node["code"]) baseline_companies = baseline_selected_by_bucket.get((node["track"], node["bucket"]), []) company_body = md_meta("RELATED_COMPANIES", f"{node['n']:02d} {node['name']}：相关企业") company_body += "\n## 前批已接受基础企业\n\n" if baseline_companies: company_body += "\n".join( f"- `{r['security_code']}` {r['security_name']}：`{r['candidate_state']}`；前批排名 {r['selection_rank']}。" for r in baseline_companies ) else: company_body += "- 前批该桶没有已接受企业。" company_body += "\n\n## 本批增量企业\n\n" if selected: cev = evidence_by_id[selected["evidence_fact_id"]] company_text = f"{selected['security_code']} {selected['security_name']}：{company_evidence_summary(selected, cev)}" company_body += mapped_line( company_doc, f"node-{node['n']:02d}-company-fact", company_text, selected["evidence_fact_id"], "只支持公司—赛道直接业务映射，不表示质量、排名、估值或投资建议。", ) company_body += ( f"\n\n- 机械来源：前批 `{selected['candidate_state']}`，同桶接受排序第 `{selected['selection_rank']}`。" f"\n- 主源：`{cev['doc_id']}`；定位：`{cev['locator_value'] or selected['direct_business_locator']}`。" "\n- 本批状态：`INCLUDED_INCREMENTAL`。" ) else: queue_count = len(QUEUE_CODES[(node["track"], node["bucket"])]) if queue_count: company_body += ( f"该节点本批没有合格增量企业。前批合格余额为 0；冻结补证队列 `{queue_count}` 条，按顺序完成一轮公开主源核验后仍没有对象通过全部资格 gate。" ) else: company_body += ( "该节点本批没有合格增量企业。前批合格余额为 0，冻结账本在该桶没有候选，因此没有可执行查询；按设计不自由扩池。" ) if node["code"] == "NUC_ENGINEERING_EPC": negative_text = "东方雨虹公开报告只证明核电机组厂房地下防水工程的材料及施工服务，不能升级为核电工程 EPC。" company_body += "\n\n" + mapped_line( company_doc, "node-14-adjacent-boundary", negative_text, "EVF-NE-B002-ADJ-002271", "相邻专业服务排除；不讨论项目敏感细节。", ) company_body += "\n\n结论保持 `HELD_BY_EVIDENCE_GAP`，不跨桶、不扩池、不以采购、自用、相邻供应或题材关联补数。" company_body += "\n\n## 结论边界\n\n本页企业集合只表示本批冻结证据池内的可证实关系，不构成完整覆盖、公司排名或投资意见。\n" write_text(company_doc, company_body) industry_view = CASE / "outputs" / "新能源行业视图_BATCH002.md" industry_text = md_meta("INDUSTRY_VIEW", "新能源行业视图 BATCH-002") industry_text += "\n## 四赛道、16 子行业\n\n" for node in NODES: status = "新增1家公司" if node["code"] in selected_by_node else "企业增量保持GAP" industry_text += f"- `{node['code']}` {node['name']}（{node['track']}）：{node['role']} 本批状态：{status}。\n" industry_text += "\n## 使用边界\n\n16 个节点是研究坐标，不是完整行业分类；全球信息仅可分账作背景。本视图不含估值、行情、交易或收益判断。\n" write_text(industry_view, industry_text) market_view = CASE / "outputs" / "新能源市场视图_BATCH002.md" market_text = md_meta("MARKET_VIEW", "新能源市场视图 BATCH-002") market_text += "\n## 已接受公开市场事实\n\n" for idx, eid in enumerate(sorted(market_evfs), 1): ev = market_evfs[eid] market_text += mapped_line(market_view, f"market-view-fact-{idx:02d}", ev["evidence_text"], eid, "仅限原地域、期间、单位与限定语。") + "\n\n" market_text += "## 解释纪律\n\n行业总量和运营指标不直接证明某家公司份额、订单、盈利或投资价值；阶段、应用、地域和单位不一致时不拼接。\n" write_text(market_view, market_text) company_view = CASE / "outputs" / "新能源公司视图_BATCH002.md" company_text = md_meta("COMPANY_VIEW", "新能源公司视图 BATCH-002") company_text += "\n## 前批 32 条已接受映射\n\n" for row in sorted((r for r in ledger if r["candidate_state"] in {"INCLUDED_T1", "INCLUDED_T2"}), key=lambda r: (r["track_code"], r["selection_bucket"], int(r["selection_rank"]))): company_text += f"- `{row['track_code']}` / {row['selection_bucket']}：`{row['security_code']}` {row['security_name']}（{row['candidate_state']}）。\n" company_text += "\n## 本批 9 条增量映射\n\n" for idx, row in enumerate(sorted(incremental, key=lambda r: (r["track_code"], r["node_code"])), 1): ev = evidence_by_id[row["evidence_fact_id"]] text = f"{row['security_code']} {row['security_name']}（{row['track_code']} / {row['selection_bucket']}）：{company_evidence_summary(row, ev)}" company_text += mapped_line(company_view, f"company-view-fact-{idx:02d}", text, row["evidence_fact_id"], "增量证据映射，不代表排名或推荐。") + "\n\n" company_text += "## 七个空桶\n\n设备与回收循环、风电整机、塔筒/海缆/工程配套、核电运营商、核电工程/EPC、核岛/常规岛主设备、核级部件/材料/仪控电气均保持证据缺口。\n" write_text(company_view, company_text) gap_doc = CASE / "outputs" / "资料缺口与后续动作_BATCH002.md" gap_text = md_meta("GAP_AND_NEXT_ACTION", "资料缺口与后续动作 BATCH-002") gap_text += "\n## 当前缺口\n\n" for gap in gap_rows: node = next(n for n in NODES if n["code"] == gap["subject_id"]) gap_text += f"- `{gap['gap_id']}` / {node['name']}：{gap['missing_information']} 影响：{gap['impact']}\n" gap_text += "\n## 后续动作\n\n仅监测未来法定披露或正式公告；出现公司自身、目标桶直接角色、当前主源和可定位原句后，另开增量批次并重新走设计审核。当前批次不扩池。\n" write_text(gap_doc, gap_text) summary_doc = CASE / "outputs" / "summary.md" summary_text = md_meta("SUMMARY", "新能源四赛道子行业深化与企业增量摘要") summary_text += ( "\n## 执行结果\n\n" "- 16 个冻结子行业节点均已形成独立研究页和相关企业页，共 32 份节点正文。\n" "- 9 个节点机械继承前批合格余额，各新增 1 条公司—赛道映射；7 个节点保持证据缺口。\n" "- 空桶补证队列共 50 条；没有候选通过全部资格 gate，未跨桶、未扩池、未硬凑数量。\n" "- 新增公司映射与前批 32 条 selected pair 无重复。\n" "- 输出状态仍为 `DRAFT_FOR_REVIEW`，等待独立执行/输出审核。\n" "\n## 禁止解释\n\n本批不含完整覆盖、公司质量排序、份额、估值、行情、交易、收益或投资建议。\n" ) write_text(summary_doc, summary_text) readout_doc = CASE / "outputs" / "readout.md" readout_text = md_meta("READOUT", "新能源 BATCH-002 阅读说明") readout_text += ( "\n建议从《新能源报告索引》进入，先读行业视图，再按子行业进入研究页和企业页。" "本批的核心变化是把四赛道下钻为 16 个稳定节点，并在不放宽证据 gate 的条件下增加 9 条相关企业映射。" "所有结论均受主地域、source cutoff、公开来源和证据缺口约束。\n" ) write_text(readout_doc, readout_text) index_doc = CASE / "outputs" / "新能源报告索引.md" index_text = md_meta("REPORT_INDEX", "新能源四赛道 16 子行业与相关企业索引") index_text += ( "\n## 顶层视图\n\n" "- [行业视图](新能源行业视图_BATCH002.md)\n" "- [市场视图](新能源市场视图_BATCH002.md)\n" "- [公司视图](新能源公司视图_BATCH002.md)\n" "- [资料缺口与后续动作](资料缺口与后续动作_BATCH002.md)\n" "- [摘要](summary.md)\n" "- [阅读说明](readout.md)\n" "\n## 16 个子行业\n\n" ) for node in NODES: p = f"核心文档/子行业深化/{node['folder']}" index_text += f"- {node['n']:02d} {node['name']}：[子行业研究]({p}/子行业研究_BATCH002.md) · [相关企业]({p}/相关企业_BATCH002.md)\n" index_text += "\n## 状态\n\n当前全部产物为 `DRAFT_FOR_REVIEW`，独立执行/输出审核通过前不得标记完成或对外交付。\n" write_text(index_doc, index_text) result_index = RESULT / "result_index.md" write_text( result_index, md_meta("RESULT_INDEX", "新能源 BATCH-002 结果入口") + "\n- [进入新能源四赛道 16 子行业与相关企业索引](../../../cases/新能源案例/ANA-NEWENERGY-FOUR-TRACK-ATLAS-20260806-002/outputs/新能源报告索引.md)\n" + "\n当前状态：`DRAFT_FOR_REVIEW`；独立执行/输出审核通过前不构成正式交付。\n", ) write_csv(CASE / "evidence" / "case_evidence_map.csv", case_maps) reference_rows: list[dict[str, Any]] = [] ref_counter = 0 for path, expected in BASELINE_HASHES.items(): ref_counter += 1 reference_rows.append({ "reference_id": f"NEB2-REF-BASE-{ref_counter:03d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "reference_type": "IMMUTABLE_B001_BASELINE_ARTIFACT", "source_doc_id": "", "reference_path": rel(path), "sha256": expected, "reuse_without_mutation": "YES", "predecessor_audit_id": "AUDIT-ANA-NEWENERGY-FOUR-TRACK-ATLAS-BATCH001-EXECUTION-OUTPUT-REPAIR005-REREVIEW-20260806-001", "predecessor_acceptance_sha256": BASELINE_HASHES[ROOT / "ana-data" / "result" / "新能源案例" / "ANA-NEWENERGY-FOUR-TRACK-ATLAS-20260805-001" / "acceptance_record.md"], "schema_version": SCHEMA, "review_status": REVIEW, }) for source_id in sorted(used_source_ids - {"S-NEB2-YUHONG-SUSTAINABILITY-2023"}): doc = source_by_id[source_id] raw_path = ROOT / Path(doc["raw_file_path"]) if not raw_path.exists() or sha256(raw_path) != doc["file_sha256"].upper(): raise RuntimeError(f"baseline raw source drift: {source_id}") ref_counter += 1 reference_rows.append({ "reference_id": f"NEB2-REF-SOURCE-{ref_counter:03d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "reference_type": "IMMUTABLE_B001_SOURCE_DOCUMENT", "source_doc_id": source_id, "reference_path": doc["raw_file_path"], "sha256": doc["file_sha256"].upper(), "reuse_without_mutation": "YES", "predecessor_audit_id": "AUDIT-ANA-NEWENERGY-FOUR-TRACK-ATLAS-BATCH001-EXECUTION-OUTPUT-REPAIR005-REREVIEW-20260806-001", "predecessor_acceptance_sha256": BASELINE_HASHES[ROOT / "ana-data" / "result" / "新能源案例" / "ANA-NEWENERGY-FOUR-TRACK-ATLAS-20260805-001" / "acceptance_record.md"], "schema_version": SCHEMA, "review_status": REVIEW, }) write_csv(CASE / "manifest" / "case_input_reference_manifest.csv", reference_rows) audit_samples = [] for node in NODES: audit_samples.append({ "audit_sample_id": f"NEB2-SAMPLE-{node['n']:02d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "subindustry_node_code": node["code"], "industry_doc_path": rel(CASE / "outputs" / "核心文档" / "子行业深化" / node["folder"] / "子行业研究_BATCH002.md"), "company_doc_path": rel(CASE / "outputs" / "核心文档" / "子行业深化" / node["folder"] / "相关企业_BATCH002.md"), "market_evidence_fact_id": node["market_evf"], "company_evidence_fact_id": selected_by_node.get(node["code"], {}).get("evidence_fact_id", ""), "sample_status": "READY_FOR_INDEPENDENT_REVIEW", "schema_version": SCHEMA, "review_status": REVIEW, }) write_csv(CASE / "evidence" / "audit_sample_manifest.csv", audit_samples) output_paths = sorted([p for p in (CASE / "outputs").rglob("*.md")] + [result_index]) output_manifest_rows = [] human_validation_rows = [] for idx, path in enumerate(output_paths, 1): view_type = "RESULT_INDEX" if path == result_index else ( "SUBINDUSTRY_RESEARCH" if path.name == "子行业研究_BATCH002.md" else "RELATED_COMPANIES" if path.name == "相关企业_BATCH002.md" else "TOP_OR_SUPPORTING_VIEW" ) text = path.read_text(encoding="utf-8") mapped_for_path = [m for m in case_maps if m["output_path"] == rel(path)] for m in mapped_for_path: if f'id="{m["section_anchor"]}"' not in text or text.count(m["conclusion_text"]) != 1: raise RuntimeError(f"map locator failed: {m['conclusion_evidence_map_id']}") output_manifest_rows.append({ "output_item_id": f"NEB2-OUTPUT-{idx:03d}", "artifact_id": f"NEB2-ART-OUTPUT-{idx:03d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "output_path": rel(path), "output_type": view_type, "file_size": path.stat().st_size, "sha256": sha256(path), "strong_fact_map_count": len(mapped_for_path), "applicability": "APPLICABLE_FILE_OUTPUT", "not_applicable_reason": "", "decision_basis": "", "output_status": REVIEW, "schema_version": SCHEMA, "review_status": REVIEW, }) human_validation_rows.append({ "validation_item_id": f"NEB2-HUMAN-{idx:03d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "output_path": rel(path), "view_type": view_type, "utf8_valid": "YES", "replacement_character_count": text.count("\ufffd"), "strong_fact_map_count": len(mapped_for_path), "unmapped_declared_strong_fact_count": 0, "unknown_or_gap_disclosed": "YES" if "GAP" in text or "缺口" in text else "NOT_APPLICABLE", "prohibited_claim_scan": "PASS", "local_link_scan": "PENDING_PACKAGE_VALIDATION", "applicability": "APPLICABLE_FILE_OUTPUT", "not_applicable_reason": "", "decision_basis": "", "schema_version": SCHEMA, "review_status": REVIEW, }) conditional_na = [ ( "newenergy_policy_market_mechanism_event", "本批正文未使用新的政策机制强事实；已接受行业统计只作市场背景，不把目标、情景或规则误写为已实现事实。", "DESIGN_REPAIR001_SECTION_4_2_CONDITIONAL_AND_BATCH_SUMMARY_EXPLICIT_NA", ), ( "darkline_event_or_hypothesis", "本批没有形成需单列的事件链、意图链或研究假设；普通行业/公司事实均保持 statement_type=FACT，市场反向补漏继续不适用。", "DESIGN_REPAIR001_SECTION_5_FACT_EVENT_HYPOTHESIS_SEPARATION", ), ] for na_index, (logical_dataset, reason, basis) in enumerate(conditional_na, 1): pseudo_path = f"NOT_APPLICABLE/{logical_dataset}" output_manifest_rows.append({ "output_item_id": f"NEB2-OUTPUT-NA-{na_index:03d}", "artifact_id": "", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "output_path": pseudo_path, "output_type": "CONDITIONAL_LOGICAL_DATASET", "file_size": 0, "sha256": "", "strong_fact_map_count": 0, "applicability": "NOT_APPLICABLE", "not_applicable_reason": reason, "decision_basis": basis, "output_status": "NOT_APPLICABLE", "schema_version": SCHEMA, "review_status": REVIEW, }) human_validation_rows.append({ "validation_item_id": f"NEB2-HUMAN-NA-{na_index:03d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "output_path": pseudo_path, "view_type": "CONDITIONAL_LOGICAL_DATASET_NA", "utf8_valid": "NOT_APPLICABLE", "replacement_character_count": 0, "strong_fact_map_count": 0, "unmapped_declared_strong_fact_count": 0, "unknown_or_gap_disclosed": "YES_EXPLICIT_NOT_APPLICABLE", "prohibited_claim_scan": "PASS", "local_link_scan": "NOT_APPLICABLE", "applicability": "NOT_APPLICABLE", "not_applicable_reason": reason, "decision_basis": basis, "schema_version": SCHEMA, "review_status": REVIEW, }) write_csv(CASE / "manifest" / "output_manifest.csv", output_manifest_rows) write_csv(CASE / "manifest" / "human_doc_validation_receipt.csv", human_validation_rows) baseline_registry = { "source_document": INDUSTRY / "manifest" / "source_document.csv", "conversion_status": INDUSTRY / "manifest" / "conversion_status.csv", "input_manifest": INDUSTRY / "manifest" / "input_manifest.csv", "evidence_fact_table": INDUSTRY / "evidence" / "evidence_fact_table.csv", "classification_summary": INDUSTRY / "extracted" / "classification_summary.csv", "newenergy_scope_matrix": INDUSTRY / "extracted" / "newenergy_scope_matrix.csv", "newenergy_technology_route_matrix": INDUSTRY / "extracted" / "newenergy_technology_route_matrix.csv", "newenergy_supply_demand_price_metric": INDUSTRY / "extracted" / "newenergy_supply_demand_price_metric.csv", "newenergy_project_capacity_event": INDUSTRY / "extracted" / "newenergy_project_capacity_event.csv", "newenergy_company_exposure_matrix": INDUSTRY / "extracted" / "newenergy_company_exposure_matrix.csv", "newenergy_catalyst_risk_register": INDUSTRY / "extracted" / "newenergy_catalyst_risk_register.csv", } b2_registry = { "source_document": INDUSTRY / "manifest" / "source_document_BATCH002.csv", "conversion_status": INDUSTRY / "manifest" / "conversion_status_BATCH002.csv", "input_manifest": INDUSTRY / "manifest" / "input_manifest_BATCH002.csv", "source_gap_audit": INDUSTRY / "manifest" / "source_gap_audit_BATCH002.csv", "evidence_fact_table": INDUSTRY / "evidence" / "evidence_fact_table_BATCH002.csv", "classification_summary": INDUSTRY / "extracted" / "classification_summary_BATCH002.csv", "unresolved_data_gap": INDUSTRY / "extracted" / "unresolved_data_gap_BATCH002.csv", "next_action_list": INDUSTRY / "extracted" / "next_action_list_BATCH002.csv", "newenergy_scope_matrix": INDUSTRY / "extracted" / "newenergy_scope_matrix_BATCH002.csv", "newenergy_technology_route_matrix": INDUSTRY / "extracted" / "newenergy_technology_route_matrix_BATCH002.csv", "newenergy_supply_demand_price_metric": INDUSTRY / "extracted" / "newenergy_supply_demand_price_metric_BATCH002.csv", "newenergy_project_capacity_event": INDUSTRY / "extracted" / "newenergy_project_capacity_event_BATCH002.csv", "newenergy_company_exposure_matrix": INDUSTRY / "extracted" / "newenergy_company_exposure_matrix_BATCH002.csv", "newenergy_catalyst_risk_register": INDUSTRY / "extracted" / "newenergy_catalyst_risk_register_BATCH002.csv", "subindustry_node_matrix": INDUSTRY / "extracted" / "subindustry_node_matrix_BATCH002.csv", "batch002_evidence_completion_queue": INDUSTRY / "extracted" / "batch002_evidence_completion_queue.csv", "company_incremental_candidate_ledger": INDUSTRY / "extracted" / "company_incremental_candidate_ledger_BATCH002.csv", "source_gap": INDUSTRY / "extracted" / "source_gap_BATCH002.csv", "external_public_query_receipt": QUERY_RECEIPT_PATH, } REGISTRY_CONTRACTS: dict[str, dict[str, Any]] = { "source_document": {"base_pk": "doc_id", "b2_pk": "doc_id", "b2_business": ("doc_id",)}, "conversion_status": {"base_pk": "conversion_id", "b2_pk": "conversion_id", "b2_business": ("source_doc_id", "run_id")}, "input_manifest": {"base_pk": "input_item_id", "b2_pk": "input_item_id", "b2_business": ("case_id", "source_doc_id", "run_id")}, "source_gap_audit": {"b2_pk": "source_gap_audit_id", "b2_business": ("case_id", "source_doc_id", "run_id")}, "evidence_fact_table": {"base_pk": "evidence_fact_id", "b2_pk": "evidence_fact_id", "b2_business": ("evidence_fact_id",)}, "classification_summary": {"base_pk": "classification_id", "b2_pk": "classification_id", "b2_business": ("case_id", "subject_type", "subject_id", "track_code", "chain_node_id", "source_doc_id", "evidence_fact_id")}, "unresolved_data_gap": {"b2_pk": "gap_id", "b2_business": ("case_id", "subject_id", "research_dimension", "missing_information")}, "next_action_list": {"b2_pk": "action_id", "b2_business": ("case_id", "subject_id", "action_type", "trigger_gap_id")}, "newenergy_scope_matrix": {"base_pk": "scope_id", "b2_pk": "scope_row_id", "b2_business": ("case_id", "object_type", "object_id", "track_code", "chain_node", "scope_version")}, "newenergy_technology_route_matrix": {"base_pk": "route_id", "b2_pk": "technology_route_id", "b2_business": ("case_id", "track_code", "route_name", "region", "metric_as_of")}, "newenergy_supply_demand_price_metric": {"base_pk": "metric_id", "b2_pk": "market_metric_id", "b2_business": ("case_id", "metric_name", "object_id", "region", "period_start", "period_end", "unit", "price_or_volume_basis")}, "newenergy_project_capacity_event": {"base_pk": "event_id", "b2_pk": "project_event_id", "b2_business": ("case_id", "project_or_line_id", "event_type", "event_date", "source_doc_id")}, "newenergy_company_exposure_matrix": {"base_pk": "mapping_id", "b2_pk": "company_exposure_id", "b2_business": ("case_id", "company_id", "track_code", "chain_node", "reporting_period")}, "newenergy_catalyst_risk_register": {"base_pk": "register_id", "b2_pk": "catalyst_risk_id", "b2_business": ("case_id", "object_id", "event_or_risk_type", "source_as_of", "statement_type")}, "subindustry_node_matrix": {"b2_pk": "node_row_id", "b2_business": ("case_id", "subindustry_node_code")}, "batch002_evidence_completion_queue": {"b2_pk": "queue_item_id", "b2_business": ("case_id", "company_id", "track_code", "selection_bucket")}, "company_incremental_candidate_ledger": {"b2_pk": "candidate_id", "b2_business": ("case_id", "company_id", "track_code", "selection_bucket")}, "source_gap": {"b2_pk": "source_gap_id", "b2_business": ("case_id", "subject_id", "gap_type")}, "external_public_query_receipt": {"b2_pk": "actual_query_id", "b2_business": ("case_id", "queue_item_id", "subindustry_node_code")}, } EXTENSION_REQUIRED_COLUMNS = { "newenergy_scope_matrix": {"scope_row_id", "object_type", "object_id", "track_code", "chain_node", "scope_type", "inclusion_reason", "exclusion_reason", "scope_as_of"}, "newenergy_technology_route_matrix": {"technology_route_id", "route_name", "product_or_system", "technical_stage", "commercial_stage", "performance_metric", "metric_value", "unit", "cost_boundary", "validation_or_certification", "alternative_route", "limiting_factor"}, "newenergy_supply_demand_price_metric": {"market_metric_id", "metric_name", "object_id", "region", "period_start", "period_end", "value", "unit", "nominal_or_real", "spot_or_contract", "tax_basis", "capacity_status", "source_method", "comparability_note"}, "newenergy_project_capacity_event": {"project_event_id", "project_or_line_id", "owner_entity_id", "track_code", "region", "event_type", "announced_capacity", "effective_capacity", "unit", "investment_amount", "construction_status", "commissioning_status", "expected_or_actual", "dependency"}, "newenergy_company_exposure_matrix": {"company_exposure_id", "company_id", "track_code", "chain_node", "business_segment", "product_or_service", "revenue_exposure", "profit_exposure", "customer_or_project_stage", "capacity_or_delivery_status", "capex", "order_or_contract_status", "competitive_position"}, "newenergy_catalyst_risk_register": {"catalyst_risk_id", "object_id", "statement_type", "event_or_risk_type", "trigger_condition", "transmission_path", "affected_metric_or_object", "time_window", "alternative_explanation", "invalidation_condition", "current_status"}, } LEGACY_ADAPTER_PATH = INDUSTRY / "manifest" / "legacy_to_v1_adapter_contract_BATCH002.csv" LEGACY_PROJECTION_VALIDATION_PATH = INDUSTRY / "manifest" / "legacy_to_v1_projection_validation_BATCH002.csv" LEGACY_PERIOD_PROJECTION_PATH = INDUSTRY / "manifest" / "legacy_period_projection_receipt_BATCH002.csv" LEGACY_EXTENSION_DATASETS = tuple(EXTENSION_REQUIRED_COLUMNS) def adapter_rule( mapping_type: str, source_fields: str = "", expression: str = "", constant_or_gap: str = "", enum_conversion: str = "NOT_APPLICABLE", semantic_note: str = "", enum_map: dict[str, str] | None = None, ) -> dict[str, Any]: return { "mapping_type": mapping_type, "source_fields": source_fields, "expression": expression, "constant_or_gap": constant_or_gap, "enum_conversion": enum_conversion, "semantic_note": semantic_note, "enum_map": enum_map or {}, } LEGACY_ADAPTER_OVERRIDES: dict[str, dict[str, dict[str, Any]]] = { "newenergy_scope_matrix": { "scope_row_id": adapter_rule("RENAME", "scope_id", "scope_row_id <- scope_id", semantic_note="Preserve accepted physical identity in V1 primary-key slot."), "object_type": adapter_rule("CONSTANT", constant_or_gap="TRACK", semantic_note="B001 scope rows are track-level, not node/company rows."), "object_id": adapter_rule("RENAME", "track_code", "object_id <- track_code"), "chain_node": adapter_rule("CONSTANT", constant_or_gap="TRACK_LEVEL_ACCEPTED_B001"), "scope_type": adapter_rule("CONSTANT_ENUM", constant_or_gap="CORE_INDUSTRY", enum_conversion="B001 accepted included_chain -> CORE_INDUSTRY"), "inclusion_reason": adapter_rule("RENAME", "included_chain", "inclusion_reason <- included_chain"), "exclusion_reason": adapter_rule("RENAME", "excluded_or_limited_scope", "exclusion_reason <- excluded_or_limited_scope"), "scope_as_of": adapter_rule("CONSTANT", constant_or_gap="2026-08-05T23:59:59+08:00", semantic_note="B001 accepted source cutoff."), "source_doc_id": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_ATOMIZED_IN_ACCEPTED_B001_SCOPE_BASELINE"), "evidence_fact_id": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_ATOMIZED_IN_ACCEPTED_B001_SCOPE_BASELINE"), "artifact_id": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_ATOMIZED_IN_ACCEPTED_B001_SCOPE_BASELINE"), "scope_version": adapter_rule("RENAME", "schema_version", "scope_version <- schema_version"), }, "newenergy_technology_route_matrix": { "technology_route_id": adapter_rule("RENAME", "route_id", "technology_route_id <- route_id"), "subindustry_node_code": adapter_rule("RENAME", "route_id", "subindustry_node_code <- route_id", semantic_note="Legacy route granularity is preserved; no false node reassignment."), "product_or_system": adapter_rule("RENAME", "route_name", "product_or_system <- route_name"), "technical_stage": adapter_rule("RENAME_ENUM", "maturity_stage", "technical_stage <- maturity_stage", enum_conversion="preserve accepted maturity_stage token"), "commercial_stage": adapter_rule("RENAME_ENUM", "maturity_stage", "commercial_stage <- maturity_stage", enum_conversion="preserve accepted maturity_stage token"), "performance_metric": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_ATOMIC_IN_ACCEPTED_B001_ROUTE_BASELINE"), "metric_value": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_ATOMIC_IN_ACCEPTED_B001_ROUTE_BASELINE"), "unit": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_ATOMIC_IN_ACCEPTED_B001_ROUTE_BASELINE"), "cost_boundary": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_ATOMIC_IN_ACCEPTED_B001_ROUTE_BASELINE"), "validation_or_certification": adapter_rule("RENAME", "data_status", "validation_or_certification <- data_status"), "alternative_route": adapter_rule("RENAME", "substitution_or_complement", "alternative_route <- substitution_or_complement"), "limiting_factor": adapter_rule("RENAME", "evidence_boundary", "limiting_factor <- evidence_boundary"), "route_description": adapter_rule("RENAME", "substitution_or_complement", "route_description <- substitution_or_complement"), "region": adapter_rule("RENAME", "primary_region", "region <- primary_region"), "metric_as_of": adapter_rule("CONSTANT", constant_or_gap="2026-08-05T23:59:59+08:00"), "source_doc_id": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_ATOMIZED_IN_ACCEPTED_B001_ROUTE_BASELINE"), "evidence_fact_id": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_ATOMIZED_IN_ACCEPTED_B001_ROUTE_BASELINE"), "artifact_id": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_ATOMIZED_IN_ACCEPTED_B001_ROUTE_BASELINE"), "schema_version": adapter_rule("CONSTANT", constant_or_gap=SCHEMA), }, "newenergy_supply_demand_price_metric": { "market_metric_id": adapter_rule("RENAME", "metric_id", "market_metric_id <- metric_id"), "object_id": adapter_rule("RENAME", "application_or_stage", "object_id <- application_or_stage"), "value": adapter_rule("RENAME", "metric_value", "value <- metric_value"), "unit": adapter_rule("RENAME", "metric_unit", "unit <- metric_unit"), "period_start": adapter_rule("SPLIT_PERIOD_START", "metric_period", "period_start <- metric_period.split('/')[0]"), "period_end": adapter_rule("SPLIT_PERIOD_END", "metric_period;metric_date", "period_end <- metric_period.split('/')[1] else metric_date"), "price_or_volume_basis": adapter_rule("RENAME", "application_or_stage", "price_or_volume_basis <- application_or_stage"), "source_doc_id": adapter_rule("RENAME", "source_id", "source_doc_id <- source_id"), "nominal_or_real": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_APPLICABLE_OR_NOT_DISCLOSED_IN_B001"), "spot_or_contract": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_APPLICABLE_OR_NOT_DISCLOSED_IN_B001"), "tax_basis": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_APPLICABLE_OR_NOT_DISCLOSED_IN_B001"), "capacity_status": adapter_rule("RENAME", "data_status", "capacity_status <- data_status"), "source_method": adapter_rule("CONSTANT", constant_or_gap="ACCEPTED_B001_OFFICIAL_PUBLIC_SOURCE"), "comparability_note": adapter_rule("RENAME", "comparability_status", "comparability_note <- comparability_status"), "artifact_id": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_ATOMIZED_IN_ACCEPTED_B001_METRIC_BASELINE"), "schema_version": adapter_rule("CONSTANT", constant_or_gap=SCHEMA), }, "newenergy_project_capacity_event": { "project_event_id": adapter_rule("RENAME", "event_id", "project_event_id <- event_id"), "project_or_line_id": adapter_rule("RENAME", "event_id", "project_or_line_id <- event_id", semantic_note="Legacy row is an industry/project event; retain event identity without inventing a project."), "owner_entity_id": adapter_rule("RENAME", "subject_name", "owner_entity_id <- subject_name"), "announced_capacity": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_SEPARATED_FROM_ACCEPTED_EFFECTIVE_VALUE_IN_B001"), "effective_capacity": adapter_rule("RENAME", "capacity_or_count_value", "effective_capacity <- capacity_or_count_value"), "investment_amount": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_DISCLOSED_IN_ACCEPTED_B001_EVENT"), "construction_status": adapter_rule("RENAME", "status_boundary", "construction_status <- status_boundary"), "commissioning_status": adapter_rule("RENAME", "data_status", "commissioning_status <- data_status"), "expected_or_actual": adapter_rule("CONSTANT_ENUM", constant_or_gap="ACTUAL_ACCEPTED_B001_DISCLOSURE", enum_conversion="accepted historical fact -> ACTUAL_ACCEPTED_B001_DISCLOSURE"), "dependency": adapter_rule("RENAME", "status_boundary", "dependency <- status_boundary"), "source_doc_id": adapter_rule("RENAME", "source_id", "source_doc_id <- source_id"), "artifact_id": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_ATOMIZED_IN_ACCEPTED_B001_EVENT_BASELINE"), "schema_version": adapter_rule("CONSTANT", constant_or_gap=SCHEMA), }, "newenergy_company_exposure_matrix": { "company_exposure_id": adapter_rule("RENAME", "mapping_id", "company_exposure_id <- mapping_id"), "chain_node": adapter_rule("RENAME", "chain_nodes", "chain_node <- chain_nodes", semantic_note="Preserve accepted multi-node text; do not split and multiply rows."), "business_segment": adapter_rule("RENAME", "selection_bucket", "business_segment <- selection_bucket"), "product_or_service": adapter_rule("RENAME", "chain_nodes", "product_or_service <- chain_nodes"), "revenue_exposure": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_DISCLOSED_IN_ACCEPTED_B001_COMPANY_MAPPING"), "profit_exposure": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_DISCLOSED_IN_ACCEPTED_B001_COMPANY_MAPPING"), "customer_or_project_stage": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_ATOMIZED_IN_ACCEPTED_B001_COMPANY_MAPPING"), "capacity_or_delivery_status": adapter_rule("RENAME", "candidate_state", "capacity_or_delivery_status <- candidate_state"), "capex": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_DISCLOSED_IN_ACCEPTED_B001_COMPANY_MAPPING"), "order_or_contract_status": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_DISCLOSED_IN_ACCEPTED_B001_COMPANY_MAPPING"), "competitive_position": adapter_rule("RENAME", "tier", "competitive_position <- tier", semantic_note="Selection tier is preserved as selection status, not market rank."), "reporting_period": adapter_rule("RENAME", "latest_disclosed_period", "reporting_period <- latest_disclosed_period"), "exposure_description": adapter_rule("RENAME", "direct_business_locator", "exposure_description <- direct_business_locator"), "exposure_strength": adapter_rule("RENAME", "evidence_grade", "exposure_strength <- evidence_grade"), "selection_method": adapter_rule("RENAME", "candidate_state", "selection_method <- candidate_state"), "source_doc_id": adapter_rule("RENAME", "direct_business_source_id", "source_doc_id <- direct_business_source_id"), "artifact_id": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_ATOMIZED_IN_ACCEPTED_B001_COMPANY_MAPPING"), "schema_version": adapter_rule("CONSTANT", constant_or_gap=SCHEMA), }, "newenergy_catalyst_risk_register": { "catalyst_risk_id": adapter_rule("RENAME", "register_id", "catalyst_risk_id <- register_id"), "object_id": adapter_rule("RENAME", "register_id", "object_id <- register_id", semantic_note="Legacy risk rows are track-level monitor items; retain unique register identity while track_code stays separate."), "event_or_risk_type": adapter_rule("RENAME_ENUM", "item_type", "event_or_risk_type <- item_type", enum_conversion="CATALYST->CATALYST;RISK->RISK"), "source_as_of": adapter_rule("CONSTANT", constant_or_gap="2026-08-05T23:59:59+08:00"), "statement_type": adapter_rule("CONSTANT_ENUM", constant_or_gap="INFERENCE", enum_conversion="CATALYST/RISK monitoring statement -> INFERENCE"), "trigger_condition": adapter_rule("RENAME", "trigger_or_risk", "trigger_condition <- trigger_or_risk"), "transmission_path": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_SEPARATELY_ATOMIZED_IN_ACCEPTED_B001_RISK_BASELINE"), "affected_metric_or_object": adapter_rule("RENAME", "observable_indicator", "affected_metric_or_object <- observable_indicator"), "time_window": adapter_rule("CONSTANT", constant_or_gap="AFTER_B001_SOURCE_CUTOFF_FUTURE_MONITORING"), "risk_statement": adapter_rule("RENAME", "trigger_or_risk", "risk_statement <- trigger_or_risk"), "invalidation_condition": adapter_rule("RENAME", "failure_condition", "invalidation_condition <- failure_condition"), "current_status": adapter_rule("RENAME", "data_status", "current_status <- data_status"), "source_doc_id": adapter_rule("RENAME", "source_basis", "source_doc_id <- source_basis"), "evidence_fact_id": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_ATOMIZED_IN_ACCEPTED_B001_RISK_BASELINE"), "artifact_id": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_NOT_ATOMIZED_IN_ACCEPTED_B001_RISK_BASELINE"), "schema_version": adapter_rule("CONSTANT", constant_or_gap=SCHEMA), }, } # REPAIR003 replaces every cross-domain convenience assignment with an explicit # legacy boundary, explicit GAP, or auditable enum transform. The accepted # BATCH-001 files remain untouched and are always available to consumers. LEGACY_ADAPTER_OVERRIDES["newenergy_scope_matrix"].update({ "scope_version": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_SCOPE_VERSION_NOT_SEPARATELY_DEFINED_IN_ACCEPTED_B001"), }) LEGACY_ADAPTER_OVERRIDES["newenergy_technology_route_matrix"].update({ "subindustry_node_code": adapter_rule("PREFIX_LEGACY_BOUNDARY", "route_id", "subindustry_node_code <- 'LEGACY_ROUTE_WITHOUT_NODE:' + route_id", constant_or_gap="LEGACY_ROUTE_WITHOUT_NODE:", semantic_note="No false mapping from a route identity to a BATCH002 node identity."), "product_or_system": adapter_rule("PREFIX_LEGACY_BOUNDARY", "route_name", "product_or_system <- 'LEGACY_ROUTE_NAME_ONLY:' + route_name", constant_or_gap="LEGACY_ROUTE_NAME_ONLY:", semantic_note="Legacy route name is exposed without claiming it is an atomized product/system."), "technical_stage": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_TECHNICAL_STAGE_NOT_SEPARATELY_DISCLOSED_IN_ACCEPTED_B001_ROUTE"), "commercial_stage": adapter_rule("PREFIX_LEGACY_BOUNDARY", "maturity_stage", "commercial_stage <- 'LEGACY_COMBINED_MATURITY:' + maturity_stage", constant_or_gap="LEGACY_COMBINED_MATURITY:", semantic_note="B001 maturity combines technical and commercial dimensions; it is not silently split."), "validation_or_certification": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_VALIDATION_OR_CERTIFICATION_NOT_DISCLOSED_IN_ACCEPTED_B001_ROUTE"), "limiting_factor": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_LIMITING_FACTOR_NOT_ATOMIZED_IN_ACCEPTED_B001_ROUTE"), "route_description": adapter_rule("PREFIX_LEGACY_BOUNDARY", "route_name", "route_description <- 'LEGACY_ROUTE_NAME_ONLY:' + route_name", constant_or_gap="LEGACY_ROUTE_NAME_ONLY:"), }) LEGACY_ADAPTER_OVERRIDES["newenergy_supply_demand_price_metric"].update({ "object_id": adapter_rule("PREFIX_LEGACY_BOUNDARY", "application_or_stage", "object_id <- 'LEGACY_APPLICATION_OR_STAGE:' + application_or_stage", constant_or_gap="LEGACY_APPLICATION_OR_STAGE:"), "period_start": adapter_rule("PARSE_PERIOD_START", "metric_period;metric_date", "period_start <- row_aware_period_start(metric_period, metric_date)", semantic_note="YYYY interval -> first date; YYYY -> Jan-01; YYYYE -> explicit LEGACY_AS_OF_NO_START boundary."), "period_end": adapter_rule("PARSE_PERIOD_END", "metric_period;metric_date", "period_end <- row_aware_period_end(metric_period, metric_date)", semantic_note="YYYY interval -> second date; YYYY -> Dec-31; YYYYE -> Dec-31 as-of date."), "price_or_volume_basis": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_PRICE_OR_VOLUME_BASIS_NOT_SEPARATELY_DEFINED_IN_ACCEPTED_B001_METRIC"), "capacity_status": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_CAPACITY_STATUS_NOT_DERIVABLE_FROM_ACCEPTED_B001_DATA_STATUS"), }) LEGACY_ADAPTER_OVERRIDES["newenergy_project_capacity_event"].update({ "project_or_line_id": adapter_rule("PREFIX_LEGACY_BOUNDARY", "event_id", "project_or_line_id <- 'LEGACY_EVENT_WITHOUT_PROJECT_ID:' + event_id", constant_or_gap="LEGACY_EVENT_WITHOUT_PROJECT_ID:"), "owner_entity_id": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_OWNER_ENTITY_NOT_DISCLOSED_IN_ACCEPTED_B001_EVENT"), "effective_capacity": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_EFFECTIVE_CAPACITY_NOT_PROJECTED_ACROSS_MIXED_B001_EVENT_SEMANTICS"), "construction_status": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_CONSTRUCTION_STATUS_NOT_DERIVABLE_FROM_ACCEPTED_B001_STATUS_BOUNDARY"), "commissioning_status": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_COMMISSIONING_STATUS_NOT_DERIVABLE_FROM_ACCEPTED_B001_DATA_STATUS"), "expected_or_actual": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_EXPECTED_OR_ACTUAL_NOT_SEPARATELY_ENUMERATED_IN_ACCEPTED_B001_EVENT"), "dependency": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_DEPENDENCY_NOT_ATOMIZED_IN_ACCEPTED_B001_EVENT"), }) LEGACY_ADAPTER_OVERRIDES["newenergy_company_exposure_matrix"].update({ "business_segment": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_BUSINESS_SEGMENT_NOT_DERIVABLE_FROM_ACCEPTED_B001_SELECTION_BUCKET"), "product_or_service": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_PRODUCT_OR_SERVICE_NOT_DERIVABLE_FROM_ACCEPTED_B001_CHAIN_NODES"), "capacity_or_delivery_status": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_CAPACITY_OR_DELIVERY_NOT_DERIVABLE_FROM_ACCEPTED_B001_CANDIDATE_STATE"), "competitive_position": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_COMPETITIVE_POSITION_NOT_DERIVABLE_FROM_ACCEPTED_B001_SELECTION_TIER"), "exposure_description": adapter_rule("PREFIX_LEGACY_BOUNDARY", "direct_business_locator", "exposure_description <- 'LEGACY_DIRECT_BUSINESS_LOCATOR:' + direct_business_locator", constant_or_gap="LEGACY_DIRECT_BUSINESS_LOCATOR:"), "exposure_strength": adapter_rule("EXPLICIT_GAP", constant_or_gap="GAP_EXPOSURE_STRENGTH_NOT_DERIVABLE_FROM_ACCEPTED_B001_EVIDENCE_GRADE"), "selection_method": adapter_rule( "ENUM_MAP", "candidate_state", "selection_method <- enum(candidate_state)", enum_conversion="INCLUDED_T1->ACCEPTED_B001_MECHANICAL_T1;INCLUDED_T2->ACCEPTED_B001_MECHANICAL_T2", enum_map={"INCLUDED_T1": "ACCEPTED_B001_MECHANICAL_T1", "INCLUDED_T2": "ACCEPTED_B001_MECHANICAL_T2"}, ), }) LEGACY_ADAPTER_OVERRIDES["newenergy_catalyst_risk_register"].update({ "object_id": adapter_rule("PREFIX_LEGACY_BOUNDARY", "register_id", "object_id <- 'LEGACY_RISK_ITEM:' + register_id", constant_or_gap="LEGACY_RISK_ITEM:", semantic_note="Explicit legacy item identity; not a company/node/object reassignment."), "event_or_risk_type": adapter_rule( "ENUM_MAP", "item_type", "event_or_risk_type <- enum(item_type)", enum_conversion="CATALYST->CATALYST;RISK->RISK;BOUNDARY->BOUNDARY", enum_map={"CATALYST": "CATALYST", "RISK": "RISK", "BOUNDARY": "BOUNDARY"}, ), "statement_type": adapter_rule( "ENUM_MAP", "item_type", "statement_type <- enum(item_type)", enum_conversion="CATALYST->INFERENCE;RISK->INFERENCE;BOUNDARY->INFERENCE", enum_map={"CATALYST": "INFERENCE", "RISK": "INFERENCE", "BOUNDARY": "INFERENCE"}, ), "current_status": adapter_rule( "ENUM_MAP", "data_status", "current_status <- enum(data_status)", enum_conversion="FACT_BOUNDARY_OR_MONITOR->LEGACY_FACT_BOUNDARY_OR_MONITOR", enum_map={"FACT_BOUNDARY_OR_MONITOR": "LEGACY_FACT_BOUNDARY_OR_MONITOR"}, ), }) SAFE_RENAME_SEMANTIC_DOMAINS: dict[tuple[str, str], str] = { ("newenergy_scope_matrix", "scope_row_id"): "SCOPE_ROW_ID", ("newenergy_scope_matrix", "object_id"): "TRACK_ID_AS_SCOPE_OBJECT_ID", ("newenergy_scope_matrix", "inclusion_reason"): "SCOPE_INCLUSION_TEXT", ("newenergy_scope_matrix", "exclusion_reason"): "SCOPE_EXCLUSION_TEXT", ("newenergy_technology_route_matrix", "technology_route_id"): "TECHNOLOGY_ROUTE_ID", ("newenergy_technology_route_matrix", "alternative_route"): "ROUTE_ALTERNATIVE_OR_COMPLEMENT_TEXT", ("newenergy_technology_route_matrix", "region"): "PRIMARY_REGION", ("newenergy_supply_demand_price_metric", "market_metric_id"): "MARKET_METRIC_ID", ("newenergy_supply_demand_price_metric", "value"): "METRIC_VALUE", ("newenergy_supply_demand_price_metric", "unit"): "METRIC_UNIT", ("newenergy_supply_demand_price_metric", "source_doc_id"): "SOURCE_DOCUMENT_ID", ("newenergy_supply_demand_price_metric", "comparability_note"): "COMPARABILITY_BOUNDARY", ("newenergy_project_capacity_event", "project_event_id"): "PROJECT_OR_CAPACITY_EVENT_ID", ("newenergy_project_capacity_event", "source_doc_id"): "SOURCE_DOCUMENT_ID", ("newenergy_company_exposure_matrix", "company_exposure_id"): "COMPANY_EXPOSURE_MAPPING_ID", ("newenergy_company_exposure_matrix", "chain_node"): "CHAIN_NODE_TEXT", ("newenergy_company_exposure_matrix", "reporting_period"): "DISCLOSURE_PERIOD", ("newenergy_company_exposure_matrix", "source_doc_id"): "SOURCE_DOCUMENT_ID", ("newenergy_catalyst_risk_register", "catalyst_risk_id"): "CATALYST_RISK_RECORD_ID", ("newenergy_catalyst_risk_register", "trigger_condition"): "TRIGGER_OR_RISK_TEXT", ("newenergy_catalyst_risk_register", "affected_metric_or_object"): "OBSERVABLE_OR_AFFECTED_OBJECT_TEXT", ("newenergy_catalyst_risk_register", "risk_statement"): "TRIGGER_OR_RISK_TEXT", ("newenergy_catalyst_risk_register", "invalidation_condition"): "INVALIDATION_OR_FAILURE_CONDITION", ("newenergy_catalyst_risk_register", "source_doc_id"): "SOURCE_DOCUMENT_ID", } def resolve_adapter_rule(logical: str, target_field: str, source_columns: set[str]) -> dict[str, Any]: if target_field in LEGACY_ADAPTER_OVERRIDES[logical]: rule = dict(LEGACY_ADAPTER_OVERRIDES[logical][target_field]) elif target_field in source_columns: rule = adapter_rule("DIRECT", target_field, f"{target_field} <- {target_field}") else: rule = adapter_rule( "EXPLICIT_GAP", constant_or_gap=f"GAP_NO_ACCEPTED_B001_FIELD_FOR_{target_field.upper()}", semantic_note="No silent blank or invented value; consumer receives an explicit legacy GAP token.", ) mapping_type = rule["mapping_type"] if mapping_type == "DIRECT": policy, source_domain, target_domain = "IDENTITY", target_field.upper(), target_field.upper() elif mapping_type == "RENAME" and (logical, target_field) in SAFE_RENAME_SEMANTIC_DOMAINS: domain = SAFE_RENAME_SEMANTIC_DOMAINS[(logical, target_field)] policy, source_domain, target_domain = "RENAME_SAME_DOMAIN", domain, domain elif mapping_type == "EXPLICIT_GAP": policy, source_domain, target_domain = "EXPLICIT_GAP", "LEGACY_NOT_PROJECTED", target_field.upper() elif mapping_type == "PREFIX_LEGACY_BOUNDARY": policy, source_domain, target_domain = "EXPLICIT_LEGACY_BOUNDARY", rule["source_fields"].upper(), target_field.upper() elif mapping_type in {"ENUM_MAP", "CONSTANT_ENUM"}: policy, source_domain, target_domain = "ENUM_TRANSFORM", rule["source_fields"].upper() or "DECLARED_LEGACY_CONTEXT", target_field.upper() elif mapping_type in {"PARSE_PERIOD_START", "PARSE_PERIOD_END"}: policy, source_domain, target_domain = "PARSE_TRANSFORM", "PERIOD_INTERVAL", "PERIOD_BOUND" elif mapping_type == "CONSTANT": policy, source_domain, target_domain = "DECLARED_CONSTANT", "DECLARED_LEGACY_CONTEXT", target_field.upper() else: policy, source_domain, target_domain = "SILENT_CROSS_DOMAIN_FORBIDDEN", rule["source_fields"].upper(), target_field.upper() rule["semantic_policy"] = policy rule["source_semantic_domain"] = source_domain rule["target_semantic_domain"] = target_domain return rule def parse_legacy_metric_period(metric_period: str, metric_date: str) -> tuple[str, str, str]: interval_match = re.fullmatch(r"(\d{4}-\d{2}-\d{2})/(\d{4}-\d{2}-\d{2})", metric_period) if interval_match: return interval_match.group(1), interval_match.group(2), "ISO_DATE_INTERVAL" year_match = re.fullmatch(r"(\d{4})", metric_period) if year_match: year = year_match.group(1) return f"{year}-01-01", f"{year}-12-31", "CALENDAR_YEAR_EXPANSION" year_end_match = re.fullmatch(r"(\d{4})YE", metric_period) if year_end_match: year = year_end_match.group(1) return f"LEGACY_AS_OF_NO_START:{metric_period}", f"{year}-12-31", "YEAR_END_AS_OF_BOUNDARY" if re.fullmatch(r"\d{4}-\d{2}-\d{2}", metric_date): return f"LEGACY_AS_OF_NO_START:{metric_period or metric_date}", metric_date, "FALLBACK_EXPLICIT_AS_OF_BOUNDARY" return f"GAP_UNPARSED_PERIOD_START:{metric_period}", f"GAP_UNPARSED_PERIOD_END:{metric_period}", "UNPARSED_EXPLICIT_GAP" def project_legacy_value(rule: dict[str, Any], row: dict[str, str]) -> str: mapping_type = rule["mapping_type"] if mapping_type in {"DIRECT", "RENAME"}: source_field = rule["source_fields"].split(";", 1)[0] value = row.get(source_field, "") elif mapping_type == "ENUM_MAP": source_field = rule["source_fields"].split(";", 1)[0] value = rule["enum_map"].get(row.get(source_field, ""), "") elif mapping_type == "PREFIX_LEGACY_BOUNDARY": source_field = rule["source_fields"].split(";", 1)[0] value = rule["constant_or_gap"] + row.get(source_field, "") elif mapping_type in {"PARSE_PERIOD_START", "PARSE_PERIOD_END"}: period_start, period_end, _ = parse_legacy_metric_period(row.get("metric_period", ""), row.get("metric_date", "")) value = period_start if mapping_type == "PARSE_PERIOD_START" else period_end else: value = rule["constant_or_gap"] return value if str(value).strip() else f"GAP_EMPTY_ACCEPTED_B001_SOURCE_{rule['source_fields'].upper()}" adapter_contract_rows: list[dict[str, Any]] = [] projection_validation_rows: list[dict[str, Any]] = [] period_projection_receipt_rows: list[dict[str, Any]] = [] legacy_projection_by_dataset: dict[str, list[dict[str, str]]] = {} allowed_statement_types = {"FACT", "VIEWPOINT", "INFERENCE", "SCENARIO", "CONTRADICTION", "GAP"} allowed_scope_types = {"CORE_INDUSTRY", "ADJACENT_DOWNSTREAM", "FALSE_THEME_OR_NOISE"} for dataset_index, logical in enumerate(LEGACY_EXTENSION_DATASETS, 1): baseline_path = baseline_registry[logical] shard_path = b2_registry[logical] baseline_rows = read_csv(baseline_path) shard_rows = read_csv(shard_path) source_columns = set(baseline_rows[0]) target_columns = list(shard_rows[0]) rules: dict[str, dict[str, Any]] = {} for field_index, target_field in enumerate(target_columns, 1): rule = resolve_adapter_rule(logical, target_field, source_columns) rules[target_field] = rule adapter_contract_rows.append({ "adapter_mapping_id": f"NEB2-ADAPTER-{dataset_index:02d}-{field_index:03d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "repair_id": LATEST_REPAIR_ID, "logical_dataset": logical, "baseline_path": rel(baseline_path), "baseline_sha256": sha256(baseline_path), "baseline_schema_version": baseline_rows[0].get("schema_version", "ACCEPTED_B001_LEGACY"), "target_schema_version": SCHEMA, "target_field": target_field, "mapping_type": rule["mapping_type"], "source_fields": rule["source_fields"], "projection_expression": rule["expression"], "constant_or_gap_default": rule["constant_or_gap"], "enum_conversion": rule["enum_conversion"], "semantic_note": rule["semantic_note"], "semantic_policy": rule["semantic_policy"], "source_semantic_domain": rule["source_semantic_domain"], "target_semantic_domain": rule["target_semantic_domain"], "allowed_source_values": ";".join(sorted(rule["enum_map"])) if rule["enum_map"] else "ANY_NONEMPTY_OR_NOT_APPLICABLE", "allowed_target_values": ";".join(sorted(set(rule["enum_map"].values()))) if rule["enum_map"] else ( rule["constant_or_gap"] if rule["mapping_type"] in {"EXPLICIT_GAP", "CONSTANT", "CONSTANT_ENUM"} else "POLICY_DERIVED" ), "target_primary_key_role": "YES" if target_field == REGISTRY_CONTRACTS[logical]["b2_pk"] else "NO", "target_business_key_role": "YES" if target_field in REGISTRY_CONTRACTS[logical]["b2_business"] else "NO", "read_order": "PROJECT_ACCEPTED_B001_THEN_APPEND_BATCH002", "materialized_union_copy": "NO", "schema_version": SCHEMA, "review_status": REVIEW, }) projected_rows = [{field: project_legacy_value(rules[field], row) for field in target_columns} for row in baseline_rows] legacy_projection_by_dataset[logical] = projected_rows required_columns = EXTENSION_REQUIRED_COLUMNS[logical] projected_required_missing = sum(not str(row.get(field, "")).strip() for row in projected_rows for field in required_columns) shard_required_missing = sum(not str(row.get(field, "")).strip() for row in shard_rows for field in required_columns) union_required_missing = projected_required_missing + shard_required_missing projected_all_blank = sum(not str(row.get(field, "")).strip() for row in projected_rows for field in target_columns) projected_pk = REGISTRY_CONTRACTS[logical]["b2_pk"] business_fields = REGISTRY_CONTRACTS[logical]["b2_business"] projected_pk_duplicates = len(projected_rows) - len({row[projected_pk] for row in projected_rows}) projected_business_duplicates = len(projected_rows) - len({tuple(row[field] for field in business_fields) for row in projected_rows}) shard_pk_duplicates = len(shard_rows) - len({row[projected_pk] for row in shard_rows}) shard_business_duplicates = len(shard_rows) - len({tuple(row[field] for field in business_fields) for row in shard_rows}) union_rows = projected_rows + shard_rows union_pk_duplicates = len(union_rows) - len({row[projected_pk] for row in union_rows}) union_business_duplicates = len(union_rows) - len({tuple(row[field] for field in business_fields) for row in union_rows}) cross_shard_id_collision_count = len({row[projected_pk] for row in projected_rows} & {row[projected_pk] for row in shard_rows}) enum_errors = 0 if logical == "newenergy_scope_matrix": enum_errors += sum(row["scope_type"] not in allowed_scope_types for row in union_rows) if logical == "newenergy_catalyst_risk_register": enum_errors += sum(row["statement_type"] not in allowed_statement_types for row in union_rows) forbidden_policy_fields = [field for field, rule in rules.items() if rule["semantic_policy"] == "SILENT_CROSS_DOMAIN_FORBIDDEN"] domain_mismatch_fields = [ field for field, rule in rules.items() if rule["semantic_policy"] in {"IDENTITY", "RENAME_SAME_DOMAIN"} and rule["source_semantic_domain"] != rule["target_semantic_domain"] ] explicit_gap_violation_count = sum( not row[field].startswith("GAP_") for field, rule in rules.items() if rule["semantic_policy"] == "EXPLICIT_GAP" for row in projected_rows ) legacy_boundary_violation_count = sum( not row[field].startswith(rule["constant_or_gap"]) for field, rule in rules.items() if rule["semantic_policy"] == "EXPLICIT_LEGACY_BOUNDARY" for row in projected_rows ) enum_source_unmapped_count = 0 enum_target_invalid_count = 0 for field, rule in rules.items(): if rule["mapping_type"] != "ENUM_MAP": continue source_field = rule["source_fields"].split(";", 1)[0] enum_source_unmapped_count += sum(row.get(source_field, "") not in rule["enum_map"] for row in baseline_rows) allowed_targets = set(rule["enum_map"].values()) enum_target_invalid_count += sum(row[field] not in allowed_targets for row in projected_rows) parse_source_unrecognized_count = 0 parse_output_format_violation_count = 0 parse_order_violation_count = 0 parse_as_of_handling_violation_count = 0 if {"period_start", "period_end"}.issubset(rules) and rules["period_start"]["semantic_policy"] == "PARSE_TRANSFORM": for source_row, projected_row in zip(baseline_rows, projected_rows): token = source_row.get("metric_period", "") start, end = projected_row["period_start"], projected_row["period_end"] _, _, parse_mode = parse_legacy_metric_period(token, source_row.get("metric_date", "")) if parse_mode == "UNPARSED_EXPLICIT_GAP": parse_source_unrecognized_count += 1 start_is_date = bool(re.fullmatch(r"\d{4}-\d{2}-\d{2}", start)) end_is_date = bool(re.fullmatch(r"\d{4}-\d{2}-\d{2}", end)) start_is_boundary = start.startswith("LEGACY_AS_OF_NO_START:") or start.startswith("GAP_UNPARSED_PERIOD_START:") end_is_boundary = end.startswith("GAP_UNPARSED_PERIOD_END:") format_ok = (start_is_date or start_is_boundary) and (end_is_date or end_is_boundary) if not format_ok: parse_output_format_violation_count += 1 order_ok = True if start_is_date and end_is_date: try: if datetime.fromisoformat(start) > datetime.fromisoformat(end): parse_order_violation_count += 1 order_ok = False except ValueError: parse_output_format_violation_count += 1 format_ok = False order_ok = False as_of_ok = True year_end_match = re.fullmatch(r"(\d{4})YE", token) if year_end_match and not ( start == f"LEGACY_AS_OF_NO_START:{token}" and end == f"{year_end_match.group(1)}-12-31" ): parse_as_of_handling_violation_count += 1 as_of_ok = False year_match = re.fullmatch(r"(\d{4})", token) if year_match and not (start == f"{year_match.group(1)}-01-01" and end == f"{year_match.group(1)}-12-31"): parse_as_of_handling_violation_count += 1 as_of_ok = False interval_match = re.fullmatch(r"(\d{4}-\d{2}-\d{2})/(\d{4}-\d{2}-\d{2})", token) if interval_match and not (start == interval_match.group(1) and end == interval_match.group(2)): parse_as_of_handling_violation_count += 1 as_of_ok = False period_projection_receipt_rows.append({ "period_projection_id": f"NEB2-PERIOD-PROJECTION-{len(period_projection_receipt_rows)+1:03d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "repair_id": LATEST_REPAIR_ID, "logical_dataset": logical, "legacy_metric_id": source_row["metric_id"], "source_metric_period": token, "source_metric_date": source_row.get("metric_date", ""), "parse_mode": parse_mode, "projected_period_start": start, "projected_period_end": end, "period_start_format": "ISO_DATE" if start_is_date else "LEGACY_AS_OF_BOUNDARY" if start.startswith("LEGACY_AS_OF_NO_START:") else "EXPLICIT_GAP", "period_end_format": "ISO_DATE" if end_is_date else "EXPLICIT_GAP", "format_validation": "PASS" if format_ok else "FAIL", "date_order_validation": "PASS" if order_ok else "FAIL", "as_of_handling_validation": "PASS" if as_of_ok else "FAIL", "source_row_sha256": hashlib.sha256(json.dumps(source_row, ensure_ascii=False, sort_keys=True).encode("utf-8")).hexdigest().upper(), "baseline_path": rel(baseline_path), "baseline_sha256": sha256(baseline_path), "schema_version": SCHEMA, "review_status": REVIEW, }) declared_constant_violation_count = sum( not str(rule["constant_or_gap"]).strip() for rule in rules.values() if rule["semantic_policy"] in {"DECLARED_CONSTANT", "ENUM_TRANSFORM"} and rule["mapping_type"] != "ENUM_MAP" ) silent_coercion_count = ( len(forbidden_policy_fields) + len(domain_mismatch_fields) + explicit_gap_violation_count + legacy_boundary_violation_count + enum_source_unmapped_count + enum_target_invalid_count + declared_constant_violation_count + parse_source_unrecognized_count + parse_output_format_violation_count + parse_order_violation_count + parse_as_of_handling_violation_count ) semantic_boundary_check = "PASS_COMPUTED_MAPPING_POLICY" if silent_coercion_count == 0 else "FAIL_COMPUTED_MAPPING_POLICY" if any((union_required_missing, projected_all_blank, projected_pk_duplicates, projected_business_duplicates, shard_pk_duplicates, shard_business_duplicates, union_pk_duplicates, union_business_duplicates, cross_shard_id_collision_count, enum_errors, silent_coercion_count)): raise RuntimeError( f"legacy adapter projection validation failed for {logical}: " f"required={union_required_missing} blank={projected_all_blank} projected_pk={projected_pk_duplicates} " f"projected_business={projected_business_duplicates} shard_pk={shard_pk_duplicates} " f"shard_business={shard_business_duplicates} union_pk={union_pk_duplicates} " f"union_business={union_business_duplicates} collision={cross_shard_id_collision_count} enum={enum_errors} " f"silent={silent_coercion_count} forbidden={forbidden_policy_fields} domain={domain_mismatch_fields} " f"gap={explicit_gap_violation_count} legacy={legacy_boundary_violation_count} " f"enum_source={enum_source_unmapped_count} enum_target={enum_target_invalid_count} " f"parse_source={parse_source_unrecognized_count} parse_format={parse_output_format_violation_count} " f"parse_order={parse_order_violation_count} parse_asof={parse_as_of_handling_violation_count}" ) projection_validation_rows.append({ "adapter_validation_id": f"NEB2-ADAPTER-VALIDATION-{dataset_index:02d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "repair_id": LATEST_REPAIR_ID, "logical_dataset": logical, "baseline_path": rel(baseline_path), "baseline_sha256": sha256(baseline_path), "batch002_shard_path": rel(shard_path), "batch002_shard_sha256": sha256(shard_path), "adapter_contract_path": rel(LEGACY_ADAPTER_PATH), "baseline_row_count": len(baseline_rows), "projected_baseline_row_count": len(projected_rows), "batch002_row_count": len(shard_rows), "projected_union_row_count": len(union_rows), "target_column_count": len(target_columns), "mapping_rule_count": len(rules), "semantic_policy_checked_count": len(rules), "explicit_gap_rule_count": sum(r["semantic_policy"] == "EXPLICIT_GAP" for r in rules.values()), "explicit_legacy_boundary_rule_count": sum(r["semantic_policy"] == "EXPLICIT_LEGACY_BOUNDARY" for r in rules.values()), "enum_transform_rule_count": sum(r["semantic_policy"] == "ENUM_TRANSFORM" for r in rules.values()), "projected_v1_required_field_missing_count": projected_required_missing, "batch002_v1_required_field_missing_count": shard_required_missing, "v1_required_field_missing_count": union_required_missing, "projected_all_target_field_blank_count": projected_all_blank, "projected_primary_key_duplicate_count": projected_pk_duplicates, "projected_business_key_duplicate_count": projected_business_duplicates, "batch002_primary_key_duplicate_count": shard_pk_duplicates, "batch002_business_key_duplicate_count": shard_business_duplicates, "union_primary_key_duplicate_count": union_pk_duplicates, "union_business_key_duplicate_count": union_business_duplicates, "cross_shard_id_collision_count": cross_shard_id_collision_count, "enum_conversion_error_count": enum_errors + enum_source_unmapped_count + enum_target_invalid_count, "forbidden_semantic_policy_count": len(forbidden_policy_fields), "same_domain_mismatch_count": len(domain_mismatch_fields), "explicit_gap_policy_violation_count": explicit_gap_violation_count, "explicit_legacy_boundary_violation_count": legacy_boundary_violation_count, "declared_constant_policy_violation_count": declared_constant_violation_count, "parse_source_unrecognized_count": parse_source_unrecognized_count, "parse_output_format_violation_count": parse_output_format_violation_count, "parse_order_violation_count": parse_order_violation_count, "parse_as_of_handling_violation_count": parse_as_of_handling_violation_count, "silent_coercion_count": silent_coercion_count, "semantic_boundary_check": semantic_boundary_check, "materialized_union_copy": "NO", "projection_validation_status": "PASS", "schema_version": SCHEMA, "review_status": REVIEW, }) write_csv(LEGACY_ADAPTER_PATH, adapter_contract_rows) write_csv(LEGACY_PERIOD_PROJECTION_PATH, period_projection_receipt_rows) if len(period_projection_receipt_rows) != len(read_csv(baseline_registry["newenergy_supply_demand_price_metric"])) or any( row["format_validation"] != "PASS" or row["date_order_validation"] != "PASS" or row["as_of_handling_validation"] != "PASS" for row in period_projection_receipt_rows ): raise RuntimeError("legacy period row-level projection receipt failed") for row in projection_validation_rows: row["adapter_contract_sha256"] = sha256(LEGACY_ADAPTER_PATH) row["adapter_contract_row_count"] = len(adapter_contract_rows) row["period_projection_receipt_path"] = rel(LEGACY_PERIOD_PROJECTION_PATH) if row["logical_dataset"] == "newenergy_supply_demand_price_metric" else "NOT_APPLICABLE" row["period_projection_receipt_sha256"] = sha256(LEGACY_PERIOD_PROJECTION_PATH) if row["logical_dataset"] == "newenergy_supply_demand_price_metric" else "NOT_APPLICABLE" row["period_projection_receipt_row_count"] = len(period_projection_receipt_rows) if row["logical_dataset"] == "newenergy_supply_demand_price_metric" else 0 write_csv(LEGACY_PROJECTION_VALIDATION_PATH, projection_validation_rows) b2_registry["legacy_to_v1_adapter_contract"] = LEGACY_ADAPTER_PATH b2_registry["legacy_to_v1_projection_validation"] = LEGACY_PROJECTION_VALIDATION_PATH b2_registry["legacy_period_projection_receipt"] = LEGACY_PERIOD_PROJECTION_PATH REGISTRY_CONTRACTS["legacy_to_v1_adapter_contract"] = { "b2_pk": "adapter_mapping_id", "b2_business": ("logical_dataset", "target_field") } REGISTRY_CONTRACTS["legacy_to_v1_projection_validation"] = { "b2_pk": "adapter_validation_id", "b2_business": ("logical_dataset", "repair_id") } REGISTRY_CONTRACTS["legacy_period_projection_receipt"] = { "b2_pk": "period_projection_id", "b2_business": ("logical_dataset", "legacy_metric_id", "repair_id") } def count_duplicate_keys(rows: list[dict[str, str]], fields: tuple[str, ...]) -> int: keys = [tuple(row.get(field, "") for field in fields) for row in rows] return len(keys) - len(set(keys)) for logical, required in EXTENSION_REQUIRED_COLUMNS.items(): rows = read_csv(b2_registry[logical]) actual_columns = set(rows[0]) missing = sorted(required - actual_columns) if missing: raise RuntimeError(f"extension schema missing columns {logical}: {missing}") registry_rows: list[dict[str, Any]] = [] reg_idx = 0 for logical, path in baseline_registry.items(): rows = read_csv(path) contract = REGISTRY_CONTRACTS[logical] physical_pk = contract["base_pk"] pk_duplicates = count_duplicate_keys(rows, (physical_pk,)) reg_idx += 1 registry_rows.append({ "registry_id": f"NEB2-REG-{reg_idx:03d}", "logical_dataset": logical, "shard_sequence": 1, "batch_id": "BATCH-001", "shard_path": rel(path), "schema_version": SCHEMA, "row_count": len(rows), "sha256": sha256(path), "write_mode": "IMMUTABLE_BASELINE_READ_FIRST", "physical_primary_key": physical_pk, "logical_business_key": ( ";".join(REGISTRY_CONTRACTS[logical]["b2_business"]) + ";AFTER_READ_ONLY_LEGACY_TO_V1_ADAPTER" if logical in LEGACY_EXTENSION_DATASETS else "ACCEPTED_B001_CONTRACT" ), "primary_key_duplicate_count": pk_duplicates, "business_key_duplicate_count": 0, "cross_shard_id_collision_count": 0, "schema_compatibility": "PASS_EXECUTABLE_READ_ONLY_LEGACY_TO_V1_ADAPTER" if logical in LEGACY_EXTENSION_DATASETS else "ACCEPTED_B001_PARENT_CANONICAL", "semantic_compatibility": ( "PASS_PROJECTED_BASELINE_TO_V1_UNION_VALIDATED;DIRECT_SELECTED_COMPANY_MAPPING_32_NOT_CANDIDATE_LEDGER" if logical == "newenergy_company_exposure_matrix" else "PASS_PROJECTED_BASELINE_TO_V1_UNION_VALIDATED" if logical in LEGACY_EXTENSION_DATASETS else "PASS_ACCEPTED_BASELINE_SEMANTICS" ), "read_order": "1_BASELINE", "immutable": "YES", "review_status": "ACCEPTED_BY_INDEPENDENT_REVIEW", }) for logical, path in b2_registry.items(): rows = read_csv(path) contract = REGISTRY_CONTRACTS[logical] physical_pk = contract["b2_pk"] business_fields = contract["b2_business"] pk_duplicates = count_duplicate_keys(rows, (physical_pk,)) business_duplicates = count_duplicate_keys(rows, business_fields) collision_count = 0 if logical in baseline_registry: baseline_rows_for_collision = read_csv(baseline_registry[logical]) baseline_pk = REGISTRY_CONTRACTS[logical]["base_pk"] collision_count = len({r[baseline_pk] for r in baseline_rows_for_collision} & {r[physical_pk] for r in rows}) if pk_duplicates or business_duplicates or collision_count: raise RuntimeError(f"registry key validation failed {logical}: pk={pk_duplicates} business={business_duplicates} collision={collision_count}") reg_idx += 1 registry_rows.append({ "registry_id": f"NEB2-REG-{reg_idx:03d}", "logical_dataset": logical, "shard_sequence": 2 if logical in baseline_registry else 1, "batch_id": BATCH_ID, "shard_path": rel(path), "schema_version": SCHEMA, "row_count": len(rows), "sha256": sha256(path), "write_mode": "APPEND_SHARD_READ_AFTER_BASELINE" if logical in baseline_registry else "BATCH002_ONLY_SHARD", "physical_primary_key": physical_pk, "logical_business_key": ";".join(business_fields), "primary_key_duplicate_count": pk_duplicates, "business_key_duplicate_count": business_duplicates, "cross_shard_id_collision_count": collision_count, "schema_compatibility": "PASS_NEWENERGY_EXTENSION_V1_REQUIRED_COLUMNS_AND_PROJECTED_UNION" if logical in EXTENSION_REQUIRED_COLUMNS else "PASS_BATCH002_PARENT_OR_AUXILIARY_CONTRACT", "semantic_compatibility": ( "PASS_PROJECTED_UNION;PASS_DIRECT_INCREMENTAL_COMPANY_EXPOSURE_9" if logical == "newenergy_company_exposure_matrix" else "PASS_PROJECTED_UNION" if logical in LEGACY_EXTENSION_DATASETS else "PASS" ), "read_order": "2_BATCH002" if logical in baseline_registry else "1_BATCH002_ONLY", "immutable": "NO_PENDING_REVIEW", "review_status": REVIEW, }) write_csv(INDUSTRY / "manifest" / "canonical_shard_registry_BATCH002.csv", registry_rows) batch_summary = CASE / "manifest" / "batch_summary.md" write_text( batch_summary, f"""# BATCH-002 批次摘要 - task_id: `{TASK_ID}` - case_id: `{CASE_ID}` - batch_id: `{BATCH_ID}` - run_id: `{RUN_ID}` - design_audit_id: `AUDIT-ANA-NEWENERGY-FOUR-TRACK-ATLAS-BATCH002-DESIGN-REPAIR001-REREVIEW-20260806-001` - review_status: `{REVIEW}` ## 结果计数 - 子行业节点：16；节点正文：32；顶层视图：3。 - 增量企业映射：9；前批 selected 去重冲突：0。 - 空桶：7；有限补证队列：50；补证后新增资格通过：0。 - BATCH-002 新 source_document：1；BATCH-001 immutable source 引用：{len(used_source_ids) - 1}。 - case evidence map：{len(case_maps)}；人读输出（含 result index）：{len(output_paths)}；output manifest 另含 2 条条件逻辑数据集 N/A 登记。 ## 七组扩展数据集 - `newenergy_scope_matrix`：REQUIRED / 已生成 / {len(scope_rows)} 行。 - `newenergy_technology_route_matrix`：REQUIRED / 已生成 / {len(technology_rows)} 行。 - `newenergy_supply_demand_price_metric`：CONDITIONAL_REQUIRED_PER_NODE / 已生成 / {len(metric_rows)} 行；仅复用可比、已接受的官方事实。 - `newenergy_project_capacity_event`：CONDITIONAL_REQUIRED / 已生成 / {len(project_event_rows)} 行；只作前批事件再语境化，不新增项目事实。 - `newenergy_policy_market_mechanism_event`：NOT_APPLICABLE；本批正文没有使用新的政策机制强事实，因此不创建空表；理由、判断依据和 review status 已登记 output manifest / human validation。 - `newenergy_company_exposure_matrix`：REQUIRED / 已生成 / {len(exposure_rows)} 行，与 included incremental exact-match。 - `newenergy_catalyst_risk_register`：REQUIRED / 已生成 / {len(risk_rows)} 行。 ## BATCH-001 legacy-to-V1 只读适配 - 六个前序已接受 `newenergy_*` baseline 不改写、不复制合并；通过 `legacy_to_v1_adapter_contract_BATCH002.csv` 对每个 V1 目标字段冻结直接映射、重命名、常量/显式 GAP、枚举转换、主键和业务键投影。 - `legacy_to_v1_projection_validation_BATCH002.csv` 实际投影后再与 BATCH-002 shard 做只读 union 校验：V1 必需字段、行数、主键/业务键重复、跨 shard ID 碰撞、枚举和语义边界均为 PASS；不生成物化 union 副本。 - `legacy_period_projection_receipt_BATCH002.csv` 对 16 条前序指标逐行保存期间 token、解析模式和投影边界：ISO 区间原样保留，`YYYY` 展开为完整自然年，`YYYYYE` 以 `LEGACY_AS_OF_NO_START:*` + 年末日期分账；格式、顺序和 as-of 处理均计算验证。 ## darkline 执行与停止统计 - 模式：`广撒网收集 / EXTERNAL_PUBLIC_EVIDENCE_COMPLETION`；输入严格限于 16 节点、9 条既有合格余额和 7 个空桶冻结队列。 - REPAIR001 实际查询层：50 个 queue pair 逐行回指完整条件、查询时间、官方 URL、HTTP/失败状态、返回对象、原件定位和终止理由；49 个唯一官方 URL 均取得 HTTP 200。零候选桶另记 1 条 NOT_APPLICABLE，不宣称执行查询。 - 查询摘要证据进入正式结论：0；所有正式强事实回到公开原件、接受证据或本批新归档原件。 - 搜索引擎仅定位原始来源；访问控制绕过：0；行情/K线/市场反向补漏产物：0。 - 核电敏感内容保存/输出：0；只保留民用公开高层角色边界。 - `darkline_event_or_hypothesis_BATCH002`：NOT_APPLICABLE；未形成需单列的事件链或研究假设；理由、判断依据和 review status 已登记 output manifest / human validation。 ## 输出边界 本批不声明完整覆盖、份额、公司质量排序、估值、交易、收益或投资建议。证据不足的 7 个节点保持 GAP。执行/输出独立审核通过前不得回写父级终态或对外交付。 """, ) def validate_links(paths: list[Path]) -> tuple[int, list[str]]: link_pattern = re.compile(r"\[[^\]]+\]\(([^)]+)\)") checked = 0 broken: list[str] = [] for path in paths: text = path.read_text(encoding="utf-8") for target in link_pattern.findall(text): if re.match(r"^[a-z]+://", target, re.I) or target.startswith("#"): continue checked += 1 actual = (path.parent / target.split("#", 1)[0]).resolve() if not actual.exists(): broken.append(f"{rel(path)} -> {target}") return checked, broken all_b2_text_paths = ( list(CASE.rglob("*.md")) + [result_index] + [p for p in (INDUSTRY / "supplement").glob("NEB2_*") if p.is_file()] + list((INDUSTRY / "extracted").glob("*_BATCH002.csv")) + [INDUSTRY / "extracted" / "batch002_evidence_completion_queue.csv"] + list((INDUSTRY / "manifest").glob("*_BATCH002.csv")) + [INDUSTRY / "evidence" / "evidence_fact_table_BATCH002.csv"] ) replacement_count = 0 utf8_errors: list[str] = [] for path in all_b2_text_paths: try: content = path.read_text(encoding="utf-8-sig" if path.suffix == ".csv" else "utf-8") replacement_count += content.count("\ufffd") except UnicodeDecodeError: utf8_errors.append(rel(path)) link_count, broken_links = validate_links(output_paths + [batch_summary]) included_pairs = {(r["company_id"], r["track_code"]) for r in incremental_ledger_rows if r["batch002_state"] == "INCLUDED_INCREMENTAL"} baseline_selected_pairs = {(r["company_id"], r["track_code"]) for r in ledger if r["candidate_state"] in {"INCLUDED_T1", "INCLUDED_T2"}} exposure_pairs = {(r["company_id"], r["track_code"]) for r in exposure_rows} if included_pairs != exposure_pairs or included_pairs & baseline_selected_pairs: raise RuntimeError("incremental exact-match or predecessor de-dup failed") all_evidence_ids = set(evidence_by_id) | {yuhong_evf["evidence_fact_id"]} map_orphans = [m["conclusion_evidence_map_id"] for m in case_maps if m["evidence_fact_id"] not in all_evidence_ids] if map_orphans: raise RuntimeError(f"case map evidence orphans: {map_orphans}") if utf8_errors or replacement_count or broken_links: raise RuntimeError(f"text validation failed utf8={utf8_errors} u+fffd={replacement_count} links={broken_links}") cutoff_date = SOURCE_CUTOFF[:10] source_publish_after_cutoff = [] for source_id in used_source_ids: publish_date = (source_document_rows[0] if source_id == "S-NEB2-YUHONG-SUSTAINABILITY-2023" else source_by_id[source_id])["publish_date"][:10] if publish_date and publish_date > cutoff_date: source_publish_after_cutoff.append(source_id) if source_publish_after_cutoff: raise RuntimeError(f"source cutoff violated: {source_publish_after_cutoff}") def duplicate_count(rows: list[dict[str, Any]], key_fields: tuple[str, ...]) -> int: keys = [tuple(str(r.get(k, "")) for k in key_fields) for r in rows] return len(keys) - len(set(keys)) uniqueness_checks = { "classification_business_key_duplicates": duplicate_count( classification_rows, ("case_id", "subject_type", "subject_id", "track_code", "chain_node_id", "source_doc_id", "evidence_fact_id"), ), "scope_business_key_duplicates": duplicate_count( scope_rows, ("case_id", "object_type", "object_id", "track_code", "chain_node", "scope_version") ), "technology_business_key_duplicates": duplicate_count( technology_rows, ("case_id", "track_code", "route_name", "region", "metric_as_of") ), "metric_business_key_duplicates": duplicate_count( metric_rows, ("case_id", "metric_name", "object_id", "region", "period_start", "period_end", "unit", "price_or_volume_basis") ), "exposure_business_key_duplicates": duplicate_count( exposure_rows, ("case_id", "company_id", "track_code", "chain_node", "reporting_period") ), "case_map_business_key_duplicates": duplicate_count( case_maps, ("output_path", "section_anchor", "conclusion_text", "evidence_fact_id") ), "conversion_business_key_duplicates": duplicate_count(conversion_rows, ("source_doc_id", "run_id")), "input_business_key_duplicates": duplicate_count(input_rows, ("case_id", "source_doc_id", "run_id")), } if any(uniqueness_checks.values()): raise RuntimeError(f"business-key uniqueness failed: {uniqueness_checks}") risk_statement_type_invalid_count = sum( row["statement_type"] not in allowed_statement_types for row in risk_rows ) if risk_statement_type_invalid_count: raise RuntimeError(f"risk statement_type enum invalid: {risk_statement_type_invalid_count}") validation_receipt = { "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "latest_repair_id": LATEST_REPAIR_ID, "validated_at": EXECUTED_AT, "immutable_baseline_hashes": {rel(p): sha256(p) for p in BASELINE_HASHES}, "node_count": len(NODES), "node_human_doc_count": 32, "human_output_count_including_result_index": len(output_paths), "incremental_included_count": len(included_pairs), "predecessor_selected_overlap_count": len(included_pairs & baseline_selected_pairs), "empty_bucket_count": len(QUEUE_CODES), "bounded_queue_count": len(queue_rows), "queue_final_state_distribution": dict(Counter(r["final_state"] for r in queue_rows)), "queue_prior_official_report_reference_count": sum(bool(r["prior_source_doc_id"]) for r in queue_rows), "queue_prior_raw_hash_reference_count": sum(bool(r["prior_raw_sha256"]) for r in queue_rows), "classification_scope_distribution": dict(Counter(r["scope_type"] for r in classification_rows)), "source_document_batch002_count": len(source_document_rows), "source_input_total_count": len(input_rows), "conversion_status_total_count": len(conversion_rows), "source_gap_audit_total_count": len(source_gap_audit_rows), "technology_route_row_count": len(technology_rows), "atomic_market_metric_row_count": len(metric_rows), "project_event_row_count": len(project_event_rows), "company_exposure_row_count": len(exposure_rows), "risk_register_row_count": len(risk_rows), "risk_statement_type_allowed_enum": sorted(allowed_statement_types), "risk_statement_type_invalid_count": risk_statement_type_invalid_count, "legacy_adapter_mapping_count": len(adapter_contract_rows), "legacy_adapter_dataset_count": len(projection_validation_rows), "legacy_adapter_contract_sha256": sha256(LEGACY_ADAPTER_PATH), "legacy_projection_validation_sha256": sha256(LEGACY_PROJECTION_VALIDATION_PATH), "legacy_period_projection_receipt_row_count": len(period_projection_receipt_rows), "legacy_period_projection_receipt_sha256": sha256(LEGACY_PERIOD_PROJECTION_PATH), "legacy_projection_validation_pass_count": sum(r["projection_validation_status"] == "PASS" for r in projection_validation_rows), "legacy_projection_union_row_count": sum(int(r["projected_union_row_count"]) for r in projection_validation_rows), "legacy_projection_required_field_missing_count": sum(int(r["v1_required_field_missing_count"]) for r in projection_validation_rows), "legacy_projection_union_primary_key_duplicate_count": sum(int(r["union_primary_key_duplicate_count"]) for r in projection_validation_rows), "legacy_projection_union_business_key_duplicate_count": sum(int(r["union_business_key_duplicate_count"]) for r in projection_validation_rows), "legacy_projection_cross_shard_id_collision_count": sum(int(r["cross_shard_id_collision_count"]) for r in projection_validation_rows), "legacy_projection_enum_conversion_error_count": sum(int(r["enum_conversion_error_count"]) for r in projection_validation_rows), "legacy_projection_parse_source_unrecognized_count": sum(int(r["parse_source_unrecognized_count"]) for r in projection_validation_rows), "legacy_projection_parse_output_format_violation_count": sum(int(r["parse_output_format_violation_count"]) for r in projection_validation_rows), "legacy_projection_parse_order_violation_count": sum(int(r["parse_order_violation_count"]) for r in projection_validation_rows), "legacy_projection_parse_as_of_handling_violation_count": sum(int(r["parse_as_of_handling_violation_count"]) for r in projection_validation_rows), "legacy_projection_materialized_union_copy": "NO", "source_publish_after_cutoff_count": len(source_publish_after_cutoff), "actual_query_receipt_row_count": len(actual_query_rows), "actual_queue_pair_query_count": sum(r["queue_item_id"] != "NOT_APPLICABLE_ZERO_CANDIDATE" for r in actual_query_rows), "actual_query_response_or_failure_recorded_count": sum(bool(str(r["response_result"])) for r in actual_query_rows), "actual_unique_official_url_probe_count": len(probe_by_url), "conditional_not_applicable_output_manifest_count": sum(r["applicability"] == "NOT_APPLICABLE" for r in output_manifest_rows), "conditional_not_applicable_human_validation_count": sum(r["applicability"] == "NOT_APPLICABLE" for r in human_validation_rows), "business_key_duplicate_checks": uniqueness_checks, "case_evidence_map_count": len(case_maps), "case_evidence_map_orphan_count": len(map_orphans), "local_link_count": link_count, "broken_local_link_count": len(broken_links), "utf8_error_count": len(utf8_errors), "replacement_character_count": replacement_count, "market_reverse_artifact_count": 0, "valuation_or_kline_artifact_count": 0, "access_control_bypass_count": 0, "nuclear_sensitive_content_count": 0, "outputs_status": REVIEW, "review_status": REVIEW, } validation_path = CASE / "manifest" / "execution_validation_receipt.json" write_json(validation_path, validation_receipt) for row in human_validation_rows: if row["applicability"] == "APPLICABLE_FILE_OUTPUT": row["local_link_scan"] = "PASS" write_csv(CASE / "manifest" / "human_doc_validation_receipt.csv", human_validation_rows) package_receipt = CASE / "manifest" / "package_exact_set_receipt.md" def new_artifact_candidates() -> list[Path]: paths: set[Path] = set() paths.update(CASE.rglob("*")) paths.update(RESULT.rglob("*")) paths.update((INDUSTRY / "extracted").glob("*_BATCH002.csv")) paths.add(INDUSTRY / "extracted" / "batch002_evidence_completion_queue.csv") paths.update((INDUSTRY / "manifest").glob("*_BATCH002.csv")) paths.add(INDUSTRY / "evidence" / "evidence_fact_table_BATCH002.csv") paths.update((INDUSTRY / "supplement").glob("NEB2_*")) paths.add(yuhong_raw) paths.add(yuhong_txt) paths.add(Path(__file__).resolve()) paths.add(ROOT / "ana-data" / "tools" / "newenergy_batch002_query_probe.py") return sorted(p for p in paths if p.is_file() and p != INDUSTRY / "manifest" / "artifact_manifest_BATCH002.csv") pre_receipt_candidates = new_artifact_candidates() expected_exact_count = len(pre_receipt_candidates) + (0 if package_receipt in pre_receipt_candidates else 1) write_text( package_receipt, f"""# BATCH-002 package exact-set 回执 - task_id: `{TASK_ID}` - case_id: `{CASE_ID}` - batch_id: `{BATCH_ID}` - run_id: `{RUN_ID}` - review_status: `{REVIEW}` - expected_artifact_exact_set_count: `{expected_exact_count}` - artifact_manifest_self_exclusion: `YES / 自身哈希不能递归登记` - package_receipt_included_in_artifact_manifest: `YES` - BATCH-001_1630_artifacts_relisted: `NO / 仅通过 case_input_reference_manifest 引用` exact-set 包括本批新行业 shard、新 raw/converted、supplement 查询回执、案例 outputs/manifest/evidence、result 入口和获批 replay tool；不包括空 img/tmp 目录，也不复制前批 artifacts。 """, ) artifact_paths = new_artifact_candidates() artifact_manifest_path = INDUSTRY / "manifest" / "artifact_manifest_BATCH002.csv" artifact_rows = [] for idx, path in enumerate(artifact_paths, 1): path_rel = rel(path) if "/raw/" in "/" + path_rel: artifact_type = "RAW_PUBLIC_PRIMARY_SOURCE" elif "/converted/" in "/" + path_rel: artifact_type = "CONVERTED_TEXT" elif path.suffix.lower() == ".md": artifact_type = "HUMAN_READABLE_OUTPUT_OR_RECEIPT" elif path.suffix.lower() == ".csv": artifact_type = "STRUCTURED_CANONICAL_OR_MANIFEST" elif path.suffix.lower() == ".json": artifact_type = "RECEIPT_OR_VALIDATION" elif path.suffix.lower() == ".py": artifact_type = "REPLAY_TOOL" else: artifact_type = "OTHER" artifact_rows.append({ "artifact_id": f"NEB2-ART-{idx:04d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "artifact_type": artifact_type, "industry_case": "新能源案例", "industry_id": "IND-NEWENERGY", "subindustry_id": "", "company_id": "", "logical_path": path_rel, "relative_path": path_rel, "absolute_path": str(path.resolve()), "file_name": path.name, "file_ext": path.suffix.lower(), "file_size": path.stat().st_size, "sha256": sha256(path), "source_doc_id": "S-NEB2-YUHONG-SUSTAINABILITY-2023" if path in {yuhong_raw, yuhong_txt} else "", "source_url": "https://www.yuhong.com.cn/uploads/soft/240422/1-240422101057.pdf" if path in {yuhong_raw, yuhong_txt} else "", "source_collected_at": EXECUTED_AT if path in {yuhong_raw, yuhong_txt} else "", "raw_pool_path": rel(yuhong_raw.parent) + "/" if path in {yuhong_raw, yuhong_txt} else "", "source_file_name": yuhong_raw.name if path in {yuhong_raw, yuhong_txt} else "", "detected_type": path.suffix.lower().lstrip("."), "archive_file_name": path.name, "extension_added_by_archive_flag": "NO", "extension_mismatch_flag": "NO", "created_at": EXECUTED_AT, "created_by": "case_analysis.analyst.new_energy", "tool_or_method": "newenergy_batch002_build.py", "tool_version": "BATCH002-V1", "parameters_summary": "exact released batch/run; immutable B001 references; DRAFT outputs", "source_snapshot_id": "BATCH001_ACCEPTED" if "BATCH001" in path.name else "BATCH002_EXECUTION", "artifact_status": REVIEW, "sensitivity_screen": "LEGAL_PUBLIC_CIVIL_NUCLEAR_HIGH_LEVEL_ONLY", "schema_version": SCHEMA, "review_status": REVIEW, }) write_csv(artifact_manifest_path, artifact_rows) manifest_set = {r["relative_path"] for r in artifact_rows} actual_set = {rel(p) for p in new_artifact_candidates()} if manifest_set != actual_set: raise RuntimeError(f"artifact exact-set mismatch missing={actual_set-manifest_set} extra={manifest_set-actual_set}") for row in artifact_rows: path = ROOT / Path(row["relative_path"]) if path.stat().st_size != int(row["file_size"]) or sha256(path) != row["sha256"]: raise RuntimeError(f"artifact hash mismatch: {row['relative_path']}") result = { "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "status": REVIEW, "nodes": 16, "node_docs": 32, "incremental_companies": len(incremental), "empty_buckets": len(QUEUE_CODES), "bounded_queue_rows": len(queue_rows), "source_documents_new": len(source_document_rows), "source_inputs_total": len(input_rows), "evidence_fact_rows_new": 1, "case_evidence_map_rows": len(case_maps), "human_outputs_including_result_index": len(output_paths), "artifact_exact_set": len(artifact_rows), "artifact_manifest_sha256": sha256(artifact_manifest_path), "output_manifest_sha256": sha256(CASE / "manifest" / "output_manifest.csv"), "validation_receipt_sha256": sha256(validation_path), "result_index_sha256": sha256(result_index), "baseline_hashes_stable": True, "utf8_errors": 0, "replacement_characters": 0, "broken_local_links": 0, "parent_terminal_writeback": "DENIED_PENDING_INDEPENDENT_EXECUTION_OUTPUT_REVIEW", } print(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True))