File size: 2,301 Bytes
0cf8154 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 | """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]
|