← back to Crazy News Channel Shadowman
daily-cartoons/generate.py
300 lines
#!/usr/bin/env python3
"""P24 daily editorial cartoons -> LOCAL review queue (TK-12153, article-driven TK-12237).
Never publishes.
1. articles today's real news (real-news-data.js) + invented P24 stories (stories-data.js,
fallback/fill) via extract_stories.mjs, skipping story_ids already cartooned
in the last 14 days of the queue or anywhere in cartoons/manifest.js ($0)
2. concepts local Ollama, ONE gag per article, satirizing that article's topic ($0)
3. poster Replicate black-forest-labs/flux-schnell (~$0.003) + PIL cream/mint duotone
4. clip OPTIONAL, --video only (default OFF): Replicate wan-2.2-i2v-fast 5s clip (~$0.10)
5. queue queue/<date>/<slug>/{poster.jpg, clip.mp4 (if --video), meta.json}; rebuilds review.html
Cost guard: P24_CARTOON_CAP_USD (default 1.00) is a hard DAILY cap across all runs,
computed from ~/.claude/cost-ledger.jsonl (app=p24-daily-cartoons). A cartoon is only
started if its full estimate (poster [+ clip if --video]) fits; otherwise the run stops
and exits 2.
Usage: generate.py [N=4] [--date YYYY-MM-DD] [--video] [--concepts-only]
Exit: 0 ok · 1 error(s)/no candidates · 2 cap refused remaining generations
"""
import argparse, datetime, io, json, os, re, subprocess, sys, urllib.request
from PIL import Image, ImageOps, ImageEnhance
import common as C
OLLAMA = os.environ.get("OLLAMA_HOST", "http://127.0.0.1:11434")
MODEL = os.environ.get("P24_CARTOON_OLLAMA_MODEL", "hf.co/bartowski/Qwen2.5-Coder-32B-Instruct-GGUF:Q4_K_M")
INK, MINT, CREAM = (11, 11, 12), (15, 157, 120), (205, 245, 224)
SPARE = 6 # extra article candidates fetched beyond N so a failed concept can skip to the next article
STYLE_PRE = ("Black and white single-panel editorial cartoon, classic 1970s newspaper op-ed page style. "
"Pen-and-ink with loose, confident, slightly jagged brush contour lines and dense directional crosshatching, "
"scratchboard-like white highlights cut out of heavy solid black. ")
STYLE_POST = (" Figures are invented anonymous caricatures, not any real person. High contrast, stark, absurd, ink-heavy, "
"hand-drawn newspaper cartoon. No color, no gradients, no 3D render, no photorealism, no vector clip art, "
"no text, no letters, no caption, no speech bubbles, no watermark, no signature.")
MOTION_PRE = "Animate this black-and-white pen-and-ink editorial cartoon, keep the hand-drawn ink style throughout, static camera. "
MOTION_POST = " Slow, dry, deadpan comedy timing."
# real-person guard: concepts naming well-known real public figures are dropped (invented figures
# only). This is a BACKSTOP, not the primary defense — the prompt itself instructs the model to
# never carry any real person/party/company/brand/place/org named IN THE SOURCE ARTICLE into the
# cartoon; this regex only catches a fixed list of the most likely current-events names.
DENY = re.compile(r"\b(trump|biden|harris|obama|pelosi|mcconnell|schumer|desantis|newsom|musk|bezos|zuckerberg|putin|"
r"xi jinping|zelensky|netanyahu|vance|aoc|ocasio|sanders|clinton|bush|reagan|kennedy|democrat|republican|"
r"gop|maga)\b", re.I)
def recent_titles(days=14):
out = []
if os.path.isdir(C.QUEUE):
for d in sorted(os.listdir(C.QUEUE))[-days:]:
dp = os.path.join(C.QUEUE, d)
if not os.path.isdir(dp): continue
for s in os.listdir(dp):
try: out.append(json.load(open(os.path.join(dp, s, "meta.json")))["title"])
except Exception: pass
return out
def used_story_ids(days=14):
"""story_ids to skip: cartooned in the last N days of the local queue, PLUS anything
already approved into cartoons/manifest.js (full history, no day window)."""
used = set()
if os.path.isdir(C.QUEUE):
for d in sorted(os.listdir(C.QUEUE))[-days:]:
dp = os.path.join(C.QUEUE, d)
if not os.path.isdir(dp): continue
for s in os.listdir(dp):
try:
m = json.load(open(os.path.join(dp, s, "meta.json")))
if m.get("story_id"): used.add(m["story_id"])
except Exception: pass
mf = os.path.join(C.SITE, "cartoons", "manifest.js")
if os.path.exists(mf):
mm = re.search(r"window\.P24_CARTOONS\s*=\s*(\[[\s\S]*\])\s*;\s*$", open(mf).read())
if mm:
try:
for c in json.loads(mm.group(1)):
if c.get("story_id"): used.add(c["story_id"])
except Exception: pass
return used
def load_articles():
"""Run extract_stories.mjs (sandboxed vm-context parse of real-news-data.js /
stories-data.js) and return the flat JSON article list. Read-only, $0."""
p = subprocess.run(["node", os.path.join(C.HERE, "extract_stories.mjs")],
capture_output=True, text=True, cwd=C.HERE, timeout=60)
if p.returncode != 0:
raise RuntimeError(f"extract_stories.mjs failed: {p.stderr.strip()[:400]}")
return json.loads(p.stdout)
def pick_articles(n, excluded):
"""Newest real-news first (by parsed publishedAt), then real-news without a parseable
date, then invented P24 stories (stories-data.js) as fallback/fill — skipping any id
in `excluded`. Returns up to n articles; fewer (or []) if candidates run out."""
arts = load_articles()
dated = sorted([a for a in arts if a.get("publishedAt")], key=lambda a: a["publishedAt"], reverse=True)
undated_real = [a for a in arts if not a.get("publishedAt") and a.get("source") == "real-news"]
fallback = [a for a in arts if a.get("source") == "p24"]
seen, out = set(excluded), []
for a in dated + undated_real + fallback:
if a["id"] in seen: continue
seen.add(a["id"])
out.append(a)
if len(out) >= n: break
return out
def story_url(article):
if article.get("source") == "real-news":
return article.get("sourceUrl")
return f"index.html#/article/{article['id']}"
def concept_prompt(article, avoid, want_motion):
fields = ('"title":"3-6 words","caption":"one witty line, max 18 words",'
'"scene":"60-110 word purely visual description of the drawing: foreground figure(s), '
'action, background, the absurd detail. No text or lettering in the image."')
if want_motion:
fields += ',"motion":"25-50 words: what moves in a 5-second deadpan animation of this exact scene"'
return f"""You are the gag writer for P24, a satirical "crazy news" site. Read this real news item and invent ONE
ORIGINAL single-panel editorial-cartoon concept that satirizes its topic IN THE ABSTRACT — the same underlying
absurdity, institution, or behavior, generalized so it stands alone as a gag.
ARTICLE HEADLINE: {article['headline']}
ARTICLE SUMMARY: {article['summary']}
ARTICLE TAGS: {', '.join(article.get('tags') or [])}
Rules: invented anonymous figures only (e.g. "a jowly senator", "a committee of identical bureaucrats");
NEVER name or depict any real person, party, company, brand, place, or organization mentioned in (or
implied by) the article above — invent a generic stand-in that represents the same role or theme instead;
one strong absurd visual metaphor per cartoon; nothing hateful.
Avoid repeating these recent titles: {json.dumps(avoid[-30:])}
Return ONLY JSON: {{{fields}}}"""
def ollama_concept_for_article(article, avoid, want_motion):
body = {"model": MODEL, "prompt": concept_prompt(article, avoid, want_motion), "format": "json", "stream": False,
"options": {"temperature": 0.9, "num_predict": 900}}
r = urllib.request.Request(f"{OLLAMA}/api/generate", data=json.dumps(body).encode(),
headers={"Content-Type": "application/json"})
resp = json.load(urllib.request.urlopen(r, timeout=300))
c = json.loads(resp["response"])
required = ["title", "caption", "scene"] + (["motion"] if want_motion else [])
if not all(isinstance(c.get(k), str) and c[k].strip() for k in required):
return None
if DENY.search(" ".join(c[k] for k in required)):
print(" [concepts] dropped (names real person/party):", c.get("title")); return None
return {k: c[k].strip() for k in required}
def gather_concepts(pool, n, want_motion):
"""One Ollama call per candidate article (up to n successes, 2 tries each);
on failure, skip that article and move to the next candidate in the pool."""
avoid = recent_titles()
picked = []
for a in pool:
if len(picked) >= n: break
c = None
for attempt in range(2):
try:
c = ollama_concept_for_article(a, avoid + [p[1]["title"] for p in picked], want_motion)
except Exception as e:
print(f" [concepts] ollama error for '{a['headline'][:60]}' (try {attempt+1}/2): {e}")
c = None
if c: break
if c:
print(f" [concepts] {a['id']} ({a['source']}) -> \"{c['title']}\"")
picked.append((a, c))
else:
print(f" [concepts] SKIP — no usable concept for: {a['headline'][:70]}")
return picked
def slugify(t):
return re.sub(r"[^a-z0-9]+", "-", t.lower()).strip("-")[:48] or "cartoon"
def duotone(raw_bytes, out_path):
im = Image.open(io.BytesIO(raw_bytes)).convert("L")
im = ImageEnhance.Contrast(im).enhance(1.25)
ImageOps.colorize(im, black=INK, white=CREAM, mid=MINT, blackpoint=0, whitepoint=255, midpoint=150)\
.convert("RGB").save(out_path, "JPEG", quality=90)
def gen_poster(image_prompt, slug):
C.guard("flux")
pr = C.api("https://api.replicate.com/v1/models/black-forest-labs/flux-schnell/predictions",
{"input": {"prompt": image_prompt, "aspect_ratio": "4:3", "output_format": "jpg",
"output_quality": 92, "num_outputs": 1, "megapixels": "1"}}, prefer_wait=True)
cost = C.log_cost("flux", f"poster {slug} pred={pr.get('id')}") # billed at create
pr = C.poll(pr, 2, 300)
if pr["status"] != "succeeded":
raise RuntimeError(f"flux {pr['status']}: {pr.get('error')}")
out = pr["output"]; out = out[0] if isinstance(out, list) else out
return urllib.request.urlopen(out, timeout=120).read(), cost, pr["id"]
def gen_clip(poster_path, motion_prompt, slug, out_path):
C.guard("wan")
import base64
img = "data:image/jpeg;base64," + base64.b64encode(open(poster_path, "rb").read()).decode()
pr = C.api("https://api.replicate.com/v1/models/wan-video/wan-2.2-i2v-fast/predictions",
{"input": {"prompt": motion_prompt, "image": img, "num_frames": 81, "frames_per_second": 16,
"resolution": "480p", "go_fast": True}})
cost = C.log_cost("wan", f"clip {slug} pred={pr.get('id')}")
print(f" [wan] prediction {pr['id']} {pr['status']}")
pr = C.poll(pr, 8, 1500)
if pr["status"] != "succeeded":
raise RuntimeError(f"wan {pr['status']}: {pr.get('error')}")
urllib.request.urlretrieve(pr["output"], out_path)
return cost, pr["id"], pr.get("metrics", {})
def main():
ap = argparse.ArgumentParser()
ap.add_argument("n", nargs="?", type=int, default=4)
ap.add_argument("--date", default=datetime.date.today().isoformat())
ap.add_argument("--video", action="store_true",
help="also render a Replicate wan-2.2 i2v clip (~$0.10/cartoon). Default OFF — still image only.")
ap.add_argument("--concepts-only", "--dry-run", dest="concepts_only", action="store_true",
help="fetch articles + Ollama concepts and print/save them; no Replicate calls, $0")
a = ap.parse_args()
per = C.EST["flux"] + (C.EST["wan"] if a.video else 0)
cap, spent = C.cap_usd(), C.spent_today()
print(f"P24 daily cartoons: N={a.n} date={a.date} video={a.video} cap=${cap:.2f} spent_today=${spent:.3f} "
f"est=${per:.3f}/cartoon, ${per*a.n:.3f} total")
if not a.concepts_only and spent + per > cap + 1e-9:
print(f"REFUSED: cap ${cap:.2f} leaves no room (spent ${spent:.3f} + est ${per:.3f}). No paid calls made.")
return 2
print("[articles] selecting candidates from real-news-data.js / stories-data.js…")
excluded = used_story_ids()
pool = pick_articles(a.n + SPARE, excluded)
if not pool:
print(f"ERROR: 0 article candidates — every story_id was already cartooned in the last 14 days "
f"of the queue or is present in cartoons/manifest.js ({len(excluded)} excluded).")
return 1
print(f" {len(pool)} candidate article(s) available (targeting {a.n}); {len(excluded)} story_id(s) excluded")
print("[concepts] asking local Ollama, one gag per article ($0)…")
picked = gather_concepts(pool, a.n, a.video)
if not picked:
print("ERROR: no usable concepts from Ollama"); return 1
if len(picked) < a.n:
print(f" only {len(picked)}/{a.n} article(s) produced a usable concept")
if a.concepts_only:
for art, c in picked:
print(f"\n== story {art['id']} ({art['source']}) — {art['headline'][:90]}")
print(f" title: {c['title']}")
print(f" caption: {c['caption']}")
print(f" scene: {c['scene']}")
if a.video: print(f" motion: {c.get('motion','')}")
os.makedirs(os.path.join(C.HERE, "logs"), exist_ok=True)
out = [{"story_id": art["id"], "story_source": art["source"], "story_title": art["headline"],
"story_url": story_url(art), **c} for art, c in picked]
outpath = os.path.join(C.HERE, "logs", f"concepts-{a.date}.json")
json.dump(out, open(outpath, "w"), indent=2)
print(f"\n[concepts-only] {len(out)} concept(s) written to {outpath}. $0 spent — no Replicate calls made.")
return 0
day = os.path.join(C.QUEUE, a.date); os.makedirs(day, exist_ok=True)
errors, refused, total = 0, False, 0.0
for art, c in picked:
slug = slugify(c["title"]); d = os.path.join(day, slug)
if os.path.exists(os.path.join(d, "meta.json")):
slug += "-" + datetime.datetime.now().strftime("%H%M%S"); d = os.path.join(day, slug)
if C.spent_today() + per > C.cap_usd() + 1e-9:
print(f"CAP: stopping before '{c['title']}' — would exceed ${C.cap_usd():.2f}"); refused = True; break
os.makedirs(d, exist_ok=True)
print(f"\n== {c['title']} ({slug}) <- story {art['id']}")
img_prompt = STYLE_PRE + c["scene"].strip() + STYLE_POST
meta = {"slug": slug, "date": a.date, "title": c["title"].strip(), "caption": c["caption"].strip(),
"image_prompt": img_prompt, "concept_model": MODEL,
"image_model": "black-forest-labs/flux-schnell",
"story_id": art["id"], "story_title": art["headline"], "story_url": story_url(art),
"story_source": art["source"],
"created_at": datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds"),
"status": "queued", "cost": {"concepts": 0.0}, "predictions": {}}
if a.video:
mot_prompt = MOTION_PRE + c.get("motion", "").strip() + MOTION_POST
meta["motion_prompt"] = mot_prompt
meta["video_model"] = "wan-video/wan-2.2-i2v-fast"
meta["video_params"] = {"num_frames": 81, "frames_per_second": 16, "resolution": "480p"}
try:
raw, cost, pid = gen_poster(img_prompt, slug)
meta["cost"]["image"] = cost; meta["predictions"]["image"] = pid
open(os.path.join(d, "raw.jpg"), "wb").write(raw)
duotone(raw, os.path.join(d, "poster.jpg"))
print(f" poster.jpg ok (actual ${cost:.3f})")
if a.video:
cost, pid, metrics = gen_clip(os.path.join(d, "poster.jpg"), meta["motion_prompt"], slug, os.path.join(d, "clip.mp4"))
meta["cost"]["video"] = cost; meta["predictions"]["video"] = pid
meta["video_metrics"] = metrics
print(f" clip.mp4 ok (actual-logged ${cost:.3f}, predict_time {metrics.get('predict_time')}s)")
except C.CapExceeded as e:
print("CAP:", e); refused = True
meta["status"] = "incomplete-cap"
except Exception as e:
print("ERROR:", e); errors += 1
meta["status"] = "error"; meta["error"] = str(e)[:500]
meta["cost"]["total"] = round(sum(v for k, v in meta["cost"].items() if k != "total"), 4)
total += meta["cost"]["total"]
json.dump(meta, open(os.path.join(d, "meta.json"), "w"), indent=2)
if refused: break
import build_review; build_review.build()
print(f"\nRun spend ${total:.3f} · spent today ${C.spent_today():.3f} / cap ${C.cap_usd():.2f} · review: {os.path.join(C.HERE,'review.html')}")
return 2 if refused else (1 if errors else 0)
if __name__ == "__main__":
sys.exit(main())