from __future__ import annotations import csv import json import ssl import urllib.error import urllib.request from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timedelta, timezone from pathlib import Path ROOT = Path(__file__).resolve().parents[2] INDUSTRY = ROOT / "ana-data" / "cases" / "新能源案例" QUEUE = INDUSTRY / "extracted" / "batch002_evidence_completion_queue.csv" SOURCE_B1 = INDUSTRY / "manifest" / "source_document.csv" SOURCE_B2 = INDUSTRY / "manifest" / "source_document_BATCH002.csv" OUTPUT = INDUSTRY / "supplement" / "NEB2_external_public_query_probe_REPAIR001.json" def read_csv(path: Path) -> list[dict[str, str]]: with path.open("r", encoding="utf-8-sig", newline="") as fh: return list(csv.DictReader(fh)) def probe(url: str) -> dict[str, str | int]: headers = { "User-Agent": "Mozilla/5.0 (compatible; MBX-NewEnergy-Audit/1.0; public-source-verification)", "Accept": "application/pdf,text/html;q=0.9,*/*;q=0.5", } context = ssl.create_default_context() attempts: list[dict[str, str | int]] = [] for method in ("HEAD", "GET_RANGE"): request_headers = dict(headers) actual_method = "HEAD" if method == "GET_RANGE": request_headers["Range"] = "bytes=0-0" actual_method = "GET" req = urllib.request.Request(url, headers=request_headers, method=actual_method) try: with urllib.request.urlopen(req, timeout=25, context=context) as response: status = int(getattr(response, "status", response.getcode())) attempts.append({"method": method, "status": status, "result": "RESPONSE"}) return { "request_method": method, "response_status": status, "response_result": "HTTP_RESPONSE_RECEIVED", "final_url": response.geturl(), "content_type": response.headers.get("Content-Type", ""), "content_length": response.headers.get("Content-Length", ""), "error_type": "", "error_detail": "", "attempts": attempts, } except urllib.error.HTTPError as exc: attempts.append({"method": method, "status": int(exc.code), "result": "HTTP_ERROR"}) if method == "HEAD" and exc.code in {403, 405, 501}: continue return { "request_method": method, "response_status": int(exc.code), "response_result": "HTTP_ERROR_RECORDED", "final_url": exc.geturl() or url, "content_type": exc.headers.get("Content-Type", "") if exc.headers else "", "content_length": exc.headers.get("Content-Length", "") if exc.headers else "", "error_type": type(exc).__name__, "error_detail": str(exc)[:500], "attempts": attempts, } except Exception as exc: # noqa: BLE001 - receipt must preserve real failure state attempts.append({"method": method, "status": "", "result": "REQUEST_FAILURE"}) if method == "HEAD": continue return { "request_method": method, "response_status": "", "response_result": "REQUEST_FAILURE_RECORDED", "final_url": url, "content_type": "", "content_length": "", "error_type": type(exc).__name__, "error_detail": str(exc)[:500], "attempts": attempts, } return { "request_method": "HEAD+GET_RANGE", "response_status": "", "response_result": "REQUEST_FAILURE_RECORDED", "final_url": url, "content_type": "", "content_length": "", "error_type": "UNKNOWN", "error_detail": "all attempts exhausted", "attempts": attempts, } if OUTPUT.exists(): existing = json.loads(OUTPUT.read_text(encoding="utf-8")) if existing.get("status") == "COMPLETED_WITH_ALL_RESPONSE_OR_FAILURE_STATES_RECORDED": print(json.dumps({"reused": True, "path": str(OUTPUT), "url_count": len(existing["url_results"])}, ensure_ascii=False)) raise SystemExit(0) source_rows = read_csv(SOURCE_B1) + read_csv(SOURCE_B2) source_by_id = {row["doc_id"]: row for row in source_rows} queue_rows = read_csv(QUEUE) url_to_sources: dict[str, set[str]] = {} for row in queue_rows: source_ids = [row["prior_source_doc_id"]] if row["new_source_doc_id"]: source_ids.append(row["new_source_doc_id"]) for source_id in source_ids: if source_id not in source_by_id: raise RuntimeError(f"unresolved source id: {source_id}") url = source_by_id[source_id]["source_url"] if not url: raise RuntimeError(f"source URL missing: {source_id}") url_to_sources.setdefault(url, set()).add(source_id) queried_at = datetime.now(timezone(timedelta(hours=8))).replace(microsecond=0).isoformat() url_results: list[dict[str, object]] = [] with ThreadPoolExecutor(max_workers=8) as pool: future_to_url = {pool.submit(probe, url): url for url in sorted(url_to_sources)} for future in as_completed(future_to_url): url = future_to_url[future] result = future.result() url_results.append( { "url": url, "source_doc_ids": sorted(url_to_sources[url]), "queried_at": queried_at, **result, } ) url_results.sort(key=lambda row: str(row["url"])) payload = { "task_id": "TASK-NEWENERGY-FOUR-TRACK-ATLAS-BATCH002-20260806-001", "case_id": "ANA-NEWENERGY-FOUR-TRACK-ATLAS-20260806-002", "batch_id": "BATCH-002", "run_id": "RUN-ANA-NEWENERGY-FOUR-TRACK-ATLAS-20260806-002-BATCH-002-001", "repair_id": "EXECUTION_OUTPUT_REPAIR001", "purpose": "Directly probe the already frozen official primary-source URLs used by the 50 bounded queue pairs; do not discover or add candidates.", "query_scope": "50 frozen queue pairs; prior official annual report URL for each pair; Yuhong official report additionally for the one adjacent EPC check", "queried_at": queried_at, "url_count": len(url_results), "response_distribution": {}, "url_results": url_results, "status": "COMPLETED_WITH_ALL_RESPONSE_OR_FAILURE_STATES_RECORDED", "review_status": "DRAFT_FOR_REVIEW", } distribution: dict[str, int] = {} for row in url_results: key = f"{row['response_result']}:{row['response_status']}" distribution[key] = distribution.get(key, 0) + 1 payload["response_distribution"] = distribution OUTPUT.parent.mkdir(parents=True, exist_ok=True) OUTPUT.write_text(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8", newline="\n") print(json.dumps({"reused": False, "path": str(OUTPUT), "url_count": len(url_results), "response_distribution": distribution}, ensure_ascii=False))