manga-recap-pipeline/worker_scene.py

# worker_scene.py — stage 6 scene graph. FastAPI :8004. no GPU, no session (cheap join).
# joins vision + identity into a named-character scene graph. speaker attribution comes straight from
# the vision/dialogue model (tail direction + turn-taking); the old OCR nearest-bbox heuristic is gone.
import json, logging
from fastapi import FastAPI
from pydantic import BaseModel
import transport

log = logging.getLogger("scene")


def _describe(appearance) -> str:
    """A: a stable descriptive phrase for an unnamed character, from its FIRST-seen appearance
    (frozen in the registry). "the one with black hair and glasses" — reused across panels so the
    label never drifts; empty string if appearance is bare (script then falls back to Person X)."""
    if isinstance(appearance, str):
        try:
            appearance = json.loads(appearance or "{}")
        except ValueError:
            appearance = {}
    if not isinstance(appearance, dict):
        return ""
    # animals/creatures get named as their species, not by clothes ("the black cat", not "the one
    # with a gold bell"). spec A: story-state tracks "props/entities seen (the cat)".
    species = (appearance.get("species") or "human").strip().lower()
    if species not in ("", "human", "person", "man", "woman"):
        color = (appearance.get("hair") or "").strip()
        feats = [str(f).strip() for f in (appearance.get("features") or []) if str(f).strip()]
        color = color or (feats[0] if feats else "")
        return f"the {color} {species}".replace("  ", " ").strip()
    hair = (appearance.get("hair") or "").strip()
    clothing = (appearance.get("clothing") or "").strip()
    feats = [str(f).strip() for f in (appearance.get("features") or []) if str(f).strip()]
    bits = []
    if hair:
        # vision sometimes already includes the word "hair" ("brown hair") -> don't double it.
        bits.append(hair if hair.lower().endswith("hair") else f"{hair} hair")
    if feats:
        bits.append(feats[0])
    elif clothing:
        # vision may still return a compound garment ("yellow shirt and yellow jacket");
        # keep only the first clause so the label stays "the one with a yellow shirt".
        bits.append(clothing.split(" and ")[0].strip())
    return "the one with " + " and ".join(bits) if bits else ""


class SceneInput(BaseModel):
    panel_id: str = ""
    panel_uri: str = ""
    vision_result: dict = {}
    identity_assignments: list = []
    characters_registry: list = []  # [{"character_id","name"}]


def build_scene(data: SceneInput):
    name_by_id = {c["character_id"]: c.get("name", "") for c in data.characters_registry}
    desc_by_id = {c["character_id"]: c.get("description") for c in data.characters_registry}
    id_by_local = {a["local_id"]: a["character_id"] for a in data.identity_assignments}

    characters, present = [], []
    for vc in data.vision_result.get("characters", []):
        cid = id_by_local.get(vc.get("local_id"))
        if not cid:
            continue
        name = name_by_id.get(cid, "")
        # frozen registry appearance first (stable across panels); fall back to this panel's vision.
        label = name or _describe(desc_by_id.get(cid)) or _describe(vc.get("appearance"))
        characters.append({"id": cid, "name": name, "label": label})
        present.append({"character_id": cid, "name": name, "bbox": vc.get("bbox", [0, 0, 0, 0])})

    # Speaker attribution is the vision/dialogue model's job now (it reads bubble tails + turn-taking).
    # speaker is a local_id -> map to character_id via identity; None for narration/sfx/off-panel.
    # The orchestrator's normalize_dialogue already rewrote `speaker` to a character_id and put the
    # typed answer in `speaker_ref`, so the local-id map no longer matches it. Read speaker_ref FIRST;
    # every lookup used to miss and every line narrated as "Someone".
    dialogue = []
    for d in data.vision_result.get("dialogue", []):
        raw = d.get("speaker")
        ref = d.get("speaker_ref") or {}
        # "unknown" is a DELIBERATE off-panel/indeterminate speaker (narrated as "someone"); it maps
        # to None just like narration. id_by_local.get already yields None for it.
        if ref.get("kind") == "character_id" and ref.get("value"):
            cid = ref["value"]
        else:
            cid = id_by_local.get(raw) or (raw if raw in name_by_id else None)
        dialogue.append({"speaker": cid, "text": d.get("text", ""),
                         "type": d.get("type", "speech"),
                         "confidence": d.get("confidence"),
                         "speaker_method": d.get("speaker_method", "unknown")})

    vchars = data.vision_result.get("characters", [])
    # `actions` is the list the orchestrator's beat builder reads for verifier evidence; `action` is the
    # joined string the script prompt renders. Emitting only the string left verification blind.
    actions = [c["action"].strip() for c in vchars if str(c.get("action") or "").strip()]
    action = "; ".join(actions)
    return {"panel_id": data.panel_id, "characters": characters, "dialogue": dialogue,
            "action": action, "actions": actions,
            "entities": data.vision_result.get("entities", []),
            "camera": data.vision_result.get("camera", {}),  # #8 direction passthrough -> render
            "transition": data.vision_result.get("transition", "cut")}  # #6 transition out


app = FastAPI()
transport.install_logging(app, "scene")


@app.post("/scene/build")
async def scene(data: SceneInput):
    return build_scene(data)


