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

#!/usr/bin/env python3

import hashlib
import html
import json
import re
import sqlite3
from pathlib import Path
from collections import Counter

DB = Path("data/zammad_history.sqlite3")
OUT = Path("data/zammad_cases.sqlite3")


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

    text = html.unescape(text)

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

    # quoted mail history
    text = re.sub(
        r"\n\s*(Am .* schrieb .*?:|On .* wrote:).*",
        "",
        text,
        flags=re.I | re.S,
    )

    # common signature endings
    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]

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

    return text.strip()


def normalize_for_fingerprint(text):
    text = text.lower()

    # Telefonnummern / E-Mail-Adressen / IDs
    text = re.sub(
        r"\b[\w.+-]+@[\w.-]+\.\w+\b",
        " EMAIL ",
        text,
    )
    text = re.sub(
        r"\b\d{4,}\b",
        " NUMBER ",
        text,
    )

    # whitespace
    text = re.sub(r"\s+", " ", text)

    return text.strip()


def fingerprint(text):
    normalized = normalize_for_fingerprint(text)

    # Wortfolge als stabile lokale Signatur.
    words = normalized.split()

    if len(words) > 120:
        words = words[:120]

    return hashlib.sha1(
        " ".join(words).encode(
            "utf-8",
            errors="ignore",
        )
    ).hexdigest()


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

# Neue Analyse-DB; die Originaldaten bleiben unangetastet.
out = sqlite3.connect(OUT)

out.executescript("""
DROP TABLE IF EXISTS cases;

CREATE TABLE cases (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    ticket_id INTEGER UNIQUE,
    ticket_number TEXT,
    title TEXT,
    group_name TEXT,
    state TEXT,
    created_at TEXT,
    updated_at TEXT,

    customer_id TEXT,

    tags_json TEXT,

    article_count INTEGER,
    customer_article_count INTEGER,

    subject_clean TEXT,
    conversation TEXT,

    fingerprint TEXT,

    content_length INTEGER,

    classification_status TEXT DEFAULT 'NEW',
    primary_intent TEXT,
    secondary_intents TEXT,
    complaint_type TEXT,
    actions TEXT,

    taxonomy_fit TEXT,
    taxonomy_candidate TEXT,

    classification_json TEXT
);

CREATE INDEX idx_cases_fingerprint
    ON cases(fingerprint);

CREATE INDEX idx_cases_status
    ON cases(classification_status);

CREATE INDEX idx_cases_group
    ON cases(group_name);

CREATE INDEX idx_cases_created
    ON cases(created_at);
""")

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

print("=" * 72)
print("ZAMMAD → CASE NORMALIZER")
print("=" * 72)
print(f"Tickets: {len(tickets)}")
print()

stats = Counter()
fingerprints = Counter()

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_parts = []
    all_parts = []

    for article in articles:

        body = clean(
            article["body_text"] or ""
        )

        if not body:
            continue

        # Interne Notizen nicht in das Kundenanliegen übernehmen.
        internal = bool(article["internal"])

        if internal:
            stats["internal_articles"] += 1
            continue

        sender = article["sender"] or ""

        part = (
            f"[{sender}] {body}"
        )

        customer_parts.append(part)
        all_parts.append(part)

    conversation = "\n\n".join(
        customer_parts
    ).strip()

    if not conversation:
        stats["empty_cases"] += 1
        continue

    subject = clean(
        ticket["title"] or ""
    )

    combined = (
        subject + "\n" + conversation
    ).strip()

    fp = fingerprint(combined)
    fingerprints[fp] += 1

    try:
        tags = json.loads(
            ticket["tags_json"] or "[]"
        )
    except Exception:
        tags = []

    out.execute("""
        INSERT INTO cases (
            ticket_id,
            ticket_number,
            title,
            group_name,
            state,
            created_at,
            updated_at,
            customer_id,
            tags_json,
            article_count,
            customer_article_count,
            subject_clean,
            conversation,
            fingerprint,
            content_length
        )
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
    """, (
        ticket["id"],
        ticket["number"],
        ticket["title"],
        ticket["group_name"],
        ticket["state"],
        ticket["created_at"],
        ticket["updated_at"],
        ticket["customer_id"],
        json.dumps(
            tags,
            ensure_ascii=False,
        ),
        len(articles),
        len(customer_parts),
        subject,
        conversation,
        fp,
        len(conversation),
    ))

    stats["cases"] += 1

    if len(customer_parts) > 1:
        stats["multi_article_cases"] += 1

    if pos % 250 == 0:
        out.commit()
        print(
            f"[{pos}/{len(tickets)}] "
            f"Cases: {stats['cases']}"
        )

out.commit()

duplicate_groups = sum(
    1
    for count in fingerprints.values()
    if count > 1
)

duplicate_cases = sum(
    count - 1
    for count in fingerprints.values()
    if count > 1
)

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

for key, value in stats.most_common():
    print(
        f"{key:28} {value}"
    )

print()
print(
    f"Eindeutige Fingerprints: "
    f"{len(fingerprints)}"
)
print(
    f"Fingerprint-Gruppen >1: "
    f"{duplicate_groups}"
)
print(
    f"potentielle Duplikate: "
    f"{duplicate_cases}"
)

print()
print("Top 20 identische/ähnliche Fingerprints:")

for fp, count in sorted(
    fingerprints.items(),
    key=lambda x: x[1],
    reverse=True,
)[:20]:

    if count < 2:
        break

    row = out.execute("""
        SELECT ticket_number, title
        FROM cases
        WHERE fingerprint = ?
        LIMIT 3
    """, (fp,)).fetchall()

    print(
        f"\n{count} Fälle"
    )

    for r in row:
        print(
            f"  #{r[0]} {r[1]}"
        )

print()
print(f"Analyse-DB: {OUT}")

src.close()
out.close()