"""Der Werkzeug-Ordner auf diesem Rechner: lesen, ablegen, vergleichen.
|
|
Das ist der Teil, den eine Weboberflaeche nicht kann, und der einzige Grund,
|
warum es diesen Client gibt.
|
|
Der Kniff gegen jedes Raten: Wir legen die Dateien selbst ab und merken uns dabei
|
ihre Pruefsumme. Beim naechsten Start wird nachgerechnet — stimmt sie, wissen wir
|
sicher, welche Fassung dort liegt. Stimmt sie nicht, hat jemand von Hand
|
eingegriffen, und genau das wird gemeldet statt geschaetzt.
|
"""
|
|
from __future__ import annotations
|
|
import hashlib
|
import json
|
from dataclasses import dataclass
|
from enum import Enum
|
from pathlib import Path
|
|
# Liegt im Werkzeug-Ordner selbst. Damit zieht der Stand mit, wenn jemand den
|
# Ordner verschiebt oder auf einen anderen Rechner kopiert.
|
STANDSDATEI = ".bibliothek-stand.json"
|
|
# Endung fuer die verdraengte Vorgaengerdatei. Ein laufendes Programm laesst sich
|
# unter Windows umbenennen, aber nicht ueberschreiben — daher dieser Umweg.
|
ALTLAST_ENDUNG = ".alt"
|
|
|
class Zustand(str, Enum):
|
FEHLT = "fehlt"
|
"""Nichts abgelegt — das Werkzeug ist auf diesem Rechner nicht vorhanden."""
|
|
AKTUELL = "aktuell"
|
"""Liegt da und entspricht genau dem, was die Bibliothek anbietet."""
|
|
VERALTET = "veraltet"
|
"""Liegt da, aber die Bibliothek bietet inzwischen eine andere Fassung an."""
|
|
FREMD = "fremd"
|
"""Liegt da, stammt aber nicht von uns — von Hand ersetzt oder veraendert."""
|
|
|
@dataclass
|
class Befund:
|
zustand: Zustand
|
dateiname: str = ""
|
abgelegteVersion: str = ""
|
pfad: str = ""
|
|
|
def pruefsumme(datei: Path) -> str:
|
"""SHA-256 im Strom — die Dateien sind teils dreistellig viele Megabyte gross."""
|
rechner = hashlib.sha256()
|
with datei.open("rb") as offen:
|
for brocken in iter(lambda: offen.read(1024 * 1024), b""):
|
rechner.update(brocken)
|
return rechner.hexdigest()
|
|
|
class Werkzeugordner:
|
def __init__(self, ordner: str | Path) -> None:
|
self.ordner = Path(ordner)
|
|
# --- Stand ---
|
|
def _standsdatei(self) -> Path:
|
return self.ordner / STANDSDATEI
|
|
def _stand_lesen(self) -> dict[str, dict[str, str]]:
|
datei = self._standsdatei()
|
if not datei.exists():
|
return {}
|
try:
|
gelesen = json.loads(datei.read_text(encoding="utf-8"))
|
except (OSError, json.JSONDecodeError):
|
return {}
|
return gelesen if isinstance(gelesen, dict) else {}
|
|
def _stand_schreiben(self, stand: dict[str, dict[str, str]]) -> None:
|
self.ordner.mkdir(parents=True, exist_ok=True)
|
self._standsdatei().write_text(
|
json.dumps(stand, indent=2, ensure_ascii=False), encoding="utf-8"
|
)
|
|
# --- Abfrage ---
|
|
def befund(self, software_id: int, angeboteneVersion: str) -> Befund:
|
"""Was liegt hier, und passt es zu dem, was die Bibliothek anbietet?"""
|
vermerk = self._stand_lesen().get(str(software_id))
|
if not vermerk:
|
return Befund(Zustand.FEHLT)
|
|
datei = self.ordner / vermerk.get("dateiname", "")
|
if not vermerk.get("dateiname") or not datei.exists():
|
return Befund(Zustand.FEHLT)
|
|
try:
|
tatsaechlich = pruefsumme(datei)
|
except OSError:
|
return Befund(Zustand.FREMD, vermerk.get("dateiname", ""), pfad=str(datei))
|
|
if tatsaechlich != vermerk.get("pruefsumme"):
|
return Befund(Zustand.FREMD, vermerk["dateiname"], vermerk.get("version", ""), str(datei))
|
|
zustand = Zustand.AKTUELL if vermerk.get("version") == angeboteneVersion else Zustand.VERALTET
|
return Befund(zustand, vermerk["dateiname"], vermerk.get("version", ""), str(datei))
|
|
# --- Ablegen ---
|
|
def ablegen(self, software_id: int, quelle: Path, dateiname: str, version: str) -> Path:
|
"""Legt eine geholte Datei ab und vermerkt ihre Pruefsumme.
|
|
Eine vorhandene gleichnamige Datei wird zuerst umbenannt statt geloescht:
|
Laeuft das Programm gerade, laesst Windows das Umbenennen zu, das
|
Ueberschreiben nicht. Der Rest wird beim naechsten Lauf weggeraeumt.
|
"""
|
self.ordner.mkdir(parents=True, exist_ok=True)
|
ziel = self.ordner / dateiname
|
|
if ziel.exists():
|
altlast = ziel.with_name(ziel.name + ALTLAST_ENDUNG)
|
altlast.unlink(missing_ok=True)
|
ziel.replace(altlast)
|
|
quelle.replace(ziel)
|
|
stand = self._stand_lesen()
|
stand[str(software_id)] = {
|
"dateiname": dateiname,
|
"version": version,
|
"pruefsumme": pruefsumme(ziel),
|
}
|
self._stand_schreiben(stand)
|
return ziel
|
|
def vergessen(self, software_id: int) -> None:
|
"""Nimmt einen Eintrag aus dem Stand — die Datei selbst bleibt liegen."""
|
stand = self._stand_lesen()
|
if stand.pop(str(software_id), None) is not None:
|
self._stand_schreiben(stand)
|
|
def altlasten_aufraeumen(self) -> int:
|
"""Loescht verdraengte Vorgaengerdateien, sofern sie niemand mehr haelt."""
|
if not self.ordner.exists():
|
return 0
|
entfernt = 0
|
for datei in self.ordner.glob("*" + ALTLAST_ENDUNG):
|
try:
|
datei.unlink()
|
entfernt += 1
|
except OSError:
|
continue # laeuft noch — beim naechsten Mal wieder versuchen
|
return entfernt
|
