hermes
075b72ec6c
- FastAPI + APScheduler + SQLite, single container - Daily generation at 06:00 Europe/Copenhagen + cold-start generation - Retry with backoff (3 attempts/15 min), stale fallback with banner - OpenAI-compatible endpoint via env (OPENAI_BASE_URL/MODEL/API_KEY) - docker-compose with named volume for joke persistence
61 lines
2.0 KiB
Python
61 lines
2.0 KiB
Python
"""Daily joke generation job with retry/backoff and stale-fallback semantics."""
|
|
import logging
|
|
from datetime import date, datetime, timedelta, timezone
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from . import db
|
|
from .config import settings
|
|
from .llm_client import generate_jokes
|
|
from .wiki_client import get_featured_article
|
|
|
|
log = logging.getLogger("jokes.generator")
|
|
|
|
# In-process retry bookkeeping (resets on container restart; cold-start
|
|
# generation covers the restart case anyway).
|
|
_attempts_today: dict[str, int] = {}
|
|
|
|
|
|
def today_local() -> date:
|
|
return datetime.now(ZoneInfo(settings.timezone)).date()
|
|
|
|
|
|
def _attempt_key(day: date) -> str:
|
|
return day.isoformat()
|
|
|
|
|
|
def generation_exhausted(day: date) -> bool:
|
|
return _attempts_today.get(_attempt_key(day), 0) >= settings.max_attempts
|
|
|
|
|
|
def run_generation() -> bool:
|
|
"""Try to generate today's jokes. Returns True on success."""
|
|
day = today_local()
|
|
if db.get_day(day) is not None:
|
|
log.info("Jokes for %s already exist, skipping", day)
|
|
return True
|
|
|
|
attempts = _attempts_today.get(_attempt_key(day), 0)
|
|
if attempts >= settings.max_attempts:
|
|
log.warning("Generation for %s already exhausted (%d attempts)", day, attempts)
|
|
return False
|
|
|
|
_attempts_today[_attempt_key(day)] = attempts + 1
|
|
try:
|
|
article = get_featured_article(day)
|
|
log.info("Fetched featured article: %s", article.title)
|
|
jokes = generate_jokes(article.title, article.extract)
|
|
db.save_batch(day, article.title, article.url, jokes)
|
|
log.info("Saved %d jokes for %s", len(jokes), day)
|
|
return True
|
|
except Exception as exc: # noqa: BLE001 — job must never crash the scheduler
|
|
used = _attempts_today[_attempt_key(day)]
|
|
log.error(
|
|
"Generation attempt %d/%d for %s failed: %s",
|
|
used, settings.max_attempts, day, exc,
|
|
)
|
|
return False
|
|
|
|
|
|
def next_retry_delay() -> timedelta:
|
|
return timedelta(minutes=settings.retry_minutes)
|