config_loader.py

Code Hygiene Score: 91

Keine Issues gefunden.

Dependencies 3

Funktionen 8

Code

"""
Configuration Loading Functions - Load prompts, profiles, contracts, structures.
"""

import json
import sys

sys.path.insert(0, "/var/www/scripts/pipeline")

from db import db


def get_prompt(name: str) -> str | None:
    """Load prompt from database by name."""
    cursor = db.execute(
        """SELECT content FROM prompts WHERE name = %s AND is_active = 1 ORDER BY version DESC LIMIT 1""",
        (name,),
    )
    result = cursor.fetchone()
    cursor.close()
    return result["content"] if result else None


def get_config_item(item_id: int, item_type: str) -> dict | None:
    """Load configuration item from content_config table."""
    if not item_id:
        return None

    cursor = db.execute(
        "SELECT name, content FROM content_config WHERE id = %s AND type = %s AND status = 'active'",
        (item_id, item_type),
    )
    result = cursor.fetchone()
    cursor.close()

    if result:
        config = json.loads(result["content"]) if isinstance(result["content"], str) else result["content"]
        return {"name": result["name"], "config": config}
    return None


def get_author_profile(profile_id: int) -> dict | None:
    """Load author profile from database."""
    return get_config_item(profile_id, "author_profile")


def get_contract(contract_id: int) -> dict | None:
    """Load content contract from database."""
    return get_config_item(contract_id, "contract")


def get_structure(structure_id: int) -> dict | None:
    """Load content structure from database."""
    result = get_config_item(structure_id, "structure")
    if result:
        # Structure has additional 'type' field in config
        result["type"] = result["config"].get("type", "article")
    return result


def get_order(order_id: int) -> dict | None:
    """Load content order with all related data."""
    cursor = db.execute(
        """SELECT co.*,
              ap.name as profile_name, ap.content as profile_config,
              cc.name as contract_name, cc.content as contract_config,
              cs.name as structure_name, cs.content as structure_config
           FROM content_orders co
           LEFT JOIN content_config ap ON co.author_profile_id = ap.id AND ap.type = 'author_profile'
           LEFT JOIN content_config cc ON co.contract_id = cc.id AND cc.type = 'contract'
           LEFT JOIN content_config cs ON co.structure_id = cs.id AND cs.type = 'structure'
           WHERE co.id = %s""",
        (order_id,),
    )
    result = cursor.fetchone()
    cursor.close()
    return result


def parse_author_profile(config: dict) -> str:
    """Parse new-style author profile (Cary format) into prompt text."""
    sections = []

    # Haltung
    haltung = config.get("haltung", {})
    if haltung:
        sections.append(f"""### Haltung:
- Grundhaltung: {haltung.get("grundhaltung", "")}
- Ausrichtung: {haltung.get("ausrichtung", "")}
- Spannungstoleranz: {haltung.get("spannungstoleranz", "")}
- Vereinfachung: {haltung.get("vereinfachung", "")}""")

    # Tonalität
    tonalitaet = config.get("tonalitaet", {})
    if tonalitaet:
        sections.append(f"""### Tonalität:
- Charakter: {tonalitaet.get("charakter", "")}
- Stil: {tonalitaet.get("stil", "")}
- Wirkung: {tonalitaet.get("wirkung", "")}
- Abgrenzung: {tonalitaet.get("abgrenzung", "")}""")

    # Sprachmodus
    sprachmodus = config.get("sprachmodus", {})
    if sprachmodus:
        sections.append(f"""### Sprachmodus:
- Denkstil: {sprachmodus.get("denkstil", "")}
- Aussagenform: {sprachmodus.get("aussagenform", "")}
- Fragenfunktion: {sprachmodus.get("fragenfunktion", "")}""")

    # Grammatik und Satzbau - WICHTIG für Verbote
    grammatik = config.get("grammatik_und_satzbau", {})
    if grammatik:
        verbote = []
        if grammatik.get("stakkato") == "ausgeschlossen":
            verbote.append("Stakkato-Sätze")
        if grammatik.get("einschuebe") == "keine":
            verbote.append("Einschübe")
        if grammatik.get("gedankenstriche") == "verboten":
            verbote.append("Gedankenstriche (–)")

        sections.append(f"""### Grammatik und Satzbau:
- Sätze: {grammatik.get("saetze", "")}
- Rhythmus: {grammatik.get("rhythmus", "")}
- VERBOTEN: {", ".join(verbote) if verbote else "keine"}""")

    # Wortwahl
    wortwahl = config.get("wortwahl", {})
    if wortwahl:
        verboten = []
        if wortwahl.get("buzzwords") == "ausgeschlossen":
            verboten.append("Buzzwords")
        if wortwahl.get("methodennamen") == "ausgeschlossen":
            verboten.append("Methodennamen")

        sections.append(f"""### Wortwahl:
- Niveau: {wortwahl.get("niveau", "")}
- Begriffe: {wortwahl.get("begriffe", "")}
- VERBOTEN: {", ".join(verboten) if verboten else "keine"}""")

    # Adressierung
    adressierung = config.get("adressierung", {})
    if adressierung:
        sections.append(f"""### Adressierung:
- Form: {adressierung.get("form", "Sie")}
- Beziehung: {adressierung.get("beziehung", "")}
- Einladung: {adressierung.get("einladung", "")}""")

    # Metaphern
    metaphern = config.get("metaphern", {})
    if metaphern:
        sections.append(f"""### Metaphern:
- Einsatz: {metaphern.get("einsatz", "")}
- Herkunft: {metaphern.get("herkunft", "")}
- Konsistenz: {metaphern.get("konsistenz", "")}""")

    return "\n\n".join(sections)


