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

#!/usr/bin/env python3

import re
import sqlite3
from collections import Counter, defaultdict

DB = "data/zammad_analysis.sqlite3"

AUTO_PATTERNS = [
    r"wir haben ihre anfrage erhalten",
    r"ihre anfrage wurde",
    r"ticket.{0,30}(erstellt|angelegt|eröffnet)",
    r"wurde.{0,30}(zugewiesen|weitergeleitet)",
    r"wird.{0,80}(bearbeitet|geprüft)",
    r"antworten sie bitte direkt auf diese e-mail",
    r"dies ist eine automatische",
    r"automatisch erstellt",
    r"automatische nachricht",
    r"vielen dank für ihre anfrage",
]

AGENT_PATTERNS = [
    r"wie besprochen",
    r"wie vereinbart",
    r"wie telefonisch",
    r"ich habe",
    r"wir haben.{0,80}(erstellt|angepasst|geändert|veranlasst)",
    r"wir senden",
    r"wir schicken",
    r"wir erstatten",
    r"wir werden ihnen",
    r"ich lasse ihnen",
    r"ich habe ihnen",
    r"gutschein",
    r"gutscheincode",
    r"kulanz",
    r"nachforschungsauftrag",
    r"nachlieferung",
    r"ersatzpflanze",
    r"erstattung",
    r"rückerstattung",
    r"stornorechnung",
    r"rücksendelabel",
    r"retourenlabel",
    r"paketnummer",
    r"sendungsnummer",
    r"tracking",
    r"mit freundlichen grüßen",
    r"viele grüße",
    r"freundliche grüße",
]

auto_re = [re.compile(x, re.I | re.S) for x in AUTO_PATTERNS]
agent_re = [re.compile(x, re.I | re.S) for x in AGENT_PATTERNS]

db = sqlite3.connect(DB)

# Spalten nur anlegen, wenn sie noch nicht existieren.
columns = {
    row[1]
    for row in db.execute("PRAGMA table_info(articles)")
}

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

if "classification_score" not in columns:
    db.execute("""
        ALTER TABLE articles
        ADD COLUMN classification_score INTEGER
    """)

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

db.commit()

rows = db.execute("""
    SELECT
        id,
        ticket_id,
        sender,
        clean_body
    FROM articles
    WHERE role = 'UNKNOWN'
      AND clean_body IS NOT NULL
      AND clean_body != ''
""").fetchall()

print("=" * 72)
print("UNKNOWN → AGENT / AUTOMATION VORSORTIERUNG")
print("=" * 72)
print(f"UNKNOWN-Artikel: {len(rows)}")
print()

stats = Counter()
examples = defaultdict(list)

for article_id, ticket_id, sender, text in rows:

    text = text or ""

    auto_hits = [
        p.pattern
        for p in auto_re
        if p.search(text)
    ]

    agent_hits = [
        p.pattern
        for p in agent_re
        if p.search(text)
    ]

    score = 0

    # Automationssignale
    score -= min(len(auto_hits) * 3, 9)

    # Signale für individuell formulierte Agentenantwort
    score += min(len(agent_hits) * 2, 12)

    # Individuelle Antworten sind häufig etwas länger.
    if len(text) > 500:
        score += 1

    if len(text) > 1200:
        score += 1

    if score >= 3:
        content_role = "AGENT_RESPONSE"
        stats[content_role] += 1

    elif score <= -3:
        content_role = "AUTOMATION"
        stats[content_role] += 1

    else:
        content_role = "UNKNOWN_REVIEW"
        stats[content_role] += 1

    reason = (
        f"agent_hits={len(agent_hits)};"
        f"auto_hits={len(auto_hits)};"
        f"score={score}"
    )

    db.execute("""
        UPDATE articles
        SET content_role = ?,
            classification_score = ?,
            classification_reason = ?
        WHERE id = ?
    """, (
        content_role,
        score,
        reason,
        article_id,
    ))

    if len(examples[content_role]) < 12:
        examples[content_role].append((
            ticket_id,
            score,
            text[:800],
        ))

db.commit()

print("ERGEBNIS")
print("-" * 72)

for role, count in stats.most_common():
    print(f"{role:20} {count:6}")

for role in (
    "AGENT_RESPONSE",
    "AUTOMATION",
    "UNKNOWN_REVIEW",
):
    print()
    print("=" * 72)
    print(role)
    print("=" * 72)

    for ticket_id, score, text in examples[role]:
        print(
            f"\nTicket #{ticket_id} "
            f"score={score}"
        )
        print(text)

print()
print("=" * 72)
print("STATUSVERTEILUNG")
print("=" * 72)

for row in db.execute("""
    SELECT
        COALESCE(content_role, role),
        COUNT(*)
    FROM articles
    GROUP BY COALESCE(content_role, role)
    ORDER BY COUNT(*) DESC
"""):
    print(f"{row[0]:20} {row[1]:6}")

db.close()