EpiSupport-Zammad-Scraper/app/epirent.py

"""Überwachung der EPIRent-Version auf der öffentlichen Download-Seite.

Die Seite (Squarespace) enthält im Klartext einen Block der Form:

    <p>Aktuelle Version</p><p>14505 vom 30.06.2026</p>

Wir lesen Version + Datum aus, vergleichen mit dem zuletzt gesehenen Stand und
melden eine Änderung wie jede andere (Feed + E-Mail).
"""
import logging
import re
from datetime import datetime, timezone

import requests

from . import config, db

log = logging.getLogger("episupport.epirent")

SOURCE = "epirent"
# "14505 vom 30.06.2026"
_PATTERN = re.compile(r"(\d{4,6})\s*vom\s*(\d{1,2}\.\d{1,2}\.\d{4})", re.I)
_TAGS = re.compile(r"<[^>]+>")


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def parse_version(html: str) -> tuple[str, str] | None:
    """(Version, Datum) aus dem Seiten-HTML lesen."""
    if not html:
        return None
    # Bevorzugt direkt nach der Überschrift "Aktuelle Version" suchen
    idx = html.find("Aktuelle Version")
    if idx != -1:
        segment = " ".join(_TAGS.sub(" ", html[idx:idx + 600]).split())
        m = _PATTERN.search(segment)
        if m:
            return m.group(1), m.group(2)
    # Rückfall: im gesamten Dokument suchen
    m = _PATTERN.search(_TAGS.sub(" ", html))
    return (m.group(1), m.group(2)) if m else None


def fetch_version(timeout: int = 30) -> tuple[str, str] | None:
    r = requests.get(
        config.EPIRENT_URL,
        timeout=timeout,
        headers={"User-Agent": "episupport-monitor/1.0"},
    )
    r.raise_for_status()
    return parse_version(r.text)


def _due(conn) -> bool:
    """Ist der nächste Abruf fällig (Intervall abgelaufen)?"""
    row = conn.execute(
        "SELECT checked_at FROM versions WHERE source=?", (SOURCE,)
    ).fetchone()
    if not row or not row["checked_at"]:
        return True
    try:
        last = datetime.fromisoformat(row["checked_at"])
    except ValueError:
        return True
    age = (datetime.now(timezone.utc) - last).total_seconds()
    return age >= config.EPIRENT_CHECK_INTERVAL_SECONDS


def check(force: bool = False) -> dict | None:
    """Version prüfen. Gibt bei Änderung ein Change-Dict zurück, sonst None.

    Muss AUSSERHALB einer offenen Poller-Transaktion laufen (eigene Verbindung).
    """
    if not config.EPIRENT_CHECK_ENABLED:
        return None
    with db.get_conn() as conn:
        if not force and not _due(conn):
            return None
        previous = conn.execute(
            "SELECT * FROM versions WHERE source=?", (SOURCE,)
        ).fetchone()

    try:
        found = fetch_version()
    except requests.RequestException as e:
        log.warning("EPIRent-Seite nicht abrufbar: %s", e)
        return None
    if not found:
        log.warning("Version auf %s nicht gefunden (Seitenaufbau geändert?).",
                    config.EPIRENT_URL)
        return None

    version, released = found
    now = _now()

    # Erster Lauf: nur merken, nicht melden
    if previous is None:
        with db.get_conn() as conn:
            conn.execute(
                "INSERT INTO versions (source, version, released_at, checked_at, "
                "updated_at) VALUES (?,?,?,?,?)",
                (SOURCE, version, released, now, now))
        log.info("EPIRent-Version erstmals erfasst: %s vom %s", version, released)
        return None

    unchanged = (previous["version"] == version
                 and previous["released_at"] == released)
    if unchanged:
        with db.get_conn() as conn:
            conn.execute("UPDATE versions SET checked_at=? WHERE source=?",
                         (now, SOURCE))
        return None

    # --- Änderung erkannt ---
    old_txt = f"{previous['version']} vom {previous['released_at']}"
    new_txt = f"{version} vom {released}"
    log.info("Neue EPIRent-Version: %s -> %s", old_txt, new_txt)
    with db.get_conn() as conn:
        conn.execute(
            "UPDATE versions SET version=?, released_at=?, checked_at=?, "
            "updated_at=? WHERE source=?",
            (version, released, now, now, SOURCE))
        conn.execute(
            "INSERT INTO changes (ticket_id, ticket_number, ticket_title, field, "
            "label, old_value, new_value, detail, detected_at) "
            "VALUES (?,?,?,?,?,?,?,?,?)",
            (0, None, "EPIRent Download-Seite", "epirent_version",
             "Neue EPIRent-Version", old_txt, new_txt, config.EPIRENT_URL, now))
    return {
        "ticket_number": None,
        "ticket_title": "EPIRent Download-Seite",
        "field": "epirent_version",
        "label": "Neue EPIRent-Version",
        "old_value": old_txt,
        "new_value": new_txt,
        "detail": config.EPIRENT_URL,
    }


def current() -> dict | None:
    with db.get_conn() as conn:
        row = conn.execute(
            "SELECT * FROM versions WHERE source=?", (SOURCE,)).fetchone()
    return dict(row) if row else None