pcvisit-filetracker/main.py

import os
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 info import run as run_info
from move_files import default_desktop, refresh_and_move
from session_details import run as run_session_details
from summary import run as run_summary
from watch import run as run_watch

app = typer.Typer(help="PC-Visit Filetracker")
DATA_DIRECTORY = Path("data")


def default_session_history() -> Path:
    user_profile = Path(os.environ.get("USERPROFILE", str(Path.home())))
    return user_profile / "AppData" / "Roaming" / "pcvisit Data" / "Session History"


@app.callback(invoke_without_command=True)
def app_callback(context: typer.Context) -> None:
    """Verwalten der PC-Visit-Filetracker-Befehle."""
    if context.invoked_subcommand is None:
        typer.echo(context.get_help())


@app.command("clients")
def clients_command(
    directory: Path | None = typer.Argument(None, help="Verzeichnis mit XML-Dateien"),
    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()
    try:
        result = run(directory, output)
    except ValueError as error:
        raise typer.BadParameter(str(error), param_hint="directory") from error
    typer.echo(f"{result.xml_count} XML-Dateien gelesen, {result.row_count} Einträge nach {output} geschrieben.")


@app.command("files")
def files_command(
    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:
        result = 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"{result.xml_count} XML-Dateien gelesen, {result.row_count} vorhandene Dateien nach {output} geschrieben."
    )


@app.command("move-files")
def move_files_command(
    directory: Path | None = typer.Argument(None, help="Verzeichnis mit XML-Dateien"),
    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"),
    ignore_timestamp: bool = typer.Option(
        False,
        "--ignore-timestamp",
        help="Ignoriert den Abgleich des Erstellungsdatums mit der Session.",
    ),
) -> None:
    """Verschiebt gefundene Dateien in Desktop/PC-Visit/Kunde."""
    directory = directory or default_session_history()
    try:
        result = refresh_and_move(
            directory,
            files_csv,
            clients_csv,
            desktop or default_desktop(),
            ignore_timestamp,
        )
    except ValueError as error:
        raise typer.BadParameter(str(error)) from error
    typer.echo(
        f"{result.moved_count} Dateien verschoben, {result.skipped_count} übersprungen, "
        f"{result.missing_client_count} ohne Kunde."
    )


@app.command("archive-logs")
def archive_logs_command(
    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:
        result = run_archive_logs(directory)
    except ValueError as error:
        raise typer.BadParameter(str(error), param_hint="directory") from error
    typer.echo(
        f"{result.checked_count} XML-Dateien geprüft, {result.archived_count} archiviert, "
        f"{result.skipped_count} übersprungen."
    )


@app.command("summary")
def summary_command(
    input_path: Path = typer.Option(
        DATA_DIRECTORY / "session_details.csv",
        "--input",
        "-i",
        help="Eingabe-CSV aus session-details",
    ),
    output: Path = typer.Option(DATA_DIRECTORY / "summary.csv", "--output", "-o", help="Zielpfad der CSV-Datei"),
    days: int = typer.Option(14, "--days", help="Anzahl der letzten Kalendertage für die Konsolenausgabe"),
) -> None:
    """Erstellt eine Zusammenfassung aus session_details.csv."""
    try:
        result = run_summary(input_path, output, days)
    except ValueError as error:
        raise typer.BadParameter(str(error), param_hint="input_path") from error
    typer.echo(
        f"{result.activity_count} Aktivitäten gelesen, {result.summary_count} Zusammenfassungen nach {output} geschrieben."
    )
    if result.recent_summary:
        typer.echo(result.recent_summary)


@app.command("info")
def info_command(
    directory: Path | None = typer.Argument(None, help="Verzeichnis mit XML-Dateien"),
    days: int = typer.Option(14, "--days", help="Anzahl der letzten Kalendertage für die Konsolenausgabe"),
) -> None:
    """Führt clients, archive-logs, session-details und summary aus."""
    directory = directory or default_session_history()
    try:
        result = run_info(
            directory,
            DATA_DIRECTORY / "clients.csv",
            DATA_DIRECTORY / "session_details.csv",
            DATA_DIRECTORY / "summary.csv",
            days,
        )
    except ValueError as error:
        raise typer.BadParameter(str(error), param_hint="directory") from error

    typer.echo(f"clients: {result.clients_result.row_count} Einträge")
    typer.echo(
        f"archive-logs: {result.archive_result.archived_count} archiviert, "
        f"{result.archive_result.skipped_count} übersprungen"
    )
    typer.echo(f"session-details: {result.session_details_result.activity_count} UserActivity-Einträge")
    typer.echo(f"summary: {result.summary_result.summary_count} Zusammenfassungen")
    if result.summary_result.recent_summary:
        typer.echo(result.summary_result.recent_summary)


@app.command("session-details")
def session_details_command(
    directory: Path | None = typer.Argument(None, help="Verzeichnis mit XML-Dateien"),
    output: Path = typer.Option(
        DATA_DIRECTORY / "session_details.csv",
        "--output",
        "-o",
        help="Zielpfad der CSV-Datei",
    ),
) -> None:
    """Sammelt UserActivity-Details aus Session-XML-Dateien."""
    directory = directory or default_session_history()
    try:
        result = run_session_details(directory, output)
    except ValueError as error:
        raise typer.BadParameter(str(error), param_hint="directory") from error
    typer.echo(
        f"{result.xml_count} XML-Dateien geprüft, {result.activity_count} UserActivity-Einträge nach {output} geschrieben."
    )


@app.command("watch")
def watch_command(
    directory: Path | None = typer.Argument(None, help="Verzeichnis mit XML-Dateien"),
    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"),
    ignore_timestamp: bool = typer.Option(
        False,
        "--ignore-timestamp",
        help="Ignoriert den Abgleich des Erstellungsdatums mit der Session.",
    ),
) -> None:
    """Überwacht XML-Dateien und führt bei Änderungen move-files aus."""
    directory = directory or default_session_history()
    try:
        run_watch(directory, files_csv, clients_csv, desktop or default_desktop(), ignore_timestamp)
    except ValueError as error:
        raise typer.BadParameter(str(error), param_hint="directory") from error


if __name__ == "__main__":
    app()