Automatisierung & Scripting

from __future__ import annotations

import argparse
import csv
import json
import logging
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

logger = logging.getLogger("kurs_report")


# ---------- Datenmodell ----------
@dataclass(frozen=True)
class Kurs:
    kursname: str
    format: str       # "Online" / "Praesenz"
    standort: str
    dauer_tage: int
    preis_euro: float


# ---------- CLI ----------
def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(
        prog="kurs_report",
        description="Erzeugt einen Kurs-Report (Beispieldaten) als JSON."
    )
    p.add_argument("--standort", default="Berlin", help="z.B. Berlin, Hamburg")
    p.add_argument("--format", default="Online", help="Online oder Praesenz")
    p.add_argument("--limit", type=int, default=10, help="Max. Einträge im Report")
    p.add_argument("--out", default="out/report.json", help="Output-Datei (JSON)")
    p.add_argument("--dry-run", action="store_true", help="Nur anzeigen, nichts schreiben")

    # Optional: CSV als Quelle
    p.add_argument(
        "--csv",
        default="",
        help="Optional: CSV-Datei mit Kursdaten (Header: kursname,format,standort,dauer_tage,preis_euro)"
    )

    # Optional: API als Quelle
    p.add_argument("--api", default="", help="Optional: API-URL, liefert JSON-Liste von Kursen")
    p.add_argument("--timeout", type=int, default=8, help="API-Timeout in Sekunden")

    return p.parse_args()


# ---------- Logging Setup ----------
def setup_logging() -> None:
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s | %(levelname)s | %(message)s"
    )


# ---------- Datenquellen ----------
def lade_beispieldaten() -> list[Kurs]:
    return [
        Kurs("Python Schulung", "Online",   "Berlin",   3, 1390.00),
        Kurs("Python Schulung", "Praesenz", "Hamburg",  3, 1490.00),
        Kurs("Python Schulung", "Online",   "Hamburg",  3, 1290.00),
        Kurs("Python Schulung", "Praesenz", "Berlin",   3, 1490.00),
        Kurs("Python Schulung", "Online",   "Koeln",    2,  990.00),
        Kurs("Python Schulung", "Praesenz", "Muenchen", 4, 1890.00),
    ]


def lade_csv(pfad: Path) -> list[Kurs]:
    kurse: list[Kurs] = []
    with pfad.open("r", encoding="utf-8", newline="") as f:
        reader = csv.DictReader(f)
        for z in reader:
            kurse.append(Kurs(
                kursname=str(z["kursname"]),
                format=str(z["format"]),
                standort=str(z["standort"]),
                dauer_tage=int(z["dauer_tage"]),
                preis_euro=float(z["preis_euro"]),
            ))
    return kurse


def lade_api(url: str, timeout_s: int) -> list[Kurs]:
    # requests nur nutzen, wenn API-Mode aktiv ist
    import requests  # type: ignore

    try:
        resp = requests.get(url, timeout=timeout_s)
        resp.raise_for_status()
        daten: Any = resp.json()

        if not isinstance(daten, list):
            raise RuntimeError("API liefert kein Listen-JSON.")

        kurse: list[Kurs] = []
        for item in daten:
            kurse.append(Kurs(
                kursname=str(item["kursname"]),
                format=str(item["format"]),
                standort=str(item["standort"]),
                dauer_tage=int(item["dauer_tage"]),
                preis_euro=float(item["preis_euro"]),
            ))
        return kurse

    except requests.Timeout:
        raise RuntimeError("Timeout: API antwortet nicht rechtzeitig.")
    except requests.RequestException as e:
        raise RuntimeError(f"API-Request fehlgeschlagen: {e}")


# ---------- Verarbeitung ----------
def filtere_kurse(kurse: list[Kurs], standort: str, format_: str) -> list[Kurs]:
    return [k for k in kurse if k.standort == standort and k.format == format_]


def baue_report(standort: str, format_: str, limit: int, kurse: list[Kurs]) -> dict[str, Any]:
    # Sortierung: teuerste zuerst (Beispiel)
    kurse_sortiert = sorted(kurse, key=lambda k: k.preis_euro, reverse=True)
    top = kurse_sortiert[: max(limit, 0)]

    return {
        "anbieter": "SemaTrain",
        "erstellt_utc": datetime.now(timezone.utc).isoformat(),
        "filter": {"standort": standort, "format": format_, "limit": limit},
        "anzahl_treffer": len(kurse),
        "top_kurse": [asdict(k) for k in top],
    }


def schreibe_json(pfad: Path, payload: dict[str, Any]) -> None:
    pfad.parent.mkdir(parents=True, exist_ok=True)
    pfad.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")


# ---------- Main ----------
def main() -> int:
    setup_logging()
    args = parse_args()

    logger.info(
        "Start | standort=%s | format=%s | limit=%s | out=%s | dry_run=%s",
        args.standort, args.format, args.limit, args.out, args.dry_run
    )

    try:
        # Quelle wählen: API > CSV > Beispiel
        if args.api:
            logger.info("Quelle: API (%s)", args.api)
            kurse = lade_api(args.api, args.timeout)
        elif args.csv:
            pfad = Path(args.csv)
            logger.info("Quelle: CSV (%s)", pfad)
            kurse = lade_csv(pfad)
        else:
            logger.info("Quelle: integrierte Beispieldaten")
            kurse = lade_beispieldaten()

        gefiltert = filtere_kurse(kurse, args.standort, args.format)
        logger.info("Treffer: %s", len(gefiltert))

        report = baue_report(args.standort, args.format, args.limit, gefiltert)

        if args.dry_run:
            logger.info("DRY-RUN: Report nicht geschrieben. Vorschau:")
            logger.info(json.dumps(report, ensure_ascii=False))
        else:
            schreibe_json(Path(args.out), report)
            logger.info("OK | Report geschrieben: %s", args.out)

        return 0

    except Exception:
        logger.exception("Fehler beim Erstellen des Reports")
        return 2

    finally:
        logger.info("Ende")


if __name__ == "__main__":
    raise SystemExit(main())