csv-Dateien verschoben

archive_logs.py

@@ -0,0 +1,86 @@
+import shutil
+import xml.etree.ElementTree as ET
+from datetime import datetime, timedelta
+from pathlib import Path
+
+
+def find_xml_files(directory: Path) -> list[Path]:
+    archive_directory = directory / "Archiv"
+    return sorted(
+        path
+        for path in directory.rglob("*")
+        if path.is_file()
+        and path.suffix.lower() == ".xml"
+        and not path.name.lower().startswith("sessions")
+        and archive_directory not in path.parents
+    )
+
+
+def get_archive_year(xml_path: Path, now: datetime | None = None) -> int | None:
+    try:
+        root = ET.parse(xml_path).getroot()
+    except ET.ParseError:
+        return None
+
+    session = root.find("Session")
+    if session is None:
+        return None
+
+    if session.get("Duration", "").strip():
+        start_time = session.get("StartTime", "")
+        try:
+            return datetime.fromisoformat(start_time.replace("Z", "+00:00")).year
+        except ValueError:
+            return None
+
+    created_at = datetime.fromtimestamp(xml_path.stat().st_birthtime)
+    now = now or datetime.now()
+    if now - created_at >= timedelta(days=30):
+        return created_at.year
+    return None
+
+
+def remove_empty_directories(directory: Path) -> int:
+    archive_directory = directory / "Archiv"
+    removed_count = 0
+    subdirectories = sorted(
+        (path for path in directory.rglob("*") if path.is_dir()),
+        key=lambda path: len(path.parts),
+        reverse=True,
+    )
+    for subdirectory in subdirectories:
+        if subdirectory == archive_directory or archive_directory in subdirectory.parents:
+            continue
+        try:
+            subdirectory.rmdir()
+        except OSError:
+            continue
+        removed_count += 1
+    return removed_count
+
+
+def run(directory: Path) -> tuple[int, int, int]:
+    if not directory.is_dir():
+        raise ValueError(f"Das Verzeichnis existiert nicht: {directory}")
+
+    checked_count = 0
+    archived_count = 0
+    skipped_count = 0
+    for xml_file in find_xml_files(directory):
+        checked_count += 1
+        year = get_archive_year(xml_file)
+        if year is None:
+            skipped_count += 1
+            continue
+
+        target_directory = directory / "Archiv" / str(year)
+        target_directory.mkdir(parents=True, exist_ok=True)
+        target = target_directory / xml_file.name
+        if target.exists():
+            skipped_count += 1
+            continue
+        shutil.move(str(xml_file), str(target))
+        archived_count += 1
+
+    remove_empty_directories(directory)
+    return checked_count, archived_count, skipped_count

clients.py

@@ -47,6 +47,7 @@ def calculate_customer(client_name: str) -> str:
 
 
 def write_clients_csv(rows: list[dict[str, str]], csv_path: Path) -> None:
+    csv_path.parent.mkdir(parents=True, exist_ok=True)
     fieldnames = ["ParticipantName", "ClientName", "StartTime", "Kunde"]
     with csv_path.open("w", newline="", encoding="latin-1", errors="replace") as csv_file:
         writer = csv.DictWriter(csv_file, fieldnames=fieldnames, delimiter=";")

data/files.csv

@@ -0,0 +1,2 @@
+ParticipantName;FilePath
+Global Cube (AB-GlobalCube);C:\Users\GAPS\Desktop\gcupdate_20260526.log

data/summary.csv

