tegwick
96fe5b022b
- --replay DIR / replay_dir: offline replay from the cache, hash-verified - ok values without source become error; Last-Modified recorded as published - latest period by parsed order (web_fetcher/periods.py); mixed formats refused - non-numeric values not_found unless value_type: string - retry only connection errors, timeouts, 429/5xx; honour Retry-After - opt-in live smoke test; 18 offline tests - CHANGELOG, README install docs, SCOPE update, capability D2/A3/C2/R2 - WEBFETCH-WP-0003 (repo hygiene) added Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 153021@bnt-lap001 Assistant-Session: 46917be4-6400-4106-b8b7-61953a0caf24
62 lines
2.5 KiB
Python
62 lines
2.5 KiB
Python
"""Period ordering for "latest value" selection.
|
|
|
|
A source's period labels are only compared when they share one recognised
|
|
format; mixed or unrecognised formats are an error, never a silent string
|
|
comparison. Spec field `period_format` pins the format ("auto" by default);
|
|
"string" opts in to plain lexicographic order.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Callable
|
|
|
|
_FORMATS: dict[str, tuple[re.Pattern[str], Callable[[re.Match[str]], tuple[int, ...]]]] = {
|
|
"year": (re.compile(r"^(\d{4})$"), lambda m: (int(m[1]),)),
|
|
"month": (re.compile(r"^(\d{4})-(\d{2})$"), lambda m: (int(m[1]), int(m[2]))),
|
|
"date": (re.compile(r"^(\d{4})-(\d{2})-(\d{2})$"), lambda m: (int(m[1]), int(m[2]), int(m[3]))),
|
|
"quarter": (re.compile(r"^(\d{4})-?Q([1-4])$"), lambda m: (int(m[1]), int(m[2]))),
|
|
"semester": (re.compile(r"^(\d{4})-?S([12])$"), lambda m: (int(m[1]), int(m[2]))),
|
|
"week": (re.compile(r"^(\d{4})-?W(\d{2})$"), lambda m: (int(m[1]), int(m[2]))),
|
|
# Fixed-width digit labels such as YYYYMM or Census BTOS YYYYPP.
|
|
"digits": (re.compile(r"^(\d+)$"), lambda m: (len(m[1]), int(m[1]))),
|
|
}
|
|
|
|
|
|
class PeriodError(ValueError):
|
|
pass
|
|
|
|
|
|
def detect(label: str) -> str | None:
|
|
label = label.strip()
|
|
for name, (pat, _) in _FORMATS.items():
|
|
if pat.match(label):
|
|
return name
|
|
return None
|
|
|
|
|
|
def sort_key(label: str, fmt: str) -> tuple[int, ...] | str:
|
|
if fmt == "string":
|
|
return label
|
|
pat, key = _FORMATS[fmt]
|
|
m = pat.match(label.strip())
|
|
if m is None:
|
|
raise PeriodError(f"period {label!r} does not match format {fmt!r}")
|
|
return key(m)
|
|
|
|
|
|
def order(labels: list[str], fmt: str = "auto") -> tuple[str, list[str]]:
|
|
"""Return (format, labels newest first). Raises PeriodError for mixed formats."""
|
|
if fmt == "auto":
|
|
found = {detect(lbl) for lbl in labels}
|
|
if None in found:
|
|
bad = [lbl for lbl in labels if detect(lbl) is None][:3]
|
|
raise PeriodError(f"unrecognised period labels {bad}; set period_format")
|
|
if len(found) > 1:
|
|
raise PeriodError(f"mixed period formats {sorted(found)}; set period_format")
|
|
fmt = found.pop() if found else "string"
|
|
if fmt == "digits" and len({len(lbl.strip()) for lbl in labels}) > 1:
|
|
raise PeriodError("digit period labels of different widths; set period_format")
|
|
elif fmt != "string" and fmt not in _FORMATS:
|
|
raise PeriodError(f"unknown period_format {fmt!r}")
|
|
return fmt, sorted(labels, key=lambda lbl: sort_key(lbl, fmt), reverse=True)
|