196 lines
8.2 KiB
Python
196 lines
8.2 KiB
Python
"""Run the fetchers for `bingham fetch` and write the results to the store.
|
|
|
|
Daily mode (no --since): arXiv RSS per category, Crossref for J. Rheol. since
|
|
the last successful Crossref fetch (less an overlap), then the enabled journal
|
|
RSS feeds. Items found this way have origin 'daily'.
|
|
|
|
arXiv RSS holds only the latest announcement, so a skipped weekday would be
|
|
lost. Each daily fetch records the announcement date it fully covered
|
|
(fetches source 'arxiv:covered'); when weekdays are missing since then, they
|
|
are filled from OAI-PMH ('arxiv:catchup'). A gap longer than
|
|
arxiv.max_catchup_days is still filled, but as origin 'backfill', so it does
|
|
not flood the next digest.
|
|
|
|
Backfill mode (--since DATE): arXiv OAI-PMH and Crossref from DATE. Items have
|
|
origin 'backfill' and stay out of the default digest. The first J. Rheol. fetch
|
|
is a backfill from jor.backfill_from.
|
|
"""
|
|
|
|
from datetime import date, timedelta
|
|
|
|
import yaml
|
|
|
|
from art import env, store
|
|
from art.feeds import cache, fetch
|
|
|
|
|
|
def _oai_set(category: str) -> str:
|
|
"""arXiv category -> OAI-PMH set: cs.CL -> cs, cond-mat.soft -> physics:cond-mat."""
|
|
archive = category.split(".")[0]
|
|
if archive in ("cs", "math", "stat", "q-bio", "q-fin", "eess", "econ"):
|
|
return archive
|
|
return f"physics:{archive}"
|
|
|
|
|
|
def _cached_fetch(name: str, fn, ttl_hours: float) -> tuple[list[dict], str]:
|
|
"""Run fn() unless the raw cache for `name` is fresh; fall back to stale cache on failure."""
|
|
items, fetched_at = cache.load_raw(name)
|
|
if cache.is_fresh(fetched_at, ttl_hours):
|
|
return items, "cached"
|
|
try:
|
|
items = fn()
|
|
except Exception as exc:
|
|
if items:
|
|
return items, f"FAILED ({exc}); using stale cache"
|
|
raise
|
|
cache.save_raw(name, items)
|
|
return items, "fetched"
|
|
|
|
|
|
def _record(conn, name: str, items: list[dict], origin: str, how: str, stats: dict) -> None:
|
|
n_new = store.upsert_items(conn, items, origin)
|
|
store.log_fetch(conn, name, len(items), n_new, True, how)
|
|
stats["fetched"] += len(items)
|
|
stats["new"] += n_new
|
|
print(f" {name}: {len(items)} items, {n_new} new ({how})")
|
|
|
|
|
|
def _oai_items(oai_set: str, since: str, cfg: dict) -> list[dict]:
|
|
"""OAI-PMH papers for the configured categories, with cross-lists marked as RSS would."""
|
|
cats = cfg["categories"]
|
|
items = fetch.fetch_arxiv_oai(oai_set, since, categories=cats, delay_s=cfg["request_delay_s"])
|
|
for it in items:
|
|
if it["categories"] and it["categories"][0] not in cats:
|
|
it["announce_type"] = "cross"
|
|
if not cfg["include_cross_lists"]:
|
|
items = [it for it in items if it["announce_type"] != "cross"]
|
|
return items
|
|
|
|
|
|
def _weekdays_between(a: date, b: date) -> list[date]:
|
|
"""Weekdays strictly between a and b (arXiv announces Monday to Friday)."""
|
|
days = (a + timedelta(days=i) for i in range(1, (b - a).days))
|
|
return [d for d in days if d.weekday() < 5]
|
|
|
|
|
|
def _catch_up(conn, rss_day: date | None, rss_ok: bool, cfg: dict, stats: dict) -> None:
|
|
"""Fill announcement days missed since the last covered one from OAI-PMH.
|
|
|
|
rss_day is the announcement date of today's RSS items (None when the feeds
|
|
were empty, as on weekends). The covered date advances only when every RSS
|
|
feed and the catch-up succeeded, so a failure is retried on the next run.
|
|
"""
|
|
last = conn.execute("SELECT MAX(detail) FROM fetches WHERE source = 'arxiv:covered' "
|
|
"AND ok = 1").fetchone()[0]
|
|
if last is None: # first daily fetch: nothing to catch up from
|
|
if rss_day and rss_ok:
|
|
store.log_fetch(conn, "arxiv:covered", 0, 0, True, rss_day.isoformat())
|
|
return
|
|
last = date.fromisoformat(last)
|
|
end = rss_day or date.today() + timedelta(days=1)
|
|
missing = _weekdays_between(last, end)
|
|
covered = rss_day or (missing[-1] if missing else None)
|
|
if missing:
|
|
since = (last + timedelta(days=1)).isoformat()
|
|
span = f"{missing[0]} to {missing[-1]}"
|
|
origin = "daily"
|
|
if (missing[-1] - last).days > cfg["max_catchup_days"]:
|
|
origin = "backfill"
|
|
print(f" arXiv catch-up: {span} exceeds {cfg['max_catchup_days']} days; "
|
|
"stored as backfill, not in the default digest")
|
|
n_items = n_new = 0
|
|
try:
|
|
for oai_set in sorted({_oai_set(c) for c in cfg["categories"]}):
|
|
items = _oai_items(oai_set, since, cfg)
|
|
before = stats["new"]
|
|
_record(conn, f"oai-catchup:{oai_set}", items, origin, f"missed {span}", stats)
|
|
n_items, n_new = n_items + len(items), n_new + stats["new"] - before
|
|
except Exception as exc:
|
|
store.log_fetch(conn, "arxiv:catchup", n_items, n_new, False, f"{span}: {exc}")
|
|
print(f" arXiv catch-up FAILED ({exc}); will retry on the next run")
|
|
return
|
|
store.log_fetch(conn, "arxiv:catchup", n_items, n_new, True, span)
|
|
if covered and rss_ok:
|
|
store.log_fetch(conn, "arxiv:covered", 0, 0, True, covered.isoformat())
|
|
|
|
|
|
def _fetch_arxiv(conn, since: str | None, ttl: float, stats: dict) -> None:
|
|
cfg = env.config()["arxiv"]
|
|
cats = cfg["categories"]
|
|
if since:
|
|
for oai_set in sorted({_oai_set(c) for c in cats}):
|
|
_record(conn, f"oai:{oai_set}", _oai_items(oai_set, since, cfg), "backfill",
|
|
f"since {since}", stats)
|
|
return
|
|
rss_days, rss_ok = set(), True
|
|
for cat in cats:
|
|
name = f"arxiv-rss:{cat}"
|
|
try:
|
|
items, how = _cached_fetch(name, lambda: fetch.fetch_arxiv_rss(cat), ttl)
|
|
except Exception as exc:
|
|
store.log_fetch(conn, name, 0, 0, False, str(exc))
|
|
print(f" {name}: FAILED ({exc})")
|
|
rss_ok = False
|
|
continue
|
|
if how.startswith("FAILED"):
|
|
rss_ok = False
|
|
rss_days |= {it["date"] for it in items}
|
|
if not cfg["include_cross_lists"]:
|
|
items = [it for it in items if it["announce_type"] != "cross"]
|
|
if not items:
|
|
how += "; empty (arXiv does not announce on weekends)" if date.today().weekday() >= 5 else "; empty"
|
|
_record(conn, name, items, "daily", how, stats)
|
|
rss_day = date.fromisoformat(max(rss_days)) if rss_days else None
|
|
_catch_up(conn, rss_day, rss_ok, cfg, stats)
|
|
|
|
|
|
def _fetch_jor(conn, since: str | None, ttl: float, stats: dict) -> None:
|
|
cfg = env.config()["jor"]
|
|
name = "crossref:jor"
|
|
last = store.last_fetch(conn, name)
|
|
if since:
|
|
start, origin = since, "backfill"
|
|
elif last is None:
|
|
start, origin = str(cfg["backfill_from"]), "backfill"
|
|
else:
|
|
start = (date.fromisoformat(last[:10]) - timedelta(days=cfg["crossref_overlap_days"])).isoformat()
|
|
origin = "daily"
|
|
try:
|
|
items = fetch.fetch_crossref(cfg["issn"][0], start, source="J. Rheol.")
|
|
_record(conn, name, items, origin, f"created since {start}", stats)
|
|
except Exception as exc:
|
|
store.log_fetch(conn, name, 0, 0, False, str(exc))
|
|
print(f" {name}: FAILED ({exc})")
|
|
for feed in _feeds(lambda f: f.get("source") == "J. Rheol."):
|
|
_fetch_feed(conn, feed, ttl, stats)
|
|
|
|
|
|
def _feeds(pred) -> list[dict]:
|
|
with open(env.data_file("feeds.yaml")) as f:
|
|
feeds = yaml.safe_load(f)["feeds"]
|
|
return [f for f in feeds if f.get("enabled") and pred(f)]
|
|
|
|
|
|
def _fetch_feed(conn, feed: dict, ttl: float, stats: dict) -> None:
|
|
name = f"rss:{feed['name']}"
|
|
try:
|
|
items, how = _cached_fetch(
|
|
name, lambda: fetch.fetch_rss(feed["name"], feed["url"], feed.get("source")), ttl)
|
|
except Exception as exc:
|
|
store.log_fetch(conn, name, 0, 0, False, str(exc))
|
|
print(f" {name}: FAILED ({exc})")
|
|
return
|
|
_record(conn, name, items, "daily", how, stats)
|
|
|
|
|
|
def run(conn, source: str = "all", since: str | None = None, refresh: bool = False) -> dict:
|
|
ttl = 0.0 if refresh else env.config().get("feed_cache_ttl_hours", 6)
|
|
stats = {"fetched": 0, "new": 0}
|
|
if source in ("arxiv", "all"):
|
|
_fetch_arxiv(conn, since, ttl, stats)
|
|
if source in ("jor", "all"):
|
|
_fetch_jor(conn, since, ttl, stats)
|
|
if source == "all" and not since:
|
|
for feed in _feeds(lambda f: f.get("source") != "J. Rheol."):
|
|
_fetch_feed(conn, feed, ttl, stats)
|
|
return stats
|