#!/usr/bin/env python3 """REPAIR004: acquire and page-verify the frozen potential-direct annual reports. This tool intentionally consumes the already reviewed REPAIR003 discovery snapshot. It does not issue new discovery queries and it does not expand the case scope. Its first phase acquires the 729 unique CNINFO adjunct PDFs referenced by the 858 frozen potential-direct pairs. Its second phase searches every readable page and writes replayable attachment- and pair-level receipts. Candidate reranking is performed only after these receipts have been inspected. """ from __future__ import annotations import argparse import csv import hashlib import json import os import re import shutil import sys import time import urllib.error import urllib.request from collections import defaultdict from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, as_completed from datetime import datetime, timezone from pathlib import Path from typing import Any, Iterable import pypdfium2 as pdfium 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" TOOL_VERSION = "REPAIR-004" POTENTIAL = "POTENTIAL_DIRECT_BUSINESS_CONTEXT_REQUIRES_PAGE_VERIFICATION" CNINFO_BASE = "https://static.cninfo.com.cn/" USER_AGENT = "Mozilla/5.0 MB-X-NewEnergy-Research/REPAIR004 public-source-audit" TRACK_KEYWORDS = { ("BATTERY", "资源与主材"): "正极材料", ("BATTERY", "电芯制造"): "锂离子电池", ("BATTERY", "系统/部件/BMS-Pack"): "电池管理系统", ("BATTERY", "设备与回收循环"): "锂电设备", ("SOLAR", "硅料/硅片与材料"): "光伏硅片", ("SOLAR", "电池片/组件"): "光伏组件", ("SOLAR", "设备/辅材/逆变器"): "光伏逆变器", ("SOLAR", "系统集成/电站建设运营"): "光伏电站", ("WIND", "材料与关键零部件"): "风电零部件", ("WIND", "整机"): "风力发电机组", ("WIND", "塔筒/海缆/工程配套"): "风电塔筒", ("WIND", "项目运营与运维服务"): "风电场", ("NUCLEAR", "运营商"): "核电运营", ("NUCLEAR", "工程/EPC"): "核电工程", ("NUCLEAR", "核岛/常规岛主设备"): "核电设备", ("NUCLEAR", "核级部件/材料/仪控电气"): "核级阀门", } CHAIN_NODES = { ("BATTERY", "资源与主材"): "锂资源;锂盐;正极材料及前驱体", ("BATTERY", "电芯制造"): "锂离子电池;电芯;动力/储能电池", ("BATTERY", "系统/部件/BMS-Pack"): "电池管理系统;BMS;模组/PACK", ("BATTERY", "设备与回收循环"): "锂电设备;动力电池回收", ("SOLAR", "硅料/硅片与材料"): "多晶硅;硅棒;光伏硅片", ("SOLAR", "电池片/组件"): "太阳能电池片;光伏组件", ("SOLAR", "设备/辅材/逆变器"): "光伏设备;辅材;光伏逆变器", ("SOLAR", "系统集成/电站建设运营"): "光伏系统集成;电站建设运营", ("WIND", "材料与关键零部件"): "风电材料;铸件;主轴;轴承;叶片", ("WIND", "整机"): "风电整机;风力发电机组", ("WIND", "塔筒/海缆/工程配套"): "风电塔筒;海缆;工程配套", ("WIND", "项目运营与运维服务"): "风电场;项目运营;运维服务", ("NUCLEAR", "运营商"): "民用核电运营", ("NUCLEAR", "工程/EPC"): "民用核电工程;EPC", ("NUCLEAR", "核岛/常规岛主设备"): "核岛主设备;常规岛主设备", ("NUCLEAR", "核级部件/材料/仪控电气"): "核级部件;材料;仪控电气", } DOWNLOAD_HEADERS = [ "attachment_id", "adjunct_path", "source_url", "announcement_title", "security_codes", "security_names", "pair_count", "pair_ids", "attempted_at", "attempt_count", "http_status", "final_url", "content_type", "response_byte_count", "response_sha256", "raw_path", "raw_byte_count", "raw_sha256", "pdf_signature", "pdf_readable", "page_count", "acquisition_result", "failure_class", "failure_detail", "tool_version", "review_status", ] PAIR_HEADERS = [ "qualification_row_id", "task_id", "case_id", "batch_id", "run_id", "company_id", "security_code", "track_code", "selection_bucket", "search_terms", "attachment_ids", "attachment_urls", "attachment_count", "retrieval_success_count", "retrieval_failure_count", "searched_pdf_count", "searched_page_count", "keyword_hit_attachment_count", "keyword_hit_pages", "keyword_hit_count", "direct_context_hit_pages", "direct_context_hit_count", "exact_context_sentences", "negative_or_insufficient_context_samples", "page_search_result", "qualification_result", "failure_or_hold_reason", "receipt_sha256", "verified_at", "tool_version", "review_status", ] def now_iso() -> str: return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds") 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 handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest() def read_csv(path: Path) -> list[dict[str, str]]: with path.open("r", encoding="utf-8-sig", newline="") as handle: return list(csv.DictReader(handle)) def write_csv(path: Path, headers: list[str], rows: Iterable[dict[str, Any]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(path.suffix + ".tmp") with tmp.open("w", encoding="utf-8-sig", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=headers, extrasaction="ignore", quoting=csv.QUOTE_ALL) writer.writeheader() for row in rows: writer.writerow({key: row.get(key, "") for key in headers}) os.replace(tmp, path) def find_industry_root(project_root: Path) -> Path: matches = [p for p in (project_root / "ana-data" / "cases").rglob("candidate_qualification_funnel.csv") if CASE_ID not in str(p)] if len(matches) != 1: raise RuntimeError(f"expected one industry funnel, found {len(matches)}") return matches[0].parent.parent def split_values(value: str) -> list[str]: return [item.strip() for item in value.split(";") if item.strip()] def sanitize_detail(value: str, limit: int = 600) -> str: return re.sub(r"[\r\n\t]+", " ", value or "").strip()[:limit] def load_announcement_metadata(industry_root: Path) -> dict[str, dict[str, str]]: result: dict[str, dict[str, str]] = {} for path in sorted((industry_root / "raw" / "official_discovery").glob("*.json")): payload = json.loads(path.read_text(encoding="utf-8-sig")) for page in payload.get("pages", []): for ann in page.get("response", {}).get("announcements", []): ann_id = str(ann.get("announcementId", "")) if not ann_id: continue result.setdefault(ann_id, { "attachment_id": ann_id, "adjunct_path": str(ann.get("adjunctUrl", "")), "announcement_title": re.sub(r"<[^>]+>", "", str(ann.get("announcementTitle", ""))), "security_code": str(ann.get("secCode", "")), "security_name": str(ann.get("secName", "")), }) return result def existing_pdf_for_attachment(industry_root: Path, attachment_id: str) -> Path | None: for path in (industry_root / "raw").rglob("*.pdf"): if attachment_id in path.name: return path return None def validate_pdf(path: Path) -> tuple[str, str, int, str]: signature = "" readable = "NO" page_count = 0 error = "" try: with path.open("rb") as handle: signature = handle.read(5).decode("latin-1", errors="replace") if signature != "%PDF-": return signature, readable, page_count, "MISSING_PDF_SIGNATURE" reader = PdfReader(str(path), strict=False) page_count = len(reader.pages) readable = "YES" if page_count > 0 else "NO" if page_count <= 0: error = "ZERO_PAGE_PDF" except Exception as exc: # recorded in formal receipt error = sanitize_detail(f"{type(exc).__name__}: {exc}") return signature, readable, page_count, error def download_one(item: dict[str, Any], raw_dir: Path, project_root: Path, retries: int = 3) -> dict[str, Any]: attachment_id = item["attachment_id"] target = raw_dir / f"{attachment_id}.pdf" attempted_at = now_iso() existing = None if item.get("force_http") else (target if target.exists() and target.stat().st_size > 0 else None) if existing is None and not item.get("force_http"): existing = existing_pdf_for_attachment(item["industry_root"], attachment_id) if existing is not None: if existing.resolve() != target.resolve(): shutil.copy2(existing, target) signature, readable, pages, parse_error = validate_pdf(target) file_hash = sha256_file(target) size = target.stat().st_size return { **item, "attempted_at": attempted_at, "attempt_count": "0", "http_status": "REUSED_EXISTING_CANONICAL", "final_url": item["source_url"], "content_type": "application/pdf", "response_byte_count": str(size), "response_sha256": file_hash, "raw_path": target.relative_to(project_root).as_posix(), "raw_byte_count": str(size), "raw_sha256": file_hash, "pdf_signature": signature, "pdf_readable": readable, "page_count": str(pages), "acquisition_result": "REUSED_EXISTING_CANONICAL_PDF" if readable == "YES" else "REUSED_FILE_UNREADABLE", "failure_class": "" if readable == "YES" else "PDF_VALIDATION_FAILED", "failure_detail": parse_error, "tool_version": TOOL_VERSION, "review_status": "DRAFT_FOR_REVIEW", } last_class = "" last_detail = "" last_status = "" last_body = b"" final_url = item["source_url"] content_type = "" for attempt in range(1, retries + 1): request = urllib.request.Request(item["source_url"], headers={"User-Agent": USER_AGENT}) try: with urllib.request.urlopen(request, timeout=120) as response: last_status = str(response.getcode()) final_url = response.geturl() content_type = response.headers.get("Content-Type", "") last_body = response.read() part = target.with_suffix(".pdf.part") part.write_bytes(last_body) os.replace(part, target) signature, readable, pages, parse_error = validate_pdf(target) file_hash = sha256_file(target) size = target.stat().st_size result = "ACQUIRED_PDF_AND_VALIDATED" if readable == "YES" else "ACQUIRED_RESPONSE_PDF_VALIDATION_FAILED" return { **item, "attempted_at": attempted_at, "attempt_count": str(attempt), "http_status": last_status, "final_url": final_url, "content_type": content_type, "response_byte_count": str(len(last_body)), "response_sha256": sha256_bytes(last_body), "raw_path": target.relative_to(project_root).as_posix(), "raw_byte_count": str(size), "raw_sha256": file_hash, "pdf_signature": signature, "pdf_readable": readable, "page_count": str(pages), "acquisition_result": result, "failure_class": "" if readable == "YES" else "PDF_VALIDATION_FAILED", "failure_detail": parse_error, "tool_version": TOOL_VERSION, "review_status": "DRAFT_FOR_REVIEW", } except urllib.error.HTTPError as exc: last_status = str(exc.code) final_url = exc.geturl() content_type = exc.headers.get("Content-Type", "") if exc.headers else "" try: last_body = exc.read() except Exception: last_body = b"" last_class = "HTTP_ERROR" last_detail = sanitize_detail(f"HTTPError {exc.code}: {exc.reason}") except Exception as exc: last_class = type(exc).__name__.upper() last_detail = sanitize_detail(f"{type(exc).__name__}: {exc}") if attempt < retries: time.sleep(min(8, 2 ** attempt)) return { **item, "attempted_at": attempted_at, "attempt_count": str(retries), "http_status": last_status, "final_url": final_url, "content_type": content_type, "response_byte_count": str(len(last_body)), "response_sha256": sha256_bytes(last_body) if last_body else "", "raw_path": "", "raw_byte_count": "0", "raw_sha256": "", "pdf_signature": "", "pdf_readable": "NO", "page_count": "0", "acquisition_result": "ACQUISITION_FAILED_AFTER_RETRIES", "failure_class": last_class, "failure_detail": last_detail, "tool_version": TOOL_VERSION, "review_status": "DRAFT_FOR_REVIEW", } def normalize_text(value: str) -> str: return re.sub(r"\s+", "", value or "") def sentence_windows(text: str, keyword: str, limit: int = 4) -> list[str]: cleaned = re.sub(r"[\u0000-\u0008\u000b\u000c\u000e-\u001f]", "", text or "") cleaned = re.sub(r"[ \t]+", " ", cleaned) parts = re.split(r"(?<=[。！？；;])|\n+", cleaned) result: list[str] = [] for index, part in enumerate(parts): if keyword not in normalize_text(part): continue joined = "".join(parts[max(0, index - 1): min(len(parts), index + 2)]).strip() joined = sanitize_detail(joined, 1000) if joined and joined not in result: result.append(joined) if len(result) >= limit: break if not result and keyword in normalize_text(cleaned): norm = normalize_text(cleaned) pos = norm.find(keyword) result.append(norm[max(0, pos - 240):pos + len(keyword) + 360]) return result def search_pdf_worker(payload: tuple[str, str, list[str]]) -> dict[str, Any]: attachment_id, raw_path, keywords = payload result: dict[str, Any] = { "attachment_id": attachment_id, "page_count": 0, "readable": "NO", "error": "", "keyword_hits": {}, } try: doc = pdfium.PdfDocument(raw_path) result["page_count"] = len(doc) result["readable"] = "YES" hit_map: dict[str, list[dict[str, Any]]] = {keyword: [] for keyword in keywords} for index in range(len(doc)): page = doc[index] textpage = page.get_textpage() text = textpage.get_text_range() normalized = normalize_text(text) for keyword in keywords: if keyword in normalized: hit_map[keyword].append({ "page": index + 1, "sentences": sentence_windows(text, keyword), }) textpage.close() page.close() doc.close() result["keyword_hits"] = hit_map except Exception as exc: result["error"] = sanitize_detail(f"{type(exc).__name__}: {exc}") return result DIRECT_PATTERNS = [ re.compile(r"(?:本公司|本集团|公司|集团).{0,120}(?:主营业务|主要业务|核心业务|主要从事|业务包括|产品包括|主要产品).{0,180}"), re.compile(r"(?:本公司|本集团|公司|集团).{0,120}(?:生产|制造|销售|提供服务|运营|承建|总承包).{0,160}"), re.compile(r"(?:本公司|本集团|公司|集团).{0,120}(?:旗下|全资子公司|控股子公司).{0,100}(?:专业从事|主营|生产|制造|销售|运营|承建|总承包).{0,160}"), re.compile(r"(?:主营业务|主要业务|核心业务|主要产品).{0,100}"), ] NEGATIVE_TOKENS = ("不涉及", "不从事", "未从事", "无相关业务", "尚未开展", "不具备") CONTEXT_ONLY_TOKENS = ("行业发展", "市场规模", "竞争对手", "供应商", "客户从事", "参股基金", "投资标的", "政策鼓励") BUCKET_ROLE_PATTERNS = { ("BATTERY", "资源与主材"): [ r"(?:公司|本公司|本集团).{0,100}(?:锂产品|锂盐|碳酸锂|氢氧化锂|正极材料|前驱体).{0,100}(?:生产|制造|销售|主营|主要从事)", r"(?:公司|本公司|本集团).{0,100}(?:生产|制造|销售|主营|主要从事).{0,100}(?:锂产品|锂盐|碳酸锂|氢氧化锂|正极材料|前驱体)", ], ("BATTERY", "电芯制造"): [ r"(?:公司|本公司|本集团).{0,80}(?:生产|制造|销售|主营|主要从事).{0,40}(?:锂离子电池(?!材料|添加剂)|电芯|动力电池(?!材料)|储能电池(?!材料))", r"(?:公司|本公司|本集团).{0,80}(?:锂离子电池(?!材料|添加剂)|电芯|动力电池(?!材料)|储能电池(?!材料)).{0,40}(?:生产|制造|销售|主营业务|主要业务)", ], ("BATTERY", "系统/部件/BMS-Pack"): [ r"(?:公司|本公司|本集团).{0,100}(?:电池管理系统|BMS|电池模组|PACK).{0,100}(?:产品|生产|制造|销售|供货|主营)", r"(?:核心产品|主要产品|产品包括).{0,100}(?:电池管理系统|BMS|电池模组|PACK)", ], ("BATTERY", "设备与回收循环"): [ r"(?:公司|本公司|本集团).{0,100}(?:锂电设备|锂电池设备|电池回收|动力电池回收).{0,100}(?:产品|生产|制造|销售|业务|主营)", r"(?:核心产品|主要产品|主营业务).{0,100}(?:锂电设备|锂电池设备|电池回收|动力电池回收)", ], ("SOLAR", "硅料/硅片与材料"): [ r"(?:公司|本公司|本集团).{0,100}(?:光伏硅片|单晶硅片|硅棒|多晶硅).{0,100}(?:生产|制造|销售|主营|主要业务)", r"(?:公司|本公司|本集团).{0,100}(?:生产|制造|销售|主营|主要从事).{0,100}(?:光伏硅片|单晶硅片|硅棒|多晶硅)", ], ("SOLAR", "电池片/组件"): [ r"(?:公司|本公司|本集团).{0,100}(?:光伏组件|太阳能电池片|光伏电池片).{0,100}(?:生产|制造|销售|主营|主要业务)", r"(?:公司|本公司|本集团).{0,100}(?:生产|制造|销售|主营|主要从事).{0,100}(?:光伏组件|太阳能电池片|光伏电池片)", ], ("SOLAR", "设备/辅材/逆变器"): [ r"(?:公司|本公司|本集团).{0,100}(?:光伏逆变器|光伏设备|光伏辅材).{0,100}(?:产品|生产|制造|销售|主营|主要业务)", r"(?:核心产品|主要产品|产品包括).{0,100}(?:光伏逆变器|光伏设备|光伏辅材)", ], ("SOLAR", "系统集成/电站建设运营"): [ r"(?:公司|本公司|本集团).{0,120}(?:光伏电站|光伏发电站|分布式光伏).{0,120}(?:建设|运营|持有|投资开发|发电收入|EPC|总承包)", r"(?:公司|本公司|本集团).{0,120}(?:建设|运营|持有|投资开发|EPC|总承包).{0,120}(?:光伏电站|光伏发电站|分布式光伏)", ], ("WIND", "材料与关键零部件"): [ r"(?:公司|本公司|本集团).{0,100}(?:风电零部件|风电铸件|风电主轴|风电轴承|风电叶片).{0,100}(?:生产|制造|销售|主营|主要业务|供应)", r"(?:公司|本公司|本集团).{0,100}(?:生产|制造|销售|主营|主要从事).{0,100}(?:风电零部件|风电铸件|风电主轴|风电轴承|风电叶片)", ], ("WIND", "整机"): [ r"(?:公司|本公司|本集团).{0,100}(?:风力发电机组|风电整机).{0,100}(?:研发生产|生产|制造|销售|主营|产品)", r"(?:公司|本公司|本集团).{0,100}(?:生产|制造|销售|主营|主要从事).{0,100}(?:风力发电机组|风电整机)", ], ("WIND", "塔筒/海缆/工程配套"): [ r"(?:公司|本公司|本集团).{0,100}(?:风电塔筒|风电塔架|海缆|海底电缆).{0,100}(?:生产|制造|销售|主营|工程|服务)", r"(?:公司|本公司|本集团).{0,100}(?:生产|制造|销售|主营|承建).{0,100}(?:风电塔筒|风电塔架|海缆|海底电缆)", ], ("WIND", "项目运营与运维服务"): [ r"(?:公司|本公司|本集团).{0,120}(?:风电场|风力发电项目).{0,120}(?:建设|运营|持有|投资开发|发电收入|运维|EPC)", r"(?:公司|本公司|本集团).{0,120}(?:建设|运营|持有|投资开发|运维|EPC).{0,120}(?:风电场|风力发电项目)", ], ("NUCLEAR", "运营商"): [ r"(?:公司|本公司|本集团).{0,120}(?:核电站|核电机组|核电项目).{0,120}(?:运营|运行|持有|投资开发|发电)", r"(?:公司|本公司|本集团).{0,120}(?:运营|运行|持有|投资开发).{0,120}(?:核电站|核电机组|核电项目)", ], ("NUCLEAR", "工程/EPC"): [ r"(?:公司|本公司|本集团).{0,120}(?:核电工程|核工程).{0,120}(?:EPC|总承包|承建|施工|服务|主营)", r"(?:公司|本公司|本集团).{0,120}(?:EPC|总承包|承建|施工).{0,120}(?:核电工程|核工程)", ], ("NUCLEAR", "核岛/常规岛主设备"): [ r"(?:公司|本公司|本集团).{0,120}(?:核电设备|核岛设备|常规岛设备).{0,120}(?:生产|制造|销售|产品|主营)", r"(?:公司|本公司|本集团).{0,120}(?:生产|制造|销售|主营).{0,120}(?:核电设备|核岛设备|常规岛设备)", ], ("NUCLEAR", "核级部件/材料/仪控电气"): [ r"(?:公司|本公司|本集团).{0,120}(?:核级阀门|核电阀门|核级材料|核电仪控).{0,120}(?:生产|制造|销售|产品|主营)", r"(?:公司|本公司|本集团).{0,120}(?:生产|制造|销售|主营).{0,120}(?:核级阀门|核电阀门|核级材料|核电仪控)", ], } def is_direct_context(sentence: str, keyword: str, track_code: str, selection_bucket: str) -> bool: norm = normalize_text(sentence) if keyword not in norm or any(token in norm for token in NEGATIVE_TOKENS): return False if any(token in norm for token in ("参股", "联营企业", "投资标的")) and not any( token in norm for token in ("控股子公司", "全资子公司") ): return False if any(token in norm for token in ("权益法", "长期股权投资", "合资公司将", "涉诉项目", "解除双方签订", "解除合同")): return False if any(token in norm for token in ("需遵守", "披露要求", "任职经历", "历任", "个人简历")): return False if selection_bucket == "电芯制造" and any(token in norm for token in ("电解液", "隔膜", "锂离子电池材料", "正极材料", "负极材料")) and "电芯" not in norm: return False if selection_bucket == "电芯制造" and any(token in norm for token in ("钢结构", "厂房工程", "基地建设项目", "工程项目")): return False if selection_bucket == "整机" and any(token in norm for token in ("转化为电能", "生产运营模式", "风力发电收入")): return False if selection_bucket == "塔筒/海缆/工程配套" and any(token in norm for token in ("募集资金", "已结项", "2009年", "2011年")): return False if any(token in norm for token in CONTEXT_ONLY_TOKENS) and not any(token in norm for token in ("本公司", "公司主营", "主要从事")): return False pos = norm.find(keyword) window = norm[max(0, pos - 240):pos + len(keyword) + 240] if not any(pattern.search(window) for pattern in DIRECT_PATTERNS): return False role_patterns = BUCKET_ROLE_PATTERNS[(track_code, selection_bucket)] return any(re.search(pattern, window) for pattern in role_patterns) def build_attachment_items( potential_rows: list[dict[str, str]], metadata: dict[str, dict[str, str]], industry_root: Path ) -> list[dict[str, Any]]: grouped: dict[str, dict[str, Any]] = {} for row in potential_rows: ann_ids = split_values(row["announcement_ids"]) urls = split_values(row["annual_report_adjunct_urls"]) for index, ann_id in enumerate(ann_ids): meta = metadata.get(ann_id, {}) adjunct = urls[index] if index < len(urls) else meta.get("adjunct_path", "") if not adjunct: raise RuntimeError(f"missing adjunct path for {ann_id}/{row['qualification_row_id']}") item = grouped.setdefault(ann_id, { "attachment_id": ann_id, "adjunct_path": adjunct, "source_url": CNINFO_BASE + adjunct.lstrip("/"), "announcement_title": meta.get("announcement_title", ""), "security_codes_set": set(), "security_names_set": set(), "pair_ids_set": set(), "keywords_set": set(), "industry_root": industry_root, }) item["security_codes_set"].add(row["security_code"]) if meta.get("security_name"): item["security_names_set"].add(meta["security_name"]) item["pair_ids_set"].add(row["qualification_row_id"]) keyword = TRACK_KEYWORDS.get((row["track_code"], row["selection_bucket"])) if not keyword: raise RuntimeError(f"no frozen keyword for {row['track_code']}/{row['selection_bucket']}") item["keywords_set"].add(keyword) result = [] for item in grouped.values(): item["security_codes"] = ";".join(sorted(item.pop("security_codes_set"))) item["security_names"] = ";".join(sorted(item.pop("security_names_set"))) item["pair_ids"] = ";".join(sorted(item.pop("pair_ids_set"))) item["pair_count"] = str(len(split_values(item["pair_ids"]))) item["keywords"] = sorted(item.pop("keywords_set")) result.append(item) return sorted(result, key=lambda x: x["attachment_id"]) def run_acquisition( items: list[dict[str, Any]], raw_dir: Path, receipt_path: Path, project_root: Path, workers: int, force_http: bool = False, ) -> list[dict[str, Any]]: receipt_by_id: dict[str, dict[str, Any]] = {} if receipt_path.exists(): receipt_by_id = {row["attachment_id"]: row for row in read_csv(receipt_path)} pending = [] for item in items: item["force_http"] = force_http prior = receipt_by_id.get(item["attachment_id"]) if not force_http and prior and prior.get("pdf_readable") == "YES" and prior.get("raw_path"): path = project_root / prior["raw_path"] if path.exists() and sha256_file(path) == prior.get("raw_sha256"): continue pending.append(item) print(f"ACQUIRE total={len(items)} reusable_receipts={len(items)-len(pending)} pending={len(pending)}", flush=True) with ThreadPoolExecutor(max_workers=workers) as executor: futures = {executor.submit(download_one, item, raw_dir, project_root): item for item in pending} completed = 0 for future in as_completed(futures): row = future.result() receipt_by_id[row["attachment_id"]] = row completed += 1 if completed % 20 == 0 or completed == len(pending): ordered = [receipt_by_id[x["attachment_id"]] for x in items if x["attachment_id"] in receipt_by_id] write_csv(receipt_path, DOWNLOAD_HEADERS, ordered) good = sum(x.get("pdf_readable") == "YES" for x in ordered) print(f"ACQUIRE progress={completed}/{len(pending)} receipts={len(ordered)} readable={good}", flush=True) ordered = [receipt_by_id[x["attachment_id"]] for x in items] write_csv(receipt_path, DOWNLOAD_HEADERS, ordered) return ordered def write_attachment_search_extract( converted_dir: Path, item: dict[str, Any], receipt: dict[str, Any], result: dict[str, Any] ) -> None: path = converted_dir / f"{item['attachment_id']}__keyword_pages.txt" lines = [ f"attachment_id={item['attachment_id']}", f"source_url={item['source_url']}", f"raw_path={receipt.get('raw_path','')}", f"raw_sha256={receipt.get('raw_sha256','')}", f"page_count={result.get('page_count',0)}", f"searched_terms={';'.join(item['keywords'])}", f"tool_version={TOOL_VERSION}", "scope=keyword-hit pages only; no full-report conversion; public annual report", "", ] if result.get("error"): lines.append(f"SEARCH_ERROR={result['error']}") hit_total = 0 for keyword in item["keywords"]: hits = result.get("keyword_hits", {}).get(keyword, []) hit_total += len(hits) lines.append(f"## keyword={keyword}; hit_page_count={len(hits)}") for hit in hits: lines.append(f"[page={hit['page']}]") if hit["sentences"]: lines.extend(hit["sentences"]) else: lines.append("KEYWORD_PRESENT_BUT_SENTENCE_WINDOW_EMPTY") lines.append("") if hit_total == 0 and not result.get("error"): lines.append("NEGATIVE_RESULT=ALL_READABLE_PAGES_SEARCHED_NO_FROZEN_KEYWORD_HIT") path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8") def run_search( items: list[dict[str, Any]], download_rows: list[dict[str, Any]], converted_dir: Path, search_cache_path: Path, project_root: Path, workers: int, ) -> dict[str, dict[str, Any]]: cache: dict[str, dict[str, Any]] = {} if search_cache_path.exists(): cache = json.loads(search_cache_path.read_text(encoding="utf-8")) download_by_id = {row["attachment_id"]: row for row in download_rows} payloads = [] for item in items: row = download_by_id[item["attachment_id"]] if row.get("pdf_readable") != "YES" or not row.get("raw_path"): cache[item["attachment_id"]] = { "attachment_id": item["attachment_id"], "page_count": 0, "readable": "NO", "error": row.get("failure_detail") or row.get("acquisition_result"), "keyword_hits": {}, } continue cached = cache.get(item["attachment_id"]) if cached and cached.get("raw_sha256") == row.get("raw_sha256") and cached.get("tool_version") == TOOL_VERSION: continue payloads.append((item["attachment_id"], str(project_root / row["raw_path"]), item["keywords"])) print(f"SEARCH total={len(items)} cached={len(items)-len(payloads)} pending={len(payloads)} workers={workers}", flush=True) with ProcessPoolExecutor(max_workers=workers) as executor: futures = {executor.submit(search_pdf_worker, payload): payload[0] for payload in payloads} completed = 0 for future in as_completed(futures): result = future.result() ann_id = result["attachment_id"] result["raw_sha256"] = download_by_id[ann_id].get("raw_sha256", "") result["tool_version"] = TOOL_VERSION cache[ann_id] = result completed += 1 if completed % 20 == 0 or completed == len(payloads): search_cache_path.parent.mkdir(parents=True, exist_ok=True) search_cache_path.write_text(json.dumps(cache, ensure_ascii=False, sort_keys=True), encoding="utf-8") hit_docs = sum(any(v for v in x.get("keyword_hits", {}).values()) for x in cache.values()) print(f"SEARCH progress={completed}/{len(payloads)} cached={len(cache)} hit_documents={hit_docs}", flush=True) for item in items: write_attachment_search_extract(converted_dir, item, download_by_id[item["attachment_id"]], cache[item["attachment_id"]]) search_cache_path.parent.mkdir(parents=True, exist_ok=True) search_cache_path.write_text(json.dumps(cache, ensure_ascii=False, sort_keys=True), encoding="utf-8") return cache def build_pair_receipts( potential_rows: list[dict[str, str]], download_rows: list[dict[str, Any]], search_results: dict[str, dict[str, Any]], verified_at: str, ) -> list[dict[str, Any]]: downloads = {row["attachment_id"]: row for row in download_rows} rows: list[dict[str, Any]] = [] for source in potential_rows: ann_ids = split_values(source["announcement_ids"]) urls = split_values(source["annual_report_adjunct_urls"]) keyword = TRACK_KEYWORDS[(source["track_code"], source["selection_bucket"])] retrieval_success = 0 retrieval_failure = 0 searched_pdf_count = 0 searched_pages = 0 keyword_pages: list[str] = [] direct_pages: list[str] = [] exact_sentences: list[str] = [] insufficient_samples: list[str] = [] keyword_hit_count = 0 direct_hit_count = 0 hit_attachment_ids: set[str] = set() errors: list[str] = [] for ann_id in ann_ids: download = downloads[ann_id] result = search_results[ann_id] if download.get("pdf_readable") == "YES": retrieval_success += 1 searched_pdf_count += 1 searched_pages += int(result.get("page_count", 0)) else: retrieval_failure += 1 errors.append(f"{ann_id}:{download.get('acquisition_result')}:{download.get('failure_class')}") continue hits = result.get("keyword_hits", {}).get(keyword, []) if hits: hit_attachment_ids.add(ann_id) for hit in hits: keyword_hit_count += 1 keyword_pages.append(f"{ann_id}:p{hit['page']}") sentences = hit.get("sentences", []) page_direct = False for sentence in sentences: if is_direct_context(sentence, keyword, source["track_code"], source["selection_bucket"]): direct_hit_count += 1 page_direct = True exact_sentences.append(f"{ann_id}:p{hit['page']}:{sentence}") elif len(insufficient_samples) < 8: insufficient_samples.append(f"{ann_id}:p{hit['page']}:{sentence}") if page_direct: direct_pages.append(f"{ann_id}:p{hit['page']}") if direct_hit_count: search_result = "FROZEN_KEYWORD_FOUND_WITH_DIRECT_SELF_BUSINESS_PAGE_CONTEXT" qualification = "PAGE_LEVEL_DIRECT_BUSINESS_CONTEXT_CONFIRMED_PENDING_EVIDENCE_REGISTRATION" hold_reason = "" elif keyword_hit_count: search_result = "FROZEN_KEYWORD_FOUND_BUT_NO_DIRECT_SELF_BUSINESS_PAGE_CONTEXT" qualification = "HELD_AFTER_ACTUAL_PDF_PAGE_SEARCH_INSUFFICIENT_DIRECT_CONTEXT" hold_reason = "KEYWORD_HITS_ARE_CONTEXT_ONLY_OR_NOT_SELF_BUSINESS" elif retrieval_success and not retrieval_failure: search_result = "ALL_REFERENCED_READABLE_PDFS_SEARCHED_NO_FROZEN_KEYWORD_HIT" qualification = "HELD_AFTER_ACTUAL_PDF_PAGE_SEARCH_TRUE_NEGATIVE" hold_reason = "NO_FROZEN_KEYWORD_HIT_IN_ANY_READABLE_REFERENCED_PDF" else: search_result = "ONE_OR_MORE_REFERENCED_PDFS_NOT_ACQUIRED_OR_UNREADABLE" qualification = "HELD_BY_REVIEWABLE_RETRIEVAL_OR_PARSE_FAILURE" hold_reason = ";".join(errors) core = { "qualification_row_id": source["qualification_row_id"], "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "company_id": source["company_id"], "security_code": source["security_code"], "track_code": source["track_code"], "selection_bucket": source["selection_bucket"], "search_terms": keyword, "attachment_ids": ";".join(ann_ids), "attachment_urls": ";".join(urls), "attachment_count": str(len(ann_ids)), "retrieval_success_count": str(retrieval_success), "retrieval_failure_count": str(retrieval_failure), "searched_pdf_count": str(searched_pdf_count), "searched_page_count": str(searched_pages), "keyword_hit_attachment_count": str(len(hit_attachment_ids)), "keyword_hit_pages": ";".join(keyword_pages), "keyword_hit_count": str(keyword_hit_count), "direct_context_hit_pages": ";".join(direct_pages), "direct_context_hit_count": str(direct_hit_count), "exact_context_sentences": "\n---CONTEXT---\n".join(exact_sentences[:12]), "negative_or_insufficient_context_samples": "\n---CONTEXT---\n".join(insufficient_samples[:8]), "page_search_result": search_result, "qualification_result": qualification, "failure_or_hold_reason": hold_reason, "verified_at": verified_at, "tool_version": TOOL_VERSION, "review_status": "DRAFT_FOR_REVIEW", } receipt_payload = json.dumps(core, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") core["receipt_sha256"] = sha256_bytes(receipt_payload) rows.append(core) return rows def parse_direct_context(row: dict[str, str]) -> tuple[str, int, str]: first = row["exact_context_sentences"].split("\n---CONTEXT---\n", 1)[0] match = re.match(r"([^:]+):p(\d+):(.*)", first, flags=re.S) if not match: raise RuntimeError(f"unparseable direct context: {row['qualification_row_id']}") return match.group(1), int(match.group(2)), re.sub(r"\s+", " ", match.group(3)).strip() def date_from_adjunct(path: str) -> str: match = re.search(r"finalpage/(\d{4}-\d{2}-\d{2})/", path) return match.group(1) if match else "" def descending_date(value: str) -> int: digits = "".join(char for char in (value or "") if char.isdigit())[:8] return -int(digits or "0") def build_qualification_sources_and_conversions( industry_root: Path, download_rows: list[dict[str, Any]], potential_rows: list[dict[str, str]], project_root: Path, ) -> tuple[list[dict[str, str]], list[dict[str, str]]]: source_path = industry_root / "manifest" / "source_document.csv" conversion_path = industry_root / "manifest" / "conversion_status.csv" existing_sources = [row for row in read_csv(source_path) if not row["doc_id"].startswith("S-QUAL-AR-")] existing_conversions = [row for row in read_csv(conversion_path) if not row["source_doc_id"].startswith("S-QUAL-AR-")] source_headers = list(existing_sources[0]) conversion_headers = list(existing_conversions[0]) candidate_rows = read_csv(industry_root / "extracted" / "company_track_candidate_ledger.csv") candidate_by_code = {row["security_code"]: row for row in candidate_rows} pair_by_id = {row["qualification_row_id"]: row for row in potential_rows} new_sources: list[dict[str, str]] = [] new_conversions: list[dict[str, str]] = [] for receipt in download_rows: attachment_id = receipt["attachment_id"] doc_id = f"S-QUAL-AR-{attachment_id}" converted_path = industry_root / "converted" / "qualification_filings" / f"{attachment_id}__keyword_pages.txt" if not converted_path.exists(): raise FileNotFoundError(converted_path) pair_ids = split_values(receipt["pair_ids"]) tracks = sorted({pair_by_id[pair_id]["track_code"] for pair_id in pair_ids}) codes = split_values(receipt["security_codes"]) company_ids = sorted({candidate_by_code[code]["company_id"] for code in codes if code in candidate_by_code}) raw_path = project_root / receipt["raw_path"] source_row = { "doc_id": doc_id, "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "doc_type": "OFFICIAL_QUALIFICATION_ANNUAL_REPORT", "title": receipt["announcement_title"] or f"2025年年度报告资格核验附件 {attachment_id}", "source_org": "巨潮资讯网/上市公司法定披露", "author": receipt["security_names"], "publish_date": date_from_adjunct(receipt["adjunct_path"]), "collected_at": receipt["attempted_at"], "source_url": receipt["source_url"], "industry_case": "新能源案例", "industry_id": "IND-NEWENERGY", "subindustry_id": tracks[0] if len(tracks) == 1 else "CROSS_TRACK", "company_id": company_ids[0] if len(company_ids) == 1 else "", "raw_pool_path": "ana-data/cases/新能源案例/raw/qualification_filings/", "raw_file_path": receipt["raw_path"], "converted_text_path": converted_path.relative_to(project_root).as_posix(), "converted_markdown_path": "", "file_sha256": receipt["raw_sha256"], "file_name": raw_path.name, "file_size": receipt["raw_byte_count"], "detected_type": "PDF", "source_language": "zh-CN", "public_access_basis": "OFFICIAL_PUBLIC_DISCLOSURE_DIRECT_URL", "access_status": "PUBLIC_DIRECT", "source_level": "S", "sensitivity_screen": "LEGAL_PUBLIC_SCREENED_HIGH_LEVEL_ONLY", "legal_access_note": "巨潮资讯公开法定披露附件；REPAIR004仅执行冻结关键词页级检索；未绕过访问控制。", "doc_status": "QUALIFICATION_INPUT_REPAIR004", "processing_status": "PDF_ACQUIRED_READABLE_KEYWORD_PAGES_EXTRACTED_INDEXED_FOR_QUALIFICATION", } new_sources.append({key: source_row.get(key, "") for key in source_headers}) conversion_row = { "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": source_row["raw_pool_path"], "raw_file_path": receipt["raw_path"], "raw_file_sha256": receipt["raw_sha256"], "detected_type": "PDF", "conversion_method": "PDFIUM_FULL_PAGE_TEXT_SEARCH_AND_KEYWORD_PAGE_EXTRACT", "parameters_summary": "all readable pages; frozen bucket keyword; exact hit page and sentence windows; no OCR; nuclear high-level public boundary", "converted_text_path": source_row["converted_text_path"], "converted_markdown_path": "", "converted_path": source_row["converted_text_path"], "converted_sha256": sha256_file(converted_path), "page_or_duration_count": receipt["page_count"], "status": "TEXT_SEARCHED_INDEXED_QUALIFICATION_COMPLETE", "error_code": "", "error_summary": "", "created_at": receipt["attempted_at"], } new_conversions.append({key: conversion_row.get(key, "") for key in conversion_headers}) sources = existing_sources + new_sources conversions = existing_conversions + new_conversions if len({row["doc_id"] for row in sources}) != len(sources): raise RuntimeError("duplicate source doc id after REPAIR004 source merge") if len({row["source_doc_id"] for row in conversions}) != len(conversions): raise RuntimeError("conversion is not one-to-one after REPAIR004 source merge") write_csv(source_path, source_headers, sources) write_csv(conversion_path, conversion_headers, conversions) return sources, conversions def build_repair004_evidence_and_candidates( industry_root: Path, pair_rows: list[dict[str, Any]], source_rows: list[dict[str, str]], ) -> tuple[list[dict[str, str]], list[dict[str, str]], dict[tuple[str, str], str]]: evidence_path = industry_root / "evidence" / "evidence_fact_table.csv" evidence_rows = [row for row in read_csv(evidence_path) if not row["evidence_fact_id"].startswith("EVF-QUAL-")] evidence_headers = list(evidence_rows[0]) candidate_path = industry_root / "extracted" / "company_track_candidate_ledger.csv" candidates = read_csv(candidate_path) source_by_id = {row["doc_id"]: row for row in source_rows} candidate_by_pair = {(row["company_id"], row["track_code"]): row for row in candidates} new_evidence_id_by_pair: dict[tuple[str, str], str] = {} for receipt in pair_rows: if receipt["direct_context_hit_count"] == "0": continue pair = (receipt["company_id"], receipt["track_code"]) candidate = candidate_by_pair[pair] attachment_id, page, sentence = parse_direct_context(receipt) doc_id = f"S-QUAL-AR-{attachment_id}" source = source_by_id[doc_id] evidence_id = f"EVF-QUAL-{receipt['track_code']}-{receipt['security_code']}-R004" new_evidence_id_by_pair[pair] = evidence_id evidence_row = { "evidence_fact_id": evidence_id, "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "doc_id": doc_id, "industry_case": "新能源案例", "industry_id": "IND-NEWENERGY", "subindustry_id": receipt["track_code"], "company_id": receipt["company_id"], "track_code": receipt["track_code"], "chain_node_id": CHAIN_NODES[(receipt["track_code"], receipt["selection_bucket"])], "subject_type": "COMPANY", "subject_id": receipt["company_id"], "source_text_path": source["converted_text_path"], "raw_pool_path": source["raw_pool_path"], "raw_file_sha256": source["file_sha256"], "source_page": str(page), "source_table_id": "", "source_sentence_index": "", "locator_type": "PDF_PAGE_AND_EXACT_CONTEXT", "locator_value": f"attachment_id={attachment_id};page={page};receipt={receipt['qualification_row_id']}", "evidence_text": sentence, "evidence_type": "OFFICIAL_ANNUAL_REPORT", "statement_type": "FACT", "business_dimension": "COMPANY_EXPOSURE", "research_dimension": "COMPANY_TRACK_QUALIFICATION", "numeric_value_raw": "", "metric_candidate_name": "", "metric_candidate_unit": "", "metric_period": "2025", "metric_date": "2025-12-31", "geography": "CN", "original_qualifier": "REPAIR004实际取得CNINFO附件并逐页检索；仅支持冻结桶的直接业务角色，不作份额、质量或投资判断。", "related_company_id": "", "viewpoint_id": "", "darkline_signal_flag": "NO", "confidence_level": "HIGH", "conclusion_strength": "DIRECT_FACT", "sensitivity_screen": "LEGAL_PUBLIC_SCREENED_HIGH_LEVEL_ONLY", "contradicts_evidence_fact_id": "", "normalization_status": "NORMALIZED", "processing_status": "READY", "data_status": "VERIFIED_PUBLIC", "review_status": "DRAFT_FOR_REVIEW", } evidence_rows.append({key: evidence_row.get(key, "") for key in evidence_headers}) candidate.update({ "chain_nodes": CHAIN_NODES[(receipt["track_code"], receipt["selection_bucket"])], "direct_business_source_id": doc_id, "direct_business_locator": f"2025年报第{page}页;attachment_id={attachment_id}", "evidence_grade": "S", "exposure_specificity": "NAMED_PRODUCT_PROJECT_AND_OPERATING_FACT", "latest_disclosed_period": "2025-12-31", }) evidence_by_pair: dict[tuple[str, str], list[dict[str, str]]] = defaultdict(list) for evidence in evidence_rows: if evidence["subject_type"] == "COMPANY" and evidence["company_id"] and evidence["track_code"]: evidence_by_pair[(evidence["company_id"], evidence["track_code"])].append(evidence) eligible_groups: dict[tuple[str, str], list[dict[str, str]]] = defaultdict(list) for candidate in candidates: pair = (candidate["company_id"], candidate["track_code"]) matches = [e for e in evidence_by_pair.get(pair, []) if e["doc_id"] == candidate["direct_business_source_id"]] if matches and candidate["evidence_grade"] in {"S", "A"} and candidate["exposure_specificity"] in { "SEGMENT_REVENUE_OR_ASSET_DISCLOSED", "NAMED_PRODUCT_PROJECT_AND_OPERATING_FACT", "GENERAL_DIRECT_BUSINESS_DESCRIPTION", }: candidate["candidate_state"] = "ELIGIBLE" candidate["selection_rank"] = "" candidate["tier"] = "" candidate["tie_break_rule"] = "source_grade>exposure_specificity>latest_period>publish_date>exchange_code>security_code" candidate["include_or_exclude_reason"] = ( "REPAIR004实际取得年度报告PDF并逐页检索；直接业务角色、页码原句、S级主源及暴露具体性gate通过，进入同桶全部ELIGIBLE机械排序。" ) eligible_groups[(candidate["track_code"], candidate["selection_bucket"])].append(candidate) else: candidate["candidate_state"] = "HELD_BY_EVIDENCE_GAP" candidate["selection_rank"] = "" candidate["tier"] = "" grade_order = {"S": 0, "A": 1} specificity_order = { "SEGMENT_REVENUE_OR_ASSET_DISCLOSED": 0, "NAMED_PRODUCT_PROJECT_AND_OPERATING_FACT": 1, "GENERAL_DIRECT_BUSINESS_DESCRIPTION": 2, } for group_rows in eligible_groups.values(): group_rows.sort(key=lambda row: ( grade_order.get(row["evidence_grade"], 9), specificity_order.get(row["exposure_specificity"], 9), descending_date(row["latest_disclosed_period"]), descending_date(source_by_id[row["direct_business_source_id"]]["publish_date"]), row["exchange_code"], row["security_code"], )) for rank, candidate in enumerate(group_rows, 1): candidate["selection_rank"] = str(rank) if rank == 1: candidate["candidate_state"] = "INCLUDED_T1" candidate["tier"] = "T1_PRIMARY" elif rank == 2: candidate["candidate_state"] = "INCLUDED_T2" candidate["tier"] = "T2_CANDIDATE" else: candidate["candidate_state"] = "ELIGIBLE_NOT_SELECTED_BATCH001" candidate["tier"] = "" candidate["include_or_exclude_reason"] += " 同桶排名超过2，保留为ELIGIBLE_NOT_SELECTED_BATCH001。" track_order = {"BATTERY": 0, "SOLAR": 1, "WIND": 2, "NUCLEAR": 3} bucket_order = {key: index for index, key in enumerate(TRACK_KEYWORDS)} state_order = {"INCLUDED_T1": 0, "INCLUDED_T2": 1, "ELIGIBLE_NOT_SELECTED_BATCH001": 2, "HELD_BY_EVIDENCE_GAP": 3} candidates.sort(key=lambda row: ( track_order[row["track_code"]], bucket_order[(row["track_code"], row["selection_bucket"])], state_order.get(row["candidate_state"], 9), int(row["selection_rank"] or 999999), row["exchange_code"], row["security_code"], )) write_csv(evidence_path, evidence_headers, evidence_rows) write_csv(candidate_path, list(candidates[0]), candidates) return evidence_rows, candidates, new_evidence_id_by_pair def update_qualification_funnel( industry_root: Path, pair_rows: list[dict[str, Any]], candidates: list[dict[str, str]], evidence_rows: list[dict[str, str]], source_rows: list[dict[str, str]], ) -> list[dict[str, str]]: path = industry_root / "extracted" / "candidate_qualification_funnel.csv" funnel = read_csv(path) headers = list(funnel[0]) pair_receipt = {(row["company_id"], row["track_code"]): row for row in pair_rows} candidate_by_pair = {(row["company_id"], row["track_code"]): row for row in candidates} evidence_by_pair = defaultdict(list) for evidence in evidence_rows: if evidence["subject_type"] == "COMPANY": evidence_by_pair[(evidence["company_id"], evidence["track_code"])].append(evidence) source_by_id = {row["doc_id"]: row for row in source_rows} eligible_states = {"INCLUDED_T1", "INCLUDED_T2", "ELIGIBLE_NOT_SELECTED_BATCH001"} for row in funnel: pair = (row["company_id"], row["track_code"]) candidate = candidate_by_pair[pair] receipt = pair_receipt.get(pair) candidate_evidence = [e for e in evidence_by_pair.get(pair, []) if e["doc_id"] == candidate["direct_business_source_id"]] eligible = candidate["candidate_state"] in eligible_states if receipt: row["annual_report_retrieval_attempt"] = "CNINFO_ADJUNCT_PDF_GET_DEDUP_729_AND_FULL_PAGE_SEARCH" row["annual_report_retrieval_result"] = ( f"ACQUIRED={receipt['retrieval_success_count']};FAILED={receipt['retrieval_failure_count']};" f"SEARCHED_PDFS={receipt['searched_pdf_count']};SEARCHED_PAGES={receipt['searched_page_count']}" ) row["page_level_verification_attempt"] = "ACTUAL_PDF_OPEN_AND_FROZEN_KEYWORD_ALL_PAGE_SEARCH_WITH_BUCKET_ROLE_GATE" row["page_level_verification_result"] = receipt["qualification_result"] row["business_context_rule_result"] = ( "PAGE_VERIFIED_DIRECT_BUSINESS_ROLE_CONFIRMED" if eligible else "PAGE_VERIFIED_INSUFFICIENT_OR_ROLE_MISMATCH" ) row["replay_status"] = "ACTUAL_ATTACHMENT_RETRIEVAL_AND_PAGE_SEARCH_COMPLETED_FOR_PAIR" row["direct_business_source_id"] = candidate["direct_business_source_id"] if eligible else "" for gate in ["direct_source_gate", "locator_gate", "company_evidence_fact_gate", "source_grade_gate", "exposure_specificity_gate"]: row[gate] = "PASS" if eligible else "FAIL" row["evidence_fact_ids"] = ";".join(sorted(e["evidence_fact_id"] for e in candidate_evidence)) if eligible else "" row["failed_gates"] = "" if eligible else "DIRECT_BUSINESS_SOURCE;PAGE_OR_TEXT_LOCATOR;COMPANY_EVIDENCE_FACT;SOURCE_GRADE_S_OR_A;EXPOSURE_SPECIFICITY" row["eligibility_result"] = "ELIGIBLE" if eligible else "HELD_BY_EVIDENCE_GAP" row["eligible_rank_in_bucket"] = candidate["selection_rank"] if eligible else "" row["final_candidate_state"] = candidate["candidate_state"] source = source_by_id.get(candidate["direct_business_source_id"], {}) row["mechanical_sort_key"] = "|".join([ candidate["evidence_grade"], candidate["exposure_specificity"], candidate["latest_disclosed_period"], source.get("publish_date", ""), candidate["exchange_code"], candidate["security_code"], ]) row["funnel_rule_version"] = "REPAIR004_ACTUAL_ADJUNCT_PDF_PAGE_SEARCH_AND_BUCKET_ROLE_GATE_V1" row["review_status"] = "DRAFT_FOR_REVIEW" write_csv(path, headers, funnel) return funnel def update_classification_and_exposure( industry_root: Path, candidates: list[dict[str, str]], evidence_rows: list[dict[str, str]], source_rows: list[dict[str, str]], ) -> list[dict[str, str]]: candidate_by_pair = {(row["company_id"], row["track_code"]): row for row in candidates} evidence_by_pair = defaultdict(list) for evidence in evidence_rows: if evidence["subject_type"] == "COMPANY": evidence_by_pair[(evidence["company_id"], evidence["track_code"])].append(evidence) source_by_id = {row["doc_id"]: row for row in source_rows} classification_path = industry_root / "extracted" / "classification_summary.csv" classifications = read_csv(classification_path) for row in classifications: pair = (row["company_id"], row["track_code"]) candidate = candidate_by_pair[pair] matching = [e for e in evidence_by_pair.get(pair, []) if e["doc_id"] == candidate["direct_business_source_id"]] row["classification_reason"] = candidate["include_or_exclude_reason"] row["data_status"] = candidate["candidate_state"] row["review_status"] = "DRAFT_FOR_REVIEW" if matching: evidence = matching[0] source = source_by_id[evidence["doc_id"]] row["subject_type"] = "COMPANY_DIRECT_BUSINESS" row["source_doc_id"] = evidence["doc_id"] row["evidence_fact_id"] = evidence["evidence_fact_id"] row["chain_node_id"] = candidate["chain_nodes"] row["raw_pool_path"] = source["raw_pool_path"] row["raw_file_sha256"] = source["file_sha256"] write_csv(classification_path, list(classifications[0]), classifications) selected = [row for row in candidates if row["candidate_state"] in {"INCLUDED_T1", "INCLUDED_T2"}] matrix_path = industry_root / "extracted" / "newenergy_company_exposure_matrix.csv" matrix_headers = list(read_csv(matrix_path)[0]) matrix_rows = [] for candidate in selected: pair = (candidate["company_id"], candidate["track_code"]) evidence = next(e for e in evidence_by_pair[pair] if e["doc_id"] == candidate["direct_business_source_id"]) row = { "mapping_id": f"MAP-{candidate['track_code']}-{candidate['security_code']}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "company_id": candidate["company_id"], "security_code": candidate["security_code"], "security_name": candidate["security_name"], "legal_name": candidate["legal_name"], "exchange_code": candidate["exchange_code"], "track_code": candidate["track_code"], "chain_nodes": candidate["chain_nodes"], "selection_bucket": candidate["selection_bucket"], "tier": candidate["tier"], "candidate_state": candidate["candidate_state"], "direct_business_source_id": candidate["direct_business_source_id"], "direct_business_locator": candidate["direct_business_locator"], "evidence_fact_id": evidence["evidence_fact_id"], "evidence_grade": candidate["evidence_grade"], "exposure_specificity": candidate["exposure_specificity"], "latest_disclosed_period": candidate["latest_disclosed_period"], "primary_region": "MAINLAND_CHINA", "scope_status": "CORE_SCOPE_DIRECT_BUSINESS", "coverage_claim": "NONE_INITIAL_CANDIDATE_POOL_ONLY", "data_status": "VERIFIED_PUBLIC", "review_status": "DRAFT_FOR_REVIEW", } matrix_rows.append({key: row.get(key, "") for key in matrix_headers}) write_csv(matrix_path, matrix_headers, matrix_rows) return matrix_rows def write_company_outputs_and_map( industry_root: Path, candidates: list[dict[str, str]], evidence_rows: list[dict[str, str]], source_rows: list[dict[str, str]], ) -> list[dict[str, str]]: import newenergy_batch001_repair as base selected = [row for row in candidates if row["candidate_state"] in {"INCLUDED_T1", "INCLUDED_T2"}] evidence_by_pair = defaultdict(list) for evidence in evidence_rows: if evidence["subject_type"] == "COMPANY": evidence_by_pair[(evidence["company_id"], evidence["track_code"])].append(evidence) source_by_id = {row["doc_id"]: row for row in source_rows} track_meta = { "BATTERY": ("锂电", "01_锂电"), "SOLAR": ("光伏", "02_光伏"), "WIND": ("风电", "03_风电"), "NUCLEAR": ("核电", "04_核电"), } selected_by_track = defaultdict(list) facts: dict[tuple[str, str], tuple[dict[str, str], dict[str, str]]] = {} for candidate in selected: pair = (candidate["company_id"], candidate["track_code"]) evidence = next(e for e in evidence_by_pair[pair] if e["doc_id"] == candidate["direct_business_source_id"]) facts[pair] = (evidence, source_by_id[evidence["doc_id"]]) selected_by_track[candidate["track_code"]].append(candidate) common_meta = ( f"> 状态：`DRAFT_FOR_REVIEW` \n> `task_id={TASK_ID}` · `case_id={CASE_ID}` · `batch_id={BATCH_ID}` · `run_id={RUN_ID}` \n" "> `primary_region=MAINLAND_CHINA` · `global_comparator=SEPARATE_CONTEXT_ONLY` · `source_cutoff_at=2026-08-05T23:59:59+08:00` \n" "> `coverage_claim=NONE_INITIAL_CANDIDATE_POOL_ONLY` · `valuation_market_interface=NOT_APPLICABLE_BATCH001`\n" ) for track, rows in selected_by_track.items(): title, directory = track_meta[track] lines = [ f"# {title}相关企业", "", common_meta, "本页是完整候选发现池经实际年度报告附件逐页核验后的本批 A 股直接业务导航。每桶只保留机械排序前两名；T1/T2不表示质量或投资优先级。", "", "| 深度 | 桶 | 代码 | 公司 | 直接业务证据 | 主源 |", "|---|---|---:|---|---|---|", ] for candidate in rows: evidence, source = facts[(candidate["company_id"], track)] fact = evidence["evidence_text"].replace("|", "\\|") lines.append( f"| {candidate['tier']} | {candidate['selection_bucket']} | {candidate['security_code']} | {candidate['security_name']} | " f"{fact} | [{source['doc_id']}]({source['source_url']})，第{evidence['source_page']}页 |" ) lines.extend([ "", "## 解释限制", "", "- T1/T2 只代表本批机械排序后的研究深度，不代表公司质量、竞争排名、估值、交易或收益判断。", "- 同一公司同一赛道多个节点只计一次；多元化公司只标注主源直接支持的赛道暴露。", "- `coverage_claim=NONE_INITIAL_CANDIDATE_POOL_ONLY`。", "", "完整字段见 [候选台账](../../../../../extracted/company_track_candidate_ledger.csv)。", "", ]) path = base.CASE_OUTPUTS / "核心文档" / "子行业图谱" / directory / "相关企业.md" path.write_text("\n".join(lines), encoding="utf-8") company_lines = [ "# 新能源公司视图", "", common_meta, "32条映射来自法定年度报告直接业务证据；完整发现池实际取得附件并逐页检索后，每赛道四桶、每桶按冻结规则选择两家。T1/T2不表示公司优劣或投资优先级。", "", ] for track in ["BATTERY", "SOLAR", "WIND", "NUCLEAR"]: title, _ = track_meta[track] company_lines.extend([f"## {title}", ""]) for candidate in selected_by_track[track]: evidence, source = facts[(candidate["company_id"], track)] company_lines.extend([ f'', f"### {candidate['security_name']}（{candidate['security_code']}，{candidate['tier']}）", "", f"- 选择桶：{candidate['selection_bucket']}", f"- 直接节点：{candidate['chain_nodes']}", f"- 事实：{evidence['evidence_text']}", f"- 来源：[{source['doc_id']}]({source['source_url']})，第{evidence['source_page']}页", "- 限制：仅证明直接业务存在；不等同于赛道收入纯度、利润弹性、公司质量或投资结论。", "", ]) (base.CASE_OUTPUTS / "新能源公司视图.md").write_text("\n".join(company_lines), encoding="utf-8") map_path = base.CASE_EVIDENCE / "case_evidence_map.csv" current_map = read_csv(map_path) map_headers = list(current_map[0]) company_fact_ids = { evidence["evidence_fact_id"] for evidence in evidence_rows if evidence["subject_type"] == "COMPANY" } result = [ row for row in current_map if row["evidence_fact_id"] not in company_fact_ids and not row["evidence_fact_id"].startswith("EVF-QUAL-") ] seq = 1 def add(output_path: Path, evidence_id: str, conclusion: str, limit: str, strength: str = "DIRECT_FACT") -> None: nonlocal seq row = { "conclusion_evidence_map_id": f"CEM-R004-{seq:04d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "conclusion_id": f"CONC-R004-{seq:04d}", "output_path": output_path.relative_to(Path.cwd()).as_posix(), "section_anchor": "PENDING_MATERIALIZATION", "conclusion_text": conclusion, "conclusion_strength": strength, "evidence_fact_id": evidence_id, "support_type": "SUPPORT", "contradiction_or_limit": limit, "review_status": "DRAFT_FOR_REVIEW", } result.append({key: row.get(key, "") for key in map_headers}) seq += 1 first_by_track = {} for candidate in selected: evidence, _source = facts[(candidate["company_id"], candidate["track_code"])] _title, directory = track_meta[candidate["track_code"]] add(base.CASE_OUTPUTS / "新能源公司视图.md", evidence["evidence_fact_id"], evidence["evidence_text"], "仅证明直接业务暴露；T1/T2不是质量、估值或投资排序。") add(base.CASE_OUTPUTS / "核心文档" / "子行业图谱" / directory / "相关企业.md", evidence["evidence_fact_id"], evidence["evidence_text"], "仅证明直接业务暴露；T1/T2不是质量、估值或投资排序。") first_by_track.setdefault(candidate["track_code"], (candidate, evidence)) for track, (candidate, evidence) in first_by_track.items(): _title, directory = track_meta[track] add(base.CASE_OUTPUTS / "新能源行业视图.md", evidence["evidence_fact_id"], f"{candidate['security_name']}的官方年报直接支持其{track}业务映射。", "公司例证不构成行业或公司全集。") add(base.CASE_OUTPUTS / "核心文档" / "子行业图谱" / directory / "产业链与技术路线.md", evidence["evidence_fact_id"], f"{candidate['security_name']}的公开产品/业务事实作为产业链节点例证。", "只支持公开产品/业务节点，不据此推导技术优劣、份额或投资结论。", "MECHANISM_ONLY") result = base.materialize_evidence_locators(result) write_csv(map_path, map_headers, result) return result def update_package_manifests( industry_root: Path, source_rows: list[dict[str, str]], conversion_rows: list[dict[str, str]], evidence_rows: list[dict[str, str]], candidates: list[dict[str, str]], case_map: list[dict[str, str]], ) -> tuple[list[dict[str, str]], list[dict[str, str]]]: import newenergy_batch001_repair as base base.COLLECTED_AT = max((row["collected_at"] for row in source_rows if row.get("collected_at")), default="2026-08-06T00:00:00+08:00") base.build_input_manifests(source_rows) base.build_source_gap_audit(source_rows, evidence_rows) universe_rows = read_csv(industry_root / "extracted" / "a_share_universe.csv") base.rewrite_batch_summary(candidates, len(source_rows), len(conversion_rows), universe_rows) base.build_human_receipt(case_map, len(source_rows)) output_rows = base.rebuild_output_manifest() artifact_rows = base.build_artifact_manifest(source_rows, output_rows, Path(__file__)) return output_rows, artifact_rows def finalize_repair004( industry_root: Path, potential_rows: list[dict[str, str]], download_rows: list[dict[str, Any]], pair_rows: list[dict[str, Any]], project_root: Path, ) -> dict[str, Any]: tool_dir = str((project_root / "ana-data" / "tools").resolve()) if tool_dir not in sys.path: sys.path.insert(0, tool_dir) source_rows, conversion_rows = build_qualification_sources_and_conversions( industry_root, download_rows, potential_rows, project_root, ) evidence_rows, candidates, _new_evidence = build_repair004_evidence_and_candidates( industry_root, pair_rows, source_rows, ) funnel = update_qualification_funnel(industry_root, pair_rows, candidates, evidence_rows, source_rows) matrix_rows = update_classification_and_exposure(industry_root, candidates, evidence_rows, source_rows) case_map = write_company_outputs_and_map(industry_root, candidates, evidence_rows, source_rows) output_rows, artifact_rows = update_package_manifests( industry_root, source_rows, conversion_rows, evidence_rows, candidates, case_map, ) selected = [row for row in candidates if row["candidate_state"] in {"INCLUDED_T1", "INCLUDED_T2"}] state_counts = {state: sum(row["candidate_state"] == state for row in candidates) for state in sorted({row["candidate_state"] for row in candidates})} return { "status": "REPAIR004_FORMAL_PACKAGE_REBUILT_DRAFT_FOR_REVIEW", "sources": len(source_rows), "conversions": len(conversion_rows), "evidence_facts": len(evidence_rows), "candidate_pairs": len(candidates), "candidate_states": state_counts, "selected": len(selected), "exposure_matrix": len(matrix_rows), "funnel": len(funnel), "case_map": len(case_map), "outputs": len(output_rows), "artifacts_excluding_manifest_self": len(artifact_rows), "artifact_manifest_sha256": sha256_file(industry_root / "manifest" / "artifact_manifest.csv"), } def validate_scope( potential_rows: list[dict[str, str]], items: list[dict[str, Any]], download_rows: list[dict[str, Any]], pair_rows: list[dict[str, Any]], ) -> None: if len(potential_rows) != 858: raise RuntimeError(f"frozen potential-direct pair count changed: {len(potential_rows)} != 858") if len(items) != 729: raise RuntimeError(f"frozen unique attachment count changed: {len(items)} != 729") if len({row["qualification_row_id"] for row in potential_rows}) != 858: raise RuntimeError("potential-direct pair ids are not unique") if len(download_rows) != 729 or len({row["attachment_id"] for row in download_rows}) != 729: raise RuntimeError("download receipt does not exactly cover the 729 attachments") if len(pair_rows) != 858 or len({row["qualification_row_id"] for row in pair_rows}) != 858: raise RuntimeError("pair receipt does not exactly cover the 858 frozen pairs") for row in download_rows: if row["pdf_readable"] == "YES": path = Path.cwd() / row["raw_path"] if not path.exists() or sha256_file(path) != row["raw_sha256"]: raise RuntimeError(f"download hash mismatch: {row['attachment_id']}") def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--download-workers", type=int, default=12) parser.add_argument("--search-workers", type=int, default=max(2, min(8, os.cpu_count() or 4))) parser.add_argument("--acquire-only", action="store_true") parser.add_argument("--finalize", action="store_true") parser.add_argument("--force-http-receipt", action="store_true") args = parser.parse_args() project_root = Path.cwd().resolve() industry_root = find_industry_root(project_root) extracted_root = industry_root / "extracted" manifest_root = industry_root / "manifest" raw_dir = industry_root / "raw" / "qualification_filings" converted_dir = industry_root / "converted" / "qualification_filings" tmp_root = industry_root / CASE_ID / "tmp" / "repair004" raw_dir.mkdir(parents=True, exist_ok=True) converted_dir.mkdir(parents=True, exist_ok=True) tmp_root.mkdir(parents=True, exist_ok=True) funnel = read_csv(extracted_root / "candidate_qualification_funnel.csv") prior_pair_receipt = extracted_root / "candidate_page_qualification_receipt.csv" if prior_pair_receipt.exists(): frozen_ids = {row["qualification_row_id"] for row in read_csv(prior_pair_receipt)} if len(frozen_ids) == 858: potential_rows = [row for row in funnel if row["qualification_row_id"] in frozen_ids] else: potential_rows = [row for row in funnel if row["business_context_rule_result"].startswith("PAGE_VERIFIED_")] else: potential_rows = [row for row in funnel if row["business_context_rule_result"] == POTENTIAL] metadata = load_announcement_metadata(industry_root) items = build_attachment_items(potential_rows, metadata, industry_root) download_receipt_path = manifest_root / "candidate_attachment_download_receipt.csv" download_rows = run_acquisition( items, raw_dir, download_receipt_path, project_root, args.download_workers, args.force_http_receipt, ) if args.acquire_only: print(json.dumps({ "status": "ACQUISITION_COMPLETE", "pairs": len(potential_rows), "attachments": len(items), "readable": sum(row["pdf_readable"] == "YES" for row in download_rows), "failed": sum(row["pdf_readable"] != "YES" for row in download_rows), "receipt_sha256": sha256_file(download_receipt_path), }, ensure_ascii=False), flush=True) return search_cache_path = tmp_root / "attachment_page_search_cache.json" search_results = run_search( items, download_rows, converted_dir, search_cache_path, project_root, args.search_workers, ) verified_at = max((row["attempted_at"] for row in download_rows if row.get("attempted_at")), default=now_iso()) pair_rows = build_pair_receipts(potential_rows, download_rows, search_results, verified_at) pair_receipt_path = extracted_root / "candidate_page_qualification_receipt.csv" write_csv(pair_receipt_path, PAIR_HEADERS, pair_rows) validate_scope(potential_rows, items, download_rows, pair_rows) if args.finalize: result = finalize_repair004(industry_root, potential_rows, download_rows, pair_rows, project_root) print(json.dumps(result, ensure_ascii=False, indent=2), flush=True) return print(json.dumps({ "status": "REPAIR004_ATTACHMENT_AND_PAGE_SEARCH_COMPLETE", "pairs": len(potential_rows), "attachments": len(items), "readable_attachments": sum(row["pdf_readable"] == "YES" for row in download_rows), "failed_attachments": sum(row["pdf_readable"] != "YES" for row in download_rows), "direct_context_pairs": sum(row["direct_context_hit_count"] != "0" for row in pair_rows), "keyword_context_only_pairs": sum(row["page_search_result"] == "FROZEN_KEYWORD_FOUND_BUT_NO_DIRECT_SELF_BUSINESS_PAGE_CONTEXT" for row in pair_rows), "true_negative_pairs": sum(row["page_search_result"] == "ALL_REFERENCED_READABLE_PDFS_SEARCHED_NO_FROZEN_KEYWORD_HIT" for row in pair_rows), "retrieval_or_parse_failure_pairs": sum(row["page_search_result"] == "ONE_OR_MORE_REFERENCED_PDFS_NOT_ACQUIRED_OR_UNREADABLE" for row in pair_rows), "download_receipt_sha256": sha256_file(download_receipt_path), "pair_receipt_sha256": sha256_file(pair_receipt_path), }, ensure_ascii=False), flush=True) if __name__ == "__main__": main()