pcvisit-filetracker/session_details.py

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

from clients import calculate_customer


@dataclass(frozen=True)
class SessionDetailsResult:
    xml_count: int
    activity_count: int


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 extract_session_details(xml_path: Path) -> list[dict[str, str]]:
    try:
        root = ET.parse(xml_path).getroot()
    except ET.ParseError:
        return []

    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", "")
    if not client_name.strip():
        return []

    customer = calculate_customer(client_name)
    activities = []
    for activity in session.findall("UserActivity"):
        date_time = activity.get("DateTime", "").strip()
        activity_type = activity.get("Type", "").strip()
        try:
            parsed_date_time = datetime.fromisoformat(date_time.replace("Z", "+00:00"))
        except ValueError:
            continue
        if date_time and activity_type:
            activities.append((parsed_date_time, date_time, activity_type))

    day_counts = {}
    for parsed_date_time, _, _ in activities:
        day_counts[parsed_date_time.date()] = day_counts.get(parsed_date_time.date(), 0) + 1

    rows = []
    for index, (parsed_date_time, date_time, activity_type) in enumerate(activities):
        next_date_time = activities[index + 1][0] if index + 1 < len(activities) else parsed_date_time
        duration = min(max(int((next_date_time - parsed_date_time).total_seconds()), 0), 900)
        if (
            len(day_counts) > 1
            and day_counts[parsed_date_time.date()] == 1
            and activity_type == "SupportAndPresentationMode"
        ):
            duration = 0
        rows.append(
            {
                "DateTime": date_time,
                "Kunde": customer,
                "UserActivityType": activity_type,
                "Duration": str(duration),
            }
        )
    return rows


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


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

    xml_files = find_xml_files(directory)
    rows = [row for xml_file in xml_files for row in extract_session_details(xml_file)]
    rows.sort(key=lambda row: row["DateTime"])
    write_session_details(rows, output)
    return SessionDetailsResult(xml_count=len(xml_files), activity_count=len(rows))