summary.py 2.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778
  1. import csv
  2. import xml.etree.ElementTree as ET
  3. from collections import defaultdict
  4. from dataclasses import dataclass
  5. from decimal import Decimal, InvalidOperation
  6. from pathlib import Path
  7. from clients import calculate_customer, find_xml_files
  8. @dataclass(frozen=True)
  9. class SummaryResult:
  10. xml_count: int
  11. customer_count: int
  12. def extract_session(xml_path: Path) -> tuple[str, Decimal] | None:
  13. try:
  14. root = ET.parse(xml_path).getroot()
  15. except ET.ParseError:
  16. return None
  17. session = root.find("Session")
  18. if session is None:
  19. return None
  20. client_description = session.find("SessionDescription[@Name='ClientName']")
  21. client_name = "" if client_description is None else client_description.get("Description", "")
  22. duration = session.get("Duration", "").strip()
  23. if not client_name.strip() or not duration:
  24. return None
  25. try:
  26. return calculate_customer(client_name), Decimal(duration)
  27. except InvalidOperation:
  28. return None
  29. def write_summary(rows: list[dict[str, str]], output: Path) -> None:
  30. output.parent.mkdir(parents=True, exist_ok=True)
  31. with output.open("w", newline="", encoding="latin-1", errors="replace") as csv_file:
  32. writer = csv.DictWriter(csv_file, fieldnames=["Kunde", "AnzahlSitzungen", "Dauer"], delimiter=";")
  33. writer.writeheader()
  34. writer.writerows(rows)
  35. def format_duration(duration: Decimal) -> str:
  36. value = format(duration, "f")
  37. if "." in value:
  38. value = value.rstrip("0").rstrip(".")
  39. return value or "0"
  40. def run(directory: Path, output: Path) -> SummaryResult:
  41. if not directory.is_dir():
  42. raise ValueError(f"Das Verzeichnis existiert nicht: {directory}")
  43. totals: defaultdict[str, list[Decimal | int]] = defaultdict(lambda: [0, Decimal("0")])
  44. checked_count = 0
  45. for xml_file in find_xml_files(directory):
  46. checked_count += 1
  47. session = extract_session(xml_file)
  48. if session is None:
  49. continue
  50. customer, duration = session
  51. totals[customer][0] += 1
  52. totals[customer][1] += duration
  53. rows = [
  54. {
  55. "Kunde": customer,
  56. "AnzahlSitzungen": str(values[0]),
  57. "Dauer": format_duration(values[1]),
  58. }
  59. for customer, values in sorted(totals.items(), key=lambda item: item[0].casefold())
  60. ]
  61. write_summary(rows, output)
  62. return SummaryResult(xml_count=checked_count, customer_count=len(rows))