Explorer
/tmp/add_carousel_functions.py
← Zurück ↓ Download
from pathlib import Path
p=Path('/opt/struktur/social-media-agent/publish_ready.py'); s=p.read_text()
needle='''def build_media_brief(package: dict, feedback_rules: list[dict], reference_asset: dict | None, generation_group_id: str) -> dict:'''
helper='''def assess_media_type(package: dict) -> dict:
    topic = _text(package.get("topic")).casefold(); pillar = _text(package.get("content_pillar") or package.get("topic_category")).casefold()
    carousel_terms = ("warum", "beginnt", "ursache", "diagnose", "schritt", "checkliste", "fehler", "messung", "zusammenhang", "plan")
    if any(term in topic or term in pillar for term in carousel_terms):
        return {"media_type": "carousel", "reason": "Mehrere zusammenhängende technische Punkte lassen sich als Ursache-Wirkung- und Vorgehensfolge erklären."}
    return {"media_type": "single_image", "reason": "Eine einzelne konkrete visuelle Aussage oder Messsituation steht im Vordergrund."}


def build_carousel_slides(package: dict) -> list[dict]:
    topic = _text(package.get("topic")); claim = _text(package.get("relevance") or package.get("final_text") or package.get("facebook_text"))
    return [
        {"slide_number": 1, "purpose": "fachlicher Hook", "statement": topic, "motif": "ruhiger technischer Innenraum mit klarer Gebäudediagnostik"},
        {"slide_number": 2, "purpose": "Problem/Ausgangssituation", "statement": "Welche Ausgangsbedingungen im konkreten Objekt geprüft werden müssen.", "motif": "Innenraum und betroffener Bauteilbereich ohne dramatische Inszenierung"},
        {"slide_number": 3, "purpose": "technischer Zusammenhang", "statement": claim or "Nutzung, Bauteile und Raumklima können gemeinsam bewertet werden.", "motif": "sichtbarer technischer Zusammenhang zwischen Bauteil und Raumklima"},
        {"slide_number": 4, "purpose": "Diagnose", "statement": "Messung und fachliche Einordnung trennen Vermutung und belastbare Entscheidung.", "motif": "plausible professionelle Mess- oder Prüfsituation"},
        {"slide_number": 5, "purpose": "Vorgehensweise/Schlussfolgerung", "statement": "Erst Bedingungen und Ursache prüfen, dann eine passende Maßnahme entscheiden.", "motif": "sachliche technische Dokumentation im Innenraum"},
    ]


def _validate_carousel_slides(slides: list[dict]) -> None:
    for slide in slides:
        text = _text(slide.get("statement"))
        if not text: raise ValueError(f"Carousel-Slide {slide.get('slide_number')} ohne Aussage")
        violation = _mas_policy_violation(text)
        if violation: raise ValueError(f"Carousel-Slide {slide.get('slide_number')} verletzt Content Policy: {violation}")
        if NUMERIC_CLAIM_RE.search(text): raise ValueError(f"Carousel-Slide {slide.get('slide_number')} enthält unbelegte Präzisionszahl")


def _render_carousel_slide_text(path: Path, slide: dict) -> None:
    from PIL import Image, ImageDraw, ImageFont
    with Image.open(path) as source:
        image = source.convert("RGB")
    draw = ImageDraw.Draw(image)
    try: font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", 32)
    except OSError: font = ImageFont.load_default()
    text = f"Slide {slide['slide_number']}\n{slide['statement']}"
    lines=[]
    for paragraph in text.split("\\n"):
        words=paragraph.split(); line=""
        for word in words:
            candidate=(line+" "+word).strip()
            if draw.textlength(candidate,font=font)>900 and line: lines.append(line); line=word
            else: line=candidate
        if line: lines.append(line)
    y=image.height-60-42*len(lines); draw.rectangle((35,y-18,image.width-35,image.height-30),fill=(15,17,23))
    for line in lines: draw.text((60,y),line,font=font,fill=(245,245,245)); y+=42
    image.save(path,format="PNG",optimize=True)


def generate_carousel_for_package(package_id: str) -> dict:
    con=sqlite3.connect(CASES_DB); con.row_factory=sqlite3.Row; row=con.execute("SELECT * FROM publication_packages WHERE package_id=?",(package_id,)).fetchone()
    if not row: con.close(); raise ValueError("publication package not found")
    package=dict(row); decision=assess_media_type(package)
    if decision["media_type"] != "carousel": con.close(); raise ValueError("Package ist nicht für Carousel geeignet")
    carousel_id=f"CAR-{uuid.uuid4()}"; feedback_text,feedback_rules=feedback_prompt_text(package); reference_asset=None
    try:
        ref_path,ref_label=internal_asset_for_package(package)
        if ref_path and ref_path.is_file(): reference_asset={"path":str(ref_path),"label":ref_label,"source":"media_catalog"}
    except Exception: pass
    brief=build_media_brief(package,feedback_rules,reference_asset,carousel_id); slides=build_carousel_slides(package); _validate_carousel_slides(slides); persist_media_brief(con,brief)
    con.execute("INSERT INTO carousels(carousel_id,package_id,media_brief_id,slide_count,media_type,media_type_reason,created_at) VALUES(?,?,?,?,?,?,?)",(carousel_id,package_id,brief["media_brief_id"],len(slides),"carousel",decision["reason"],_now())); con.commit(); reports=[]
    for slide in slides:
        prompt=media_brief_prompt(brief,1)+f" Carousel-Slide {slide['slide_number']} Zweck: {slide['purpose']}. Sichtbares Motiv: {slide['motif']}. Keine Schrift oder Zahlen im Bild; Text wird später deterministisch als Layout aufgebracht."
        target=(BASE_DIR/"generated_visuals"/f"{package['signal_id']}_{carousel_id[-8:]}_slide-{slide['slide_number']}.png").resolve()
        try:
            generated=generate_with_codex(prompt,target,reference_image=Path(reference_asset["path"]) if reference_asset else None); qa=run_image_qa(target,topic, _text(package.get("final_text") or package.get("facebook_text"))); _render_carousel_slide_text(target,slide); data=target.read_bytes(); sha=hashlib.sha256(data).hexdigest();
            con.execute("INSERT INTO carousel_slides(slide_id,carousel_id,package_id,slide_number,slide_count,purpose,statement,image_path,origin_path,sha256,file_bytes,generator,prompt,media_brief_id,qa_status,qa_report_json,qa_failure_reasons,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",(f"SLIDE-{uuid.uuid4()}",carousel_id,package_id,slide["slide_number"],len(slides),slide["purpose"],slide["statement"],str(target),generated["origin_path"],sha,len(data),"Codex built-in image_gen",prompt,brief["media_brief_id"],"passed" if qa_passes(qa) else "failed",json.dumps(qa,ensure_ascii=False),"; ".join(qa.get("failure_reasons",[])),_now())); reports.append({"slide":slide,"path":str(target),"sha256":sha,"bytes":len(data),"qa":qa})
        except (CodexImageError,CodexImageQAError,OSError,ValueError) as exc: reports.append({"slide":slide,"qa":{"overall":"FAIL","failure_reasons":[str(exc)]},"error":str(exc)})
        con.commit()
    con.execute("UPDATE publication_packages SET media_type='carousel',media_type_reason=?,human_visual_approval='offen' WHERE package_id=?",(decision["reason"],package_id)); con.commit(); con.close(); return {"carousel_id":carousel_id,"media_brief_id":brief["media_brief_id"],"slides":reports,"media_type":decision["media_type"]}


'''
assert s.count(needle)==1; s=s.replace(needle,helper+needle); p.write_text(s); print('carousel generation functions added')