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))
|
