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

#!/usr/bin/env python3

import html
import re
import sqlite3
from pathlib import Path

SRC = Path("data/zammad_history.sqlite3")
DST = Path("data/zammad_analysis.sqlite3")


def clean_body(text):
    if not text:
        return ""

    text = html.unescape(text)

    # HTML entfernen
    text = re.sub(
        r"<(script|style).*?</\1>",
        " ",
        text,
        flags=re.I | re.S,
    )
    text = re.sub(r"<br\s*/?>", "\n", text, flags=re.I)
    text = re.sub(r"<[^>]+>", " ", text)

    # typische Mail-Header in Zitaten
    text = re.sub(
        r"\n\s*(Am .*? schrieb .*?:|"
        r"On .*? wrote:|"
        r"Gesendet: .*?\n|"
        r"Von: .*?\n|"
        r"From: .*?\n|"
        r"Betreff: .*?\n|"
        r"Subject: .*?\n)",
        "\n",
        text,
        flags=re.I,
    )

    # klassische Signaturen nur am Ende abschneiden
    text = re.split(
        r"\n\s*(Mit freundlichen Grüßen|"
        r"Viele Grüße|"
        r"Beste Grüße|"
        r"Freundliche Grüße)\b",
        text,
        maxsplit=1,
        flags=re.I,
    )[0]

    # Zitatzeilen
    lines = []
    for line in text.splitlines():
        if line.lstrip().startswith(">"):
            continue
        lines.append(line)

    text = "\n".join(lines)

    text = re.sub(r"[ \t]+", " ", text)
    text = re.sub(r"\n{3,}", "\n\n", text)

    return text.strip()


def role(sender, internal):
    if internal:
        return "INTERNAL"

    sender = (sender or "").lower()

    if sender == "customer":
        return "CUSTOMER"

    if sender == "agent":
        return "AGENT"

    if sender == "system":
        return "UNKNOWN"

    return "UNKNOWN"


# ------------------------------------------------------------

src = sqlite3.connect(SRC)
src.row_factory = sqlite3.Row

if DST.exists():
    DST.unlink()

dst = sqlite3.connect(DST)

dst.executescript("""
CREATE TABLE articles (
    id INTEGER PRIMARY KEY,
    ticket_id INTEGER NOT NULL,

    sender TEXT,
    role TEXT NOT NULL,
    internal INTEGER NOT NULL DEFAULT 0,
    type TEXT,

    subject TEXT,

    raw_body TEXT,
    clean_body TEXT,

    created_at TEXT
);

CREATE INDEX idx_articles_ticket
    ON articles(ticket_id);

CREATE INDEX idx_articles_role
    ON articles(role);

CREATE TABLE cases (
    ticket_id INTEGER PRIMARY KEY,

    ticket_number TEXT,
    title TEXT,

    group_name TEXT,
    state TEXT,

    created_at TEXT,
    updated_at TEXT,

    article_count INTEGER,
    customer_count INTEGER,
    agent_count INTEGER,
    unknown_count INTEGER,
    internal_count INTEGER,

    customer_text TEXT,
    agent_text TEXT,
    unknown_text TEXT,

    conversation TEXT,

    classification_status TEXT DEFAULT 'NEW',

    communication_classification TEXT,
    primary_intent TEXT,
    secondary_intents TEXT,
    complaint_type TEXT,
    actions TEXT,

    taxonomy_fit TEXT,
    taxonomy_candidate TEXT,

    summary TEXT,
    classification_json TEXT
);

CREATE INDEX idx_cases_status
    ON cases(classification_status);
""")

tickets = src.execute("""
    SELECT *
    FROM tickets
    ORDER BY id
""").fetchall()

print("=" * 72)
print("ZAMMAD ANALYSE-DATENBANK")
print("=" * 72)
print(f"Tickets: {len(tickets)}")
print()

article_total = 0

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

    articles = src.execute("""
        SELECT *
        FROM articles
        WHERE ticket_id = ?
        ORDER BY created_at, id
    """, (ticket["id"],)).fetchall()

    customer = []
    agent = []
    unknown = []
    conversation = []

    counts = {
        "CUSTOMER": 0,
        "AGENT": 0,
        "UNKNOWN": 0,
        "INTERNAL": 0,
    }

    for a in articles:

        raw = a["body_text"] or ""
        clean = clean_body(raw)

        r = role(
            a["sender"],
            a["internal"],
        )

        counts[r] += 1
        article_total += 1

        dst.execute("""
            INSERT INTO articles (
                id,
                ticket_id,
                sender,
                role,
                internal,
                type,
                subject,
                raw_body,
                clean_body,
                created_at
            )
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """, (
            a["id"],
            a["ticket_id"],
            a["sender"],
            r,
            int(a["internal"] or 0),
            a["type"],
            a["subject"],
            raw,
            clean,
            a["created_at"],
        ))

        if not clean:
            continue

        if r == "CUSTOMER":
            customer.append(clean)
            conversation.append(
                "[CUSTOMER]\n" + clean
            )

        elif r == "AGENT":
            agent.append(clean)
            conversation.append(
                "[AGENT]\n" + clean
            )

        elif r == "UNKNOWN":
            unknown.append(clean)
            conversation.append(
                "[UNKNOWN]\n" + clean
            )

        # INTERNAL bewusst nicht in conversation

    dst.execute("""
        INSERT INTO cases (
            ticket_id,
            ticket_number,
            title,
            group_name,
            state,
            created_at,
            updated_at,

            article_count,
            customer_count,
            agent_count,
            unknown_count,
            internal_count,

            customer_text,
            agent_text,
            unknown_text,
            conversation
        )
        VALUES (
            ?, ?, ?, ?, ?, ?, ?,
            ?, ?, ?, ?, ?,
            ?, ?, ?, ?
        )
    """, (
        ticket["id"],
        ticket["number"],
        ticket["title"],
        ticket["group_name"],
        ticket["state"],
        ticket["created_at"],
        ticket["updated_at"],

        len(articles),
        counts["CUSTOMER"],
        counts["AGENT"],
        counts["UNKNOWN"],
        counts["INTERNAL"],

        "\n\n".join(customer),
        "\n\n".join(agent),
        "\n\n".join(unknown),
        "\n\n".join(conversation),
    ))

    if pos % 500 == 0:
        dst.commit()

        print(
            f"[{pos:4}/{len(tickets)}] "
            f"Artikel: {article_total}"
        )

dst.commit()

print()
print("=" * 72)
print("FERTIG")
print("=" * 72)

for row in dst.execute("""
    SELECT role, COUNT(*)
    FROM articles
    GROUP BY role
    ORDER BY COUNT(*) DESC
"""):
    print(
        f"Artikel {row[0]:10} {row[1]}"
    )

print()

row = dst.execute("""
    SELECT
        COUNT(*),
        SUM(customer_count > 0),
        SUM(agent_count > 0),
        SUM(unknown_count > 0),
        SUM(internal_count > 0)
    FROM cases
""").fetchone()

print(f"Cases:              {row[0]}")
print(f"mit Kundenbeitrag:  {row[1]}")
print(f"mit Agentantwort:   {row[2]}")
print(f"mit UNKNOWN:        {row[3]}")
print(f"mit internen Notiz: {row[4]}")

print()
print(f"DB: {DST}")

src.close()
dst.close()