#!/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'<a id="company-{candidate["security_code"]}"></a>',
|
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()
|
