← back to Crazy News Channel
daily-edition/build.py
428 lines
#!/usr/bin/env python3
"""P24 Daily Edition (TK-12354) — ALL of today's news + the TOP 6 stories, each with an
original editorial-ink cartoon signed "Shadow Man" (the conrad.agentabrams.com house style).
python3 daily-edition/build.py # build/refresh today's edition
python3 daily-edition/build.py --no-images # $0: list + concepts only, no Replicate calls
python3 daily-edition/build.py --dry # print the top-6 picks and exit, writes nothing
News source: the allnewsdaily wire (~/Projects/allnewsdaily/data/wire.json, rebuilt every
10 min by com.steve.allnewsdaily-wire-push), falling back to this repo's real-news-data.js
(Google News RSS) when the wire is missing or older than WIRE_MAX_AGE_H.
Top 6 are LOCKED on the first run of the day: later same-day runs refresh the all-news list
for $0 and never regenerate illustrations (so the daily spend is ~6 x $0.003 = $0.018).
Illustrations: concept (gag + purely visual scene) from a local LLM ($0 — Mac1 Ollama
qwen3:14b, then the local MLX server), falling back to a deterministic keyword template;
image = Replicate flux-schnell (~$0.003) -> ink post-process -> "Shadow Man" signature.
Hard daily cap P24_EDITION_CAP_USD (default 0.10), summed from ~/.claude/cost-ledger.jsonl
(app=p24-daily-edition); every paid call is logged via the cost-tracker skill.
Standing rules honoured: never name/depict real people/brands in a cartoon (invented
archetypes only, DENY + proper-noun backstops); never name the research cartoonist anywhere
(BANNED); every rendered cartoon carries the "Shadow Man" signature.
"""
import argparse, datetime, hashlib, io, json, os, re, subprocess, sys, urllib.request
HERE = os.path.dirname(os.path.abspath(__file__))
SITE = os.environ.get("P24_SITE_DIR") or os.path.dirname(HERE)
sys.path.insert(0, os.path.join(os.path.dirname(HERE), "daily-cartoons"))
import common as C # noqa: E402 (shared Replicate + cost-ledger helpers)
C.APP = "p24-daily-edition"
C.cap_usd = lambda: float(os.environ.get("P24_EDITION_CAP_USD", "0.10"))
WIRE = os.path.expanduser(os.environ.get("P24_WIRE_JSON", "~/Projects/allnewsdaily/data/wire.json"))
WIRE_MAX_AGE_H = float(os.environ.get("P24_WIRE_MAX_AGE_H", "36"))
ITEM_MAX_AGE_H = float(os.environ.get("P24_ITEM_MAX_AGE_H", "36"))
OUT_DIR = os.path.join(SITE, "daily")
DATA_JS = os.path.join(SITE, "daily-edition-data.js")
TOP_N = 6
OLLAMA_HOSTS = [h for h in os.environ.get("P24_EDITION_OLLAMA", "http://192.168.1.133:11434").split(",") if h]
OLLAMA_MODEL = os.environ.get("P24_EDITION_OLLAMA_MODEL", "qwen3:14b")
MLX = os.environ.get("P24_EDITION_MLX", "http://127.0.0.1:8000")
MLX_MODEL = os.environ.get("P24_EDITION_MLX_MODEL", "mlx-community/Qwen3-30B-A3B-Instruct-2507-4bit")
SIGNATURE = "Shadow Man"
STYLE = "bold black ink editorial cartoon, heavy brush line, crosshatching, stark black and white, single panel satire"
STYLE_TAIL = (" Invented anonymous caricature figures only. No color, no text, no letters, no words, "
"no speech bubbles, no caption, no watermark, no signature.")
BANNED = re.compile(r"conrad|in the style of|style of [A-Z]|pulitzer", re.I)
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|google|apple|amazon|microsoft|meta|tesla|oracle|nvidia|openai)\b", re.I)
FONT_SIG = "/System/Library/Fonts/Supplemental/Bradley Hand Bold.ttf"
STOP = set("a an the of to in on for and or at by with from as is are was be it its this that new says after over into "
"amid report reports than more about up out who what how why will can".split())
# ---------------------------------------------------------------- news
def _parse_date(s):
if not s:
return None
for fmt in ("%a, %d %b %Y %H:%M:%S %z", "%a, %d %b %Y %H:%M:%S %Z"):
try:
return datetime.datetime.strptime(s.strip(), fmt).astimezone(datetime.timezone.utc)
except Exception:
pass
try:
return datetime.datetime.fromisoformat(s.replace("Z", "+00:00")).astimezone(datetime.timezone.utc)
except Exception:
return None
def clean_topic(t):
t = re.sub(r"^\s*(report|breaking|update|live)\s*:\s*", "", t or "", flags=re.I).strip()
return (t[:1].upper() + t[1:]) if t else t
def story_id(link):
return "d-" + hashlib.sha1((link or "").encode()).hexdigest()[:10]
def load_wire():
"""-> (source_label, updated_iso, [items]) or None when missing/stale/invalid."""
try:
w = json.load(open(WIRE))
except Exception as e:
print(f"[news] wire unavailable ({e}); falling back"); return None
upd = _parse_date(w.get("updatedAt"))
if not upd or (datetime.datetime.now(datetime.timezone.utc) - upd).total_seconds() > WIRE_MAX_AGE_H * 3600:
print(f"[news] wire stale (updatedAt={w.get('updatedAt')}); falling back"); return None
items = []
sp = w.get("splash")
if sp and sp.get("link"):
items.append(dict(sp, section="Top story", section_key="top", splash=True))
for col in w.get("columns") or []:
for it in col.get("items") or []:
items.append(dict(it, section=col.get("title") or col.get("key") or "News", section_key=col.get("key") or "news"))
return "allnewsdaily wire", upd.isoformat(), items
def load_realnews_fallback():
p = subprocess.run(["node", os.path.join(SITE, "daily-cartoons", "extract_stories.mjs")],
capture_output=True, text=True, timeout=60)
if p.returncode != 0:
raise SystemExit(f"[news] no wire AND extract_stories failed: {p.stderr[:300]}")
items = []
for a in json.loads(p.stdout):
if a.get("source") != "real-news":
continue
items.append({"outlet": a.get("sourceName") or "", "link": a.get("sourceUrl"), "topic": a.get("headline"),
"date": a.get("publishedAt"), "image": None,
"section": (a.get("tags") or ["News"])[0].title(), "section_key": "real"})
return "Google News RSS (fallback)", None, items
def normalize(items):
now = datetime.datetime.now(datetime.timezone.utc)
seen, out = set(), []
for it in items:
link, topic = it.get("link"), clean_topic(it.get("topic"))
if not link or not topic or link in seen:
continue
d = _parse_date(it.get("date"))
if d and (now - d).total_seconds() > ITEM_MAX_AGE_H * 3600:
continue
seen.add(link)
out.append({"id": story_id(link), "headline": topic, "url": link, "outlet": it.get("outlet") or "",
"published": d.isoformat() if d else None, "section": it.get("section") or "News",
"section_key": it.get("section_key") or "news", "splash": bool(it.get("splash"))})
return out
def tokens(s):
s = s.lower().replace("’", "").replace("'", "")
return {w.rstrip("s") if len(w) > 4 else w for w in re.findall(r"[a-z0-9]+", s) if w not in STOP and len(w) > 2}
def same_story(a, b):
ta, tb = tokens(a), tokens(b)
shared = len(ta & tb)
return shared >= 2 or shared / max(1, len(ta | tb)) >= 0.3
def pick_top(items, n=TOP_N):
"""Splash first, then round-robin across sections (newest first in each), skipping
near-duplicates of an already-picked (or already-skipped) story, so one big story
covered by five outlets fills ONE slot, not five."""
by_sec = {}
for it in items:
if it["splash"]:
continue
by_sec.setdefault(it["section_key"], []).append(it)
for v in by_sec.values():
v.sort(key=lambda x: x["published"] or "", reverse=True)
order = [it for it in items if it["splash"]]
queues = list(by_sec.values())
while any(queues):
for q in queues:
if q:
order.append(q.pop(0))
picked, seen_heads = [], []
for it in order:
if any(same_story(it["headline"], h) for h in seen_heads):
seen_heads.append(it["headline"])
continue
seen_heads.append(it["headline"])
picked.append(it)
if len(picked) >= n:
break
return picked
# ---------------------------------------------------------------- concepts
def concept_prompt(story):
return f"""/no_think You are the gag writer for P24, a satirical news site. Invent ONE ORIGINAL single-panel
editorial-cartoon concept about the TOPIC of this real headline, generalized so it stands alone as a gag.
HEADLINE: {story['headline']}
Rules: invented anonymous archetypes only ("a sweating official", "a committee of identical bureaucrats",
"a tiny homeowner"); NEVER name or depict any real person, party, company, brand, country, city or
organization from the headline — use a generic stand-in for the same role; one strong absurd visual
metaphor; nothing hateful. The drawing must contain NO written words, signs, labels, numbers, dates
or quoted text of any kind — tell the joke with the picture alone.
Return ONLY JSON: {{"title":"3-6 words","caption":"one witty line, max 16 words","scene":"40-80 word purely
visual description: foreground figure(s), action, background, the absurd detail"}}"""
def _ollama(host, prompt):
body = {"model": OLLAMA_MODEL, "prompt": prompt, "format": "json", "stream": False,
"options": {"temperature": 0.75, "num_predict": 450}}
r = urllib.request.Request(f"{host}/api/generate", data=json.dumps(body).encode(),
headers={"Content-Type": "application/json"})
return json.loads(json.load(urllib.request.urlopen(r, timeout=120))["response"])
def _mlx(prompt):
body = {"model": MLX_MODEL, "messages": [{"role": "user", "content": prompt}], "max_tokens": 500, "temperature": 0.9}
r = urllib.request.Request(f"{MLX}/v1/chat/completions", data=json.dumps(body).encode(),
headers={"Content-Type": "application/json"})
txt = json.load(urllib.request.urlopen(r, timeout=25))["choices"][0]["message"]["content"]
return json.loads(txt[txt.index("{"): txt.rindex("}") + 1])
def proper_nouns(headline):
"""Capitalized non-initial words of a SENTENCE-case headline. Title-case headlines capitalize
everything, so there the check is skipped (the prompt rule + DENY list still apply)."""
words = re.findall(r"[A-Za-z][A-Za-z'’.-]+", headline)
caps = [w for i, w in enumerate(words) if i > 0 and w[0].isupper()]
if len(words) > 1 and len(caps) / (len(words) - 1) > 0.5:
return set()
return {w.strip(".’'").lower() for w in caps if len(w) > 2} - STOP
QUOTED = re.compile(r"""(?:^|(?<=[\s(]))["'“‘][^"'”’]{1,80}["'”’]""")
TEXTY = re.compile(r"\b(sign|signs|banner|label|labels|placard|headline|reads|reading|written|lettering|words?|slogan|poster)\b", re.I)
def scrub_scene(scene):
"""Drop sentences that would make flux paint lettering (quoted text, signs, labels)."""
keep = [x for x in re.split(r"(?<=[.!?])\s+", scene.strip()) if x and not QUOTED.search(x) and not TEXTY.search(x)]
return " ".join(keep)
def valid_concept(c, story):
if not all(isinstance(c.get(k), str) and c[k].strip() for k in ("title", "caption", "scene")):
return False
c["scene"] = scrub_scene(c["scene"])
if len(c["scene"].split()) < 15:
return False
blob = " ".join(c[k] for k in ("title", "caption", "scene"))
if DENY.search(blob) or BANNED.search(blob):
return False
low = blob.lower()
return not any(re.search(r"\b" + re.escape(p) + r"\b", low) for p in proper_nouns(story["headline"]))
TEMPLATES = [
(r"\b(storm|flood|flooding|rain|hurricane|wildfire|fire|heat|snow|weather|climate|nor.?easter)\b", "Forecast: More of This",
"Officials remain confident the situation is under a light drizzle of control.",
"a tiny official in a suit calmly holding a cocktail umbrella while a colossal wave curls over a row of houses behind him, a weather vane spinning off its roof"),
(r"\b(court|judge|lawsuit|sues|trial|ruling|legal|antitrust)\b", "Justice, Eventually",
"The verdict is expected sometime after the heat death of the universe.",
"a towering stack of legal paperwork reaching into the clouds, a very small judge climbing it with a rope and pickaxe, lawyers waving from base camp"),
(r"\b(rates?|inflation|markets?|stocks?|economy|prices?|banks?|mortgage|tariffs?|jobs|business|pays?|salary|income|grosses|\$[0-9])", "The Market Speaks",
"Analysts describe the outlook as 'a rollercoaster, but with fewer seatbelts.'",
"an ordinary family clinging to a runaway rollercoaster made of a price chart, a smiling banker selling tickets at the bottom"),
(r"\b(election|senate|congress|vote|polls?|campaign|lawmakers|administration|policy|bill|rules|regulations?)\b", "Business as Usual",
"Both sides agree on one thing: it's the other side's fault.",
"two identical politicians in a tug of war over a crumbling capitol dome, the rope fraying, a tiny citizen standing underneath holding a teacup"),
(r"\b(war|strikes?|military|missiles?|troops|attack|ceasefire|deal|talks|nuclear)\b", "Peace Talks, Round 47",
"The negotiating table has been extended again.",
"generals on both sides of an absurdly long negotiating table that stretches past the horizon, doves nesting in their helmets"),
(r"\bai\b|robot|tech|app|chip|data center|software|phone|internet|cyber", "Progress Report",
"The machine assures everyone it has their best interests memorized.",
"a giant friendly robot gently holding a tiny office worker like a pet, feeding him emails through a funnel, server racks humming behind"),
(r"\b(health|virus|vaccines?|hospitals?|drugs?|disease|medical)\b", "Second Opinion",
"Take two headlines and call me in the morning.",
"a doctor handing a patient an enormous pill the size of a boulder, a line of worried patients stretching out the door"),
(r"\b(games?|team|season|league|coach|playoffs?|championship|sports?)\b", "Game Day",
"The fans have announced they will now be coaching from the stands.",
"a stadium where the crowd has climbed onto the field to play while the players watch from the bleachers eating hot dogs"),
]
def template_concept(story):
h = story["headline"].lower()
for rx, title, cap, scene in TEMPLATES:
if re.search(rx, h):
return {"title": title, "caption": cap, "scene": scene, "engine": "template"}
return {"title": "Film at Eleven", "caption": "Details are developing, as they always seem to be.",
"scene": "a harried newsman at a desk buried under an avalanche of paper headlines, only his hand visible holding a coffee cup aloft",
"engine": "template"}
DEAD_ENGINES = set() # an engine that errors once is skipped for the rest of the run
def concept_for(story):
p = concept_prompt(story)
for name, fn in [(f"ollama:{h}", (lambda h=h: _ollama(h, p))) for h in OLLAMA_HOSTS] + [("mlx", lambda: _mlx(p))]:
if name in DEAD_ENGINES:
continue
for attempt in range(3):
try:
c = fn()
except json.JSONDecodeError as e:
print(f" [concept] {name} bad JSON (try {attempt+1}): {str(e)[:80]}"); continue
except Exception as e:
print(f" [concept] {name} error, disabling for this run: {str(e)[:120]}")
DEAD_ENGINES.add(name); break
if valid_concept(c, story):
return {"title": c["title"].strip(), "caption": c["caption"].strip(),
"scene": c["scene"].strip(), "engine": name}
print(f" [concept] {name} returned an invalid/unsafe concept (try {attempt+1})")
return template_concept(story)
# ---------------------------------------------------------------- image
BOTTOM_CROP = 0.10 # flux paints fake scribbled "artist signatures" along the bottom edge — cut them off
def sign(art):
"""Stamp the "Shadow Man" signature lower-right, in place (house style of conrad.agentabrams.com)."""
from PIL import ImageDraw, ImageFont
W, H = art.size
d = ImageDraw.Draw(art)
sf = ImageFont.truetype(FONT_SIG, max(24, W // 28))
bb = d.textbbox((0, 0), SIGNATURE, font=sf)
sw, sh = bb[2] - bb[0], bb[3] - bb[1]
x, y = W - sw - W // 30, H - sh - H // 20
d.rectangle([x - 10, y - 6, x + sw + 10, y + sh + 14], fill=(255, 255, 255))
d.text((x, y - bb[1]), SIGNATURE, font=sf, fill=(10, 10, 10))
d.line([x, y + sh + 7, x + sw, y + sh + 3], fill=(10, 10, 10), width=3)
def ink_and_sign(raw, out_path):
from PIL import Image, ImageEnhance, ImageOps
im = Image.open(io.BytesIO(raw))
g = ImageEnhance.Contrast(ImageOps.grayscale(im)).enhance(1.5)
g = g.point(lambda p: 255 if p > 205 else (0 if p < 40 else p))
art = g.convert("RGB")
art = art.crop((0, 0, art.size[0], int(art.size[1] * (1 - BOTTOM_CROP))))
if art.size[0] > 1024:
art = art.resize((1024, int(art.size[1] * 1024 / art.size[0])))
sign(art)
art.save(out_path, "JPEG", quality=84, optimize=True, progressive=True)
def gen_image(scene, tag):
prompt = f"{scene.rstrip('.')}. {STYLE}.{STYLE_TAIL}"
if BANNED.search(prompt):
raise RuntimeError("banned term in prompt")
C.guard("flux")
pr = C.api("https://api.replicate.com/v1/models/black-forest-labs/flux-schnell/predictions",
{"input": {"prompt": 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"p24 daily edition {tag} pred={pr.get('id')}")
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"], prompt
# ---------------------------------------------------------------- main
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--no-images", action="store_true")
ap.add_argument("--dry", action="store_true")
ap.add_argument("--date", help="edition date YYYY-MM-DD (default: today, local)")
a = ap.parse_args()
date = a.date or datetime.date.today().isoformat()
src = load_wire() or load_realnews_fallback()
label, updated, raw = src
items = normalize(raw)
if not items:
print("[news] 0 items after normalize — refusing to publish an empty edition"); return 1
day_dir = os.path.join(OUT_DIR, date)
ed_path = os.path.join(day_dir, "edition.json")
prev = json.load(open(ed_path)) if os.path.exists(ed_path) else None
if prev and prev.get("top"):
top = prev["top"] # locked for the day
print(f"[top] locked from first run of {date}: {len(top)} stories")
else:
top = [dict(s) for s in pick_top(items)]
print(f"[top] picked {len(top)} new stories for {date}")
for i, s in enumerate(top, 1):
print(f" {i}. [{s['section']}] {s['headline'][:90]} — {s['outlet']}")
if a.dry:
return 0
os.makedirs(day_dir, exist_ok=True)
spent = 0.0
for i, s in enumerate(top, 1):
if not s.get("concept"):
s["concept"] = concept_for(s)
print(f" [concept] #{i} {s['concept']['engine']}: \"{s['concept']['title']}\"")
img_rel = f"daily/{date}/{i}-{s['id']}.jpg"
img_abs = os.path.join(SITE, img_rel)
if os.path.exists(img_abs):
s["image"] = img_rel
continue
if a.no_images:
s.setdefault("image", None)
continue
try:
raw_img, cost, pid, prompt = gen_image(s["concept"]["scene"], f"#{i} {s['id']}")
ink_and_sign(raw_img, img_abs)
spent += cost
s.update(image=img_rel, image_prompt=prompt, prediction_id=pid, image_cost_usd=cost,
image_created_at=datetime.datetime.now().astimezone().isoformat(timespec="seconds"))
print(f" [image] #{i} ok ${cost:.3f} -> {img_rel}")
except C.CapExceeded as e:
print(f" [image] #{i} CAP REFUSED: {e}"); s.setdefault("image", None)
except Exception as e:
print(f" [image] #{i} FAILED: {str(e)[:200]}"); s.setdefault("image", None)
top_ids = {s["id"] for s in top}
now = datetime.datetime.now().astimezone().isoformat(timespec="seconds")
edition = {
"date": date, "built_at": now, "created_at": (prev or {}).get("created_at", now),
"source": label, "source_updated_at": updated,
"illustrator": SIGNATURE, "top": top,
"all": [dict(it, top=it["id"] in top_ids) for it in items],
"counts": {"all": len(items), "top": len(top), "illustrated": sum(1 for s in top if s.get("image"))},
"spend_usd_this_run": round(spent, 4), "spend_usd_today": C.spent_today(),
}
tmp = ed_path + ".tmp"
json.dump(edition, open(tmp, "w"), indent=1)
os.replace(tmp, ed_path)
dates = sorted((d for d in os.listdir(OUT_DIR) if re.match(r"^\d{4}-\d{2}-\d{2}$", d)), reverse=True)
json.dump({"dates": dates}, open(os.path.join(OUT_DIR, "index.json"), "w"))
js = ("// AUTO-GENERATED by daily-edition/build.py (TK-12354) — do not hand-edit.\n"
"window.P24_DAILY = " + json.dumps(edition, ensure_ascii=False) + ";\n")
open(DATA_JS + ".tmp", "w").write(js)
os.replace(DATA_JS + ".tmp", DATA_JS)
print(f"[done] {date}: {edition['counts']} · source={label} · spent this run ${spent:.3f}, today ${edition['spend_usd_today']:.3f}")
return 0 if edition["counts"]["illustrated"] == len(top) or a.no_images else 3
if __name__ == "__main__":
sys.exit(main())