archive_logs.py 2.8 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798
  1. import shutil
  2. import xml.etree.ElementTree as ET
  3. from dataclasses import dataclass
  4. from datetime import datetime, timedelta
  5. from pathlib import Path
  6. @dataclass(frozen=True)
  7. class ArchiveLogsResult:
  8. checked_count: int
  9. archived_count: int
  10. skipped_count: int
  11. def find_xml_files(directory: Path) -> list[Path]:
  12. archive_directory = directory / "Archiv"
  13. return sorted(
  14. path
  15. for path in directory.rglob("*")
  16. if path.is_file()
  17. and path.suffix.lower() == ".xml"
  18. and not path.name.lower().startswith("sessions")
  19. and archive_directory not in path.parents
  20. )
  21. def get_archive_year(xml_path: Path, now: datetime | None = None) -> int | None:
  22. try:
  23. root = ET.parse(xml_path).getroot()
  24. except ET.ParseError:
  25. return None
  26. session = root.find("Session")
  27. if session is None:
  28. return None
  29. if session.get("Duration", "").strip():
  30. start_time = session.get("StartTime", "")
  31. try:
  32. return datetime.fromisoformat(start_time.replace("Z", "+00:00")).year
  33. except ValueError:
  34. return None
  35. created_at = datetime.fromtimestamp(xml_path.stat().st_birthtime)
  36. now = now or datetime.now()
  37. if now - created_at >= timedelta(days=30):
  38. return created_at.year
  39. return None
  40. def remove_empty_directories(directory: Path) -> int:
  41. archive_directory = directory / "Archiv"
  42. removed_count = 0
  43. subdirectories = sorted(
  44. (path for path in directory.rglob("*") if path.is_dir()),
  45. key=lambda path: len(path.parts),
  46. reverse=True,
  47. )
  48. for subdirectory in subdirectories:
  49. if subdirectory == archive_directory or archive_directory in subdirectory.parents:
  50. continue
  51. try:
  52. subdirectory.rmdir()
  53. except OSError:
  54. continue
  55. removed_count += 1
  56. return removed_count
  57. def run(directory: Path) -> ArchiveLogsResult:
  58. if not directory.is_dir():
  59. raise ValueError(f"Das Verzeichnis existiert nicht: {directory}")
  60. checked_count = 0
  61. archived_count = 0
  62. skipped_count = 0
  63. for xml_file in find_xml_files(directory):
  64. checked_count += 1
  65. year = get_archive_year(xml_file)
  66. if year is None:
  67. skipped_count += 1
  68. continue
  69. target_directory = directory / "Archiv" / str(year)
  70. target_directory.mkdir(parents=True, exist_ok=True)
  71. target = target_directory / xml_file.name
  72. if target.exists():
  73. skipped_count += 1
  74. continue
  75. shutil.move(str(xml_file), str(target))
  76. archived_count += 1
  77. remove_empty_directories(directory)
  78. return ArchiveLogsResult(
  79. checked_count=checked_count,
  80. archived_count=archived_count,
  81. skipped_count=skipped_count,
  82. )