def parse_structure(config: dict) -> tuple[str, str, list[str]]:
    """Parse structure profile into prompt text and format info.

    Supports both formats:
    - Blog/HTML format: gesamtaufbau, einstieg, hauptteil.bloecke, schluss, formatierung.ausschluss
    - LinkedIn format: aufbau.hook, aufbau.hauptteil, aufbau.schluss, formatierung.*, verboten[]
    """
    sections = []
    output_format = None
    erlaubte_tags = []

    # WICHTIG field (LinkedIn format)
    wichtig = config.get("WICHTIG")
    if wichtig:
        sections.append(f"### WICHTIG:\n{wichtig}")

    # Ausgabe-Format
    ausgabe = config.get("ausgabe", {})
    if ausgabe:
        output_format = ausgabe.get("format", "markdown")
        erlaubte_tags = ausgabe.get("erlaubte_tags", [])
        verbotene_tags = ausgabe.get("verbotene_tags", [])

        if output_format == "body-html":
            sections.append(f"""### Ausgabe-Format: HTML
- **Nur diese Tags verwenden:** {", ".join(erlaubte_tags)}
- **Verboten:** {", ".join(verbotene_tags)}
- {ausgabe.get("hinweis", "Sauberes semantisches HTML")}""")
        elif output_format == "reiner Text":
            # LinkedIn plain text format
            zeichen_info = ""
            if ausgabe.get("zeichen_min") and ausgabe.get("zeichen_max"):
                zeichen_info = f"\n- Länge: {ausgabe.get('zeichen_min')}-{ausgabe.get('zeichen_max')} Zeichen"
            if ausgabe.get("woerter"):
                zeichen_info += f" ({ausgabe.get('woerter')} Wörter)"
            sections.append(f"""### Ausgabe-Format: Reiner Text
- KEIN Markdown, KEINE Formatierung
- Keine **Fettschrift**, keine *Kursivschrift*
- Keine Überschriften mit # oder ##
- Nur Fließtext und einfache Zeilenumbrüche{zeichen_info}""")

    # Try Blog format first: gesamtaufbau
    aufbau = config.get("gesamtaufbau", {})
    if aufbau:
        sections.append(f"""### Gesamtaufbau:
- Form: {aufbau.get("form", "")}
- Dramaturgie: {aufbau.get("dramaturgie", "")}
- Linearität: {aufbau.get("linearitaet", "")}
- Themensprünge: {aufbau.get("themenspruenge", "")}""")

    # Try LinkedIn format: aufbau (with nested hook, hauptteil, etc.)
    linkedin_aufbau = config.get("aufbau", {})
    if linkedin_aufbau and not aufbau:  # Only if gesamtaufbau not present
        aufbau_sections = []

        # Hook
        hook = linkedin_aufbau.get("hook", {})
        if hook:
            hook_text = f"### Hook ({hook.get('position', 'Anfang')}):\n"
            hook_text += f"- Stil: {hook.get('stil', '')}\n"
            if hook.get("beispiele"):
                hook_text += f"- Beispiele: {'; '.join(hook['beispiele'][:2])}\n"
            if hook.get("verboten"):
                hook_text += f"- Verboten: {', '.join(hook['verboten'])}"
            aufbau_sections.append(hook_text)

        # Praxisbezug
        praxis = linkedin_aufbau.get("praxisbezug", {})
        if praxis:
            praxis_text = f"### Praxisbezug ({praxis.get('position', '')}):\n"
            praxis_text += f"- Stil: {praxis.get('stil', '')}\n"
            if praxis.get("muster"):
                praxis_text += f"- Muster: {'; '.join(praxis['muster'][:2])}"
            aufbau_sections.append(praxis_text)

        # Hauptteil
        hauptteil = linkedin_aufbau.get("hauptteil", {})
        if hauptteil:
            hauptteil_text = "### Hauptteil:\n"
            if hauptteil.get("struktur"):
                hauptteil_text += f"- Struktur: {hauptteil.get('struktur')}\n"
            if hauptteil.get("theoretische_einordnung"):
                hauptteil_text += f"- Theorie: {hauptteil.get('theoretische_einordnung')}\n"
            if hauptteil.get("differenzierung"):
                hauptteil_text += f"- Differenzierung: {hauptteil.get('differenzierung')}\n"
            if hauptteil.get("dreier_struktur"):
                hauptteil_text += f"- Dreier-Struktur: {hauptteil.get('dreier_struktur')}"
            aufbau_sections.append(hauptteil_text)

        # Handlungsempfehlung
        handlung = linkedin_aufbau.get("handlungsempfehlung", {})
        if handlung:
            handlung_text = "### Handlungsempfehlung:\n"
            handlung_text += f"- Stil: {handlung.get('stil', '')}\n"
            if handlung.get("muster"):
                handlung_text += f"- Muster: {'; '.join(handlung['muster'][:2])}"
            aufbau_sections.append(handlung_text)

        # Schluss (LinkedIn style)
        schluss = linkedin_aufbau.get("schluss", {})
        if schluss:
            schluss_text = "### Schluss:\n"
            if schluss.get("soft_cta"):
                schluss_text += f"- Stil: {schluss.get('soft_cta')}\n"
            if schluss.get("muster"):
                schluss_text += f"- Muster: {'; '.join(schluss['muster'][:2])}\n"
            if schluss.get("verboten"):
                schluss_text += f"- Verboten: {', '.join(schluss['verboten'])}"
            aufbau_sections.append(schluss_text)

        sections.extend(aufbau_sections)

    # Blog format: Einstieg
    einstieg = config.get("einstieg", {})
    if einstieg:
        sections.append(f"""### Einstieg:
- Funktion: {einstieg.get("funktion", "")}
- Inhaltstyp: {einstieg.get("inhaltstyp", "")}
- Ausschluss: {einstieg.get("ausschluss", "")}""")

    # Blog format: Hauptteil-Blöcke
    hauptteil = config.get("hauptteil", {})
    bloecke = hauptteil.get("bloecke", [])
    if bloecke:
        bloecke_text = "\n".join(
            [f"  {i + 1}. {b.get('fokus', '')} → {b.get('ziel', '')}" for i, b in enumerate(bloecke)]
        )
        sections.append(f"""### Hauptteil-Struktur:
{bloecke_text}""")

    # Blog format: Schluss
    schluss = config.get("schluss", {})
    if schluss:
        sections.append(f"""### Schluss:
- Typ: {schluss.get("typ", "")}
- Funktion: {schluss.get("funktion", "")}
- Abgrenzung: {schluss.get("abgrenzung", "")}""")

    # Formatierung - handle both styles
    formatierung = config.get("formatierung", {})
    if formatierung:
        # Blog format: ausschluss array
        ausschluss = formatierung.get("ausschluss", [])
        if ausschluss:
            sections.append(f"""### Formatierung verboten:
{", ".join(ausschluss)}""")

        # LinkedIn format: individual fields
        verbote = []
        if formatierung.get("emojis") == "verboten" or "verboten" in str(formatierung.get("emojis", "")):
            verbote.append("Emojis")
        if formatierung.get("fettschrift") == "verboten" or "verboten" in str(formatierung.get("fettschrift", "")):
            verbote.append("Fettschrift (**text**)")
        if formatierung.get("markdown") == "verboten" or "verboten" in str(formatierung.get("markdown", "")):
            verbote.append("Markdown-Formatierung")
        if formatierung.get("hashtags") == "keine":
            verbote.append("Hashtags")

        if verbote and not ausschluss:  # Only add if not already handled by ausschluss
            sections.append(f"""### Formatierung STRIKT verboten:
{", ".join(verbote)}""")

    # Top-level verboten array (LinkedIn format)
    verboten = config.get("verboten", [])
    if verboten:
        verboten_list = "\n".join([f"- {v}" for v in verboten])
        sections.append(f"""### VERBOTEN (strikt einhalten!):
{verboten_list}""")

    return "\n\n".join(sections), output_format, erlaubte_tags
← Übersicht