mirror of
https://github.com/Jeuners/Logpy-AgentOne.git
synced 2026-09-09 15:02:33 +02:00
989d3ba85b
- Audio-Wiedergabe im Leitstand brach durch Auto-Refresh (setInterval) ab, Karten-DOM wurde per innerHTML neu gebaut waehrend ein <audio> lief. takt() pausiert jetzt, solange eine Aufnahme abgespielt wird. Zusaetzlich HTTP-Range-Unterstuetzung fuer /audio/ nachgeruestet (war vorher immer volle Datei statt 206 Partial Content). - Sicher-Modus: Rufnummern per Klick blurren, fuer Screenshots/Aufnahmen. - Notfall-Alarm: pulsierender Karten-Rahmen + Banner + minuetlicher Ton, solange ein Notfall nicht im letzten Stapel (Erledigt) liegt. - Favicon. - Neues unveraenderliches audit.log (append-only, macOS uappnd-Flag) als Nachweis fuer den Umgang mit Anrufdaten - protokoll.schreibe() spiegelt jetzt jedes Ereignis dorthin, zusaetzlich zum rotierenden verlauf.log. - pipe.loeschen: loescht erledigte Anrufe nach LOESCHFRIST_TAGE (lokal + Nextcloud), unbearbeitete Anrufe (nicht im letzten Stapel) bleiben immer unangetastet. Gedacht fuer taeglichen Cron-Lauf. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015vtvcyNkapuDh583643hNs
100 lines
3.5 KiB
Python
100 lines
3.5 KiB
Python
"""Loescht Anrufe nach Ablauf der Aufbewahrungsfrist (LOESCHFRIST_TAGE).
|
|
|
|
Nur Anrufe im letzten Stapel (per Default "Erledigt") werden angefasst -
|
|
unbearbeitete Anrufe bleiben unangetastet, egal wie alt: der Eingang ist
|
|
tabu. Loescht lokal UND (falls hochgeladen) die Nextcloud-Kopie - sonst
|
|
bringt lokales Loeschen fuer die Datenminimierung nichts. Jede Loeschung
|
|
landet unveraenderlich in audit.log (ueber protokoll.schreibe).
|
|
|
|
python3 -m pipe.loeschen # loescht faellige Anrufe
|
|
python3 -m pipe.loeschen --pruefen # zeigt nur, was faellig waere
|
|
|
|
Gedacht fuer taeglichen Cron/launchd-Lauf - siehe README fuer die Einrichtung.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import shutil
|
|
from datetime import datetime, timedelta
|
|
|
|
from . import config, protokoll, stapel, store
|
|
|
|
|
|
def _faellig(zeit: datetime) -> bool:
|
|
if not config.LOESCHFRIST_TAGE:
|
|
return False
|
|
return datetime.now() - zeit > timedelta(days=config.LOESCHFRIST_TAGE)
|
|
|
|
|
|
def laeuft(pruefen: bool = False) -> int:
|
|
"""Loescht (oder zeigt bei pruefen=True nur) faellige Anrufe. Gibt die Anzahl zurück."""
|
|
if not config.LOESCHFRIST_TAGE:
|
|
print("LOESCHFRIST_TAGE ist 0 - Loeschung ist aus, nichts zu tun.")
|
|
return 0
|
|
if not config.ABLAGE_LOKAL.is_dir():
|
|
return 0
|
|
|
|
letzter_stapel = stapel.erlaubt()[-1]
|
|
anzahl = 0
|
|
|
|
for meta_pfad in sorted(config.ABLAGE_LOKAL.glob("*/*/meta.json")):
|
|
ordner = meta_pfad.parent
|
|
try:
|
|
daten = json.loads(meta_pfad.read_text(encoding="utf-8"))
|
|
zeit = datetime.fromisoformat(daten["empfangen"])
|
|
except Exception:
|
|
continue
|
|
|
|
if stapel.lies(ordner) != letzter_stapel:
|
|
continue # unbearbeitet - Eingang ist tabu, egal wie alt
|
|
if not _faellig(zeit):
|
|
continue
|
|
|
|
hatte_cloud_kopie = (ordner / store.MARKER).is_file()
|
|
wartet_noch_auf_upload = (
|
|
config.nextcloud_aktiv() and config.NEXTCLOUD_UPLOAD and not hatte_cloud_kopie
|
|
)
|
|
if wartet_noch_auf_upload:
|
|
# Noch nicht gesichert - nicht loeschen, sonst waere die lokale
|
|
# Kopie die einzige und faellt komplett weg. Naechster Sync-Lauf
|
|
# (pipe.resync) holt das nach, dann greift die Loeschung.
|
|
continue
|
|
|
|
alter_tage = (datetime.now() - zeit).days
|
|
kennung = str(ordner.relative_to(config.ABLAGE_LOKAL))
|
|
|
|
if pruefen:
|
|
print(f"wuerde loeschen: {kennung} ({alter_tage} Tage alt)")
|
|
anzahl += 1
|
|
continue
|
|
|
|
if hatte_cloud_kopie:
|
|
store.loeschen_bei_nextcloud(ordner)
|
|
shutil.rmtree(ordner)
|
|
protokoll.schreibe(
|
|
"loeschung",
|
|
f"{ordner.name} geloescht (Frist {config.LOESCHFRIST_TAGE}d abgelaufen, "
|
|
f"war {alter_tage} Tage alt)",
|
|
id=kennung, alter_tage=alter_tage, hatte_nextcloud_kopie=hatte_cloud_kopie,
|
|
)
|
|
anzahl += 1
|
|
|
|
return anzahl
|
|
|
|
|
|
def main() -> None:
|
|
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
|
ap.add_argument("--pruefen", action="store_true",
|
|
help="nur anzeigen, was faellig waere - nichts loeschen")
|
|
args = ap.parse_args()
|
|
|
|
anzahl = laeuft(pruefen=args.pruefen)
|
|
if config.LOESCHFRIST_TAGE:
|
|
verb = "waeren faellig" if args.pruefen else "geloescht"
|
|
print(f"{anzahl} Anruf(e) {verb} (Frist: {config.LOESCHFRIST_TAGE} Tage, "
|
|
f"nur Stapel '{stapel.erlaubt()[-1]}').")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|