import html import logging from contextlib import asynccontextmanager from datetime import date from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.cron import CronTrigger from apscheduler.triggers.interval import IntervalTrigger from fastapi import FastAPI, Request from fastapi.responses import HTMLResponse, JSONResponse, Response from zoneinfo import ZoneInfo from . import db from .config import settings from .generator import generation_exhausted, run_generation, today_local from .templates import CSS, PAGE_TEMPLATE from .trmnl import trmnl_endpoint logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s", ) log = logging.getLogger("jokes.app") scheduler = AsyncIOScheduler(timezone=ZoneInfo(settings.timezone)) async def tick() -> None: """Runs every RETRY_MINUTES: generate today's jokes if missing, unless we've exhausted retries for today.""" day = today_local() if db.get_day(day) is not None: return if generation_exhausted(day): return run_generation() @asynccontextmanager async def lifespan(app: FastAPI): db.init_db() # Cold start: generate immediately if today's jokes are missing. if db.get_day(today_local()) is None: log.info("Cold start: generating today's jokes now") run_generation() # Daily schedule. scheduler.add_job( tick, CronTrigger( hour=settings.generate_hour, minute=settings.generate_minute, timezone=ZoneInfo(settings.timezone), ), id="daily-generation", ) # Retry loop (no-ops when today's jokes exist or retries are exhausted). scheduler.add_job( tick, IntervalTrigger(minutes=settings.retry_minutes), id="retry-generation", ) scheduler.start() yield scheduler.shutdown(wait=False) app = FastAPI(title="Wiki Jokes", lifespan=lifespan) SECURITY_HEADERS = { "Content-Security-Policy": ( "default-src 'none'; style-src 'self'; img-src 'self' data:; " "base-uri 'none'; form-action 'none'; frame-ancestors 'none'" ), "X-Content-Type-Options": "nosniff", "X-Frame-Options": "DENY", "Referrer-Policy": "no-referrer", "Strict-Transport-Security": "max-age=31536000; includeSubDomains", "Permissions-Policy": "camera=(), microphone=(), geolocation=()", } @app.middleware("http") async def security_headers(request: Request, call_next): response = await call_next(request) for key, value in SECURITY_HEADERS.items(): response.headers[key] = value return response @app.get("/styles.css") def styles(): return Response(content=CSS, media_type="text/css") @app.get("/favicon.ico") def favicon(): # Tiny 1x1 transparent GIF; stops 404 noise and scanner probing. import base64 return Response( content=base64.b64decode("R0lGODlhAQABAIAAAAAAAP///yH5BAEAAAAALAAAAAABAAEAAAIBRAA7"), media_type="image/gif", ) @app.get("/", response_class=HTMLResponse) def index(): today = today_local() batch = db.get_day(today) stale = False if batch is None: batch = db.get_latest() stale = True if batch is None: return HTMLResponse( "

Wiki Jokes

No jokes yet — the daily generator is " "working on it. Check back soon!

", status_code=503, ) items = "\n".join( f'
  • {html.escape(j)}
  • ' for j in batch["jokes"] ) stale_note = ( '

    ⚠️ Today\'s jokes are still being prepared — ' "showing the latest available batch.

    " if stale else "" ) return PAGE_TEMPLATE.format( day=batch["day"], stale_note=stale_note, jokes=items, article_title=html.escape(batch["article_title"]), article_url=html.escape(batch["article_url"], quote=True), ) @app.get("/trmnl.json") def trmnl(request: Request): """JSON feed for TRMNL e-ink devices (polling strategy).""" return trmnl_endpoint(request) @app.get("/health") def health(): today = today_local() has_today = db.get_day(today) is not None return JSONResponse( { "status": "ok", "today": today.isoformat(), "has_todays_jokes": has_today, "retries_exhausted": generation_exhausted(today), } )