@app.get("/health")
async def health():
    return {"status": "ok"}


if __name__ == "__main__":
    out = build_scene(SceneInput(
        panel_id="p001",
        vision_result={"characters": [{"local_id": "person_1", "bbox": [0, 0, 10, 10], "action": "waving"}]},
        identity_assignments=[{"local_id": "person_1", "character_id": "c1"}],
        characters_registry=[{"character_id": "c1", "name": "Teto"}],
    ))
    assert out["characters"] == [{"id": "c1", "name": "Teto", "label": "Teto"}]

    # A: unnamed character -> stable descriptive label from frozen registry appearance
    outu = build_scene(SceneInput(
        panel_id="p003",
        vision_result={"characters": [{"local_id": "person_1", "bbox": [0, 0, 10, 10]}]},
        identity_assignments=[{"local_id": "person_1", "character_id": "c9"}],
        characters_registry=[{"character_id": "c9", "name": "",
                              "description": '{"hair":"black","features":["glasses"]}'}],
    ))
    assert outu["characters"][0]["name"] == "" and outu["characters"][0]["label"] == "the one with black hair and glasses"
    assert _describe({}) == ""     # bare appearance -> empty, script falls back to Person X
    # animals are named as their species (color + species), never by clothing/"the one with..."
    assert _describe({"species": "cat", "hair": "black"}) == "the black cat"
    assert _describe({"species": "cat", "features": ["gold bell"]}) == "the gold bell cat"
    assert _describe({"species": "cat"}) == "the cat"
    assert _describe({"species": "human", "hair": "black"}) == "the one with black hair"
    # compound garment collapses to the first clause (no "shirt and jacket" redundancy)
    assert _describe({"clothing": "yellow shirt and yellow jacket"}) == "the one with yellow shirt"
    assert out["action"] == "waving"

    # vision-owned dialogue: typed, proper-cased, speaker mapped local_id -> character_id
    out2 = build_scene(SceneInput(
        panel_id="p002",
        vision_result={
            "characters": [{"local_id": "person_1", "bbox": [0, 0, 10, 10]}],
            "dialogue": [{"speaker": "person_1", "type": "thought", "text": "So this is it."},
                         {"speaker": "", "type": "narration", "text": "Three years later."}],
            "entities": [{"name": "Everyday", "kind": "shop"}],
        },
        identity_assignments=[{"local_id": "person_1", "character_id": "c1"}],
        characters_registry=[{"character_id": "c1", "name": "Teto"}],
    ))
    assert out2["dialogue"][0]["speaker"] == "c1" and out2["dialogue"][0]["type"] == "thought"
    assert out2["dialogue"][1]["speaker"] is None and out2["dialogue"][1]["type"] == "narration"
    assert out2["entities"][0]["name"] == "Everyday"

    # explicit "unknown" (off-panel/indeterminate) maps to None -> narrated as "someone", never
    # pinned to a visible character. (the "line defaults to the MC" bug.)
    out4 = build_scene(SceneInput(
        panel_id="p005b",
        vision_result={
            "characters": [{"local_id": "person_1", "bbox": [0, 0, 10, 10]}],
            "dialogue": [{"speaker": "unknown", "type": "speech", "text": "A Teto and Egen test?"}],
        },
        identity_assignments=[{"local_id": "person_1", "character_id": "c1"}],
        characters_registry=[{"character_id": "c1", "name": "MC"}],
    ))
    assert out4["dialogue"][0]["speaker"] is None, out4["dialogue"][0]
    # Attribution provenance is load-bearing: a weak guess must remain distinguishable downstream.
    out5 = build_scene(SceneInput(
        panel_id="p006",
        vision_result={"dialogue": [{"speaker": "person_1", "type": "speech", "text": "Maybe.",
                                      "confidence": 0.3, "speaker_method": "turn_taking"}]},
        identity_assignments=[{"local_id": "person_1", "character_id": "c1"}],
    ))
    assert out5["dialogue"][0]["confidence"] == 0.3
    assert out5["dialogue"][0]["speaker_method"] == "turn_taking"

    # orchestrator-normalized rows: `speaker` is already a character_id and `speaker_ref` is typed.
    # the local-id map cannot resolve either — speaker_ref must win, or everything narrates as Someone.
    out6 = build_scene(SceneInput(
        panel_id="p007",
        vision_result={"characters": [{"local_id": "person_1", "bbox": [0, 0, 10, 10]}],
                       "dialogue": [{"speaker": "c1", "speaker_ref": {"kind": "character_id", "value": "c1"},
                                     "type": "speech", "text": "Hey."},
                                    {"speaker": None, "speaker_ref": {"kind": "unknown", "value": None},
                                     "type": "speech", "text": "..."}]},
        identity_assignments=[{"local_id": "person_1", "character_id": "c1"}],
        characters_registry=[{"character_id": "c1", "name": "Teto"}],
    ))
    assert out6["dialogue"][0]["speaker"] == "c1", out6["dialogue"][0]
    assert out6["dialogue"][1]["speaker"] is None, out6["dialogue"][1]
    # actions survive as a list for the verifier, not only as the joined prompt string
    assert out["actions"] == ["waving"] and out["action"] == "waving"
    print("worker_scene self-check ok")