pcvisit-filetracker/files.py

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


@dataclass(frozen=True)
class FilesResult:
    xml_count: int
    row_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 is_file_from_session_day(file_path: Path, session_start_time: str) -> bool:
    try:
        session_date = datetime.fromisoformat(session_start_time.replace("Z", "+00:00")).date()
    except ValueError:
        return False
    return datetime.fromtimestamp(file_path.stat().st_birthtime).date() == session_date


def is_in_pc_visit_directory(file_path: Path, desktop_directory: Path) -> bool:
    try:
        relative_parts = file_path.relative_to(desktop_directory).parts
    except ValueError:
        return False
    return bool(relative_parts) and relative_parts[0].casefold() == "pc-visit"


def find_file_on_desktop(
    file_path: str,
    session_start_time: str,
    desktop_directory: Path | None = None,
    ignore_timestamp: bool = False,
) -> Path | None:
    desktop_directory = desktop_directory or Path(os.environ.get("USERPROFILE", str(Path.home()))) / "Desktop"
    path_parts = [part for part in file_path.replace("\\", "/").split("/") if part]
    desktop_index = next(
        (index for index, part in enumerate(path_parts) if part.casefold() == "desktop"),
        None,
    )
    if desktop_index is not None:
        relative_path = Path(*path_parts[desktop_index + 1 :])
        exact_match = desktop_directory / relative_path
        if (
            exact_match.is_file()
            and not is_in_pc_visit_directory(exact_match, desktop_directory)
            and (ignore_timestamp or is_file_from_session_day(exact_match, session_start_time))
        ):
            return exact_match

    file_name = Path(path_parts[-1]).name if path_parts else ""
    if not file_name or not desktop_directory.is_dir():
        return None
    return next(
        (
            path
            for path in desktop_directory.rglob(file_name)
            if path.is_file()
            and not is_in_pc_visit_directory(path, desktop_directory)
            and (ignore_timestamp or is_file_from_session_day(path, session_start_time))
        ),
        None,
    )


def extract_received_files(xml_path: Path, ignore_timestamp: bool = False) -> list[dict[str, str]]:
    root = ET.parse(xml_path).getroot()
    session = root.find("Session")
    if session is None:
        return []

    participant_names = [
        participant.get("Name", "").strip()
        for participant in session.findall("Participant")
        if participant.get("Role") not in {"ROLE_SUPPORTER", "ROLE_UNKNOWN"} and participant.get("Name", "").strip()
    ]
    if not participant_names:
        return []

    rows = []
    for received_file in session.findall("ReceivedFile"):
        original_path = received_file.get("Name", "").strip()
        start_time = received_file.get("DateTime", "").strip()
        local_path = find_file_on_desktop(original_path, start_time, ignore_timestamp=ignore_timestamp)
        if local_path is not None:
            rows.append({"ParticipantName": participant_names[0], "FilePath": str(local_path)})
    return rows


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


def run(directory: Path, output: Path, ignore_timestamp: bool = False) -> FilesResult:
    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_received_files(xml_file, ignore_timestamp=ignore_timestamp)
    ]
    unique_rows = {(row["ParticipantName"], row["FilePath"]): row for row in rows}
    write_files_csv(
        sorted(unique_rows.values(), key=lambda row: (row["ParticipantName"].casefold(), row["FilePath"].casefold())),
        output,
    )
    return FilesResult(xml_count=len(xml_files), row_count=len(unique_rows))