ad/screenshots.py

# retoor <retoor@molodetz.nl>
import html
import os
import shutil
import subprocess
import sys
import time
import tomllib
from datetime import datetime, timezone
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent
SHOTS_DATA_DIR = REPO_ROOT / "data" / f".tmp-shots-{os.getpid()}"
OUTPUT_DIR = Path(os.environ.get("SCREENSHOTS_DIR", str(REPO_ROOT.parent / "static" / "pizdetz")))
ENV = os.environ.get("SCREENSHOTS_ENV", "test")
BASE_URL = os.environ.get("SCREENSHOTS_BASE_URL", "").rstrip("/")
PORT = os.environ.get("SCREENSHOTS_PORT", "8094")
LOCAL_BASE = f"http://127.0.0.1:{PORT}"
ADMIN_USERNAME = os.environ.get("SCREENSHOTS_ADMIN_USERNAME", "retoor")
ADMIN_PASSWORD = os.environ.get("SCREENSHOTS_ADMIN_PASSWORD", "")
BOOT_ADMIN_PASSWORD = "shots-admin-pass-9d3c"
FULL_DESKTOP = {"width": 1440, "height": 900}
KNOWN_ENVS = ["test", "staging", "production"]

PUBLIC_SHOTS = [
    ("home", "/", "Home"),
    ("roll", "/roll", "Roll"),
    ("standard", "/standard", "Standard"),
    ("flyers", "/flyers", "Flyers"),
    ("memes", "/memes", "Memes"),
    ("people", "/people", "People"),
    ("person", "/people/retoor", "Person"),
    ("join", "/join", "Join"),
    ("terms", "/terms", "Terms"),
    ("privacy", "/privacy", "Privacy"),
    ("login", "/auth/login", "Log in"),
]

DOCS_SHOTS = [
    ("docs", "/docs", "Docs"),
    ("docs-welcome", "/docs/welcome", "Docs welcome"),
    ("docs-api", "/docs/api", "Docs API index"),
    ("docs-api-content", "/docs/api/content", "Docs API content"),
    ("docs-component-code", "/docs/component-code", "Docs component code"),
]

ADMIN_SHOTS = [
    ("admin", "/admin", "Admin"),
    ("admin-posts", "/admin/posts", "Admin posts"),
    ("admin-post-new", "/admin/posts/new", "Admin editor"),
    ("admin-joins", "/admin/joins", "Admin joins"),
    ("admin-users", "/admin/users", "Admin users"),
    ("admin-settings", "/admin/settings", "Admin settings"),
    ("admin-services", "/admin/services", "Admin services"),
    ("admin-services-backup", "/admin/services/backup", "Admin service backup"),
    ("admin-backups", "/admin/backups", "Admin backups"),
    ("admin-audit", "/admin/audit", "Admin audit"),
    ("admin-stats", "/admin/stats", "Admin stats"),
    ("admin-trash", "/admin/trash", "Admin trash"),
    ("notifications", "/notifications", "Notifications"),
    ("profile-api-key", "/profile/api-key", "Profile API key"),
    ("docs-api-admin", "/docs/api/admin", "Docs API admin"),
]

RESPONSIVE_VIEWPORTS = [
    ("1440x900", {"width": 1440, "height": 900}),
    ("768x1024", {"width": 768, "height": 1024}),
    ("390x844", {"width": 390, "height": 844}),
    ("360x740", {"width": 360, "height": 740}),
    ("320x568", {"width": 320, "height": 568}),
]

RESPONSIVE_PAGES = [
    ("home", "/", "Home"),
    ("roll", "/roll", "Roll"),
    ("flyers", "/flyers", "Flyers"),
    ("join", "/join", "Join"),
    ("docs", "/docs", "Docs"),
]


