Server-Bibliothek
vor 18 Std. 98bce1b0da675715b2e5482c0ad61ba21135f62f
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
"""Bruecke zwischen Oberflaeche (pywebview) und Backend.
 
Jede oeffentliche Methode ist aus dem Frontend als
`window.pywebview.api.<methode>(...)` erreichbar und liefert JSON-taugliche Werte.
Fehler werden nicht geworfen, sondern als `{ok: false, meldung: ...}` gemeldet —
ein Ausnahmefehler ueber die Bruecke kaeme in der Oberflaeche nur als
unverstaendlicher Text an.
"""
 
from __future__ import annotations
 
import subprocess
import tempfile
from dataclasses import asdict
from pathlib import Path
from typing import Any
 
from . import bibliothek, konfiguration
from .werkzeuge import Werkzeugordner, Zustand
 
# Installationsprogramme kommen nicht zwischen die eigenstaendigen Werkzeuge —
# sie werden einmal ausgefuehrt und sind danach entbehrlich.
UNTERORDNER_INSTALLER = "Installationsprogramme"
 
# Vorerst melden nur unsere eigenen Werkzeuge Updates. Fremdprogramme kuemmern
# sich teils selbst darum, und wo nicht, fehlt uns ein verlaesslicher Weg, ihre
# installierte Fassung zu erkennen. Bewusst eng gehalten, leicht zu weiten.
EINSTUFUNG_MIT_UPDATEPRUEFUNG = "intern"
 
 
class Api:
    def __init__(self) -> None:
        self._einstellungen = konfiguration.laden()
        self._ordner = Werkzeugordner(self._einstellungen.werkzeugOrdner)
        self._ordner.altlasten_aufraeumen()
 
    # --- Einstellungen ---
 
    def einstellungen(self) -> dict[str, Any]:
        return {
            "einstellungen": asdict(self._einstellungen),
            "vorgaben": {
                "serverAdresse": konfiguration.STANDARD_SERVER,
                "werkzeugOrdner": konfiguration.STANDARD_WERKZEUG_ORDNER,
            },
        }
 
    def einstellungen_speichern(self, serverAdresse: str, werkzeugOrdner: str) -> dict[str, Any]:
        self._einstellungen = konfiguration.speichern(
            konfiguration.Einstellungen(serverAdresse=serverAdresse, werkzeugOrdner=werkzeugOrdner)
        )
        self._ordner = Werkzeugordner(self._einstellungen.werkzeugOrdner)
        return {"ok": True, **self.einstellungen()}
 
    # --- Katalog ---
 
    def katalog(self) -> dict[str, Any]:
        """Katalog der Bibliothek, angereichert um den Befund auf diesem Rechner."""
        try:
            eintraege = bibliothek.katalog(self._einstellungen.serverAdresse)
        except bibliothek.BibliothekNichtErreichbar as fehler:
            return {"ok": False, "meldung": str(fehler), "eintraege": []}
 
        angereichert = []
        for eintrag in eintraege:
            eigenstaendig = eintrag.get("programmart") == "eigenstaendig"
            befund = (
                self._ordner.befund(int(eintrag["id"]), str(eintrag.get("version") or ""))
                if eigenstaendig
                else None
            )
            angereichert.append(
                {
                    **eintrag,
                    "lokal": asdict(befund) if befund else None,
                    "ueberwacht": eigenstaendig
                    and eintrag.get("einstufung") == EINSTUFUNG_MIT_UPDATEPRUEFUNG,
                }
            )
        return {"ok": True, "eintraege": angereichert, "ordner": str(self._ordner.ordner)}
 
    def veraltete(self) -> list[dict[str, str]]:
        """Kurzfassung fuer die Meldung beim Start: was ist ueberwacht und nicht aktuell."""
        stand = self.katalog()
        if not stand["ok"]:
            return []
        return [
            {"name": eintrag["name"], "zustand": eintrag["lokal"]["zustand"]}
            for eintrag in stand["eintraege"]
            if eintrag["ueberwacht"]
            and eintrag["lokal"]
            and eintrag["lokal"]["zustand"] in (Zustand.VERALTET.value, Zustand.FREMD.value)
        ]
 
    # --- Holen ---
 
    def herunterladen(self, software_id: int, dateiname: str, version: str, programmart: str) -> dict[str, Any]:
        """Holt eine Datei und legt sie am passenden Ort ab.
 
        Eigenstaendige Programme kommen in den Werkzeug-Ordner und werden im Stand
        vermerkt; Installationsprogramme daneben in einen eigenen Unterordner, ohne
        Vermerk — was installiert ist, sagt der Ordner ohnehin nicht.
        """
        if not dateiname:
            return {"ok": False, "meldung": "Für diesen Eintrag liegt keine Datei bereit."}
 
        eigenstaendig = programmart == "eigenstaendig"
        with tempfile.TemporaryDirectory(prefix="bibliothek-") as zwischenlager:
            zwischenziel = Path(zwischenlager) / dateiname
            try:
                bibliothek.datei_holen(self._einstellungen.serverAdresse, int(software_id), zwischenziel)
            except bibliothek.BibliothekNichtErreichbar as fehler:
                return {"ok": False, "meldung": str(fehler)}
 
            try:
                if eigenstaendig:
                    abgelegt = self._ordner.ablegen(int(software_id), zwischenziel, dateiname, version)
                else:
                    ziel = self._ordner.ordner / UNTERORDNER_INSTALLER / dateiname
                    ziel.parent.mkdir(parents=True, exist_ok=True)
                    ziel.unlink(missing_ok=True)
                    zwischenziel.replace(ziel)
                    abgelegt = ziel
            except OSError as fehler:
                return {"ok": False, "meldung": f"Ablegen fehlgeschlagen: {fehler}"}
 
        return {
            "ok": True,
            "pfad": str(abgelegt),
            "meldung": (
                f"Abgelegt unter {abgelegt}. Lief das Programm gerade, wirkt die neue "
                "Fassung nach dem nächsten Start."
                if eigenstaendig
                else f"Abgelegt unter {abgelegt}. Nach der Installation kann die Datei gelöscht werden."
            ),
        }
 
    # --- Windows-Handreichungen ---
 
    def ordner_oeffnen(self, unterordner: str = "") -> dict[str, Any]:
        ziel = self._ordner.ordner / unterordner if unterordner else self._ordner.ordner
        ziel.mkdir(parents=True, exist_ok=True)
        subprocess.Popen(["explorer", str(ziel)])  # noqa: S603,S607 - fester Befehl
        return {"ok": True}
 
    def verknuepfung_anlegen(self, pfad: str, name: str) -> dict[str, Any]:
        """Legt eine Verknuepfung auf dem Desktop an — das kann kein Browser."""
        ziel = Path(pfad)
        if not ziel.exists():
            return {"ok": False, "meldung": "Die Datei liegt nicht (mehr) an diesem Ort."}
        skript = (
            "$verknuepfung = (New-Object -ComObject WScript.Shell)."
            "CreateShortcut([Environment]::GetFolderPath('Desktop') + '\\' + $env:VERKNUEPFUNG_NAME + '.lnk');"
            "$verknuepfung.TargetPath = $env:VERKNUEPFUNG_ZIEL;"
            "$verknuepfung.WorkingDirectory = Split-Path $env:VERKNUEPFUNG_ZIEL;"
            "$verknuepfung.Save()"
        )
        try:
            subprocess.run(  # noqa: S603
                ["powershell", "-NoProfile", "-NonInteractive", "-Command", skript],
                check=True,
                capture_output=True,
                timeout=30,
                env={**_umgebung(), "VERKNUEPFUNG_ZIEL": str(ziel), "VERKNUEPFUNG_NAME": _dateiname_saeubern(name)},
            )
        except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as fehler:
            return {"ok": False, "meldung": f"Verknüpfung fehlgeschlagen: {fehler}"}
        return {"ok": True, "meldung": f"Verknüpfung „{name}“ liegt auf dem Desktop."}
 
 
def _umgebung() -> dict[str, str]:
    import os
 
    return dict(os.environ)
 
 
def _dateiname_saeubern(name: str) -> str:
    verboten = '\\/:*?"<>|'
    gesaeubert = "".join(zeichen for zeichen in name if zeichen not in verboten).strip()
    return gesaeubert or "Programm"