tegwick
29ceb77938
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a09c77-b1a9-7eb3-b501-6cb24b458c97
74 lines
3.5 KiB
Python
74 lines
3.5 KiB
Python
"""Idempotent daily/weekly source capture and digest jobs, with durable run health."""
|
|
from datetime import datetime, timedelta
|
|
import fcntl
|
|
import json
|
|
from pathlib import Path
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from .ingest import ingest
|
|
from .store import atomic, put
|
|
|
|
|
|
def run_due(root, config, now=None, runner=ingest):
|
|
from .__main__ import draft
|
|
root = Path(root)
|
|
zone = ZoneInfo(config['timezone'])
|
|
now = now or datetime.now(zone)
|
|
if now.tzinfo is None:
|
|
raise ValueError('scheduler time requires a timezone')
|
|
local = now.astimezone(zone)
|
|
state_dir = root / '.runtime'
|
|
state_dir.mkdir(parents=True, exist_ok=True)
|
|
with (state_dir / 'scheduler.lock').open('a') as lock:
|
|
try:
|
|
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except BlockingIOError:
|
|
return {'ok': True, 'status': 'already_running', 'jobs': []}
|
|
state_path = state_dir / 'scheduler.json'
|
|
state = json.loads(state_path.read_text()) if state_path.exists() else {}
|
|
results = []
|
|
for job in config['jobs']:
|
|
hour, minute = (int(x) for x in job['at'].split(':'))
|
|
if not (0 <= hour < 24 and 0 <= minute < 60):
|
|
raise ValueError('invalid schedule time')
|
|
due = local.replace(hour=hour, minute=minute, second=0, microsecond=0)
|
|
if job['cadence'] == 'daily':
|
|
if due > local:
|
|
due -= timedelta(days=1)
|
|
period = due.date().isoformat()
|
|
elif job['cadence'] == 'weekly':
|
|
due -= timedelta(days=(due.weekday()-job['weekday']) % 7)
|
|
if due > local:
|
|
due -= timedelta(days=7)
|
|
year, week, _ = due.isocalendar()
|
|
period = f'{year}-W{week:02}'
|
|
else:
|
|
raise ValueError('unsupported cadence')
|
|
key = job['id'] + ':' + period
|
|
if state.get(key, {}).get('ok'):
|
|
continue
|
|
try:
|
|
registry_path = (root / job['sources']).resolve()
|
|
if not registry_path.is_relative_to(root.resolve()):
|
|
raise ValueError('source registry escapes repository')
|
|
registry = json.loads(registry_path.read_text())
|
|
result = runner(root, registry)
|
|
if not result['ok']:
|
|
raise ValueError('one or more sources failed; see ingestion-runs')
|
|
# Capture-only changes remain unreviewed. Do not fabricate factual events.
|
|
try:
|
|
output = draft(root, period)
|
|
digest_status = 'created'
|
|
except FileExistsError:
|
|
output = root / 'digests' / job['cadence'] / f'{period}.md'
|
|
digest_status = 'existing_preserved'
|
|
entry = {'ok': True, 'finished_at': local.isoformat(), 'period': period,
|
|
'digest': str(output.relative_to(root)), 'digest_status': digest_status}
|
|
except (ValueError, OSError, KeyError) as exc:
|
|
entry = {'ok': False, 'finished_at': local.isoformat(), 'period': period,
|
|
'error': str(exc)}
|
|
state[key] = entry
|
|
put(root, 'scheduler-runs', {'job_id': job['id'], **entry})
|
|
atomic(state_path, json.dumps(state, indent=2).encode() + b'\n')
|
|
results.append({'job_id': job['id'], **entry})
|
|
return {'ok': all(r['ok'] for r in results), 'status': 'checked', 'jobs': results}
|