"""Archive external web sources into the ana evidence layout.
|
|
This script is intentionally small and dependency-free. It fetches explicit
|
URLs, stores snapshots under an industry case directory, and writes a manifest.
|
|
It does not bypass paywalls, log in, execute JavaScript, or store credentials.
|
Use it only after the case design says external collection is needed.
|
"""
|
|
from __future__ import annotations
|
|
import argparse
|
import csv
|
import hashlib
|
import html
|
import mimetypes
|
import re
|
import sys
|
from datetime import datetime, timezone
|
from pathlib import Path
|
from urllib.error import HTTPError, URLError
|
from urllib.parse import urlparse
|
from urllib.request import Request, urlopen
|
|
|
SCRIPT_PATH = Path(__file__).resolve()
|
PROJECT_ROOT = SCRIPT_PATH.parents[2]
|
DEFAULT_USER_AGENT = "project-info-ana-source-collector/1.0"
|
|
|
def now_iso() -> str:
|
return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")
|
|
|
def safe_name(value: str, fallback: str = "source", max_len: int = 80) -> str:
|
value = re.sub(r"[^A-Za-z0-9._-]+", "_", value.strip())
|
value = value.strip("._-")
|
if not value:
|
value = fallback
|
return value[:max_len]
|
|
|
def sha256_bytes(data: bytes) -> str:
|
return hashlib.sha256(data).hexdigest()
|
|
|
def read_url_file(path: Path) -> list[str]:
|
urls: list[str] = []
|
for line in path.read_text(encoding="utf-8").splitlines():
|
line = line.strip()
|
if not line or line.startswith("#"):
|
continue
|
urls.append(line)
|
return urls
|
|
|
def infer_extension(url: str, content_type: str) -> str:
|
path_ext = Path(urlparse(url).path).suffix
|
if path_ext and len(path_ext) <= 10:
|
return path_ext.lower()
|
media_type = content_type.split(";", 1)[0].strip().lower()
|
if media_type == "text/html":
|
return ".html"
|
if media_type == "text/plain":
|
return ".txt"
|
if media_type == "application/json":
|
return ".json"
|
if media_type in ("text/csv", "application/csv"):
|
return ".csv"
|
if media_type == "application/pdf":
|
return ".pdf"
|
guessed = mimetypes.guess_extension(media_type)
|
return guessed or ".bin"
|
|
|
def html_to_text(data: bytes, content_type: str) -> str:
|
encoding = "utf-8"
|
match = re.search(r"charset=([A-Za-z0-9._-]+)", content_type, re.I)
|
if match:
|
encoding = match.group(1)
|
text = data.decode(encoding, errors="replace")
|
text = re.sub(r"(?is)<script.*?</script>", " ", text)
|
text = re.sub(r"(?is)<style.*?</style>", " ", text)
|
text = re.sub(r"(?s)<[^>]+>", " ", text)
|
text = html.unescape(text)
|
text = re.sub(r"[ \t\r\f\v]+", " ", text)
|
text = re.sub(r"\n\s*\n+", "\n\n", text)
|
return text.strip() + "\n"
|
|
|
def fetch(url: str, timeout: int, user_agent: str) -> tuple[int, str, bytes]:
|
req = Request(url, headers={"User-Agent": user_agent})
|
with urlopen(req, timeout=timeout) as response:
|
status = int(getattr(response, "status", 200))
|
content_type = response.headers.get("Content-Type", "")
|
data = response.read()
|
return status, content_type, data
|
|
|
def append_manifest(path: Path, rows: list[dict[str, str]]) -> None:
|
if not rows:
|
return
|
path.parent.mkdir(parents=True, exist_ok=True)
|
fieldnames = list(rows[0].keys())
|
exists = path.exists()
|
with path.open("a", encoding="utf-8-sig", newline="") as fh:
|
writer = csv.DictWriter(fh, fieldnames=fieldnames)
|
if not exists:
|
writer.writeheader()
|
writer.writerows(rows)
|
|
|
def project_relative(path: Path) -> str:
|
try:
|
return path.resolve().relative_to(PROJECT_ROOT.resolve()).as_posix()
|
except ValueError:
|
return path.resolve().as_posix()
|
|
|
def collect(args: argparse.Namespace) -> int:
|
urls = list(args.url or [])
|
if args.url_file:
|
urls.extend(read_url_file(Path(args.url_file)))
|
urls = [url.strip() for url in urls if url.strip()]
|
if not urls:
|
raise SystemExit("no URLs provided")
|
|
industry_root = PROJECT_ROOT / "ana-data" / "cases" / args.industry_case
|
target_dir = industry_root / args.target
|
manifest_dir = industry_root / "manifest"
|
target_dir.mkdir(parents=True, exist_ok=True)
|
manifest_dir.mkdir(parents=True, exist_ok=True)
|
|
run_safe = safe_name(args.run_id, "run")
|
manifest_path = manifest_dir / f"web_source_manifest_{run_safe}.csv"
|
rows: list[dict[str, str]] = []
|
|
for index, url in enumerate(urls, start=1):
|
fetched_at = now_iso()
|
source_id = f"{run_safe}_WEB_{index:04d}"
|
parsed = urlparse(url)
|
base = safe_name(f"{parsed.netloc}_{Path(parsed.path).stem}", f"web_{index:04d}")
|
status = ""
|
content_type = ""
|
file_path = ""
|
text_path = ""
|
body_sha = ""
|
body_size = "0"
|
fetch_status = "ERROR"
|
error = ""
|
|
try:
|
http_status, content_type, data = fetch(url, args.timeout, args.user_agent)
|
status = str(http_status)
|
body_sha = sha256_bytes(data)
|
body_size = str(len(data))
|
ext = infer_extension(url, content_type)
|
out_path = target_dir / f"{source_id}_{base}_{body_sha[:12]}{ext}"
|
out_path.write_bytes(data)
|
file_path = project_relative(out_path)
|
fetch_status = "FETCHED"
|
|
if args.extract_text and (
|
content_type.lower().startswith("text/html")
|
or content_type.lower().startswith("text/plain")
|
or ext in (".html", ".htm", ".txt")
|
):
|
text = html_to_text(data, content_type)
|
text_out = target_dir / f"{source_id}_{base}_{body_sha[:12]}.txt"
|
text_out.write_text(text, encoding="utf-8")
|
text_path = project_relative(text_out)
|
except HTTPError as exc:
|
status = str(exc.code)
|
error = str(exc)
|
except URLError as exc:
|
error = str(exc.reason)
|
except Exception as exc: # noqa: BLE001 - manifest should retain the failure.
|
error = str(exc)
|
|
rows.append(
|
{
|
"source_id": source_id,
|
"case_id": args.case_id,
|
"batch_id": args.batch_id,
|
"run_id": args.run_id,
|
"target": args.target,
|
"source_url": url,
|
"fetched_at": fetched_at,
|
"http_status": status,
|
"content_type": content_type,
|
"bytes": body_size,
|
"sha256": body_sha,
|
"relative_path": file_path,
|
"text_relative_path": text_path,
|
"fetch_status": fetch_status,
|
"error": error,
|
"user_agent": args.user_agent,
|
}
|
)
|
|
append_manifest(manifest_path, rows)
|
print(f"manifest={project_relative(manifest_path)} rows={len(rows)}")
|
return 0
|
|
|
def parse_args(argv: list[str]) -> argparse.Namespace:
|
parser = argparse.ArgumentParser(description=__doc__)
|
parser.add_argument("--industry-case", required=True, help="industry case folder name")
|
parser.add_argument("--case-id", required=True)
|
parser.add_argument("--batch-id", required=True)
|
parser.add_argument("--run-id", required=True)
|
parser.add_argument("--url", action="append", help="URL to fetch; can repeat")
|
parser.add_argument("--url-file", help="UTF-8 text file, one URL per line")
|
parser.add_argument("--target", choices=["raw", "supplement"], default="supplement")
|
parser.add_argument("--timeout", type=int, default=30)
|
parser.add_argument("--user-agent", default=DEFAULT_USER_AGENT)
|
parser.add_argument("--extract-text", action=argparse.BooleanOptionalAction, default=True)
|
return parser.parse_args(argv)
|
|
|
def main(argv: list[str] | None = None) -> int:
|
args = parse_args(argv or sys.argv[1:])
|
return collect(args)
|
|
|
if __name__ == "__main__":
|
raise SystemExit(main())
|
