| """Corpus record format shared by every source module. |
| |
| A source module defines NAME, GROUP and LICENSE constants plus: |
| fetch(raw_dir: Path) -> None downloads into raw_dir (idempotent) |
| iter_records(raw_dir: Path) -> Iterator[Record] parses offline |
| """ |
|
|
| from __future__ import annotations |
|
|
| import json |
| from dataclasses import asdict, dataclass |
| from typing import Literal |
|
|
| Group = Literal["astro", "arxiv", "textbook", "general"] |
|
|
|
|
| @dataclass(frozen=True) |
| class Record: |
| source: str |
| id: str |
| license: str |
| group: Group |
| text: str |
|
|
| def to_json(self) -> str: |
| return json.dumps(asdict(self), ensure_ascii=False) |
|
|
| @classmethod |
| def from_json(cls, line: str) -> Record: |
| return cls(**json.loads(line)) |
|
|
|
|
| def download(url: str, dest, timeout: float = 120, retries: int = 3) -> None: |
| """Download `url` to `dest` unless it already exists (written via a .part file). |
| |
| A body shorter than the server's Content-Length is treated as a failed attempt and retried. |
| """ |
| import http.client |
| import shutil |
| import time |
| import urllib.request |
| from pathlib import Path |
|
|
| dest = Path(dest) |
| if dest.exists(): |
| return |
| dest.parent.mkdir(parents=True, exist_ok=True) |
| part = dest.with_name(dest.name + ".part") |
| req = urllib.request.Request(url, headers={"User-Agent": "zenith-corpus-builder/0.1 (research)"}) |
| for attempt in range(1, retries + 1): |
| try: |
| with urllib.request.urlopen(req, timeout=timeout) as resp, open(part, "wb") as f: |
| expected = resp.headers.get("Content-Length") |
| shutil.copyfileobj(resp, f, length=1 << 20) |
| got = part.stat().st_size |
| if expected is not None and got != int(expected): |
| raise IOError(f"truncated download of {url}: {got} of {expected} bytes") |
| part.replace(dest) |
| return |
| except (OSError, http.client.HTTPException) as e: |
| part.unlink(missing_ok=True) |
| if attempt == retries: |
| raise IOError(f"truncated or failed download of {url} after {retries} attempts: {e}") from e |
| time.sleep(2 * attempt) |
|
|
|
|
| def local_name(tag: str) -> str: |
| """XML tag without its namespace.""" |
| return tag.rsplit("}", 1)[-1] |
|
|