def server_env(data_dir):
    return {
        "MOLODETZ_DATA_DIR": str(data_dir),
        "MOLODETZ_DATABASE_URL": f"sqlite:///{data_dir / 'molodetz.db'}",
        "MOLODETZ_ADMIN_USERNAME": ADMIN_USERNAME,
        "MOLODETZ_ADMIN_EMAIL": "retoor@molodetz.nl",
        "MOLODETZ_ADMIN_PASSWORD": BOOT_ADMIN_PASSWORD,
        "MOLODETZ_DISABLE_SERVICES": "1",
        "MOLODETZ_DISABLE_RATE_LIMIT": "1",
        "MOLODETZ_PORT": PORT,
        "MOLODETZ_INTERNAL_BASE_URL": LOCAL_BASE,
        "MOLODETZ_SITEMAP_TTL": "0",
        "MOLODETZ_LANDING_TTL": "0",
        "MOLODETZ_UNREAD_TTL": "0",
        "MOLODETZ_STATIC_VERSION": "shots",
        "SECRET_KEY": "shots-secret-key-not-for-production",
    }


def wait_ready(process, base):
    import httpx

    deadline = time.monotonic() + 60.0
    while time.monotonic() < deadline:
        if process.poll() is not None:
            raise RuntimeError("shots server exited during startup")
        try:
            if httpx.get(f"{base}/health", timeout=1.0).status_code == 200:
                return
        except httpx.HTTPError:
            pass
        time.sleep(0.2)
    raise RuntimeError("shots server did not become ready")


def start_server(data_dir):
    log = open(data_dir / "server.log", "w")
    process = subprocess.Popen(
        [sys.executable, "-m", "uvicorn", "molodetz.main:app", "--host", "127.0.0.1", "--port", PORT],
        cwd=REPO_ROOT,
        env={**os.environ, **server_env(data_dir)},
        stdout=log,
        stderr=subprocess.STDOUT,
    )
    try:
        wait_ready(process, LOCAL_BASE)
    except Exception:
        process.terminate()
        log.close()
        raise
    return process, log


def stop_server(process, log):
    process.terminate()
    try:
        process.wait(timeout=10)
    except subprocess.TimeoutExpired:
        process.kill()
    log.close()


def goto_ready(page, url):
    from playwright.sync_api import TimeoutError as PlaywrightTimeoutError

    page.goto(url, wait_until="domcontentloaded")
    try:
        page.wait_for_selector("html[data-app=ready]")
    except PlaywrightTimeoutError:
        page.reload(wait_until="domcontentloaded")
        page.wait_for_selector("html[data-app=ready]")


def login_admin(page, base, password):
    page.goto(base + "/auth/login", wait_until="domcontentloaded")
    page.fill("input[name=username]", ADMIN_USERNAME)
    page.fill("input[name=password]", password)
    page.click("button[type=submit]")
    page.wait_for_url("**/admin", wait_until="domcontentloaded")


def seed_join(base):
    import httpx

    response = httpx.post(
        base + "/join",
        data={
            "name": "Gallery Writer",
            "contact": "gallery@example.invalid",
            "repo_url": "https://example.com/gallery",
            "message": "I want to write along.",
            "website": "",
        },
        timeout=15.0,
    )
    if response.status_code not in (302, 303):
        raise RuntimeError(f"seed join failed: {response.status_code}")


def first_post_slug(base):
    import httpx

    response = httpx.get(base + "/roll", headers={"Accept": "application/json"}, timeout=15.0)
    posts = response.json().get("posts", [])
    if not posts:
        raise RuntimeError("roll has no posts")
    return posts[0]["slug"]


def capture(page, out_dir, base, counter, name, path, title, section, viewport, full_page):
    goto_ready(page, base + path)
    filename = f"{counter:02d}-{name}-{viewport}.png"
    page.screenshot(path=str(out_dir / filename), full_page=full_page)
    return {"section": section, "title": title, "path": path, "viewport": viewport, "file": filename}


def check_overflow(page, path, viewport):
    width, inner = page.evaluate("[document.documentElement.scrollWidth, window.innerWidth]")
    if width > inner:
        return f"{path} at {viewport}: scroll {width} wider than viewport {inner}"
    return ""


