pcvisit-filetracker/summary.py

import csv
from collections import defaultdict
from dataclasses import dataclass
from datetime import date, datetime, timedelta
from decimal import Decimal, InvalidOperation
from pathlib import Path


@dataclass(frozen=True)
class SummaryResult:
    activity_count: int
    summary_count: int
    recent_summary: str


def read_session_details(input_path: Path) -> list[dict[str, str]]:
    if not input_path.is_file():
        raise ValueError(f"Die CSV-Datei existiert nicht: {input_path}")

    with input_path.open("r", newline="", encoding="latin-1") as csv_file:
        return list(csv.DictReader(csv_file, delimiter=";"))


def write_summary(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=["Datum", "Startzeit", "Kunde", "Dauer"], delimiter=";")
        writer.writeheader()
        writer.writerows(rows)


def format_duration(duration: Decimal) -> str:
    value = format(duration, "f")
    if "." in value:
        value = value.rstrip("0").rstrip(".")
    return value or "0"


def format_elapsed_seconds(duration: Decimal) -> str:
    total_seconds = max(int(duration), 0)
    minutes, seconds = divmod(total_seconds, 60)
    return f"{minutes}m" if seconds == 0 else f"{minutes}m {seconds:02d}s"


def format_recent_summary(rows: list[dict[str, str]], days: int = 14, today: date | None = None) -> str:
    if days < 1:
        raise ValueError("Die Anzahl der Tage muss mindestens 1 sein.")

    today = today or date.today()
    first_day = today - timedelta(days=days - 1)
    weekday_names = ["Mo", "Di", "Mi", "Do", "Fr", "Sa", "So"]
    lines: list[str] = []
    current_date: date | None = None
    for row in rows:
        row_date = date.fromisoformat(row["Datum"])
        if row_date < first_day or row_date > today:
            continue
        if not row.get("_has_visible_activity", True):
            continue
        if row_date != current_date:
            if lines:
                lines.append("")
            lines.append(f"{weekday_names[row_date.weekday()]}, {row_date:%d.%m.%Y}")
            current_date = row_date
        lines.append(f"- {row['Startzeit'][:5]} {row['Kunde']} ({format_elapsed_seconds(Decimal(row['Dauer']))})")
    return "\n".join(lines)


def run(input_path: Path, output: Path, days: int = 14) -> SummaryResult:
    totals: defaultdict[tuple[str, str], list[Decimal | str | bool]] = defaultdict(lambda: [Decimal("0"), "", False])
    activity_count = 0
    for row in read_session_details(input_path):
        date_time = row.get("DateTime", "").strip()
        customer = row.get("Kunde", "").strip()
        duration_text = row.get("Duration", "").strip()
        try:
            parsed_date_time = datetime.fromisoformat(date_time.replace("Z", "+00:00"))
            duration = Decimal(duration_text)
        except (InvalidOperation, ValueError):
            continue
        if not customer or not duration_text:
            continue

        date = parsed_date_time.date().isoformat()
        key = (date, customer)
        values = totals[key]
        values[0] += duration
        time = parsed_date_time.time().isoformat()
        if not values[1] or time < values[1]:
            values[1] = time
        if row.get("UserActivityType", "").strip() != "SupportAndPresentationMode" or duration != 0:
            values[2] = True
        activity_count += 1

    rows = [
        {
            "Datum": date,
            "Startzeit": start_time,
            "Kunde": customer,
            "Dauer": format_duration(duration),
        }
        for (date, customer), (duration, start_time, _) in sorted(
            totals.items(), key=lambda item: (item[0][0], item[1][1], item[0][1].casefold())
        )
    ]
    write_summary(rows, output)
    recent_rows = [{**row, "_has_visible_activity": totals[(row["Datum"], row["Kunde"])][2]} for row in rows]
    return SummaryResult(
        activity_count=activity_count,
        summary_count=len(rows),
        recent_summary=format_recent_summary(recent_rows, days),
    )