156 lines
4.4 KiB
Python
156 lines
4.4 KiB
Python
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(
|
|
"<h1>Wiki Jokes</h1><p>No jokes yet — the daily generator is "
|
|
"working on it. Check back soon!</p>",
|
|
status_code=503,
|
|
)
|
|
items = "\n".join(
|
|
f' <li class="joke">{html.escape(j)}</li>' for j in batch["jokes"]
|
|
)
|
|
stale_note = (
|
|
'<p class="stale">⚠️ Today\'s jokes are still being prepared — '
|
|
"showing the latest available batch.</p>"
|
|
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),
|
|
}
|
|
)
|