3cx-telefonie-middleware/tools/rebuild_zammad_cases.py

#!/usr/bin/env python3

import re
import sqlite3

DB = "data/zammad_analysis.sqlite3"

db = sqlite3.connect(DB)
db.row_factory = sqlite3.Row

# Felder bei Bedarf anlegen
columns = {
    row[1]
    for row in db.execute("PRAGMA table_info(cases)")
}

if "analysis_status" not in columns:
    db.execute("""
        ALTER TABLE cases
        ADD COLUMN analysis_status TEXT
    """)

if "analysis_reason" not in columns:
    db.execute("""
        ALTER TABLE cases
        ADD COLUMN analysis_reason TEXT
    """)

db.commit()

# Eindeutig automatische/spamartige Inhalte
spam_patterns = [
    r"\bspam\b",
    r"\bwerbe[- ]?mail\b",
    r"\bmarketing[- ]?mail\b",
    r"\bunsubscribe\b",
    r"\bnewsletter\b",
]

auto_patterns = [
    r"wir haben ihre anfrage erhalten",
    r"ihre anfrage wurde.{0,80}(erhalten|erstellt|angelegt)",
    r"ticket.{0,80}(erstellt|angelegt|eröffnet)",
    r"ticket.{0,80}(verfolgen|verfolgung)",
    r"link.{0,80}(ticket|anfrage)",
    r"dies ist eine automatische",
    r"automatische nachricht",
    r"vielen dank für ihre anfrage",
]

spam_re = [re.compile(x, re.I | re.S) for x in spam_patterns]
auto_re = [re.compile(x, re.I | re.S) for x in auto_patterns]

tickets = db.execute("""
    SELECT ticket_id, ticket_number, title
    FROM cases
    ORDER BY ticket_id
""").fetchall()

stats = {}

def count(status):
    stats[status] = stats.get(status, 0) + 1


for pos, ticket in enumerate(tickets, 1):

    articles = db.execute("""
        SELECT
            role,
            content_role,
            analysis_body
        FROM articles
        WHERE ticket_id = ?
        ORDER BY id
    """, (ticket["ticket_id"],)).fetchall()

    customer = []
    agent = []

    for article in articles:

        text = (article["analysis_body"] or "").strip()

        if not text:
            continue

        if article["role"] == "CUSTOMER":
            customer.append(text)

        elif article["role"] == "AGENT":
            agent.append(text)

        elif article["content_role"] == "AGENT_RESPONSE":
            agent.append(text)

        # AUTOMATION / INTERNAL / sonstige UNKNOWN:
        # nicht in die Analyse übernehmen.

    customer_text = "\n\n".join(customer).strip()
    agent_text = "\n\n".join(agent).strip()

    # --------------------------------------------------------
    # Status bestimmen
    # --------------------------------------------------------

    if not customer_text:
        status = "AUTOMATION_ONLY"
        reason = "Kein verwertbarer Kundenbeitrag"

    elif (
        len(customer_text) < 30
        and not agent_text
    ):
        status = "NON_ANALYZABLE"
        reason = "Kundenbeitrag zu kurz"

    elif (
        any(p.search(customer_text) for p in auto_re)
        and len(customer_text) < 250
    ):
        status = "AUTOMATION_ONLY"
        reason = "Automatische Ticketnachricht"

    else:
        status = "ANALYZABLE"
        reason = "Verwertbarer Kundenfall"

    count(status)

    classification_status = (
        "NEW"
        if status == "ANALYZABLE"
        else "SKIP"
    )

    db.execute("""
        UPDATE cases
        SET
            customer_text = ?,
            agent_text = ?,
            unknown_text = '',
            analysis_status = ?,
            analysis_reason = ?,
            classification_status = ?
        WHERE ticket_id = ?
    """, (
        customer_text,
        agent_text,
        status,
        reason,
        classification_status,
        ticket["ticket_id"],
    ))

    if pos % 500 == 0:
        db.commit()
        print(
            f"[{pos}/{len(tickets)}] "
            f"ANALYZABLE={stats.get('ANALYZABLE', 0)} "
            f"SKIP={pos - stats.get('ANALYZABLE', 0)}",
            flush=True,
        )

db.commit()

print()
print("=" * 72)
print("ZAMMAD CASE REBUILD FERTIG")
print("=" * 72)

for status, number in sorted(
    stats.items(),
    key=lambda x: x[1],
    reverse=True,
):
    print(f"{status:24} {number:6}")

print()

row = db.execute("""
    SELECT COUNT(*)
    FROM cases
    WHERE analysis_status = 'ANALYZABLE'
""").fetchone()

print(
    f"Für Qwen vorgesehen: {row[0]}"
)

print()
print("Kontrolle #91005 / #91006")
print("-" * 72)

for row in db.execute("""
    SELECT
        ticket_number,
        title,
        analysis_status,
        analysis_reason,
        customer_text
    FROM cases
    WHERE ticket_number IN ('91005', '91006')
    ORDER BY ticket_number
"""):
    print(
        f"#{row['ticket_number']} "
        f"{row['analysis_status']} "
        f"({row['analysis_reason']})"
    )
    print(
        f"  {row['title'] or ''}"
    )
    print(
        f"  {(row['customer_text'] or '')[:300]}"
    )

db.close()