pcvisit-filetracker/clients.py

import csv
import re
import xml.etree.ElementTree as ET
from datetime import datetime
from pathlib import Path


def extract_clients(xml_path: Path) -> list[dict[str, str]]:
    root = ET.parse(xml_path).getroot()
    session = root.find("Session")
    if session is None:
        return []

    client_description = session.find("SessionDescription[@Name='ClientName']")
    client_name = "" if client_description is None else client_description.get("Description", "")
    start_time = session.get("StartTime", "")

    return [
        {
            "ParticipantName": participant.get("Name", ""),
            "ClientName": client_name,
            "StartTime": start_time,
        }
        for participant in session.findall("Participant")
        if participant.get("Role") not in {"ROLE_SUPPORTER", "ROLE_UNKNOWN"}
        and participant.get("Name", "").strip()
        and client_name.strip()
    ]


def calculate_customer(client_name: str) -> str:
    match = re.search(r"\bC(?:7|11)\b", client_name, re.IGNORECASE)
    customer = client_name[: match.end()] if match else client_name
    customer = (
        customer.replace("&", "u.")
        .replace("ä", "ae")
        .replace("ö", "oe")
        .replace("ü", "ue")
        .replace("Ä", "Ae")
        .replace("Ö", "Oe")
        .replace("Ü", "Ue")
        .replace("ß", "ss")
    )
    customer = re.sub(r'[<>:"/\\|?*\x00-\x1f]', "_", customer)
    customer = customer.rstrip(" .")
    return customer or "_"


def write_clients_csv(rows: list[dict[str, str]], csv_path: Path) -> None:
    csv_path.parent.mkdir(parents=True, exist_ok=True)
    fieldnames = ["ParticipantName", "ClientName", "StartTime", "Kunde"]
    with csv_path.open("w", newline="", encoding="latin-1", errors="replace") as csv_file:
        writer = csv.DictWriter(csv_file, fieldnames=fieldnames, delimiter=";")
        writer.writeheader()
        writer.writerows(rows)


def read_clients_csv(csv_path: Path) -> list[dict[str, str]]:
    if not csv_path.exists():
        return []

    with csv_path.open("r", newline="", encoding="latin-1") as csv_file:
        reader = csv.DictReader(csv_file, delimiter=";")
        return [
            {
                "ParticipantName": row.get("ParticipantName", "").strip(),
                "ClientName": row.get("ClientName", ""),
                "StartTime": row.get("StartTime", ""),
            }
            for row in reader
            if row.get("ParticipantName", "").strip() and row.get("ClientName", "").strip()
        ]


def parse_start_time(start_time: str) -> datetime:
    if not start_time:
        return datetime.min
    try:
        return datetime.fromisoformat(start_time.replace("Z", "+00:00")).replace(tzinfo=None)
    except ValueError:
        return datetime.min


def merge_clients(existing_rows: list[dict[str, str]], new_rows: list[dict[str, str]]) -> list[dict[str, str]]:
    newest_rows: dict[str, dict[str, str]] = {}
    for row in [*existing_rows, *new_rows]:
        participant_name = row["ParticipantName"].strip()
        client_name = row["ClientName"].strip()
        if not participant_name or not client_name:
            continue

        normalized_row = {
            "ParticipantName": participant_name,
            "ClientName": client_name,
            "StartTime": row.get("StartTime", ""),
        }
        current = newest_rows.get(participant_name)
        if current is None or parse_start_time(normalized_row["StartTime"]) >= parse_start_time(current["StartTime"]):
            newest_rows[participant_name] = normalized_row

    return [
        {**row, "Kunde": calculate_customer(row["ClientName"])}
        for row in sorted(newest_rows.values(), key=lambda row: row["ParticipantName"].casefold())
    ]


def find_xml_files(directory: Path) -> list[Path]:
    return sorted(
        path
        for path in directory.rglob("*")
        if path.is_file() and path.suffix.lower() == ".xml" and not path.name.lower().startswith("sessions")
    )


def run(directory: Path, output: Path) -> tuple[int, int]:
    if not directory.is_dir():
        raise ValueError(f"Das Verzeichnis existiert nicht: {directory}")

    xml_files = find_xml_files(directory)
    new_rows = [row for xml_file in xml_files for row in extract_clients(xml_file)]
    rows = merge_clients(read_clients_csv(output), new_rows)
    write_clients_csv(rows, output)
    return len(xml_files), len(rows)