#!/usr/bin/env python3
|
"""Rebuild the governed BATCH-001 new-energy evidence package after HOLD/3.
|
|
The script is intentionally deterministic at the transformation layer. Network
|
snapshots are timestamped and hashed; all downstream CSV/Markdown files are
|
rebuilt from those snapshots plus the already archived 38 official sources.
|
"""
|
|
from __future__ import annotations
|
|
import csv
|
import hashlib
|
import html
|
import io
|
import json
|
import math
|
import re
|
import shutil
|
import sys
|
import time
|
import urllib.parse
|
import urllib.request
|
from collections import defaultdict
|
from datetime import datetime
|
from html.parser import HTMLParser
|
from pathlib import Path
|
from zoneinfo import ZoneInfo
|
|
import openpyxl # noqa: F401 - dependency check and XLSX header validation
|
from pypdf import PdfReader
|
|
|
TASK_ID = "TASK-NEWENERGY-FOUR-TRACK-ATLAS-20260805-001"
|
CASE_ID = "ANA-NEWENERGY-FOUR-TRACK-ATLAS-20260805-001"
|
BATCH_ID = "BATCH-001"
|
RUN_ID = "RUN-ANA-NEWENERGY-FOUR-TRACK-ATLAS-20260805-001-BATCH-001-001"
|
SCHEMA_VERSION = "NEWENERGY_EXTENSION_V1"
|
REVIEW_STATUS = "DRAFT_FOR_REVIEW"
|
SOURCE_CUTOFF = "2026-08-05T23:59:59+08:00"
|
AS_OF_DATE = "2026-08-05"
|
COLLECTED_AT = datetime.now(ZoneInfo("Asia/Shanghai")).replace(microsecond=0).isoformat()
|
ARTIFACT_TOOL_VERSION = "REPAIR-005"
|
ARTIFACT_PARAMETERS_SUMMARY = (
|
"B1 REPAIR005; reuse 729 acquired PDFs; replay 858 pairs through the "
|
"company-self and target-bucket semantic role gate; clear non-eligible "
|
"derived fields and dangling evidence FKs; mechanical rerank; exact-set hash coverage"
|
)
|
ARTIFACT_STATUS = "READY_FOR_REPAIR005_FOCUSED_REREVIEW"
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
INDUSTRY_ROOT = PROJECT_ROOT / "ana-data/cases/新能源案例"
|
CASE_ROOT = INDUSTRY_ROOT / CASE_ID
|
RESULT_ROOT = PROJECT_ROOT / f"ana-data/result/新能源案例/{CASE_ID}"
|
RAW_ROOT = INDUSTRY_ROOT / "raw"
|
CONVERTED_ROOT = INDUSTRY_ROOT / "converted"
|
EXTRACTED_ROOT = INDUSTRY_ROOT / "extracted"
|
SUPPLEMENT_ROOT = INDUSTRY_ROOT / "supplement"
|
EVIDENCE_ROOT = INDUSTRY_ROOT / "evidence"
|
MANIFEST_ROOT = INDUSTRY_ROOT / "manifest"
|
CASE_EVIDENCE = CASE_ROOT / "evidence"
|
CASE_MANIFEST = CASE_ROOT / "manifest"
|
CASE_OUTPUTS = CASE_ROOT / "outputs"
|
|
OLD_CASE_EVIDENCE = CASE_EVIDENCE
|
OLD_CASE_MANIFEST = CASE_MANIFEST
|
|
HEADERS = {
|
"artifact_manifest": [
|
"artifact_id", "task_id", "case_id", "batch_id", "run_id", "artifact_type",
|
"industry_case", "industry_id", "subindustry_id", "company_id", "logical_path",
|
"relative_path", "absolute_path", "file_name", "file_ext", "file_size", "sha256",
|
"source_doc_id", "source_url", "source_collected_at", "raw_pool_path",
|
"source_file_name", "detected_type", "archive_file_name",
|
"extension_added_by_archive_flag", "extension_mismatch_flag", "created_at",
|
"created_by", "tool_or_method", "tool_version", "parameters_summary",
|
"source_snapshot_id", "artifact_status", "sensitivity_screen",
|
],
|
"conversion_status": [
|
"conversion_id", "task_id", "case_id", "batch_id", "run_id", "source_doc_id",
|
"raw_pool_path", "raw_file_path", "raw_file_sha256", "detected_type",
|
"conversion_method", "parameters_summary", "converted_text_path",
|
"converted_markdown_path", "converted_path", "converted_sha256",
|
"page_or_duration_count", "status", "error_code", "error_summary", "created_at",
|
],
|
"universe": [
|
"universe_row_id", "exchange_code", "security_code", "security_name", "legal_name",
|
"listing_status", "listing_date", "board", "source_doc_id", "source_record_locator",
|
"as_of_date", "source_url", "raw_path", "raw_sha256", "review_status",
|
],
|
"human_receipt": [
|
"validation_item_id", "task_id", "case_id", "batch_id", "run_id", "view_type",
|
"output_path", "document_status", "not_applicable_reason", "evidence_boundary",
|
"source_count", "evidence_count", "conclusion_count", "unknown_count",
|
"review_status", "validated_by", "validated_at",
|
],
|
}
|
|
QUERY_CONFIG = [
|
("BATTERY", "A", "资源与主材", "正极材料"),
|
("BATTERY", "B", "电芯制造", "锂离子电池"),
|
("BATTERY", "C", "系统/部件/BMS-Pack", "电池管理系统"),
|
("BATTERY", "D", "设备与回收循环", "锂电设备"),
|
("SOLAR", "A", "硅料/硅片与材料", "光伏硅片"),
|
("SOLAR", "B", "电池片/组件", "光伏组件"),
|
("SOLAR", "C", "设备/辅材/逆变器", "光伏逆变器"),
|
("SOLAR", "D", "系统集成/电站建设运营", "光伏电站"),
|
("WIND", "A", "材料与关键零部件", "风电零部件"),
|
("WIND", "B", "整机", "风力发电机组"),
|
("WIND", "C", "塔筒/海缆/工程配套", "风电塔筒"),
|
("WIND", "D", "项目运营与运维服务", "风电场"),
|
("NUCLEAR", "A", "运营商", "核电运营"),
|
("NUCLEAR", "B", "工程/EPC", "核电工程"),
|
("NUCLEAR", "C", "核岛/常规岛主设备", "核电设备"),
|
("NUCLEAR", "D", "核级部件/材料/仪控电气", "核级阀门"),
|
]
|
|
TRACK_ORDER = {"BATTERY": 0, "SOLAR": 1, "WIND": 2, "NUCLEAR": 3}
|
BUCKET_ORDER = {(track, bucket): i for i, (track, _, bucket, _) in enumerate(QUERY_CONFIG)}
|
|
|
def rel(path: Path) -> str:
|
return path.resolve().relative_to(PROJECT_ROOT.resolve()).as_posix()
|
|
|
def sha256_bytes(data: bytes) -> str:
|
return hashlib.sha256(data).hexdigest()
|
|
|
def sha256_file(path: Path) -> str:
|
h = hashlib.sha256()
|
with path.open("rb") as f:
|
for chunk in iter(lambda: f.read(1024 * 1024), b""):
|
h.update(chunk)
|
return h.hexdigest()
|
|
|
def write_bytes(path: Path, data: bytes) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
path.write_bytes(data)
|
|
|
def write_text(path: Path, text: str) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
path.write_text(text.replace("\r\n", "\n"), encoding="utf-8", newline="\n")
|
|
|
def read_csv(path: Path) -> list[dict[str, str]]:
|
with path.open("r", encoding="utf-8-sig", newline="") as f:
|
return list(csv.DictReader(f))
|
|
|
def write_csv(path: Path, fieldnames: list[str], rows: list[dict[str, object]]) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
with path.open("w", encoding="utf-8", newline="") as f:
|
w = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore", quoting=csv.QUOTE_ALL)
|
w.writeheader()
|
for row in rows:
|
w.writerow({k: "" if row.get(k) is None else row.get(k, "") for k in fieldnames})
|
|
|
def fetch(url: str, *, method: str = "GET", body: bytes | None = None, referer: str = "") -> bytes:
|
headers = {
|
"User-Agent": "Mozilla/5.0",
|
"Accept": "application/json, text/plain, */*",
|
"Referer": referer or url,
|
"X-Requested-With": "XMLHttpRequest",
|
}
|
req = urllib.request.Request(url, data=body, headers=headers, method=method)
|
for attempt in range(5):
|
try:
|
with urllib.request.urlopen(req, timeout=45) as response:
|
return response.read()
|
except Exception:
|
if attempt == 4:
|
raise
|
time.sleep(1.5 * (attempt + 1))
|
raise RuntimeError("unreachable")
|
|
|
class TextExtractor(HTMLParser):
|
def __init__(self) -> None:
|
super().__init__()
|
self.parts: list[str] = []
|
|
def handle_data(self, data: str) -> None:
|
value = re.sub(r"\s+", " ", data).strip()
|
if value:
|
self.parts.append(value)
|
|
|
def html_to_text(data: bytes) -> str:
|
text = data.decode("utf-8", errors="replace")
|
parser = TextExtractor()
|
parser.feed(text)
|
return "\n".join(parser.parts) + "\n"
|
|
|
def clean_html_text(value: str) -> str:
|
return html.unescape(re.sub(r"<[^>]+>", "", value or "")).strip()
|
|
|
def clean_annual_title(title: str) -> bool:
|
compact = re.sub(r"\s+", "", clean_html_text(title))
|
if "2025年年度报告" not in compact:
|
return False
|
excluded = ["摘要", "审计", "问询", "回复", "说明", "意见", "更正公告", "英文版"]
|
return not any(token in compact for token in excluded)
|
|
|
def ensure_roots() -> None:
|
for p in [
|
CONVERTED_ROOT / "official_filings", CONVERTED_ROOT / "official_market",
|
CONVERTED_ROOT / "official_discovery", EXTRACTED_ROOT, SUPPLEMENT_ROOT,
|
EVIDENCE_ROOT, MANIFEST_ROOT, CASE_EVIDENCE, CASE_MANIFEST,
|
RAW_ROOT / "official_universe", RAW_ROOT / "official_discovery",
|
]:
|
p.mkdir(parents=True, exist_ok=True)
|
|
|
def collect_sse_universe() -> tuple[list[dict[str, str]], Path, str]:
|
endpoint = "https://query.sse.com.cn/sseQuery/commonQuery.do"
|
payloads = []
|
records: list[dict[str, str]] = []
|
for stock_type, board in [("1", "SSE_MAIN"), ("8", "SSE_STAR")]:
|
params = {
|
"STOCK_TYPE": stock_type, "REG_PROVINCE": "", "CSRC_CODE": "",
|
"STOCK_CODE": "", "sqlId": "COMMON_SSE_CP_GPJCTPZ_GPLB_GP_L",
|
"COMPANY_STATUS": "2,4,5,7,8", "type": "inParams", "isPagination": "true",
|
"pageHelp.cacheSize": "1", "pageHelp.beginPage": "1",
|
"pageHelp.pageSize": "5000", "pageHelp.pageNo": "1",
|
}
|
url = endpoint + "?" + urllib.parse.urlencode(params)
|
obj = json.loads(fetch(url, referer="https://www.sse.com.cn/assortment/stock/home/"))
|
payloads.append({"request_url": url, "response": obj})
|
for item in obj.get("result", []):
|
code = str(item.get("A_STOCK_CODE", "")).strip()
|
if not re.fullmatch(r"\d{6}", code):
|
continue
|
records.append({
|
"exchange_code": "SSE", "security_code": code,
|
"security_name": str(item.get("COMPANY_ABBR", "")).strip(),
|
"legal_name": str(item.get("FULL_NAME", "")).strip(),
|
"listing_status": "LISTED", "listing_date": str(item.get("LIST_DATE", "")).strip(),
|
"board": board, "locator": f"STOCK_TYPE={stock_type};A_STOCK_CODE={code}",
|
})
|
wrapper = {
|
"snapshot_id": "S-UNIVERSE-SSE-20260805", "as_of_date": AS_OF_DATE,
|
"source": "上海证券交易所股票列表官方查询接口", "payloads": payloads,
|
}
|
path = RAW_ROOT / "official_universe/SSE_A_SHARE_UNIVERSE_20260805.json"
|
write_text(path, json.dumps(wrapper, ensure_ascii=False, indent=2) + "\n")
|
return records, path, "https://www.sse.com.cn/assortment/stock/home/"
|
|
|
def collect_szse_universe() -> tuple[list[dict[str, str]], Path, str]:
|
endpoint = "https://www.szse.cn/api/report/ShowReport"
|
referer = "https://www.szse.cn/market/stock/company/"
|
params = {"SHOWTYPE": "xlsx", "CATALOGID": "1110x", "TABKEY": "tab1"}
|
url = endpoint + "?" + urllib.parse.urlencode(params)
|
raw_bytes = fetch(url, referer=referer)
|
if not raw_bytes.startswith(b"PK\x03\x04"):
|
raise RuntimeError("SZSE company-list download is not an XLSX/ZIP payload")
|
path = RAW_ROOT / "official_universe/SZSE_A_SHARE_UNIVERSE_20260805.xlsx"
|
write_bytes(path, raw_bytes)
|
# The official workbook advertises an incorrect A1:A1 worksheet dimension;
|
# normal mode is required for openpyxl to discover all physical rows.
|
workbook = openpyxl.load_workbook(io.BytesIO(raw_bytes), read_only=False, data_only=True)
|
worksheet = workbook.active
|
records: list[dict[str, str]] = []
|
for row_no, values in enumerate(worksheet.iter_rows(values_only=True), 1):
|
code = str(values[0] or "").strip().zfill(6)
|
if not re.fullmatch(r"(?:00|30)\d{4}", code):
|
continue
|
records.append({
|
"exchange_code": "SZSE", "security_code": code,
|
"security_name": "", "legal_name": "", "listing_status": "LISTED", "listing_date": "",
|
"board": "SZSE_CHINEXT" if code.startswith("30") else "SZSE_MAIN",
|
"locator": f"xlsx_row={row_no};security_code={code}",
|
})
|
return records, path, referer
|
|
|
def collect_bse_universe() -> tuple[list[dict[str, str]], Path, str]:
|
endpoint = "https://www.bse.cn/nqxxController/nqxxCnzq.do"
|
referer = "https://www.bse.cn/nq/listedcompany.html"
|
payloads = []
|
records: list[dict[str, str]] = []
|
page = 0
|
total_pages = None
|
while total_pages is None or page < total_pages:
|
body = urllib.parse.urlencode([
|
("page", str(page)), ("typejb", "T"), ("xxfcbj[]", "2"), ("xxzqdm", ""),
|
("sortfield", "xxzqdm"), ("sorttype", "asc"), ("callback", "cb"),
|
]).encode("ascii")
|
raw = fetch(endpoint, method="POST", body=body, referer=referer).decode("utf-8")
|
if not raw.startswith("cb(") or not raw.endswith(")"):
|
raise RuntimeError("Unexpected BSE JSONP envelope")
|
obj = json.loads(raw[3:-1])
|
page_obj = obj[0]
|
content = page_obj["content"]
|
total_pages = int(page_obj["totalPages"])
|
payloads.append({"request": urllib.parse.parse_qs(body.decode("ascii")), "response": obj})
|
for item in content:
|
code = str(item.get("xxzqdm", "")).strip()
|
if not re.fullmatch(r"92\d{4}", code):
|
continue
|
records.append({
|
"exchange_code": "BSE", "security_code": code,
|
"security_name": str(item.get("xxzqjc", "")).strip(), "legal_name": "",
|
"listing_status": "LISTED", "listing_date": str(item.get("fxssrq", "")).strip(),
|
"board": "BSE", "locator": f"page={page};xxzqdm={code}",
|
})
|
page += 1
|
wrapper = {
|
"snapshot_id": "S-UNIVERSE-BSE-20260805", "as_of_date": AS_OF_DATE,
|
"source": "北京证券交易所股票列表官方查询接口", "payloads": payloads,
|
}
|
path = RAW_ROOT / "official_universe/BSE_A_SHARE_UNIVERSE_20260805.json"
|
write_text(path, json.dumps(wrapper, ensure_ascii=False, indent=2) + "\n")
|
return records, path, referer
|
|
|
def normalize_universe() -> tuple[list[dict[str, str]], dict[str, dict[str, str]], list[dict[str, str]]]:
|
sources = []
|
all_records: list[dict[str, str]] = []
|
for collector, source_id, org in [
|
(collect_sse_universe, "S-UNIVERSE-SSE-20260805", "上海证券交易所"),
|
(collect_szse_universe, "S-UNIVERSE-SZSE-20260805", "深圳证券交易所"),
|
(collect_bse_universe, "S-UNIVERSE-BSE-20260805", "北京证券交易所"),
|
]:
|
records, raw_path, source_url = collector()
|
raw_hash = sha256_file(raw_path)
|
for rec in records:
|
rec.update({"source_doc_id": source_id, "source_url": source_url,
|
"raw_path": rel(raw_path), "raw_sha256": raw_hash})
|
all_records.extend(records)
|
sources.append({
|
"doc_id": source_id, "title": f"{org}A股上市公司基准快照（{AS_OF_DATE}）",
|
"source_org": org, "author": org, "source_url": source_url,
|
"raw_path": raw_path, "raw_hash": raw_hash, "doc_type": "OFFICIAL_A_SHARE_UNIVERSE_SNAPSHOT",
|
"subindustry_id": "", "sensitivity": "LEGAL_PUBLIC_SCREENED",
|
})
|
dedup: dict[tuple[str, str], dict[str, str]] = {}
|
for rec in all_records:
|
key = (rec["exchange_code"], rec["security_code"])
|
if key in dedup:
|
raise RuntimeError(f"Duplicate universe key: {key}")
|
dedup[key] = rec
|
rows = []
|
for i, rec in enumerate(sorted(dedup.values(), key=lambda r: (r["exchange_code"], r["security_code"])), 1):
|
rows.append({
|
"universe_row_id": f"UNIV-{i:05d}", **rec, "as_of_date": AS_OF_DATE,
|
"review_status": REVIEW_STATUS,
|
})
|
write_csv(EXTRACTED_ROOT / "a_share_universe.csv", HEADERS["universe"], rows)
|
by_code: dict[str, dict[str, str]] = {}
|
for rec in rows:
|
code = rec["security_code"]
|
if code in by_code:
|
raise RuntimeError(f"Cross-exchange duplicate security code: {code}")
|
by_code[code] = rec
|
return rows, by_code, sources
|
|
|
def collect_discovery_queries(universe_by_code: dict[str, dict[str, str]]) -> tuple[dict[tuple[str, str], list[dict]], list[dict[str, str]], list[dict]]:
|
endpoint = "https://www.cninfo.com.cn/new/fulltextSearch/full"
|
referer = "https://www.cninfo.com.cn/new/fulltextSearch?searchType=1"
|
hits: dict[tuple[str, str], list[dict]] = defaultdict(list)
|
source_records: list[dict[str, str]] = []
|
summaries: list[dict] = []
|
for track, bucket_code, bucket, keyword in QUERY_CONFIG:
|
pages = []
|
page = 1
|
total_pages = None
|
total_records = None
|
archived_records = 0
|
while total_records is None or archived_records < total_records:
|
params = {
|
"searchkey": keyword, "sdate": "2026-03-01", "edate": "2026-05-15",
|
"isfulltext": "true", "sortName": "pubdate", "sortType": "asc",
|
"pageNum": str(page), "pageSize": "100", "type": "szb,cyb,hzb,kcb,bjs",
|
}
|
url = endpoint + "?" + urllib.parse.urlencode(params)
|
response_bytes = fetch(url, referer=referer)
|
obj = json.loads(response_bytes)
|
if total_pages is None:
|
total_pages = int(obj.get("totalpages") or 0)
|
total_records = int(obj.get("totalRecordNum") or 0)
|
elif int(obj.get("totalRecordNum") or 0) != total_records:
|
raise RuntimeError(f"CNINFO totalRecordNum changed during pagination: {track}/{bucket_code}")
|
announcements = obj.get("announcements") or []
|
if not announcements and archived_records < total_records:
|
raise RuntimeError(
|
f"CNINFO empty page before total reached: {track}/{bucket_code} "
|
f"page={page} archived={archived_records} total={total_records}"
|
)
|
# Preserve the annual-report full-text hit context used by the
|
# REPAIR003 qualification attempt. CNINFO occasionally returns
|
# U+FFFD inside its own snippets; make those source positions
|
# explicit without retaining the replacement character itself.
|
compact_announcements = []
|
for ann in announcements:
|
announcement_content = str(ann.get("announcementContent") or "")
|
compact_announcements.append({
|
key: ann.get(key)
|
for key in [
|
"announcementId", "announcementTime", "announcementTitle",
|
"secCode", "secName", "adjunctUrl",
|
]
|
})
|
compact_announcements[-1]["announcementContent_sanitized"] = announcement_content.replace(
|
"\ufffd", "<CNINFO_SOURCE_U_FFFD>"
|
)
|
compact_announcements[-1]["source_replacement_character_count"] = announcement_content.count("\ufffd")
|
pages.append({
|
"request_url": url,
|
"response_byte_count": len(response_bytes),
|
"response_sha256": hashlib.sha256(response_bytes).hexdigest(),
|
"response": {
|
"totalpages": obj.get("totalpages"),
|
"totalRecordNum": obj.get("totalRecordNum"),
|
"announcements": compact_announcements,
|
},
|
})
|
for ann in announcements:
|
if not clean_annual_title(str(ann.get("announcementTitle", ""))):
|
continue
|
code = str(ann.get("secCode", "")).strip()
|
if code not in universe_by_code:
|
continue
|
announcement_content = str(ann.get("announcementContent") or "")
|
hit = {
|
"track": track, "bucket_code": bucket_code, "bucket": bucket,
|
"keyword": keyword, "source_doc_id": f"S-DISCOVERY-{track}-{bucket_code}-20260805",
|
"security_code": code, "announcement_id": str(ann.get("announcementId", "")),
|
"security_name": str(ann.get("secName", "")).strip(),
|
"announcement_time": str(ann.get("announcementTime", "")),
|
"announcement_title": clean_html_text(str(ann.get("announcementTitle", ""))),
|
"adjunct_url": str(ann.get("adjunctUrl", "")),
|
"announcement_content_sanitized": announcement_content.replace(
|
"\ufffd", "<CNINFO_SOURCE_U_FFFD>"
|
),
|
"source_replacement_character_count": announcement_content.count("\ufffd"),
|
}
|
hits[(code, track)].append(hit)
|
archived_records += len(announcements)
|
page += 1
|
if archived_records != total_records:
|
raise RuntimeError(
|
f"CNINFO archived record mismatch: {track}/{bucket_code} "
|
f"archived={archived_records} total={total_records}"
|
)
|
wrapper = {
|
"snapshot_id": f"S-DISCOVERY-{track}-{bucket_code}-20260805",
|
"as_of_date": AS_OF_DATE, "source_cutoff_at": SOURCE_CUTOFF,
|
"purpose": "candidate_discovery_and_uniform_annual_report_context_qualification_attempt_not_direct_business_evidence",
|
"query_contract": {
|
"keyword": keyword, "track_code": track, "selection_bucket": bucket,
|
"sdate": "2026-03-01", "edate": "2026-05-15", "isfulltext": True,
|
"sortName": "pubdate", "sortType": "asc", "pageSize": 100,
|
"type": "szb,cyb,hzb,kcb,bjs",
|
"annual_report_filter": "clean title contains 2025年年度报告; exclude 摘要/审计/问询/回复/说明/意见/更正公告/英文版",
|
"a_share_identity_filter": "must join official SSE/SZSE/BSE as-of universe",
|
},
|
"server_total_record_num": total_records, "server_total_pages": total_pages,
|
"server_total_pages_semantics": "LAST_FULL_PAGE_NUMBER; FINAL_REMAINDER_IS_PAGE_PLUS_ONE",
|
"archived_record_count": archived_records,
|
"archived_page_count": len(pages),
|
"terminal_page_num": len(pages),
|
"terminal_page_record_count": len(pages[-1]["response"]["announcements"]) if pages else 0,
|
"termination_reason": "ARCHIVED_RECORD_COUNT_EQUALS_SERVER_TOTAL_RECORD_NUM",
|
"pages": pages,
|
}
|
raw_path = RAW_ROOT / f"official_discovery/{track}_{bucket_code}_CNINFO_2025_AR_QUERY_20260805.json"
|
write_text(raw_path, json.dumps(wrapper, ensure_ascii=False, indent=2) + "\n")
|
source_id = f"S-DISCOVERY-{track}-{bucket_code}-20260805"
|
raw_hash = sha256_file(raw_path)
|
hit_pairs = {key for key, values in hits.items() if any(v["source_doc_id"] == source_id for v in values)}
|
summary = {
|
"source_doc_id": source_id, "track_code": track, "bucket_code": bucket_code,
|
"selection_bucket": bucket, "keyword": keyword, "server_total_record_num": total_records,
|
"server_total_pages": total_pages, "annual_report_a_share_unique_pair_hits": len(hit_pairs),
|
"archived_record_count": archived_records, "archived_page_count": len(pages),
|
"terminal_page_num": len(pages),
|
"terminal_page_record_count": len(pages[-1]["response"]["announcements"]) if pages else 0,
|
"termination_reason": "ARCHIVED_RECORD_COUNT_EQUALS_SERVER_TOTAL_RECORD_NUM",
|
"raw_path": rel(raw_path), "raw_sha256": raw_hash,
|
}
|
summaries.append(summary)
|
source_records.append({
|
"doc_id": source_id, "title": f"巨潮资讯2025年年度报告全文候选发现回执：{track}/{bucket}/{keyword}",
|
"source_org": "巨潮资讯网", "author": "巨潮资讯网", "source_url": referer,
|
"raw_path": raw_path, "raw_hash": raw_hash, "doc_type": "OFFICIAL_DISCOVERY_QUERY_RECEIPT",
|
"subindustry_id": track, "sensitivity": "LEGAL_PUBLIC_SCREENED",
|
})
|
return hits, source_records, summaries
|
|
|
def build_candidate_ledger(
|
current_candidates: list[dict[str, str]], hits: dict[tuple[str, str], list[dict]],
|
universe_by_code: dict[str, dict[str, str]], source_docs_by_id: dict[str, dict[str, str]],
|
) -> list[dict[str, str]]:
|
current_by_key = {(r["security_code"], r["track_code"]): dict(r) for r in current_candidates}
|
all_keys = set(current_by_key) | set(hits)
|
rows: list[dict[str, str]] = []
|
for code, track in all_keys:
|
if (code, track) in current_by_key:
|
row = current_by_key[(code, track)]
|
row["candidate_discovery_channel"] = "OFFICIAL_A_SHARE_UNIVERSE_CNINFO_FULLTEXT_AND_DIRECT_REPORT_VERIFICATION"
|
row["include_or_exclude_reason"] = (
|
"直接业务年报页级证据通过；同桶只对ELIGIBLE对象按来源等级、暴露具体性、披露期间、"
|
"年报发布日期、交易所和证券代码机械排序；未取得页级直接业务证据的查询命中保持HELD，不参与排序"
|
)
|
rows.append(row)
|
continue
|
universe = universe_by_code[code]
|
pair_hits = sorted(hits[(code, track)], key=lambda x: (BUCKET_ORDER[(track, x["bucket"])], x["announcement_time"], x["announcement_id"]))
|
chosen = pair_hits[0]
|
source_ids = sorted({x["source_doc_id"] for x in pair_hits})
|
locators = sorted({f"{x['announcement_id']}@{x['announcement_time']}" for x in pair_hits if x["announcement_id"]})
|
exchange = universe["exchange_code"]
|
rows.append({
|
"candidate_id": f"CAND-DISC-{track}-{code}", "task_id": TASK_ID, "case_id": CASE_ID,
|
"batch_id": BATCH_ID, "run_id": RUN_ID,
|
"company_id": f"CN_A:{exchange}:{code}", "security_code": code,
|
"security_name": universe["security_name"] or chosen.get("security_name", ""), "legal_name": universe["legal_name"],
|
"exchange_code": exchange, "listed_status": "LISTED", "identity_as_of": AS_OF_DATE,
|
"track_code": track, "chain_nodes": "", "candidate_discovery_channel": "CNINFO_OFFICIAL_FULLTEXT_2025_ANNUAL_REPORT_QUERY",
|
"candidate_source_id": chosen["source_doc_id"], "direct_business_source_id": "",
|
"direct_business_locator": "QUERY_HITS=" + ";".join(locators[:8]),
|
"evidence_grade": "UNVERIFIED_QUERY_HIT", "exposure_specificity": "UNKNOWN",
|
"latest_disclosed_period": "2025-12-31", "selection_bucket": chosen["bucket"],
|
"candidate_state": "HELD_BY_EVIDENCE_GAP", "selection_rank": "", "tier": "",
|
"tie_break_rule": "NOT_ELIGIBLE_NO_DIRECT_BUSINESS_PAGE_EVIDENCE",
|
"include_or_exclude_reason": (
|
f"官方A股基准内公司；2025年年度报告全文检索命中{len(pair_hits)}次（回执{';'.join(source_ids)}）；"
|
"未完成公司—赛道页级直接业务证据验证，保持HELD，不进入T1/T2机械排序"
|
),
|
"coverage_claim": "NONE_DISCOVERY_POOL_NOT_COMPLETE_COVERAGE_CLAIM", "review_status": REVIEW_STATUS,
|
})
|
def sort_key(row: dict[str, str]):
|
state_rank = {"INCLUDED_T1": 0, "INCLUDED_T2": 1}.get(row["candidate_state"], 2)
|
return (TRACK_ORDER[row["track_code"]], BUCKET_ORDER[(row["track_code"], row["selection_bucket"])], state_rank,
|
int(row["selection_rank"] or 999), row["exchange_code"], row["security_code"])
|
rows.sort(key=sort_key)
|
seen = set()
|
for row in rows:
|
key = (row["company_id"], row["track_code"])
|
if key in seen:
|
raise RuntimeError(f"Duplicate candidate pair: {key}")
|
seen.add(key)
|
return rows
|
|
|
def apply_candidate_qualification_funnel(
|
candidates: list[dict[str, str]],
|
hits: dict[tuple[str, str], list[dict]],
|
universe_by_code: dict[str, dict[str, str]],
|
source_rows: list[dict[str, str]],
|
evidence_rows: list[dict[str, str]],
|
) -> tuple[list[dict[str, str]], list[dict[str, str]]]:
|
"""Retrieve and assess official annual-report context for every pair.
|
|
The full discovery pool receives the same replayable attempt: acquire the
|
official CNINFO annual-report hit context and attachment locator, classify
|
its business context, then test whether a stable page/text locator and a
|
company evidence fact exist. A keyword hit is never silently promoted to
|
direct business evidence.
|
"""
|
source_by_id = {row["doc_id"]: row for row in source_rows}
|
company_evidence: dict[tuple[str, str], list[dict[str, str]]] = defaultdict(list)
|
for evidence in evidence_rows:
|
if evidence.get("company_id") and evidence.get("doc_id"):
|
company_evidence[(evidence["company_id"], evidence["doc_id"])].append(evidence)
|
|
gate_by_pair: dict[tuple[str, str], dict[str, object]] = {}
|
allowed_specificity = {
|
"SEGMENT_REVENUE_OR_ASSET_DISCLOSED",
|
"NAMED_PRODUCT_PROJECT_AND_OPERATING_FACT",
|
"GENERAL_DIRECT_BUSINESS_DESCRIPTION",
|
}
|
direct_context_tokens = [
|
"公司主要从事", "主营业务", "主要业务为", "公司从事", "公司产品",
|
"公司生产", "公司业务", "业务包括", "产品包括", "本公司", "项目",
|
]
|
negative_context_tokens = ["不涉及", "不从事", "无相关业务", "未从事"]
|
for row in candidates:
|
pair = (row["company_id"], row["track_code"])
|
query_hits = hits.get((row["security_code"], row["track_code"]), [])
|
direct_source_id = row.get("direct_business_source_id", "")
|
evidence_matches = company_evidence.get((row["company_id"], direct_source_id), [])
|
query_contexts = [
|
clean_html_text(hit.get("announcement_content_sanitized", ""))
|
for hit in query_hits
|
if hit.get("announcement_content_sanitized", "")
|
]
|
verified_contexts = [evidence["evidence_text"] for evidence in evidence_matches if evidence.get("evidence_text")]
|
all_contexts = query_contexts + verified_contexts
|
context_joined = "\n---CONTEXT---\n".join(all_contexts)
|
context_sha256 = hashlib.sha256(context_joined.encode("utf-8")).hexdigest() if context_joined else ""
|
annual_context_pass = bool(query_contexts or verified_contexts)
|
context_screen_pass = annual_context_pass
|
if verified_contexts:
|
context_result = "VERIFIED_PAGE_EVIDENCE_CONTEXT"
|
elif any(token in context_joined for token in negative_context_tokens):
|
context_result = "EXPLICIT_NEGATIVE_OR_NON_DIRECT_CONTEXT"
|
elif any(token in context_joined for token in direct_context_tokens):
|
context_result = "POTENTIAL_DIRECT_BUSINESS_CONTEXT_REQUIRES_PAGE_VERIFICATION"
|
elif query_contexts:
|
context_result = "FULLTEXT_CONTEXT_INSUFFICIENT_FOR_DIRECT_BUSINESS"
|
else:
|
context_result = "NO_ANNUAL_REPORT_CONTEXT_ACQUIRED"
|
identity_pass = row["security_code"] in universe_by_code
|
source_pass = bool(direct_source_id and direct_source_id in source_by_id)
|
locator_pass = bool(
|
row.get("direct_business_locator")
|
and any(
|
evidence.get("source_page")
|
or evidence.get("locator_value")
|
or evidence.get("source_sentence_index")
|
for evidence in evidence_matches
|
)
|
)
|
evidence_pass = bool(evidence_matches)
|
grade_pass = row.get("evidence_grade") in {"S", "A"}
|
specificity_pass = row.get("exposure_specificity") in allowed_specificity
|
eligible = all([
|
identity_pass, annual_context_pass, context_screen_pass,
|
source_pass, locator_pass, evidence_pass, grade_pass, specificity_pass,
|
])
|
failed = []
|
for label, passed in [
|
("A_SHARE_IDENTITY", identity_pass),
|
("ANNUAL_REPORT_CONTEXT_ACQUIRED", annual_context_pass),
|
("BUSINESS_CONTEXT_SCREEN_PERFORMED", context_screen_pass),
|
("DIRECT_BUSINESS_SOURCE", source_pass),
|
("PAGE_OR_TEXT_LOCATOR", locator_pass),
|
("COMPANY_EVIDENCE_FACT", evidence_pass),
|
("SOURCE_GRADE_S_OR_A", grade_pass),
|
("EXPOSURE_SPECIFICITY", specificity_pass),
|
]:
|
if not passed:
|
failed.append(label)
|
gate_by_pair[pair] = {
|
"query_hits": query_hits,
|
"query_contexts": query_contexts,
|
"context_sha256": context_sha256,
|
"context_result": context_result,
|
"context_excerpt": context_joined[:5000],
|
"annual_context_pass": annual_context_pass,
|
"context_screen_pass": context_screen_pass,
|
"source_replacement_count": sum(int(hit.get("source_replacement_character_count", 0)) for hit in query_hits),
|
"identity_pass": identity_pass,
|
"source_pass": source_pass,
|
"locator_pass": locator_pass,
|
"evidence_pass": evidence_pass,
|
"grade_pass": grade_pass,
|
"specificity_pass": specificity_pass,
|
"eligible": eligible,
|
"failed": failed,
|
"evidence_matches": evidence_matches,
|
}
|
if eligible:
|
row["candidate_state"] = "ELIGIBLE"
|
row["selection_rank"] = ""
|
row["tier"] = ""
|
row["include_or_exclude_reason"] = (
|
"REPAIR003统一年度报告语境核验通过：官方年报语境已取得并判定，且A股身份、直接业务主源、"
|
"页级/文本定位、公司证据事实、S/A来源等级和暴露具体性均通过；进入全部ELIGIBLE机械排序。"
|
)
|
else:
|
row["candidate_state"] = "HELD_BY_EVIDENCE_GAP"
|
row["selection_rank"] = ""
|
row["tier"] = ""
|
row["tie_break_rule"] = "NOT_ELIGIBLE_REPAIR003_ANNUAL_REPORT_CONTEXT_AND_PAGE_FUNNEL"
|
row["include_or_exclude_reason"] = (
|
f"REPAIR003已取得/处理官方年度报告全文命中语境，语境判定={context_result}；"
|
"资格 gate 未通过：" + ",".join(failed)
|
+ "；无稳定页级直接业务证据时保持HELD且不参与T1/T2排序。"
|
)
|
|
grade_order = {"S": 0, "A": 1}
|
specificity_order = {
|
"SEGMENT_REVENUE_OR_ASSET_DISCLOSED": 0,
|
"NAMED_PRODUCT_PROJECT_AND_OPERATING_FACT": 1,
|
"GENERAL_DIRECT_BUSINESS_DESCRIPTION": 2,
|
}
|
|
def descending_date(value: str) -> int:
|
digits = re.sub(r"\D", "", value or "")[:8]
|
return -int(digits or "0")
|
|
eligible_groups: dict[tuple[str, str], list[dict[str, str]]] = defaultdict(list)
|
for row in candidates:
|
if row["candidate_state"] == "ELIGIBLE":
|
eligible_groups[(row["track_code"], row["selection_bucket"])].append(row)
|
for group_rows in eligible_groups.values():
|
group_rows.sort(key=lambda row: (
|
grade_order.get(row["evidence_grade"], 9),
|
specificity_order.get(row["exposure_specificity"], 9),
|
descending_date(row["latest_disclosed_period"]),
|
descending_date(source_by_id[row["direct_business_source_id"]].get("publish_date", "")),
|
row["exchange_code"],
|
row["security_code"],
|
))
|
for rank, row in enumerate(group_rows, 1):
|
row["selection_rank"] = str(rank)
|
row["tie_break_rule"] = (
|
"source_grade>exposure_specificity>latest_period>publish_date>exchange_code>security_code"
|
)
|
if rank == 1:
|
row["candidate_state"] = "INCLUDED_T1"
|
row["tier"] = "T1_PRIMARY"
|
elif rank == 2:
|
row["candidate_state"] = "INCLUDED_T2"
|
row["tier"] = "T2_CANDIDATE"
|
else:
|
row["candidate_state"] = "ELIGIBLE_NOT_SELECTED_BATCH001"
|
row["tier"] = ""
|
row["include_or_exclude_reason"] += " 超过本批同桶两家上限，保留为ELIGIBLE_NOT_SELECTED_BATCH001。"
|
|
state_order = {"INCLUDED_T1": 0, "INCLUDED_T2": 1, "ELIGIBLE_NOT_SELECTED_BATCH001": 2, "HELD_BY_EVIDENCE_GAP": 3}
|
candidates.sort(key=lambda row: (
|
TRACK_ORDER[row["track_code"]],
|
BUCKET_ORDER[(row["track_code"], row["selection_bucket"])],
|
state_order.get(row["candidate_state"], 9),
|
int(row["selection_rank"] or 999999),
|
row["exchange_code"],
|
row["security_code"],
|
))
|
|
funnel_headers = [
|
"qualification_row_id", "task_id", "case_id", "batch_id", "run_id",
|
"company_id", "security_code", "track_code", "selection_bucket",
|
"query_hit_count", "query_source_ids", "announcement_ids", "annual_report_adjunct_urls",
|
"annual_report_retrieval_attempt", "annual_report_retrieval_result",
|
"fulltext_context_locator", "fulltext_context_count", "fulltext_context_sha256",
|
"fulltext_context_excerpt_sanitized", "source_replacement_character_count",
|
"business_context_rule_result", "page_level_verification_attempt",
|
"page_level_verification_result", "a_share_identity_gate",
|
"direct_business_source_id", "direct_source_gate", "locator_gate",
|
"company_evidence_fact_gate", "source_grade_gate", "exposure_specificity_gate",
|
"evidence_fact_ids", "failed_gates", "eligibility_result",
|
"eligible_rank_in_bucket", "final_candidate_state", "mechanical_sort_key",
|
"funnel_rule_version", "replay_status", "review_status",
|
]
|
funnel_rows = []
|
for seq, row in enumerate(candidates, 1):
|
pair = (row["company_id"], row["track_code"])
|
gate = gate_by_pair[pair]
|
query_hits = gate["query_hits"]
|
evidence_matches = gate["evidence_matches"]
|
direct_src = source_by_id.get(row.get("direct_business_source_id", ""), {})
|
mechanical_key = "|".join([
|
row.get("evidence_grade", ""), row.get("exposure_specificity", ""),
|
row.get("latest_disclosed_period", ""), direct_src.get("publish_date", ""),
|
row.get("exchange_code", ""), row.get("security_code", ""),
|
])
|
funnel_rows.append({
|
"qualification_row_id": f"QUAL-{seq:05d}", "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"],
|
"track_code": row["track_code"], "selection_bucket": row["selection_bucket"],
|
"query_hit_count": str(len(query_hits)),
|
"query_source_ids": ";".join(sorted({hit["source_doc_id"] for hit in query_hits})),
|
"announcement_ids": ";".join(sorted({hit["announcement_id"] for hit in query_hits if hit["announcement_id"]})),
|
"annual_report_adjunct_urls": ";".join(sorted({hit["adjunct_url"] for hit in query_hits if hit["adjunct_url"]})),
|
"annual_report_retrieval_attempt": "CNINFO_OFFICIAL_2025_ANNUAL_REPORT_FULLTEXT_CONTEXT_FETCH",
|
"annual_report_retrieval_result": (
|
"ACQUIRED_FULLTEXT_QUERY_CONTEXT"
|
if gate["query_contexts"]
|
else "ACQUIRED_ARCHIVED_ANNUAL_REPORT_PAGE_CONTEXT"
|
if evidence_matches
|
else "FAILED_NO_ANNUAL_REPORT_CONTEXT"
|
),
|
"fulltext_context_locator": ";".join(
|
sorted({f"{hit['source_doc_id']}:{hit['announcement_id']}" for hit in query_hits})
|
),
|
"fulltext_context_count": str(len(gate["query_contexts"])),
|
"fulltext_context_sha256": gate["context_sha256"],
|
"fulltext_context_excerpt_sanitized": gate["context_excerpt"],
|
"source_replacement_character_count": str(gate["source_replacement_count"]),
|
"business_context_rule_result": gate["context_result"],
|
"page_level_verification_attempt": "CHECK_ARCHIVED_PDF_OR_STABLE_TEXT_LOCATOR_AFTER_FULLTEXT_CONTEXT",
|
"page_level_verification_result": (
|
"PASS_VERIFIED_DIRECT_BUSINESS_LOCATOR"
|
if gate["locator_pass"]
|
else "INSUFFICIENT_NO_STABLE_PAGE_LEVEL_DIRECT_BUSINESS_LOCATOR"
|
),
|
"a_share_identity_gate": "PASS" if gate["identity_pass"] else "FAIL",
|
"direct_business_source_id": row.get("direct_business_source_id", ""),
|
"direct_source_gate": "PASS" if gate["source_pass"] else "FAIL",
|
"locator_gate": "PASS" if gate["locator_pass"] else "FAIL",
|
"company_evidence_fact_gate": "PASS" if gate["evidence_pass"] else "FAIL",
|
"source_grade_gate": "PASS" if gate["grade_pass"] else "FAIL",
|
"exposure_specificity_gate": "PASS" if gate["specificity_pass"] else "FAIL",
|
"evidence_fact_ids": ";".join(sorted(evidence["evidence_fact_id"] for evidence in evidence_matches)),
|
"failed_gates": ";".join(gate["failed"]),
|
"eligibility_result": "ELIGIBLE" if gate["eligible"] else "HELD_BY_EVIDENCE_GAP",
|
"eligible_rank_in_bucket": row["selection_rank"] if gate["eligible"] else "",
|
"final_candidate_state": row["candidate_state"],
|
"mechanical_sort_key": mechanical_key,
|
"funnel_rule_version": "REPAIR003_ANNUAL_REPORT_CONTEXT_QUALIFICATION_V1",
|
"replay_status": "ATTEMPT_COMPLETED_FOR_DISCOVERED_PAIR",
|
"review_status": REVIEW_STATUS,
|
})
|
write_csv(EXTRACTED_ROOT / "candidate_qualification_funnel.csv", funnel_headers, funnel_rows)
|
return candidates, funnel_rows
|
|
|
def make_source_documents(existing_rows: list[dict[str, str]], new_sources: list[dict[str, str]]) -> list[dict[str, str]]:
|
headers = list(existing_rows[0].keys())
|
rows = []
|
for old in existing_rows:
|
row = dict(old)
|
if row["doc_id"] in {"S-MARKET-NUCLEAR-CNEA-2024", "S-MARKET-NUCLEAR-CEC-2024"}:
|
row["doc_id"] = "S-MARKET-NUCLEAR-CEC-2024"
|
row["source_org"] = "中国电力企业联合会"
|
row["author"] = "中国电力企业联合会"
|
row["raw_file_path"] = row["raw_file_path"].replace("NUCLEAR_CNEA_2024_OPERATION.html", "NUCLEAR_CEC_2024_OPERATION.html")
|
row["file_name"] = "NUCLEAR_CEC_2024_OPERATION.html"
|
row["legal_access_note"] = "中国电力企业联合会公开页；页面明确标注来源；仅作民用核电公开历史对照。"
|
rows.append(row)
|
for src in new_sources:
|
raw_path: Path = src["raw_path"]
|
rows.append({
|
"doc_id": src["doc_id"], "task_id": TASK_ID, "case_id": CASE_ID,
|
"batch_id": BATCH_ID, "run_id": RUN_ID, "doc_type": src["doc_type"],
|
"title": src["title"], "source_org": src["source_org"], "author": src["author"],
|
"publish_date": AS_OF_DATE, "collected_at": COLLECTED_AT, "source_url": src["source_url"],
|
"industry_case": "新能源案例", "industry_id": "IND-NEWENERGY",
|
"subindustry_id": src["subindustry_id"], "company_id": "",
|
"raw_pool_path": "ana-data/cases/新能源案例/raw/", "raw_file_path": rel(raw_path),
|
"converted_text_path": "", "converted_markdown_path": "", "file_sha256": src["raw_hash"],
|
"file_name": raw_path.name, "file_size": str(raw_path.stat().st_size),
|
"detected_type": "XLSX" if raw_path.suffix.lower() == ".xlsx" else "JSON",
|
"source_language": "zh-CN", "public_access_basis": "OFFICIAL_PUBLIC_QUERY_DIRECT",
|
"access_status": "PUBLIC_DIRECT", "source_level": "S", "sensitivity_screen": src["sensitivity"],
|
"legal_access_note": "官方公开查询回执；只用于A股身份或候选发现，不替代公司—赛道直接业务页级证据。",
|
"doc_status": "INCLUDED_DISCOVERY_ONLY", "processing_status": "RAW_ARCHIVED_DISCOVERY_INDEXED",
|
})
|
# Preserve the existing schema order; all appended rows use those fields.
|
for row in rows:
|
for h in headers:
|
row.setdefault(h, "")
|
return rows
|
|
|
def convert_sources(source_rows: list[dict[str, str]], evidence_rows: list[dict[str, str]], discovery_summaries: list[dict]) -> list[dict[str, str]]:
|
pages_by_doc: dict[str, set[int]] = defaultdict(set)
|
for ev in evidence_rows:
|
if ev.get("source_page", "").isdigit():
|
pages_by_doc[ev["doc_id"]].add(int(ev["source_page"]))
|
summary_by_source = {s["source_doc_id"]: s for s in discovery_summaries}
|
evidence_docs = {ev["doc_id"] for ev in evidence_rows}
|
conversions = []
|
for row in source_rows:
|
raw_path = PROJECT_ROOT / row["raw_file_path"]
|
if not raw_path.exists():
|
raise FileNotFoundError(raw_path)
|
doc_id = row["doc_id"]
|
detected = row["detected_type"].upper()
|
if detected == "PDF":
|
out = CONVERTED_ROOT / f"official_filings/{doc_id}__cited_pages.txt"
|
reader = PdfReader(str(raw_path))
|
page_numbers = sorted(pages_by_doc.get(doc_id) or {1})
|
parts = [
|
f"source_doc_id={doc_id}", f"raw_path={rel(raw_path)}", f"raw_sha256={sha256_file(raw_path)}",
|
f"conversion_scope=CITED_PAGES_ONLY:{','.join(map(str, page_numbers))}",
|
"note=正式证据只引用列出的页码；本转换不是全文覆盖声明。", "",
|
]
|
for page_no in page_numbers:
|
if not 1 <= page_no <= len(reader.pages):
|
raise RuntimeError(f"Page {page_no} out of range for {doc_id}")
|
parts.extend([f"===== PDF_PAGE {page_no} =====", reader.pages[page_no - 1].extract_text() or "", ""])
|
write_text(out, "\n".join(parts))
|
method = "PYPDF_CITED_PAGE_TEXT_EXTRACTION"
|
params = f"UTF-8; cited_pages={','.join(map(str, page_numbers))}; page markers retained"
|
page_count = str(len(reader.pages))
|
elif detected == "HTML":
|
out = CONVERTED_ROOT / f"official_market/{doc_id}.txt"
|
text = html_to_text(raw_path.read_bytes())
|
write_text(out, f"source_doc_id={doc_id}\nraw_path={rel(raw_path)}\n\n{text}")
|
method = "HTML_TEXT_NORMALIZATION"
|
params = "Python HTMLParser; whitespace normalized; source URL retained in source_document"
|
page_count = "N/A_HTML"
|
elif detected == "JSON":
|
out = CONVERTED_ROOT / f"official_discovery/{doc_id}.txt"
|
if doc_id in summary_by_source:
|
s = summary_by_source[doc_id]
|
text = "\n".join(f"{k}={v}" for k, v in s.items()) + "\n"
|
else:
|
obj = json.loads(raw_path.read_text(encoding="utf-8"))
|
payload_count = len(obj.get("payloads", []))
|
text = f"source_doc_id={doc_id}\nas_of_date={AS_OF_DATE}\npayload_count={payload_count}\n"
|
write_text(out, text)
|
method = "JSON_RECEIPT_SUMMARY"
|
params = "Full official JSON payload retained in raw; converted text is a deterministic index summary"
|
page_count = "N/A_JSON"
|
elif detected == "XLSX":
|
out = CONVERTED_ROOT / f"official_discovery/{doc_id}.txt"
|
# The official workbook has a stale A1:A1 dimension, so read-only
|
# mode would silently expose only its header row.
|
workbook = openpyxl.load_workbook(raw_path, read_only=False, data_only=True)
|
worksheet = workbook.active
|
codes = []
|
for values in worksheet.iter_rows(values_only=True):
|
value = str(values[0] or "").strip().zfill(6)
|
if re.fullmatch(r"(?:00|30)\d{4}", value):
|
codes.append(value)
|
write_text(
|
out,
|
f"source_doc_id={doc_id}\nraw_path={rel(raw_path)}\nrecord_count={len(codes)}\n"
|
+ "\n".join(codes)
|
+ "\n",
|
)
|
method = "XLSX_FIRST_COLUMN_CODE_EXTRACTION"
|
params = "openpyxl normal mode due stale A1:A1 dimension; official workbook; first column security codes retained in source order"
|
page_count = f"ROWS={len(codes)}"
|
else:
|
raise RuntimeError(f"Unsupported source type {detected} for {doc_id}")
|
row["converted_text_path"] = rel(out)
|
row["processing_status"] = "TEXT_CONVERTED_EVIDENCE_EXTRACTED" if doc_id in evidence_docs else "TEXT_CONVERTED_INDEXED_DISCOVERY_ONLY"
|
conversions.append({
|
"conversion_id": f"CONV-{doc_id}", "task_id": TASK_ID, "case_id": CASE_ID,
|
"batch_id": BATCH_ID, "run_id": RUN_ID, "source_doc_id": doc_id,
|
"raw_pool_path": row["raw_pool_path"], "raw_file_path": row["raw_file_path"],
|
"raw_file_sha256": row["file_sha256"], "detected_type": detected,
|
"conversion_method": method, "parameters_summary": params,
|
"converted_text_path": rel(out), "converted_markdown_path": "", "converted_path": rel(out),
|
"converted_sha256": sha256_file(out), "page_or_duration_count": page_count,
|
"status": "CONVERTED_EVIDENCE_EXTRACTED" if doc_id in evidence_docs else "CONVERTED_INDEXED_DISCOVERY_ONLY",
|
"error_code": "", "error_summary": "", "created_at": COLLECTED_AT,
|
})
|
return conversions
|
|
|
def update_evidence_facts(rows: list[dict[str, str]], source_rows: list[dict[str, str]]) -> list[dict[str, str]]:
|
source_by_id = {r["doc_id"]: r for r in source_rows}
|
for row in rows:
|
if row["doc_id"] in {"S-MARKET-NUCLEAR-CNEA-2024", "S-MARKET-NUCLEAR-CEC-2024"}:
|
row["doc_id"] = "S-MARKET-NUCLEAR-CEC-2024"
|
row["original_qualifier"] = "中国电力企业联合会公开运行情况；归档页明确标注来源"
|
row["source_text_path"] = source_by_id[row["doc_id"]]["converted_text_path"]
|
if row["doc_id"] in {
|
"S-MARKET-BATTERY-MIIT-2025-0104",
|
"S-MARKET-RENEWABLE-NEA-2025",
|
"S-MARKET-WINDSOLAR-NEA-2025",
|
"S-MARKET-NUCLEAR-CEC-2024",
|
}:
|
row["processing_status"] = "EVIDENCE_EXTRACTED_AND_LINKED_TO_OUTPUT"
|
return rows
|
|
|
def expand_classification(current_rows: list[dict[str, str]], candidates: list[dict[str, str]], source_rows: list[dict[str, str]]) -> list[dict[str, str]]:
|
headers = list(current_rows[0].keys())
|
current_by_pair = {(r["company_id"], r["track_code"]): dict(r) for r in current_rows}
|
source_by_id = {r["doc_id"]: r for r in source_rows}
|
rows = []
|
for cand in candidates:
|
key = (cand["company_id"], cand["track_code"])
|
if key in current_by_pair:
|
rows.append(current_by_pair[key])
|
continue
|
src = source_by_id[cand["candidate_source_id"]]
|
rows.append({
|
"classification_id": f"CLASS-DISC-{cand['track_code']}-{cand['security_code']}",
|
"task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID,
|
"subject_type": "COMPANY_QUERY_HIT", "subject_id": cand["company_id"],
|
"industry_id": "IND-NEWENERGY", "subindustry_id": cand["track_code"],
|
"company_id": cand["company_id"], "track_code": cand["track_code"], "chain_node_id": "",
|
"scope_type": "", "classification_reason": cand["include_or_exclude_reason"],
|
"source_doc_id": cand["candidate_source_id"], "evidence_fact_id": "",
|
"raw_pool_path": src["raw_pool_path"], "raw_file_sha256": src["file_sha256"],
|
"data_status": cand["candidate_state"], "review_status": REVIEW_STATUS,
|
})
|
for row in rows:
|
for h in headers:
|
row.setdefault(h, "")
|
return rows
|
|
|
def build_input_manifests(source_rows: list[dict[str, str]]) -> tuple[list[dict[str, str]], list[dict[str, str]]]:
|
header = [
|
"input_item_id", "task_id", "case_id", "batch_id", "run_id", "source_doc_id",
|
"include_decision", "exclude_reason", "raw_pool_path", "raw_file_path", "raw_file_sha256",
|
"source_url", "source_level", "public_access_basis", "sensitivity_screen",
|
"processing_status", "review_status",
|
]
|
industry_rows = []
|
case_rows = []
|
for src in source_rows:
|
discovery = src["doc_type"] in {"OFFICIAL_A_SHARE_UNIVERSE_SNAPSHOT", "OFFICIAL_DISCOVERY_QUERY_RECEIPT"}
|
row = {
|
"input_item_id": f"INPUT-{src['doc_id']}", "task_id": TASK_ID, "case_id": CASE_ID,
|
"batch_id": BATCH_ID, "run_id": RUN_ID, "source_doc_id": src["doc_id"],
|
"include_decision": "INCLUDE_DISCOVERY_ONLY" if discovery else "INCLUDE",
|
"exclude_reason": "", "raw_pool_path": src["raw_pool_path"],
|
"raw_file_path": src["raw_file_path"], "raw_file_sha256": src["file_sha256"],
|
"source_url": src["source_url"], "source_level": src["source_level"],
|
"public_access_basis": src["public_access_basis"], "sensitivity_screen": src["sensitivity_screen"],
|
"processing_status": src["processing_status"], "review_status": REVIEW_STATUS,
|
}
|
industry_rows.append(row)
|
case_row = dict(row)
|
case_row["input_item_id"] = f"CASE-INPUT-{src['doc_id']}"
|
case_rows.append(case_row)
|
write_csv(MANIFEST_ROOT / "input_manifest.csv", header, industry_rows)
|
case_header = ["case_input_item_id"] + header[1:]
|
converted_case = [{"case_input_item_id": r.pop("input_item_id"), **r} for r in case_rows]
|
write_csv(CASE_MANIFEST / "case_input_manifest.csv", case_header, converted_case)
|
return industry_rows, converted_case
|
|
|
def build_source_gap_audit(source_rows: list[dict[str, str]], evidence_rows: list[dict[str, str]]) -> list[dict[str, str]]:
|
header = [
|
"source_gap_audit_id", "task_id", "case_id", "batch_id", "run_id", "source_doc_id",
|
"source_received_flag", "raw_archived_flag", "converted_flag", "indexed_flag",
|
"evidence_linked_flag", "raw_pool_path", "raw_file_sha256", "gap_type", "impact",
|
"status", "review_status",
|
]
|
evidence_docs = {r["doc_id"] for r in evidence_rows}
|
rows = []
|
for src in source_rows:
|
linked = src["doc_id"] in evidence_docs
|
rows.append({
|
"source_gap_audit_id": f"SGA-{src['doc_id']}", "task_id": TASK_ID, "case_id": CASE_ID,
|
"batch_id": BATCH_ID, "run_id": RUN_ID, "source_doc_id": src["doc_id"],
|
"source_received_flag": "YES", "raw_archived_flag": "YES", "converted_flag": "YES",
|
"indexed_flag": "YES", "evidence_linked_flag": "YES" if linked else "DISCOVERY_ONLY_NOT_FORMAL_EVIDENCE",
|
"raw_pool_path": src["raw_pool_path"], "raw_file_sha256": src["file_sha256"],
|
"gap_type": "NONE", "impact": "NONE",
|
"status": "COMPLETE_EVIDENCE_LINKED" if linked else "COMPLETE_DISCOVERY_ONLY",
|
"review_status": REVIEW_STATUS,
|
})
|
write_csv(MANIFEST_ROOT / "source_gap_audit.csv", header, rows)
|
return rows
|
|
|
def update_outputs() -> None:
|
replacements = {
|
"中国核能行业协会历史对照": "中国电力企业联合会历史对照",
|
"[中国核能行业协会：2024年全国核电运行情况]": "[中国电力企业联合会：2024年全国核电运行情况]",
|
"../evidence/evidence_fact.csv": "../../evidence/evidence_fact_table.csv",
|
"../evidence/company_track_candidate_ledger.csv": "../../extracted/company_track_candidate_ledger.csv",
|
"../evidence/newenergy_company_exposure_matrix.csv": "../../extracted/newenergy_company_exposure_matrix.csv",
|
"../evidence/newenergy_supply_demand_price_metric.csv": "../../extracted/newenergy_supply_demand_price_metric.csv",
|
"../evidence/newenergy_technology_route_matrix.csv": "../../extracted/newenergy_technology_route_matrix.csv",
|
"../evidence/newenergy_project_capacity_event.csv": "../../extracted/newenergy_project_capacity_event.csv",
|
"../manifest/source_document.csv": "../../manifest/source_document.csv",
|
"../../../../evidence/company_track_candidate_ledger.csv": "../../../../../extracted/company_track_candidate_ledger.csv",
|
"../../../../evidence/newenergy_technology_route_matrix.csv": "../../../../../extracted/newenergy_technology_route_matrix.csv",
|
"../../../../evidence/newenergy_supply_demand_price_metric.csv": "../../../../../extracted/newenergy_supply_demand_price_metric.csv",
|
}
|
for path in CASE_OUTPUTS.rglob("*.md"):
|
text = path.read_text(encoding="utf-8")
|
for old, new in replacements.items():
|
text = text.replace(old, new)
|
text = re.sub(
|
r"(?:\.\./)+manifest/source_document\.csv",
|
"../../manifest/source_document.csv",
|
text,
|
)
|
write_text(path, text)
|
result = RESULT_ROOT / "result_index.md"
|
text = result.read_text(encoding="utf-8")
|
text = re.sub(
|
r"(?:\.\./)+cases/[^/]+/" + re.escape(CASE_ID) + r"/outputs/",
|
f"../../../cases/{INDUSTRY_ROOT.name}/{CASE_ID}/outputs/",
|
text,
|
)
|
write_text(result, text)
|
|
|
def expand_case_evidence_map(
|
current_rows: list[dict[str, str]], candidates: list[dict[str, str]], metrics: list[dict[str, str]],
|
evidence_rows: list[dict[str, str]],
|
) -> list[dict[str, str]]:
|
headers = list(current_rows[0].keys())
|
rows = []
|
for row in current_rows:
|
row = dict(row)
|
if row["evidence_fact_id"] == "EVF-MKT-NUC-2024":
|
row["contradiction_or_limit"] = row["contradiction_or_limit"].replace("中国核能行业协会", "中国电力企业联合会")
|
row["conclusion_text"] = row["conclusion_text"].replace("中国核能行业协会", "中国电力企业联合会")
|
elif row["evidence_fact_id"] == "EVF-MKT-NUC-01":
|
row["contradiction_or_limit"] = row["contradiction_or_limit"].replace("中国电力企业联合会", "中国核能行业协会")
|
row["conclusion_text"] = row["conclusion_text"].replace("中国电力企业联合会", "中国核能行业协会")
|
rows.append(row)
|
selected = [r for r in candidates if r["candidate_state"] in {"INCLUDED_T1", "INCLUDED_T2"}]
|
ev_by_company = {r["company_id"]: r for r in evidence_rows if r["subject_type"] == "COMPANY"}
|
track_dir = {"BATTERY": "01_锂电", "SOLAR": "02_光伏", "WIND": "03_风电", "NUCLEAR": "04_核电"}
|
seq = 1
|
existing = {(r["output_path"], r["evidence_fact_id"], r["conclusion_text"]) for r in rows}
|
|
def add(output_path: str, evidence_fact_id: str, conclusion_text: str, anchor: str, strength: str = "DIRECT_FACT", limit: str = "") -> None:
|
nonlocal seq
|
key = (output_path, evidence_fact_id, conclusion_text)
|
if key in existing:
|
return
|
rows.append({
|
"conclusion_evidence_map_id": f"CEM-REPAIR-{seq:04d}", "task_id": TASK_ID,
|
"case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID,
|
"conclusion_id": f"CONC-REPAIR-{seq:04d}", "output_path": output_path,
|
"section_anchor": anchor, "conclusion_text": conclusion_text,
|
"conclusion_strength": strength, "evidence_fact_id": evidence_fact_id,
|
"support_type": "SUPPORT", "contradiction_or_limit": limit,
|
"review_status": REVIEW_STATUS,
|
})
|
existing.add(key)
|
seq += 1
|
|
first_by_track = {}
|
for cand in selected:
|
ev = ev_by_company[cand["company_id"]]
|
related = rel(CASE_OUTPUTS / f"核心文档/子行业图谱/{track_dir[cand['track_code']]}/相关企业.md")
|
add(related, ev["evidence_fact_id"], ev["evidence_text"], f"company-{cand['security_code']}",
|
limit="仅证明直接业务暴露；T1/T2只是本批研究深度，不是质量、估值或投资排序。")
|
first_by_track.setdefault(cand["track_code"], (cand, ev))
|
for track, (cand, ev) in first_by_track.items():
|
industry_view = rel(CASE_OUTPUTS / "新能源行业视图.md")
|
tech_doc = rel(CASE_OUTPUTS / f"核心文档/子行业图谱/{track_dir[track]}/产业链与技术路线.md")
|
add(industry_view, ev["evidence_fact_id"], f"{cand['security_name']}的官方年报直接支持其{track}业务映射。",
|
f"track-{track.lower()}", limit="公司例证不构成行业或公司全集。")
|
add(tech_doc, ev["evidence_fact_id"], f"{cand['security_name']}的公开产品/业务事实作为产业链节点例证。",
|
"official-company-example", strength="MECHANISM_ONLY",
|
limit="只支持公开产品/业务节点，不据此推导技术优劣、份额或投资结论。")
|
for metric in metrics:
|
track = metric["track_code"]
|
if track == "CROSS_TRACK":
|
industry_view = rel(CASE_OUTPUTS / "新能源行业视图.md")
|
text = f"{metric['metric_name']}={metric['metric_value']}{metric['metric_unit']}（{metric['metric_period']}）"
|
add(
|
industry_view,
|
metric["evidence_fact_id"],
|
text,
|
f"metric-{metric['metric_id'].lower()}",
|
limit=metric["qualifier"],
|
)
|
continue
|
market_doc = rel(CASE_OUTPUTS / f"核心文档/子行业图谱/{track_dir[track]}/市场与供需.md")
|
text = f"{metric['metric_name']}={metric['metric_value']}{metric['metric_unit']}（{metric['metric_period']}）"
|
add(market_doc, metric["evidence_fact_id"], text, f"metric-{metric['metric_id'].lower()}",
|
limit=metric["qualifier"])
|
primary_market_ev = {"BATTERY": "EVF-MKT-BAT-01", "SOLAR": "EVF-MKT-SOL-01",
|
"WIND": "EVF-MKT-WIND-01", "NUCLEAR": "EVF-MKT-NUC-01"}
|
ev_by_id = {r["evidence_fact_id"]: r for r in evidence_rows}
|
for track, evidence_id in primary_market_ev.items():
|
overview = rel(CASE_OUTPUTS / f"核心文档/子行业图谱/{track_dir[track]}/总览.md")
|
ev = ev_by_id[evidence_id]
|
add(overview, evidence_id, ev["evidence_text"], "market-fact-summary",
|
limit="按来源原始期间、地域和阶段使用；不得外推或跨口径拼接。")
|
nuclear_2024 = ev_by_id["EVF-MKT-NUC-2024"]
|
for name in ["市场与供需.md", "总览.md"]:
|
path = rel(CASE_OUTPUTS / f"核心文档/子行业图谱/04_核电/{name}")
|
add(path, nuclear_2024["evidence_fact_id"], nuclear_2024["evidence_text"], "nuclear-2024-history",
|
limit="来源为中国电力企业联合会；仅作2024历史对照，不与2025口径静默拼接。")
|
return rows
|
|
|
def materialize_evidence_locators(case_map: list[dict[str, str]]) -> list[dict[str, str]]:
|
"""Make every map row resolve to an explicit anchor and verbatim statement."""
|
begin = "<!-- BEGIN GENERATED EVIDENCE LOCATORS REPAIR004 -->"
|
end = "<!-- END GENERATED EVIDENCE LOCATORS REPAIR004 -->"
|
by_output: dict[str, list[dict[str, str]]] = defaultdict(list)
|
for row in case_map:
|
row["conclusion_text"] = re.sub(r"\s+", " ", row["conclusion_text"]).strip()
|
if row["evidence_fact_id"] == "EVF-MKT-NUC-2024":
|
row["conclusion_text"] = row["conclusion_text"].replace("中国核能行业协会", "中国电力企业联合会")
|
row["contradiction_or_limit"] = row["contradiction_or_limit"].replace("中国核能行业协会", "中国电力企业联合会")
|
elif row["evidence_fact_id"] == "EVF-MKT-NUC-01":
|
row["conclusion_text"] = row["conclusion_text"].replace("中国电力企业联合会", "中国核能行业协会")
|
row["contradiction_or_limit"] = row["contradiction_or_limit"].replace("中国电力企业联合会", "中国核能行业协会")
|
by_output[row["output_path"]].append(row)
|
|
global_seq = 1
|
for output_path in sorted(by_output):
|
path = PROJECT_ROOT / output_path
|
if not path.exists():
|
raise FileNotFoundError(path)
|
text = path.read_text(encoding="utf-8")
|
text = re.sub(
|
r"\n## 证据定位索引（REPAIR\d+）\n\n?"
|
r"<!-- BEGIN GENERATED EVIDENCE LOCATORS REPAIR\d+ -->.*?"
|
r"<!-- END GENERATED EVIDENCE LOCATORS REPAIR\d+ -->\n?",
|
"\n",
|
text,
|
flags=re.S,
|
).rstrip()
|
lines = ["", "", "## 证据定位索引（REPAIR004）", "", begin, ""]
|
for row in sorted(
|
by_output[output_path],
|
key=lambda item: (item["evidence_fact_id"], item["conclusion_evidence_map_id"], item["conclusion_text"]),
|
):
|
anchor = f"evidence-locator-{global_seq:04d}"
|
global_seq += 1
|
row["section_anchor"] = anchor
|
row["review_status"] = REVIEW_STATUS
|
lines.extend([
|
f'<a id="{anchor}"></a>',
|
f'- `{row["evidence_fact_id"]}`：{row["conclusion_text"]}',
|
"",
|
])
|
lines.append(end)
|
write_text(path, text + "\n".join(lines) + "\n")
|
return case_map
|
|
|
def rewrite_batch_summary(candidate_rows: list[dict[str, str]], source_count: int, conversion_count: int, universe_rows: list[dict[str, str]]) -> None:
|
counts = defaultdict(int)
|
for row in candidate_rows:
|
counts[row["candidate_state"]] += 1
|
exchange_counts = defaultdict(int)
|
for row in universe_rows:
|
exchange_counts[row["exchange_code"]] += 1
|
text = f"""# {CASE_ID} / {BATCH_ID} 批次摘要
|
|
- `task_id={TASK_ID}`
|
- `run_id={RUN_ID}`
|
- `schema_version={SCHEMA_VERSION}`
|
- `primary_region=MAINLAND_CHINA`
|
- `global_comparator=SEPARATE_CONTEXT_ONLY`
|
- `source_cutoff_at={SOURCE_CUTOFF}`
|
- `track_count=4`
|
- `official_a_share_universe_count={len(universe_rows)}`（SSE={exchange_counts['SSE']}，SZSE={exchange_counts['SZSE']}，BSE={exchange_counts['BSE']}）
|
- `candidate_discovery_pair_count={len(candidate_rows)}`
|
- `candidate_state_count=INCLUDED_T1:{counts['INCLUDED_T1']},INCLUDED_T2:{counts['INCLUDED_T2']},HELD_BY_EVIDENCE_GAP:{counts['HELD_BY_EVIDENCE_GAP']}`
|
- `selected_company_track_mapping_count={counts['INCLUDED_T1'] + counts['INCLUDED_T2']}`
|
- `tier_count=T1_PRIMARY:{counts['INCLUDED_T1']},T2_CANDIDATE:{counts['INCLUDED_T2']}`
|
- `source_document_count={source_count}`（原38份研究来源 + 3份交易所A股基准 + 16份候选发现回执）
|
- `conversion_status_count={conversion_count}`
|
- `valuation_market_interface=NOT_APPLICABLE_BATCH001`
|
- `market_reverse_gap_scan=NOT_APPLICABLE_BATCH001`
|
- `coverage_claim=NONE_INITIAL_CANDIDATE_POOL_ONLY`
|
- `research_mode=darkline_广撒网收集`
|
- `output_status=DRAFT_FOR_REVIEW`
|
- `review_status=PENDING_REPAIR005_FOCUSED_REREVIEW`
|
|
本批仍只形成中国大陆四赛道基础行业图谱与 A 股直接业务导航。候选发现先以三家交易所官方 A 股全集为身份基准，再以巨潮资讯 16 组固定全文检索生成完整命中账本；只有已有法定年报页级直接业务证据的对象进入 ELIGIBLE 排序，其余命中全部保持 `HELD_BY_EVIDENCE_GAP`。T1/T2 只表示本批研究深度。未进行估值、行情、交易、收益、完整覆盖或公司质量排名。核电仅使用民用公开高层信息并执行关键基础设施敏感信息停止边界。
|
"""
|
write_text(CASE_MANIFEST / "batch_summary.md", text)
|
|
|
def write_discovery_receipt(summaries: list[dict], candidates: list[dict[str, str]], universe_rows: list[dict[str, str]]) -> None:
|
state_counts = defaultdict(int)
|
track_counts = defaultdict(int)
|
for row in candidates:
|
state_counts[row["candidate_state"]] += 1
|
track_counts[row["track_code"]] += 1
|
lines = [
|
"# BATCH-001 候选发现与机械选择回执", "",
|
f"- `task_id={TASK_ID}`", f"- `case_id={CASE_ID}`", f"- `batch_id={BATCH_ID}`", f"- `run_id={RUN_ID}`",
|
f"- `as_of_date={AS_OF_DATE}`", f"- `source_cutoff_at={SOURCE_CUTOFF}`", "- `coverage_claim=NONE`", "",
|
"## 1. A股身份基准", "",
|
f"官方交易所基准合计 `{len(universe_rows)}` 条；唯一键为 `(exchange_code, security_code)`。三份完整官方响应保存于行业统一 raw，规范化全集保存于 `extracted/a_share_universe.csv`。该基准只证明证券身份，不证明新能源业务。", "",
|
"## 2. 发现查询", "",
|
"查询时间窗固定为 `2026-03-01..2026-05-15`，只查 `szb,cyb,hzb,kcb,bjs`，全文检索、发布日期升序、每页100条；从 pageNum=1 开始，持续取得直到 archived_record_count 精确等于 server_totalRecordNum。接口 totalpages 表示最后一个完整页号，存在残页时继续抓取 totalpages+1。只保留标题为2025年年度报告全文且能回连官方A股基准的命中。搜索引擎未作为纳入证据。", "",
|
"| 赛道 | 桶 | 关键词 | 服务器记录 | 已归档记录 | API页值 | 实际页数 | 终止页条数 | A股年报唯一命中 | raw SHA-256 |", "|---|---|---|---:|---:|---:|---:|---:|---:|---|",
|
]
|
for s in summaries:
|
lines.append(f"| `{s['track_code']}` | {s['selection_bucket']} | {s['keyword']} | {s['server_total_record_num']} | {s['archived_record_count']} | {s['server_total_pages']} | {s['archived_page_count']} | {s['terminal_page_record_count']} | {s['annual_report_a_share_unique_pair_hits']} | `{s['raw_sha256']}` |")
|
lines.extend([
|
"", "## 3. 全量账本与选择", "",
|
f"规范化后唯一 `(company_id, track_code)` 共 `{len(candidates)}` 条：BATTERY={track_counts['BATTERY']}、SOLAR={track_counts['SOLAR']}、WIND={track_counts['WIND']}、NUCLEAR={track_counts['NUCLEAR']}。",
|
f"其中 `INCLUDED_T1={state_counts['INCLUDED_T1']}`、`INCLUDED_T2={state_counts['INCLUDED_T2']}`、`HELD_BY_EVIDENCE_GAP={state_counts['HELD_BY_EVIDENCE_GAP']}`。所有命中均保留，未入选/证据不足对象没有删除。", "",
|
"`extracted/candidate_qualification_funnel.csv` 对完整唯一命中池逐行保存实际年度报告语境处理回执：公告 ID/附件 URL → 官方全文命中语境 → 语境 locator/hash/规则判定 → 稳定页级直接业务定位 → 公司 evidence fact → S/A来源等级 → 暴露具体性。每个发现 pair 均记录取得、处理、失败/不足结果，不再只做既有 evidence join。机械顺序仍为：直接业务来源等级 → 暴露具体性 → 披露期间 → 发布日期 → 交易所 → 证券代码。只有所有 gate 通过的对象进入 ELIGIBLE 排序；全文命中本身不升级为业务证据。", "",
|
"## 4. 边界", "",
|
"该回执只证明本次冻结查询的命中集合和选择过程可重跑，不声称覆盖全部新能源公司。核电查询只用于民用公开高层业务发现，不读取、保存或推断关键基础设施敏感细节。", "",
|
])
|
write_text(SUPPLEMENT_ROOT / "candidate_discovery_receipt.md", "\n".join(lines))
|
write_csv(MANIFEST_ROOT / "candidate_discovery_query_summary.csv", list(summaries[0].keys()), summaries)
|
|
|
def build_human_receipt(case_map: list[dict[str, str]], source_count: int) -> None:
|
header = HEADERS["human_receipt"]
|
by_path = defaultdict(list)
|
for row in case_map:
|
by_path[row["output_path"]].append(row)
|
md_paths = sorted(CASE_OUTPUTS.rglob("*.md"))
|
rows = []
|
for i, path in enumerate(md_paths, 1):
|
p = rel(path)
|
name = path.name
|
if name == "新能源行业视图.md":
|
view_type = "INDUSTRY_VIEW"
|
elif name == "新能源市场视图.md":
|
view_type = "MARKET_VIEW"
|
elif name == "新能源公司视图.md":
|
view_type = "COMPANY_VIEW"
|
elif name == "新能源报告索引.md":
|
view_type = "REPORT_INDEX"
|
elif name in {"summary.md", "readout.md"}:
|
view_type = name.removesuffix(".md").upper()
|
else:
|
view_type = "SUBINDUSTRY_OR_GAP_DOC"
|
mappings = by_path.get(p, [])
|
rows.append({
|
"validation_item_id": f"HDV-NEWENERGY-{i:03d}", "task_id": TASK_ID, "case_id": CASE_ID,
|
"batch_id": BATCH_ID, "run_id": RUN_ID, "view_type": view_type, "output_path": p,
|
"document_status": "DRAFT_FOR_REPAIR005_FOCUSED_REREVIEW", "not_applicable_reason": "",
|
"evidence_boundary": "OFFICIAL_PUBLIC_SOURCES; STRONG_FACTS_MAPPED; GAP_AND_MECHANISM_LIMITS_RETAINED",
|
"source_count": str(source_count), "evidence_count": str(len({x['evidence_fact_id'] for x in mappings})),
|
"conclusion_count": str(len(mappings)), "unknown_count": "0",
|
"review_status": "PENDING_REPAIR005_FOCUSED_REREVIEW",
|
"validated_by": "case_analysis.analyst.new_energy", "validated_at": COLLECTED_AT,
|
})
|
write_csv(CASE_MANIFEST / "human_doc_validation_receipt.csv", header, rows)
|
|
|
def rebuild_output_manifest() -> list[dict[str, str]]:
|
path = CASE_MANIFEST / "output_manifest.csv"
|
rows = read_csv(path)
|
for row in rows:
|
output = PROJECT_ROOT / row["output_path"]
|
if not output.exists():
|
raise FileNotFoundError(output)
|
row["output_sha256"] = sha256_file(output)
|
row["data_status"] = REVIEW_STATUS
|
row["review_status"] = "PENDING_REPAIR005_FOCUSED_REREVIEW"
|
write_csv(path, list(rows[0].keys()), rows)
|
return rows
|
|
|
def write_exact_set_receipt_placeholder() -> Path:
|
path = CASE_MANIFEST / "package_exact_set_receipt.md"
|
write_text(path, "# Formal package exact-set receipt\n\nPENDING_ENUMERATION\n")
|
return path
|
|
|
def formal_files(extra_tool: Path) -> list[Path]:
|
roots = [
|
RAW_ROOT, CONVERTED_ROOT, EXTRACTED_ROOT, SUPPLEMENT_ROOT, EVIDENCE_ROOT, MANIFEST_ROOT,
|
CASE_ROOT, RESULT_ROOT,
|
]
|
files = set()
|
artifact_manifest = MANIFEST_ROOT / "artifact_manifest.csv"
|
for root in roots:
|
if not root.exists():
|
continue
|
for path in root.rglob("*"):
|
if not path.is_file() or path.name == ".gitkeep" or path.resolve() == artifact_manifest.resolve():
|
continue
|
if any(part.lower() in {"tmp", "img"} for part in path.parts):
|
continue
|
files.add(path.resolve())
|
files.add(extra_tool.resolve())
|
return sorted(files, key=lambda p: rel(p))
|
|
|
def write_exact_set_receipt(files: list[Path]) -> None:
|
counts = defaultdict(int)
|
for path in files:
|
r = rel(path)
|
if "/raw/" in f"/{r}/":
|
counts["raw"] += 1
|
elif "/converted/" in f"/{r}/":
|
counts["converted"] += 1
|
elif "/extracted/" in f"/{r}/":
|
counts["extracted"] += 1
|
elif "/evidence/" in f"/{r}/":
|
counts["evidence"] += 1
|
elif "/manifest/" in f"/{r}/":
|
counts["manifest"] += 1
|
elif "/outputs/" in f"/{r}/":
|
counts["outputs"] += 1
|
elif "/result/" in f"/{r}/":
|
counts["result"] += 1
|
else:
|
counts["other"] += 1
|
text = f"""# BATCH-001 正式产物 exact-set 回执
|
|
- `task_id={TASK_ID}`
|
- `case_id={CASE_ID}`
|
- `batch_id={BATCH_ID}`
|
- `run_id={RUN_ID}`
|
- `generated_at={COLLECTED_AT}`
|
- `formal_file_count_excluding_artifact_manifest={len(files)}`
|
- `raw={counts['raw']}`
|
- `converted={counts['converted']}`
|
- `extracted={counts['extracted']}`
|
- `evidence={counts['evidence']}`
|
- `manifest={counts['manifest']}`
|
- `outputs={counts['outputs']}`
|
- `result={counts['result']}`
|
- `other_replay_tool={counts['other']}`
|
|
`artifact_manifest.csv` 对以上 exact-set 中每个文件逐项记录 bytes 与 SHA-256。该 manifest 自身因自引用哈希不可满足而明确自排除；复审时直接对 manifest 文件另算 SHA-256。`.gitkeep` 仅为目录脚手架，不属于正式产物并明确排除。`tmp/` 与 `img/` 本批没有被正式结论引用，均不进入 formal exact-set。
|
"""
|
write_text(CASE_MANIFEST / "package_exact_set_receipt.md", text)
|
|
|
def artifact_type(path: Path) -> str:
|
r = rel(path)
|
if "/raw/" in f"/{r}/":
|
return "RAW_DOCUMENT"
|
if "/converted/" in f"/{r}/":
|
return "CONVERTED_TEXT"
|
if "/supplement/" in f"/{r}/":
|
return "SUPPLEMENT_SOURCE"
|
if "/extracted/" in f"/{r}/":
|
return "EXTRACTED_FACT"
|
if "/evidence/" in f"/{r}/":
|
return "EVIDENCE_INDEX"
|
if "/outputs/" in f"/{r}/":
|
return "REPORT_MARKDOWN"
|
if "/manifest/" in f"/{r}/":
|
return "MANIFEST"
|
if "/result/" in f"/{r}/":
|
return "PACKAGE"
|
if r.endswith(".py"):
|
return "REPLAY_TOOL"
|
return "OUTPUT_TABLE"
|
|
|
def build_artifact_manifest(source_rows: list[dict[str, str]], output_rows: list[dict[str, str]], tool_path: Path) -> list[dict[str, str]]:
|
write_exact_set_receipt_placeholder()
|
files = formal_files(tool_path)
|
write_exact_set_receipt(files)
|
files = formal_files(tool_path)
|
source_by_raw = {r["raw_file_path"]: r for r in source_rows}
|
source_by_converted = {r["converted_text_path"]: r for r in source_rows if r["converted_text_path"]}
|
output_by_path = {r["output_path"]: r for r in output_rows}
|
rows = []
|
generic_i = 1
|
for path in files:
|
r = rel(path)
|
src = source_by_raw.get(r) or source_by_converted.get(r)
|
out = output_by_path.get(r)
|
if out:
|
artifact_id = out["artifact_id"]
|
elif src and r == src["raw_file_path"]:
|
artifact_id = f"ART-{src['doc_id']}-RAW"
|
elif src:
|
artifact_id = f"ART-{src['doc_id']}-CONVERTED"
|
else:
|
artifact_id = f"ART-REPAIR-{generic_i:04d}"
|
generic_i += 1
|
rows.append({
|
"artifact_id": artifact_id, "task_id": TASK_ID, "case_id": CASE_ID,
|
"batch_id": BATCH_ID, "run_id": RUN_ID, "artifact_type": artifact_type(path),
|
"industry_case": "新能源案例", "industry_id": "IND-NEWENERGY",
|
"subindustry_id": src["subindustry_id"] if src else "",
|
"company_id": src["company_id"] if src else "", "logical_path": r, "relative_path": r,
|
"absolute_path": path.resolve().as_posix(), "file_name": path.name,
|
"file_ext": path.suffix.lower(), "file_size": str(path.stat().st_size), "sha256": sha256_file(path),
|
"source_doc_id": src["doc_id"] if src else "", "source_url": src["source_url"] if src else "",
|
"source_collected_at": src["collected_at"] if src else COLLECTED_AT,
|
"raw_pool_path": src["raw_pool_path"] if src else "", "source_file_name": path.name,
|
"detected_type": path.suffix.lower().lstrip(".").upper() or "NO_EXT", "archive_file_name": path.name,
|
"extension_added_by_archive_flag": "NO", "extension_mismatch_flag": "NO",
|
"created_at": COLLECTED_AT, "created_by": "case_analysis.analyst.new_energy",
|
"tool_or_method": "NEWENERGY_BATCH001_REPAIR_REBUILD",
|
"tool_version": ARTIFACT_TOOL_VERSION, "parameters_summary": ARTIFACT_PARAMETERS_SUMMARY,
|
"source_snapshot_id": f"SNAP-{src['doc_id']}" if src else "",
|
"artifact_status": ARTIFACT_STATUS,
|
"sensitivity_screen": src["sensitivity_screen"] if src else "LEGAL_PUBLIC_SCREENED",
|
})
|
if len({r["artifact_id"] for r in rows}) != len(rows):
|
raise RuntimeError("Duplicate artifact_id")
|
write_csv(MANIFEST_ROOT / "artifact_manifest.csv", HEADERS["artifact_manifest"], rows)
|
return rows
|
|
|
def remove_case_level_duplicates() -> None:
|
for name in [
|
"company_track_candidate_ledger.csv", "classification_summary.csv", "evidence_fact.csv",
|
"newenergy_scope_matrix.csv", "newenergy_technology_route_matrix.csv",
|
"newenergy_supply_demand_price_metric.csv", "newenergy_project_capacity_event.csv",
|
"newenergy_company_exposure_matrix.csv", "newenergy_catalyst_risk_register.csv",
|
]:
|
path = CASE_EVIDENCE / name
|
if path.exists():
|
path.unlink()
|
for name in ["artifact_manifest.csv", "source_document.csv", "input_manifest.csv", "source_gap_audit.csv"]:
|
path = CASE_MANIFEST / name
|
if path.exists():
|
path.unlink()
|
|
|
def validate(
|
candidates: list[dict[str, str]], universe_rows: list[dict[str, str]], source_rows: list[dict[str, str]],
|
conversions: list[dict[str, str]], artifacts: list[dict[str, str]], case_map: list[dict[str, str]],
|
evidence_rows: list[dict[str, str]], query_summaries: list[dict], qualification_rows: list[dict[str, str]],
|
) -> None:
|
errors = []
|
if len(universe_rows) != len({(r["exchange_code"], r["security_code"]) for r in universe_rows}):
|
errors.append("universe duplicate")
|
exchange_counts: dict[str, int] = defaultdict(int)
|
for row in universe_rows:
|
exchange_counts[row["exchange_code"]] += 1
|
expected_exchange_counts = {"SSE": 2310, "SZSE": 2898, "BSE": 333}
|
if dict(exchange_counts) != expected_exchange_counts:
|
errors.append(f"universe exchange counts {dict(exchange_counts)} != {expected_exchange_counts}")
|
if len(candidates) != len({(r["company_id"], r["track_code"]) for r in candidates}):
|
errors.append("candidate duplicate")
|
if len(query_summaries) != 16:
|
errors.append(f"query summary count {len(query_summaries)}")
|
for summary in query_summaries:
|
if summary["archived_record_count"] != summary["server_total_record_num"]:
|
errors.append(f"query incomplete {summary['source_doc_id']}")
|
expected_terminal = summary["server_total_record_num"] % 100 or 100
|
if summary["terminal_page_record_count"] != expected_terminal:
|
errors.append(f"query terminal page {summary['source_doc_id']}")
|
if len(qualification_rows) != len(candidates):
|
errors.append(f"qualification count {len(qualification_rows)} != {len(candidates)}")
|
if len({(r["company_id"], r["track_code"]) for r in qualification_rows}) != len(qualification_rows):
|
errors.append("qualification duplicate")
|
for row in qualification_rows:
|
if row["replay_status"] != "ATTEMPT_COMPLETED_FOR_DISCOVERED_PAIR":
|
errors.append(f"qualification attempt missing {row['qualification_row_id']}")
|
if row["annual_report_retrieval_result"] == "FAILED_NO_ANNUAL_REPORT_CONTEXT":
|
errors.append(f"annual report context missing {row['qualification_row_id']}")
|
if not row["fulltext_context_sha256"] or not row["business_context_rule_result"]:
|
errors.append(f"annual report context unprocessed {row['qualification_row_id']}")
|
if "\ufffd" in row["fulltext_context_excerpt_sanitized"]:
|
errors.append(f"qualification replacement char {row['qualification_row_id']}")
|
if row["eligibility_result"] == "ELIGIBLE" and row["page_level_verification_result"] != "PASS_VERIFIED_DIRECT_BUSINESS_LOCATOR":
|
errors.append(f"eligible page locator {row['qualification_row_id']}")
|
if row["eligibility_result"] != "ELIGIBLE" and not row["failed_gates"]:
|
errors.append(f"held without failed gate {row['qualification_row_id']}")
|
selected = [r for r in candidates if r["candidate_state"] in {"INCLUDED_T1", "INCLUDED_T2"}]
|
if len(selected) != 32:
|
errors.append(f"selected != 32 ({len(selected)})")
|
if not any(r["candidate_state"] == "HELD_BY_EVIDENCE_GAP" for r in candidates):
|
errors.append("no held discovery candidates")
|
for track, _, bucket, _ in QUERY_CONFIG:
|
bucket_rows = [r for r in selected if r["track_code"] == track and r["selection_bucket"] == bucket]
|
if sorted((r["candidate_state"], r["selection_rank"]) for r in bucket_rows) != [("INCLUDED_T1", "1"), ("INCLUDED_T2", "2")]:
|
errors.append(f"selection bucket invalid {track}/{bucket}")
|
if len(source_rows) != len({r["doc_id"] for r in source_rows}):
|
errors.append("source duplicate")
|
if len(conversions) != len(source_rows):
|
errors.append("conversion count mismatch")
|
for row in conversions:
|
path = PROJECT_ROOT / row["converted_path"]
|
if not path.exists() or sha256_file(path) != row["converted_sha256"]:
|
errors.append(f"conversion hash {row['conversion_id']}")
|
evidence_doc_ids = {row["doc_id"] for row in evidence_rows}
|
source_by_id = {row["doc_id"]: row for row in source_rows}
|
conversion_by_id = {row["source_doc_id"]: row for row in conversions}
|
for doc_id in evidence_doc_ids:
|
if source_by_id[doc_id]["processing_status"] != "TEXT_CONVERTED_EVIDENCE_EXTRACTED":
|
errors.append(f"source evidence processing {doc_id}")
|
if conversion_by_id[doc_id]["status"] != "CONVERTED_EVIDENCE_EXTRACTED":
|
errors.append(f"conversion evidence processing {doc_id}")
|
formal = formal_files(Path(__file__))
|
if len(artifacts) != len(formal):
|
errors.append(f"artifact exact-set mismatch {len(artifacts)} != {len(formal)}")
|
if any(row["tool_version"] != ARTIFACT_TOOL_VERSION for row in artifacts):
|
errors.append("artifact tool version mismatch")
|
artifact_paths = {r["relative_path"] for r in artifacts}
|
missing = {rel(p) for p in formal} - artifact_paths
|
if missing:
|
errors.append(f"artifact missing {sorted(missing)[:5]}")
|
required_output_fragments = ["新能源行业视图.md"] + [f"{d}/{n}" for d in ["01_锂电", "02_光伏", "03_风电", "04_核电"] for n in ["总览.md", "产业链与技术路线.md", "市场与供需.md", "相关企业.md"]]
|
mapped_paths = {r["output_path"] for r in case_map}
|
for fragment in required_output_fragments:
|
if not any(p.endswith(fragment) for p in mapped_paths):
|
errors.append(f"unmapped output {fragment}")
|
if "EVF-MKT-NUC-2024" not in {r["evidence_fact_id"] for r in case_map}:
|
errors.append("nuclear 2024 evidence unmapped")
|
for row in case_map:
|
output = PROJECT_ROOT / row["output_path"]
|
if not output.exists():
|
errors.append(f"map output missing {row['conclusion_evidence_map_id']}")
|
continue
|
text = output.read_text(encoding="utf-8")
|
if f'<a id="{row["section_anchor"]}"></a>' not in text:
|
errors.append(f"map anchor missing {row['conclusion_evidence_map_id']}")
|
if row["conclusion_text"] not in text:
|
errors.append(f"map conclusion missing {row['conclusion_evidence_map_id']}")
|
cec_source = source_by_id.get("S-MARKET-NUCLEAR-CEC-2024", {})
|
if cec_source.get("source_org") != "中国电力企业联合会" or cec_source.get("author") != "中国电力企业联合会":
|
errors.append("CEC source attribution")
|
nuclear_2024 = next((row for row in evidence_rows if row["evidence_fact_id"] == "EVF-MKT-NUC-2024"), None)
|
if not nuclear_2024 or "中国电力企业联合会" not in nuclear_2024["original_qualifier"]:
|
errors.append("CEC evidence attribution")
|
nuclear_2025_maps = [row for row in case_map if row["evidence_fact_id"] == "EVF-MKT-NUC-01"]
|
nuclear_2025_attributed_maps = [
|
row for row in nuclear_2025_maps
|
if "中国核能行业协会" in (row["conclusion_text"] + " " + row["contradiction_or_limit"])
|
or "中国电力企业联合会" in (row["conclusion_text"] + " " + row["contradiction_or_limit"])
|
]
|
if len(nuclear_2025_attributed_maps) != 4:
|
errors.append(f"nuclear 2025 attributed map count {len(nuclear_2025_attributed_maps)}")
|
for row in nuclear_2025_attributed_maps:
|
scope_text = row["conclusion_text"] + " " + row["contradiction_or_limit"]
|
if "中国核能行业协会" not in scope_text or "中国电力企业联合会" in scope_text:
|
errors.append(f"nuclear 2025 attribution {row['conclusion_evidence_map_id']}")
|
for row in [item for item in case_map if item["evidence_fact_id"] == "EVF-MKT-NUC-2024"]:
|
scope_text = row["conclusion_text"] + " " + row["contradiction_or_limit"]
|
if "中国电力企业联合会" not in scope_text or "中国核能行业协会" in scope_text:
|
errors.append(f"nuclear 2024 map attribution {row['conclusion_evidence_map_id']}")
|
for path in [INDUSTRY_ROOT / x for x in ["converted", "extracted", "supplement", "evidence", "manifest"]]:
|
if not path.exists():
|
errors.append(f"missing industry root {path}")
|
if errors:
|
raise RuntimeError("Validation failed:\n- " + "\n- ".join(errors))
|
|
|
def main() -> None:
|
ensure_roots()
|
# Reruns read the industry canonical produced by an earlier successful build;
|
# the first run falls back to the HOLD-era case-local files.
|
candidate_input = EXTRACTED_ROOT / "company_track_candidate_ledger.csv"
|
if not candidate_input.exists():
|
candidate_input = OLD_CASE_EVIDENCE / "company_track_candidate_ledger.csv"
|
current_candidates = [
|
row for row in read_csv(candidate_input)
|
if row["candidate_state"] in {"INCLUDED_T1", "INCLUDED_T2"}
|
]
|
selected_keys = {(row["company_id"], row["track_code"]) for row in current_candidates}
|
|
classification_input = EXTRACTED_ROOT / "classification_summary.csv"
|
if not classification_input.exists():
|
classification_input = OLD_CASE_EVIDENCE / "classification_summary.csv"
|
current_classification = [
|
row for row in read_csv(classification_input)
|
if (row["company_id"], row["track_code"]) in selected_keys
|
]
|
|
evidence_input = EVIDENCE_ROOT / "evidence_fact_table.csv"
|
if not evidence_input.exists():
|
evidence_input = OLD_CASE_EVIDENCE / "evidence_fact.csv"
|
evidence_rows = read_csv(evidence_input)
|
current_case_map = read_csv(OLD_CASE_EVIDENCE / "case_evidence_map.csv")
|
source_input = MANIFEST_ROOT / "source_document.csv"
|
if not source_input.exists():
|
source_input = OLD_CASE_MANIFEST / "source_document.csv"
|
source_rows = [
|
row for row in read_csv(source_input)
|
if row["doc_type"] not in {"OFFICIAL_A_SHARE_UNIVERSE_SNAPSHOT", "OFFICIAL_DISCOVERY_QUERY_RECEIPT"}
|
]
|
output_rows = read_csv(OLD_CASE_MANIFEST / "output_manifest.csv")
|
|
old_cec = RAW_ROOT / "official_market/NUCLEAR_CNEA_2024_OPERATION.html"
|
new_cec = RAW_ROOT / "official_market/NUCLEAR_CEC_2024_OPERATION.html"
|
if old_cec.exists() and not new_cec.exists():
|
old_cec.replace(new_cec)
|
elif old_cec.exists() and new_cec.exists():
|
if sha256_file(old_cec) != sha256_file(new_cec):
|
raise RuntimeError("CEC raw rename collision")
|
old_cec.unlink()
|
|
universe_rows, universe_by_code, universe_sources = normalize_universe()
|
hits, query_sources, query_summaries = collect_discovery_queries(universe_by_code)
|
all_new_sources = universe_sources + query_sources
|
source_rows = make_source_documents(source_rows, all_new_sources)
|
source_by_id = {r["doc_id"]: r for r in source_rows}
|
|
# Correct predecessor IDs before downstream joins.
|
for row in evidence_rows:
|
if row["doc_id"] == "S-MARKET-NUCLEAR-CNEA-2024":
|
row["doc_id"] = "S-MARKET-NUCLEAR-CEC-2024"
|
candidates = build_candidate_ledger(current_candidates, hits, universe_by_code, source_by_id)
|
|
# Generate conversions, then backfill canonical source/evidence paths.
|
conversions = convert_sources(source_rows, evidence_rows, query_summaries)
|
evidence_rows = update_evidence_facts(evidence_rows, source_rows)
|
source_by_id = {r["doc_id"]: r for r in source_rows}
|
candidates, qualification_rows = apply_candidate_qualification_funnel(
|
candidates, hits, universe_by_code, source_rows, evidence_rows
|
)
|
|
# Correct the CEC ID anywhere it remains in inherited case-level files.
|
for root in [CASE_EVIDENCE, CASE_MANIFEST]:
|
for path in root.glob("*.csv"):
|
text = path.read_text(encoding="utf-8")
|
text = text.replace("S-MARKET-NUCLEAR-CNEA-2024", "S-MARKET-NUCLEAR-CEC-2024")
|
text = text.replace("NUCLEAR_CNEA_2024_OPERATION.html", "NUCLEAR_CEC_2024_OPERATION.html")
|
write_text(path, text)
|
|
classification = expand_classification(current_classification, candidates, source_rows)
|
metric_input = EXTRACTED_ROOT / "newenergy_supply_demand_price_metric.csv"
|
if not metric_input.exists():
|
metric_input = OLD_CASE_EVIDENCE / "newenergy_supply_demand_price_metric.csv"
|
metrics = read_csv(metric_input)
|
for row in metrics:
|
if row["source_id"] == "S-MARKET-NUCLEAR-CNEA-2024":
|
row["source_id"] = "S-MARKET-NUCLEAR-CEC-2024"
|
|
# Move common canonical tables to their industry-level authority.
|
write_csv(MANIFEST_ROOT / "source_document.csv", list(source_rows[0].keys()), source_rows)
|
write_csv(MANIFEST_ROOT / "conversion_status.csv", HEADERS["conversion_status"], conversions)
|
build_input_manifests(source_rows)
|
build_source_gap_audit(source_rows, evidence_rows)
|
write_csv(EVIDENCE_ROOT / "evidence_fact_table.csv", list(evidence_rows[0].keys()), evidence_rows)
|
write_csv(EXTRACTED_ROOT / "company_track_candidate_ledger.csv", list(candidates[0].keys()), candidates)
|
write_csv(EXTRACTED_ROOT / "classification_summary.csv", list(classification[0].keys()), classification)
|
|
for name in [
|
"newenergy_scope_matrix.csv", "newenergy_technology_route_matrix.csv",
|
"newenergy_project_capacity_event.csv", "newenergy_company_exposure_matrix.csv",
|
"newenergy_catalyst_risk_register.csv",
|
]:
|
input_path = EXTRACTED_ROOT / name
|
if not input_path.exists():
|
input_path = OLD_CASE_EVIDENCE / name
|
rows = read_csv(input_path)
|
write_csv(EXTRACTED_ROOT / name, list(rows[0].keys()), rows)
|
write_csv(EXTRACTED_ROOT / "newenergy_supply_demand_price_metric.csv", list(metrics[0].keys()), metrics)
|
|
update_outputs()
|
case_map = expand_case_evidence_map(current_case_map, candidates, metrics, evidence_rows)
|
case_map = materialize_evidence_locators(case_map)
|
write_csv(CASE_EVIDENCE / "case_evidence_map.csv", list(case_map[0].keys()), case_map)
|
write_discovery_receipt(query_summaries, candidates, universe_rows)
|
rewrite_batch_summary(candidates, len(source_rows), len(conversions), universe_rows)
|
build_human_receipt(case_map, len(source_rows))
|
output_rows = rebuild_output_manifest()
|
|
remove_case_level_duplicates()
|
artifacts = build_artifact_manifest(source_rows, output_rows, Path(__file__))
|
validate(
|
candidates, universe_rows, source_rows, conversions, artifacts, case_map,
|
evidence_rows, query_summaries, qualification_rows,
|
)
|
|
result = {
|
"status": "PASS_LOCAL_REPAIR_BUILD",
|
"universe": len(universe_rows), "candidate_pairs": len(candidates),
|
"candidate_states": dict(sorted((s, sum(1 for r in candidates if r["candidate_state"] == s)) for s in {r["candidate_state"] for r in candidates})),
|
"sources": len(source_rows), "conversions": len(conversions), "evidence_facts": len(evidence_rows),
|
"case_evidence_map": len(case_map), "formal_artifacts_excluding_manifest_self": len(artifacts),
|
"artifact_manifest_sha256": sha256_file(MANIFEST_ROOT / "artifact_manifest.csv"),
|
}
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
|
if __name__ == "__main__":
|
main()
|