def capture_full(browser, out_dir, base, password, records, counter):
    anon = browser.new_context(viewport=FULL_DESKTOP)
    page = anon.new_page()
    page.set_default_timeout(60000)
    for name, path, title in PUBLIC_SHOTS:
        records.append(capture(page, out_dir, base, counter, name, path, title, "Public", "full", True))
        counter += 1
    slug = first_post_slug(base)
    records.append(capture(page, out_dir, base, counter, "post", f"/posts/{slug}", "Post", "Public", "full", True))
    counter += 1
    for name, path, title in DOCS_SHOTS:
        records.append(capture(page, out_dir, base, counter, name, path, title, "Docs", "full", True))
        counter += 1
    anon.close()
    admin = browser.new_context(viewport=FULL_DESKTOP)
    admin_page = admin.new_page()
    admin_page.set_default_timeout(60000)
    login_admin(admin_page, base, password)
    for name, path, title in ADMIN_SHOTS:
        records.append(capture(admin_page, out_dir, base, counter, name, path, title, "Administration", "full", True))
        counter += 1
    admin.close()
    return counter


def capture_responsive(browser, out_dir, base, records, failures, counter):
    for viewport, size in RESPONSIVE_VIEWPORTS:
        context = browser.new_context(viewport=size)
        page = context.new_page()
        page.set_default_timeout(60000)
        for name, path, title in RESPONSIVE_PAGES:
            goto_ready(page, base + path)
            failure = check_overflow(page, path, viewport)
            if failure:
                failures.append(failure)
            filename = f"{counter:02d}-{name}-{viewport}.png"
            page.screenshot(path=str(out_dir / filename), full_page=False)
            records.append({"section": "Responsive", "title": title, "path": path, "viewport": viewport, "file": filename})
            counter += 1
        context.close()
    context = browser.new_context(viewport={"width": 390, "height": 844})
    page = context.new_page()
    page.set_default_timeout(60000)
    goto_ready(page, base + "/")
    page.click("[data-nav-toggle]")
    filename = f"{counter:02d}-home-nav-open-390x844.png"
    page.screenshot(path=str(out_dir / filename), full_page=False)
    records.append({"section": "Responsive", "title": "Home navigation open", "path": "/", "viewport": "390x844", "file": filename})
    context.close()
    return counter + 1


def app_version():
    with (REPO_ROOT / "pyproject.toml").open("rb") as handle:
        return tomllib.load(handle)["project"]["version"]


def render_index(records, failures, checked, env, version, stamped):
    sections = []
    for section in ["Public", "Docs", "Administration", "Responsive"]:
        figures = []
        for record in records:
            if record["section"] != section:
                continue
            figures.append(
                '<figure><a href="{f}"><img src="{f}" alt="{t}" loading="lazy"></a>'
                "<figcaption>{t} <code>{p}</code> ({v})</figcaption></figure>".format(
                    f=html.escape(record["file"]),
                    t=html.escape(record["title"]),
                    p=html.escape(record["path"]),
                    v=html.escape(record["viewport"]),
                )
            )
        note = ""
        if section == "Responsive":
            passed = checked - len(failures)
            items = "".join(f"<li>{html.escape(failure)}</li>" for failure in failures)
            failures_html = f"<ul>{items}</ul>" if items else ""
            note = f"<p class=\"checks\">{passed} of {checked} responsive checks passed, no horizontal overflow.</p>{failures_html}"
        sections.append("<h2>{s}</h2>{n}<div class=\"shots\">{f}</div>".format(s=section, n=note, f="".join(figures)))
    return """<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Molodetz in pictures ({e})</title>
<meta name="description" content="Screenshot gallery of the Molodetz site.">
<style>
body{{margin:0 auto;max-width:1120px;padding:48px 24px 96px;font-family:Georgia,'Times New Roman',serif;color:#1a1815;background:#fff}}
header{{border-bottom:1px solid #e2ddd4;padding-bottom:24px}}
header nav{{font-size:14px;margin-bottom:16px}}
header nav a{{color:#555}}
h1{{font-size:34px;margin:0 0 8px;font-weight:600}}
header p{{margin:4px 0;font-size:15px;color:#555}}
h2{{font-size:22px;margin:44px 0 4px}}
.checks{{font-size:14px;color:#555;margin:8px 0 0}}
ul{{font-size:14px;color:#8a1f1f}}
.shots{{display:grid;grid-template-columns:repeat(auto-fill,minmax(300px,1fr));gap:28px 24px;margin-top:16px}}
figure{{margin:0}}
figure a{{display:block;border:1px solid #e2ddd4;border-radius:4px;overflow:hidden}}
figure img{{display:block;width:100%;height:220px;object-fit:cover;object-position:top}}
figcaption{{font-size:13px;color:#555;margin-top:8px}}
figcaption code{{font-family:ui-monospace,Menlo,monospace;font-size:12px}}
footer{{margin-top:56px;border-top:1px solid #e2ddd4;padding-top:16px;font-size:13px;color:#555}}
</style>
</head>
<body>
<header>
<nav><a href="../index.html">All environments</a></nav>
<h1>Molodetz in pictures ({e})</h1>
<p>{n} screenshots of version {v}, taken {d}.</p>
<p>Every thumbnail links to the full image.</p>
</header>
{s}
<footer>
<p>Molodetz screenshot gallery. Generated from a seeded review build.</p>
</footer>
</body>
</html>
""".format(n=len(records), e=html.escape(env), v=html.escape(version), d=html.escape(stamped), s="".join(sections))


