# worker_script.py — stage 7 narration. FastAPI :8005. calls the warm gemma4 server (:8090). import os, re, json from fastapi import FastAPI from pydantic import BaseModel import requests import transport app = FastAPI() transport.install_logging(app, "script") GEMMA4_URL = os.environ.get("GEMMA4_URL", "http://127.0.0.1:8090") class ScriptInput(BaseModel): scene_graph: dict chapter_context: str = "" panel_id: str = "" session_id: str = "" names_by_id: dict = {} # character_id -> name, resolved from the final registry (A backfill) genders_by_id: dict = {} # character_id -> m|f|unknown, from the registry (F2, for pronouns) brief: str = "" # #3: static whole-chapter synopsis (background for consistency/tone) recent: list = [] # #1: last few panels' narration text (local flow, no repeats) introduced: list = [] # #2: names already narrated -> refer by name/pronoun, don't re-describe panel_count: int = 1 # scene-level narration: #panels in this beat -> scales the length budget class SummaryInput(BaseModel): prior: str = "" # the running "story so far" before this batch recent: list = [] # narration of the panels since the last refresh session_id: str = "" def _strip_thought(text: str) -> str: parts = re.split(r"<\|?channel\|?>", text) return parts[-1].strip() if len(parts) > 1 else text # how each bubble type reads in the narration. narration/sfx have no speaker. _VERB = {"speech": "says", "thought": "thinks", "shout": "shouts"} def _render_line(d, name_by_id): typ = d.get("type", "speech") txt = d.get("text", "") if typ == "narration": return f'Caption (scene fact, weave in — do not announce it): "{txt}"' if typ == "sfx": return f"Sound effect: {txt}" who = name_by_id.get(d.get("speaker"), "Someone") if d.get("speaker") else "Someone" return f'{who} {_VERB.get(typ, "says")} "{txt}"' def _name_map(characters, names_by_id=None): """id -> display name. Final registry name first (names_by_id, so a name learned in a LATER panel backfills to this one); else the name baked at scene time; else the frozen descriptive label ("the one with black hair and glasses"); else a stable positional label ("Person A"). The model never sees raw DB ids like character_11004ac9.""" names_by_id = names_by_id or {} name_by_id, n = {}, 0 for c in characters: if names_by_id.get(c["id"]) or c.get("name"): name_by_id[c["id"]] = names_by_id.get(c["id"]) or c["name"] elif c.get("label"): name_by_id[c["id"]] = c["label"] else: name_by_id[c["id"]] = f"Person {chr(65 + n)}" n += 1 return name_by_id _PRONOUN = {"m": "he", "f": "she"} def _chars_line(name_by_id, genders_by_id): """render 'Name (he), Other (she)' so the narrator gets pronouns right (F2).""" genders_by_id = genders_by_id or {} out = [] for cid, disp in name_by_id.items(): p = _PRONOUN.get((genders_by_id.get(cid) or "").lower()) out.append(f"{disp} ({p})" if p else disp) return ", ".join(out) or "none" def build_prompt(sg: dict, chapter_context: str, names_by_id=None, brief: str = "", recent=None, introduced=None, genders_by_id=None, panel_count: int = 1) -> str: name_by_id = _name_map(sg.get("characters", []), names_by_id) chars = _chars_line(name_by_id, genders_by_id) dialogue = "\n".join(_render_line(d, name_by_id) for d in sg.get("dialogue", [])) or "none" ents = ", ".join(e.get("name", "") for e in sg.get("entities", []) if e.get("name")) # cast+entities roster (whole-manga memory), kept distinct from the running plot summary below — # they used to both render as "Story so far", so the static roster masqueraded as the story. ctx = f"Cast & places: {chapter_context}\n\n" if chapter_context else "" # #3 running "story so far": only what has happened up to now, so it grounds tone/naming # without any risk of narrating ahead (future panels aren't in it yet). ov = (f"Story so far (background — what has happened up to now; stay consistent with it, do " f"NOT restate it verbatim): {brief}\n\n") if brief else "" # #1 local flow: keep continuity with what was just said, don't restate it — and don't reuse its # sentence scaffolding (LLMs happily repeat openers/structure even when the facts differ). rec = ("Just narrated (continue smoothly from this; do NOT repeat these facts, and do NOT reuse " "their sentence structure or opening words — vary the rhythm):\n" + "\n".join(f"- {r}" for r in recent) + "\n\n") if recent else "" # camera->pacing: let the director's shot drive sentence rhythm (the biggest "feels AI" tell is # narration pacing that ignores the cut). fast/tight shot + hard cut -> punchy; held/drifting shot # or soft dissolve -> can breathe. cam = (sg.get("camera") or {}).get("effect") or "" trans = sg.get("transition") or "" pace = (f"Shot pacing — camera: {cam or 'static'}, transition into this panel: {trans or 'cut'}. " "Match your line's rhythm to the shot: a hard cut or tight/fast camera wants a short, " "punchy sentence; a held or slowly drifting shot or a soft dissolve can run a little " "longer and calmer.\n") if (cam or trans) else "" # #2 no re-introductions once a character has appeared. intro = ("Already introduced — refer to them by name or pronoun, do NOT re-describe their " "appearance: " + ", ".join(introduced) + "\n") if introduced else "" # scene-level narration: one beat can span several panels, so scale the budget with the beat size # (one flowing recap of the whole moment, not a line per panel). solo panel keeps a tight cap. if panel_count > 1: unit = f"this {panel_count}-panel manga beat (one continuous moment)" budget = f"{min(1 + panel_count, 4)} short sentences, max ~{min(20 * panel_count, 60)} words" thin = "beat" else: unit, budget, thin = "this manga panel", "1-2 short sentences, max ~25 words", "panel" return ( ov + rec + ctx + intro + f"Write TIGHT recap narration for {unit}: {budget}, " "present tense. Tell it like you're recapping to a friend — natural, with momentum. Output " "only the narration — no preamble, no notes, no alternatives, never ask for more info. If the " f"{thin} has little dialogue or action, one brief scene-setting sentence from what is given.\n" "Voice rules (avoid machinery):\n" "- Lead with what happens or what's said. When a character speaks, use their ACTUAL words in " "a short quote — don't paraphrase into 'he remarks that…' / 'she asks if…'.\n" "- Do NOT open with a physical-action gerund ('Looking up,', 'Adjusting his glasses,', " "'Leaning forward,'). Vary sentence openings.\n" "- Do NOT invent intent or filler ('prepares to…', 'ready to assert…', 'seeking " "confirmation'). Only what is actually shown.\n" "- Refer to characters by the names below. For anyone WITHOUT a real name, use a SHORT handle " "(e.g. 'the guy in glasses') at most ONCE, then 'he'/'she' — never restate their full " "appearance from panel to panel.\n" "- A thought reads as internal, a shout as raised. Caption/box text is a scene fact — weave " "it in; never say 'a narration box' or 'the screen displays'.\n" + (f"Spell these proper nouns exactly: {ents}\n" if ents else "") + "\n" + pace + f"Characters present: {chars}\n" f"Dialogue:\n{dialogue}\n" f"Action: {sg.get('action','')}\n\n" "Narration:" ) _SYSTEM = ("You are a manga recap narrator. You output only the finished narration paragraph — " "never your reasoning, drafts, options, or requests for more information.") def call_gemma4(prompt: str, system: str = _SYSTEM, max_tokens: int = 160, temperature: float = 0.6) -> str: payload = {"messages": [{"role": "system", "content": system}, {"role": "user", "content": prompt}], "temperature": temperature, "max_tokens": max_tokens} r = requests.post(f"{GEMMA4_URL}/v1/chat/completions", json=payload, timeout=300) r.raise_for_status() return _strip_thought(r.json()["choices"][0]["message"]["content"]) def build_summary_prompt(prior: str, recent) -> str: prior_s = prior.strip() or "(nothing yet — this is the start of the chapter)" new = "\n".join(f"- {r}" for r in recent if r) or "- (none)" return ( "You maintain a running 'story so far' summary for a manga recap. Given the previous " "summary and the narration of the panels since, produce an UPDATED summary in 3-6 " "sentences: compact and factual, consistent names/places, fold in the new events and drop " "stale detail. Only what has happened so far — never speculate ahead. Output only the " "summary.\n\n" f"Previous summary:\n{prior_s}\n\nNewly narrated:\n{new}\n\nUpdated summary:" ) @app.post("/script/summary") async def summary(data: SummaryInput): return {"brief": call_gemma4(build_summary_prompt(data.prior, data.recent))} @app.post("/script") async def script(data: ScriptInput): text = call_gemma4(build_prompt(data.scene_graph, data.chapter_context, data.names_by_id, data.brief, data.recent, data.introduced, data.genders_by_id, data.panel_count)) return {"panel_id": data.panel_id, "text": text} # --- roster normalization (spec E2-lite): dedup the accumulated cast+entities in one pass --- _NORM_SYSTEM = ("You clean up a manga's cast and entity lists. You output ONLY one JSON object, no " "prose, no reasoning.") class NormalizeInput(BaseModel): names: list = [] # canonical character names already in the registry (distinct people) entities: list = [] # [{"name","kind"}] accumulated over the chapter, full of near-dups session_id: str = "" def _extract_json(raw: str) -> dict: text = _strip_thought(raw) m = re.search(r"\{.*\}", text, re.DOTALL) if not m: raise ValueError(f"no json: {text[:200]}") return json.loads(m.group(0)) def build_normalize_prompt(names, entities) -> str: ents = "\n".join(f"- {e.get('name','')} ({e.get('kind','')})" for e in entities) or "- (none)" nms = ", ".join(n for n in names if n) or "(none)" return ( "Clean up these places/organizations/brands extracted from a manga chapter (may be noisy, " "duplicated). Do THREE things:\n" "1. MERGE entries that are the same thing into ONE canonical spelling — casing/spacing " "variants and truncations (e.g. 'Product Planning' and 'Product Planning Team' -> one).\n" "2. KEEP genuinely different things separate — do NOT merge distinct brands (e.g. EGENPICK " "and TETOPICK stay two).\n" "3. DROP non-entities: comparison phrases ('Teto Pick vs Ege'), fragments, and any entry that " "is actually one of the CHARACTERS listed below (a person's name is not a place/org).\n" "Also propose casing fixes for the character names (Title Case proper nouns; NEVER merge two " "different people — only fix the spelling of the SAME name).\n\n" f"Characters (people — for reference; drop entities that are actually these): {nms}\n" f"Entities:\n{ents}\n\n" 'Respond with ONLY this JSON:\n' '{"entities":[{"name":"Everyday","kind":"shop"}],"name_fixes":{"CHOI HAESEON":"Choi Haeseon"}}' ) @app.post("/normalize") async def normalize(data: NormalizeInput): prompt = build_normalize_prompt(data.names, data.entities) try: res = _extract_json(call_gemma4(prompt, system=_NORM_SYSTEM, max_tokens=768, temperature=0.1)) except (ValueError, json.JSONDecodeError): return {"entities": data.entities, "name_fixes": {}} # fail safe: leave the roster unchanged res.setdefault("entities", data.entities) res.setdefault("name_fixes", {}) return res @app.get("/health") async def health(): return {"status": "ok"} if __name__ == "__main__": # self-check: prompt includes character names + quoted dialogue. sg = {"characters": [{"id": "c1", "name": "Gojo Satoru"}, {"id": "character_deadbeef", "label": "the one with white hair"}, {"id": "character_nolabel"}], "dialogue": [{"speaker": "c1", "type": "shout", "text": "stand proud"}, {"speaker": "character_deadbeef", "type": "speech", "text": "no"}, {"speaker": "character_nolabel", "type": "speech", "text": "hm"}, {"speaker": "", "type": "narration", "text": "Years passed."}], "entities": [{"name": "Jujutsu High", "kind": "org"}], "action": "approaches"} p = build_prompt(sg, "prior scene") assert "Gojo Satoru shouts \"stand proud\"" in p assert 'the one with white hair says "no"' in p # unnamed + label -> descriptive phrase assert 'Person A says "hm"' in p # unnamed, no label -> positional fallback assert "character_deadbeef" not in p # raw db id must never reach the model # A backfill: a name learned later (registry) overrides the scene-time label for THIS panel pb = build_prompt(sg, "prior scene", {"character_deadbeef": "Nanami"}) assert 'Nanami says "no"' in pb and "the one with white hair" not in pb assert 'Caption (scene fact' in p and '"Years passed."' in p assert "Jujutsu High" in p and "prior scene" in p # scene-level narration: a multi-panel beat scales the budget and talks about a "beat", not a panel. assert "this manga panel" in p and "25 words" in p # solo default pbeat = build_prompt(sg, "prior scene", panel_count=3) assert "3-panel manga beat" in pbeat and "60 words" in pbeat and "beat has little" in pbeat # camera->pacing renders only when the graph carries a shot; absent by default. assert "Shot pacing" not in p pcam = build_prompt({**sg, "camera": {"effect": "dolly_to_subject"}, "transition": "dissolve"}, "s") assert "dolly_to_subject" in pcam and "dissolve" in pcam and "Shot pacing" in pcam # anti-repetition names structure, not just facts pr = build_prompt(sg, "s", recent=["He drew his blade."]) assert "sentence structure" in pr # F2 gender -> pronoun annotation on the characters-present line pg = build_prompt(sg, "prior scene", {"c1": "Gojo Satoru"}, genders_by_id={"c1": "m"}) assert "Gojo Satoru (he)" in pg # #1/#2/#3: brief (background), sliding window, introduced-set all render into the prompt pc = build_prompt(sg, "prior scene", brief="A duel unfolds.", recent=["He drew his blade.", "The crowd fell silent."], introduced=["Gojo Satoru"]) assert "A duel unfolds" in pc and "Story so far" in pc assert "- He drew his blade." in pc and "The crowd fell silent." in pc assert "Already introduced" in pc and "Gojo Satoru" in pc # #3 incremental: prior summary + new narration -> updated-summary prompt sp = build_summary_prompt("Hero left the village.", ["He reached the city gate."]) assert "Hero left the village." in sp and "He reached the city gate." in sp assert "Updated summary:" in sp sp0 = build_summary_prompt("", []) # cold start, no narration yet assert "nothing yet" in sp0 # normalization: prompt lists the entities + names; parser reads back a canonical roster np = build_normalize_prompt(["Lim Seonho"], [{"name": "Product Planning", "kind": "team"}, {"name": "Product Planning Team", "kind": "org"}]) assert "Product Planning Team" in np and "Lim Seonho" in np and "MERGE" in np nr = _extract_json('{"entities":[{"name":"Everyday","kind":"shop"}],' '"name_fixes":{"CHOI HAESEON":"Choi Haeseon"}}') assert nr["entities"][0]["name"] == "Everyday" and nr["name_fixes"]["CHOI HAESEON"] == "Choi Haeseon" print("worker_script self-check ok")