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), )