def env_previews(out_dir):
    preferred = ["01-home-full.png", "18-admin-full.png"]
    found = [name for name in preferred if (out_dir / name).exists()]
    for shot in sorted(out_dir.glob("*-home-390x844.png")):
        if shot.name not in found:
            found.append(shot.name)
        if len(found) == 3:
            break
    for shot in sorted(out_dir.glob("*.png")):
        if shot.name not in found:
            found.append(shot.name)
        if len(found) == 3:
            break
    return found


def render_hub(entries):
    cards = []
    for env, count, previews in entries:
        thumbs = "".join(
            '<a href="{e}/index.html"><img src="{e}/{f}" alt="{e}" loading="lazy"></a>'.format(e=env, f=html.escape(name))
            for name in previews
        )
        cards.append(
            '<section><h2><a href="{e}/index.html">{e}</a></h2><p>{n} screenshots.</p><div class="previews">{t}</div></section>'.format(
                e=html.escape(env), n=count, t=thumbs
            )
        )
    return """<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Molodetz screenshot environments</title>
<meta name="description" content="Screenshot galleries of the Molodetz site per environment.">
<style>
body{{margin:0 auto;max-width:1120px;padding:48px 24px 96px;font-family:Georgia,'Times New Roman',serif;color:#1a1815;background:#fff}}
header{{border-bottom:1px solid #e2ddd4;padding-bottom:24px}}
h1{{font-size:34px;margin:0 0 8px;font-weight:600}}
header p{{margin:4px 0;font-size:15px;color:#555}}
section{{margin-top:40px}}
section h2{{font-size:24px;margin:0 0 4px;text-transform:capitalize}}
section h2 a{{color:#1a1815}}
section p{{font-size:14px;color:#555;margin:4px 0 12px}}
.previews{{display:grid;grid-template-columns:repeat(3,1fr);gap:16px}}
.previews a{{display:block;border:1px solid #e2ddd4;border-radius:4px;overflow:hidden}}
.previews img{{display:block;width:100%;height:200px;object-fit:cover;object-position:top}}
footer{{margin-top:56px;border-top:1px solid #e2ddd4;padding-top:16px;font-size:13px;color:#555}}
footer code{{font-family:ui-monospace,Menlo,monospace;font-size:12px}}
</style>
</head>
<body>
<header>
<h1>Molodetz screenshot environments</h1>
<p>One gallery per environment. Regenerate with <code>make screenshots</code>, <code>make screenshots-staging</code>, <code>make screenshots-production</code>.</p>
</header>
{c}
<footer>
<p>Molodetz screenshot hub. Rebuilt on every gallery run from the environments present.</p>
</footer>
</body>
</html>
""".format(c="".join(cards))