@@ -0,0 +1,182 @@
+Kunde;AnzahlSitzungen;Dauer
+3H Rheindorf C7;2;682
+ABL Service C7;3;1412
+Altermann C7;3;229
+Arndt C7;2;2542
+Arnhoelter C7;5;2177971
+Aurego C11;9;1234367
+Aurego C7;5;9475
+Auto Epple C7;2;6244
+Auto Epple Opel SKR C7;4;127536
+Auto Epple Opel SKR51 C11;7;19554
+Automag C11;35;2550506
+Barbarossa OPTIMA C7;1;27
+Barth u. Frey C11;25;3520121
+Becker-Tiemann Bielefeld C11;62;7443680
+Bekirovski C11;4;1285205
+Berkemeier C11;20;1690319
+Berkemeier Muenster C11;7;342423
+Berkemeier M??nster C11;1;372
+Berkemeier Saerbeck C11;4;773855
+Berkemeier_C11;1;179983
+Bichmann C11;24;2510308
+Bieling C11;19;3439069
+Bieling C7;1;269650
+Boehm C11;19;1101484
+Boettche C11;20;3502293
+Boettche C7;6;310147
+Brass C11;39;4872564
+Brass C7;2;430671
+Brass_Jacob C11;3;41026
+Bunk C11;8;619434
+cognos-12-0-2;4;507999
+Cordes C7;2;57
+Croll u. Sondermann  C7;3;904
+Czeczelits C11;11;435119
+Czeczelits C7;5;76252
+Darmas _Tiemeyer C11;1;158
+Darmas C11;18;1370770
+Delta C7;4;142951
+Dresen C7;28;1433736
+Ebber C7;6;480139
+Echterstrasse C7;1;110272
+Ella;2;18790
+Elmer  C7;1;261362
+Elspass C7;1;495
+Englhart C7;3;2916
+Fluegel C7;3;3307
+Foerster C7;4;740
+Fricke C7;1;296
+gc-server6;13;1161008
+gekuendigt - Rastetter WERWISO WS19 (AHRAAP03#1);2;5962
+gekuendigt - Rastetter WERWISO_DATEV WS19 (AHRAAP03#1);2;858
+Gerds C11;24;2638555
+Gieraths C11;10;92007
+global-cube-demo;1;1800
+global-cube2 (Server 2016) WS16;1;1391
+global-cube3 (Server2019) WS19;2;80186
+global-cube4 (Win 10);3;134976
+global-cube6 (Server 2022) WS22;1;57
+global-cube7 (Server 2025) WS25;2;12516
+Goendle AT C11;5;657997
+Goendle C11;1;213
+Gottstein C11;15;4994976
+Gottstein C12 ARI WS19 (GSGlobalCube);1;92550
+Gottwald C7;3;2156
+Greiner C11;17;1158245
+Greiner C7;1;16
+Griesbeck Deggendorf AUTOSYS WS22 (SRV03#225);3;62046
+Griesbeck Oberpfalz C7;2;354
+Guema Wertheim C7;4;13700
+Guenther u Sustrate C11;4;529865
+Guenther u. Schmitt C11;5;411443
+Guenther u. Schmitt C7;1;427600
+Haeusler C11;23;2899109
+Hannuschka C7;1;127
+Hannuschka Interims-PC (PC121#004);5;185156
+Hasler C11;10;1872303
+Hedtke alt C11;1;12894
+Hedtke C11;18;3564260
+Heidenreich C11;6;510255
+Heinen Dortmund C11;18;2952440
+Hermann C11;21;1343491
+Hinz C7;2;296
+HIRO C11;6;466222
+Hoefler C7;2;432383
+Hoffmann Saalfeld C7;1;159
+Hoffmann, Saalfeld C7;2;110242
+Holz C7;3;16073
+Jacob C11;14;3802822
+Jansen C11;11;437654
+Joest C11;14;4004353
+Jonas C7;5;2058
+Kieschnick C7;1;35
+Kleinemeier C11;4;439616
+Koenig Berlin C11;136;3144539
+Koenig Berlin C12 Eisenhut (global-cube3#84);3;328667
+Koenig Berlin C7;10;583223
+Koenig C12 Eisenhut (global-cube3#84);3;69277
+Koenig u. Partner C7;3;75004
+Koerner C11;5;53926
+Konrad C7;1;137
+K??nig Berlin C11;1;101906
+La Linea C7;1;140
+Lackermann C11;10;1423468
+Lausse C11;7;443870
+Lausse C7;1;391
+Ley Bergneustadt C11;29;9565958
+Liegert C11;10;877571
+Limburg C7;2;109996
+Linck C11;11;3270340
+Linck C12 CARLO WS19 (global-cube3);4;6805
+Loebau C7;1;73
+Loeffler C11;7;1630977
+Luchtenberg C7;10;194
+Lutz C11;14;1635641
+Meures C7;2;281
+MGS  C7;1;98
+Moll C11;13;929339
+Motor Center Heinen C11;2;453363
+Mueller Freising C11;8;753080
+Mueller Strausberg C7;1;413
+MVC Wien C7;18;1083247
+Nefzger C11;3;88817
+Neustadt C7;4;921126
+Nolte (EPC4#6);1;184
+Nolte C11;65;2042969
+Peter Nordhessen (ehem. Heidenreich) C11;3;2867
+Pichel Chemnitz Dracar+ WS19 (GC01PICH);2;328
+Pichel Hartmannsdorf Dracar+ WS19 (GC01PIHD#1);5;416494
+Premium Cars C7;1;433832
+Premium Cars Peters C11;22;1380986
+Premium Cars WS12 (PCP-vApp);1;46
+Psotka C7;2;75
+Rastetter C11;3;5946
+Rastetter WERWISO WS19 (AHRAAP03#1);1;40
+Rau Brunsbuettel C7;2;29540
+Reichstein u. Opitz C11;9;23577
+Reichstein u. Opitz C7;1;344
+Reisacher C11;125;5029988
+Reisacher C7;88;13043511
+Roell C7;1;996
+Rueschkamp C11;38;4995550
+Russig C7;2;76698
+Schachtschneider C11;11;491867
+Schmelter C7;1;152
+Schmid C11;7;1067412
+Schmidt Starnberg C7;15;393273
+Schmidt Wilhelmshafen C7;2;1054735
+Schwarz C7;1;152
+Siebrecht C7;4;8575
+Sievers-Group C11;1;501150
+sievers-group C11;1;174398
+Sirch C11;22;1879338
+Stange C11;5;65804
+SundK alt Hamburg C11;7;112511
+SundK C11;11;1333511
+SundK Hamburg C11;5;180894
+SundK Hamburg C12 WERWISO_DATEV WS25;19;1816931
+Tabor C11;7;1880560
+test-c12;3;23582
+Timmermanns C7;31;4411846
+Tobaben C7;3;70632
+Toennemann C11;13;1594440
+Tretter C7;7;235867
+Ulmen C11;20;2731413
+Ulmen C7;1;8755
+VCRR LUEG C11;18;3620104
+VM Toni (VPN-DK, Van Eupen, Winter, Elspass, Schneider-Gruppe);21;817086
+Vogl C11;14;2051471
+von der Weppen C11;4;29082
+VPN2 (VPN-DK, Iveco, Hymer);12;1171716
+Wahl-Group C7;2;3136
+Weis C11;8;50368
+Weller Bietigheim C7;5;388435
+Weller Bietigheim OPTIMA C11;17;2823443
+Weller Wetzlar C7;1;8
+Willi Mueller C11;8;97249
+Winter Bretnig C11;28;3598951
+Zobjack C7;3;361907
+Zschernitz C11;1;348
+Zumbuelt C11;12;1691764
+Zumbuelt C7;3;82774

files.py

@@ -0,0 +1,114 @@
+import csv
+import os
+import xml.etree.ElementTree as ET
+from datetime import datetime
+from pathlib import Path
+
+
+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) -> tuple[int, int]:
+    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 len(xml_files), len(unique_rows)

main.py

@@ -3,9 +3,14 @@ from pathlib import Path
 
 import typer
 
+from archive_logs import run as run_archive_logs
 from clients import run
+from files import run as run_files
+from move_files import default_desktop, move_files
+from summary import run as run_summary
 
 app = typer.Typer(help="PC-Visit Filetracker")
+DATA_DIRECTORY = Path("data")
 
 
 def default_session_history() -> Path:
@@ -23,7 +28,7 @@ def app_callback(context: typer.Context) -> None:
 @app.command()
 def clients(
     directory: Path | None = typer.Argument(None, help="Verzeichnis mit XML-Dateien"),
-    output: Path = typer.Option(Path("clients.csv"), "--output", "-o", help="Zielpfad der CSV-Datei"),
+    output: Path = typer.Option(DATA_DIRECTORY / "clients.csv", "--output", "-o", help="Zielpfad der CSV-Datei"),
 ) -> None:
     """Importiert Client-Daten aus Session-XML-Dateien."""
     directory = directory or default_session_history()
@@ -34,5 +39,71 @@ def clients(
     typer.echo(f"{xml_count} XML-Dateien gelesen, {row_count} Einträge nach {output} geschrieben.")
 
 
+@app.command()
+def files(
+    directory: Path | None = typer.Argument(None, help="Verzeichnis mit XML-Dateien"),
+    output: Path = typer.Option(DATA_DIRECTORY / "files.csv", "--output", "-o", help="Zielpfad der CSV-Datei"),
+    ignore_timestamp: bool = typer.Option(
+        False,
+        "--ignore-timestamp",
+        help="Ignoriert den Abgleich des Erstellungsdatums mit der Session.",
+    ),
+) -> None:
+    """Importiert vorhandene Dateien aus ReceivedFile-Elementen."""
+    directory = directory or default_session_history()
+    try:
+        xml_count, row_count = run_files(directory, output, ignore_timestamp=ignore_timestamp)
+    except ValueError as error:
+        raise typer.BadParameter(str(error), param_hint="directory") from error
+    typer.echo(f"{xml_count} XML-Dateien gelesen, {row_count} vorhandene Dateien nach {output} geschrieben.")
+
+
+@app.command("move-files")
+def move_files_command(
+    files_csv: Path = typer.Option(
+        DATA_DIRECTORY / "files.csv", "--files-csv", help="CSV-Datei mit den gefundenen Dateien"
+    ),
+    clients_csv: Path = typer.Option(DATA_DIRECTORY / "clients.csv", "--clients-csv", help="CSV-Datei mit den Kunden"),
+    desktop: Path | None = typer.Option(None, "--desktop", help="Desktop-Verzeichnis als Zielbasis"),
+) -> None:
+    """Verschiebt gefundene Dateien in Desktop/PC-Visit/Kunde."""
+    try:
+        moved_count, skipped_count, missing_client_count = move_files(
+            files_csv, clients_csv, desktop or default_desktop()
+        )
+    except ValueError as error:
+        raise typer.BadParameter(str(error)) from error
+    typer.echo(
+        f"{moved_count} Dateien verschoben, {skipped_count} übersprungen, " f"{missing_client_count} ohne Kunde."
+    )
+
+
+@app.command("archive-logs")
+def archive_logs(
+    directory: Path | None = typer.Argument(None, help="Verzeichnis mit XML-Dateien"),
+) -> None:
+    """Archiviert Session-XML-Dateien nach Jahr."""
+    directory = directory or default_session_history()
+    try:
+        checked_count, archived_count, skipped_count = run_archive_logs(directory)
+    except ValueError as error:
+        raise typer.BadParameter(str(error), param_hint="directory") from error
+    typer.echo(f"{checked_count} XML-Dateien geprüft, {archived_count} archiviert, " f"{skipped_count} übersprungen.")
+
+
+@app.command()
+def summary(
+    directory: Path | None = typer.Argument(None, help="Verzeichnis mit XML-Dateien"),
+    output: Path = typer.Option(DATA_DIRECTORY / "summary.csv", "--output", "-o", help="Zielpfad der CSV-Datei"),
+) -> None:
+    """Erstellt eine Sitzungs- und Dauersumme pro Kunde."""
+    directory = directory or default_session_history()
+    try:
+        xml_count, customer_count = run_summary(directory, output)
+    except ValueError as error:
+        raise typer.BadParameter(str(error), param_hint="directory") from error
+    typer.echo(f"{xml_count} XML-Dateien geprüft, {customer_count} Kunden nach {output} geschrieben.")
+
+
 if __name__ == "__main__":
     app()

move_files.py

@@ -0,0 +1,56 @@
+import csv
+import os
+import shutil
+from pathlib import Path
+
+
+def default_desktop() -> Path:
+    user_profile = Path(os.environ.get("USERPROFILE", str(Path.home())))
+    return user_profile / "Desktop"
+
+
+def read_csv(csv_path: Path) -> list[dict[str, str]]:
+    if not csv_path.is_file():
+        raise ValueError(f"Die CSV-Datei existiert nicht: {csv_path}")
+
+    with csv_path.open("r", newline="", encoding="latin-1") as csv_file:
+        return list(csv.DictReader(csv_file, delimiter=";"))
+
+
+def move_files(
+    files_csv: Path,
+    clients_csv: Path,
+    desktop_directory: Path | None = None,
+) -> tuple[int, int, int]:
+    desktop_directory = desktop_directory or default_desktop()
+    client_rows = read_csv(clients_csv)
+    client_names = {
+        row.get("ParticipantName", "").strip(): row.get("Kunde", "").strip()
+        for row in client_rows
+        if row.get("ParticipantName", "").strip() and row.get("Kunde", "").strip()
+    }
+
+    moved_count = 0
+    skipped_count = 0
+    missing_client_count = 0
+    for row in read_csv(files_csv):
+        participant_name = row.get("ParticipantName", "").strip()
+        source = Path(row.get("FilePath", "").strip())
+        customer = client_names.get(participant_name)
+        if not customer:
+            missing_client_count += 1
+            continue
+        if not source.is_file():
+            skipped_count += 1
+            continue
+
+        target_directory = desktop_directory / "PC-Visit" / customer
+        target_directory.mkdir(parents=True, exist_ok=True)
+        target = target_directory / source.name
+        if source.resolve() == target.resolve() or target.exists():
+            skipped_count += 1
+            continue
+        shutil.move(str(source), str(target))
+        moved_count += 1
+
+    return moved_count, skipped_count, missing_client_count

summary.py

@@ -0,0 +1,71 @@
+import csv
+import xml.etree.ElementTree as ET
+from collections import defaultdict
+from decimal import Decimal, InvalidOperation
+from pathlib import Path
+
+from clients import calculate_customer, find_xml_files
+
+
+def extract_session(xml_path: Path) -> tuple[str, Decimal] | None:
+    try:
+        root = ET.parse(xml_path).getroot()
+    except ET.ParseError:
+        return None
+
+    session = root.find("Session")
+    if session is None:
+        return None
+
+    client_description = session.find("SessionDescription[@Name='ClientName']")
+    client_name = "" if client_description is None else client_description.get("Description", "")
+    duration = session.get("Duration", "").strip()
+    if not client_name.strip() or not duration:
+        return None
+
+    try:
+        return calculate_customer(client_name), Decimal(duration)
+    except InvalidOperation:
+        return None
+
+
+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=["Kunde", "AnzahlSitzungen", "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 run(directory: Path, output: Path) -> tuple[int, int]:
+    if not directory.is_dir():
+        raise ValueError(f"Das Verzeichnis existiert nicht: {directory}")
+
+    totals: defaultdict[str, list[Decimal | int]] = defaultdict(lambda: [0, Decimal("0")])
+    checked_count = 0
+    for xml_file in find_xml_files(directory):
+        checked_count += 1
+        session = extract_session(xml_file)
+        if session is None:
+            continue
+        customer, duration = session
+        totals[customer][0] += 1
+        totals[customer][1] += duration
+
+    rows = [
+        {
+            "Kunde": customer,
+            "AnzahlSitzungen": str(values[0]),
+            "Dauer": format_duration(values[1]),
+        }
+        for customer, values in sorted(totals.items(), key=lambda item: item[0].casefold())
+    ]
+    write_summary(rows, output)
+    return checked_count, len(rows)