def rebuild_hub():
    entries = []
    for env in KNOWN_ENVS:
        out_dir = OUTPUT_DIR / env
        if not (out_dir / "index.html").exists():
            continue
        count = len(list(out_dir.glob("*.png")))
        entries.append((env, count, env_previews(out_dir)))
    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
    (OUTPUT_DIR / "index.html").write_text(render_hub(entries), encoding="utf-8")
    return [env for env, _, _ in entries]


def clear_output(out_dir):
    out_dir.mkdir(parents=True, exist_ok=True)
    for stale in out_dir.glob("*.png"):
        stale.unlink()
    index = out_dir / "index.html"
    if index.exists():
        index.unlink()


def wait_live(base):
    import httpx

    try:
        response = httpx.get(f"{base}/health", timeout=15.0)
    except httpx.HTTPError as exc:
        raise RuntimeError(f"base url {base} is not reachable: {exc}") from exc
    if response.status_code != 200:
        raise RuntimeError(f"base url {base} health is {response.status_code}")


def main():
    from playwright.sync_api import sync_playwright

    if ENV not in KNOWN_ENVS:
        raise RuntimeError(f"unknown SCREENSHOTS_ENV {ENV!r}, want one of {', '.join(KNOWN_ENVS)}")
    boot = not BASE_URL
    if boot and ENV != "test":
        raise RuntimeError("booted runs always target the test gallery, set SCREENSHOTS_ENV=test")
    if not boot and ENV == "test":
        raise RuntimeError("base-url runs target staging or production, set SCREENSHOTS_ENV accordingly")
    if not boot and not ADMIN_PASSWORD:
        raise RuntimeError("base-url runs need SCREENSHOTS_ADMIN_PASSWORD for the admin shots")
    base = LOCAL_BASE if boot else BASE_URL
    password = BOOT_ADMIN_PASSWORD if boot else ADMIN_PASSWORD
    out_dir = OUTPUT_DIR / ENV
    process = None
    log = None
    records = []
    failures = []
    try:
        if boot:
            if REPO_ROOT / "data" not in SHOTS_DATA_DIR.parents:
                raise RuntimeError(f"refusing test data outside {REPO_ROOT / 'data'}")
            shutil.rmtree(SHOTS_DATA_DIR, ignore_errors=True)
            SHOTS_DATA_DIR.mkdir(parents=True, exist_ok=True)
            process, log = start_server(SHOTS_DATA_DIR)
            seed_join(base)
        else:
            wait_live(base)
        headless = os.environ.get("PLAYWRIGHT_HEADLESS", "1") == "1"
        slow_mo = int(os.environ.get("PLAYWRIGHT_SLOW_MO", "0") or 0)
        clear_output(out_dir)
        with sync_playwright() as playwright:
            browser = playwright.chromium.launch(headless=headless, slow_mo=slow_mo)
            counter = capture_full(browser, out_dir, base, password, records, 1)
            capture_responsive(browser, out_dir, base, records, failures, counter)
            browser.close()
        checked = len(RESPONSIVE_PAGES) * len(RESPONSIVE_VIEWPORTS)
        stamped = datetime.now(timezone.utc).date().isoformat()
        (out_dir / "index.html").write_text(render_index(records, failures, checked, ENV, app_version(), stamped), encoding="utf-8")
        present = rebuild_hub()
    finally:
        if process is not None:
            stop_server(process, log)
            shutil.rmtree(SHOTS_DATA_DIR, ignore_errors=True)
    print(f"wrote {len(records)} screenshots for {ENV} to {out_dir}, hub covers {', '.join(present)}")
    for failure in failures:
        print(f"OVERFLOW {failure}")
    return 1 if failures else 0


if __name__ == "__main__":
    sys.exit(main())