Julian Appel
a19558fc6c
Python-Server ohne Abhängigkeiten, der die dBase/FoxPro-Dateien des originalen KFE-Viewers (KFE.DBF, VORTEXT, GRUPPEN, Parameter, Muster-LVs) direkt einliest, plus Single-Page-Frontend mit Live-Suche, Facettenfiltern, einstellbarer Kalkulation (Lohnsätze, Materialzuschlag, DEL-Notiz), Detailansicht und Merkliste mit CSV-Export. Docker-Image enthält nur die App; die lizenzpflichtigen KFE-Daten werden zur Laufzeit nach /data gemountet und sind nicht Teil des Repos. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
322 lines
12 KiB
Python
322 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""KFE-Viewer – leichtgewichtiger Web-Viewer für die ZVEH-Kalkulationshilfe (KFE).
|
||
|
||
Liest die dBase/FoxPro-Dateien des originalen KFE-Viewers (KFE.DBF, VORTEXT.DBF, ...)
|
||
direkt ein und stellt sie über einen kleinen HTTP-Server als Web-App bereit.
|
||
Keine Abhängigkeiten außer der Python-Standardbibliothek (>= 3.8).
|
||
|
||
python server.py # startet auf http://127.0.0.1:8765
|
||
python server.py --port 9000 --no-browser
|
||
python server.py --data PFAD/ZU/Daten # anderes Datenverzeichnis
|
||
python server.py --static kfe.html # erzeugt eine einzelne HTML-Datei ohne Server
|
||
"""
|
||
|
||
import argparse
|
||
import configparser
|
||
import gzip
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import re
|
||
import signal
|
||
import struct
|
||
import sys
|
||
import time
|
||
import webbrowser
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
from pathlib import Path
|
||
|
||
HERE = Path(__file__).resolve().parent
|
||
# Standard: ./data neben dem Skript, sonst die Verzeichnisstruktur der KFE-CD daneben
|
||
DEFAULT_DATA = next((p for p in (HERE / "data",
|
||
HERE.parent / "2020-05-12-NEU" / "0_software" / "KFE_ViewerCD3" / "Daten")
|
||
if (p / "KFE.DBF").exists()), HERE / "data")
|
||
|
||
# Mengeneinheiten-Codes der ZVEHNORM (abgeglichen mit den DATANORM-Exporten)
|
||
MEH = {"01": "Stk", "12": "cm", "13": "m", "23": "kg", "33": "l", "42": "cm²", "43": "m²", "53": "Std"}
|
||
|
||
|
||
# --------------------------------------------------------------------------- DBF
|
||
|
||
def read_dbf(path, encoding="cp850"):
|
||
"""Liest eine dBase/FoxPro-Tabelle. Memofelder werden als Blocknummer (int) geliefert."""
|
||
with open(path, "rb") as f:
|
||
head = f.read(32)
|
||
count, header_len, record_len = struct.unpack("<IHH", head[4:12])
|
||
fields = []
|
||
while True:
|
||
fd = f.read(32)
|
||
if not fd or fd[0] == 0x0D:
|
||
break
|
||
name = fd[:11].split(b"\0")[0].decode("ascii")
|
||
fields.append((name, chr(fd[11]), fd[16]))
|
||
f.seek(header_len)
|
||
data = f.read(count * record_len)
|
||
|
||
rows = []
|
||
for i in range(count):
|
||
rec = data[i * record_len:(i + 1) * record_len]
|
||
if not rec or rec[0:1] == b"*":
|
||
continue
|
||
pos, row = 1, {}
|
||
for name, typ, length in fields:
|
||
raw = rec[pos:pos + length]
|
||
pos += length
|
||
if typ == "M":
|
||
if length == 4:
|
||
row[name] = struct.unpack("<I", raw)[0]
|
||
else:
|
||
s = raw.strip(b" \0")
|
||
row[name] = int(s) if s.isdigit() else 0
|
||
elif typ == "L":
|
||
row[name] = raw in (b"T", b"t", b"Y", b"y")
|
||
else:
|
||
row[name] = raw.decode(encoding, "replace").strip()
|
||
rows.append(row)
|
||
return rows
|
||
|
||
|
||
class Memo:
|
||
"""Zugriff auf eine FoxPro-Memodatei (.FPT)."""
|
||
|
||
def __init__(self, path, encoding="cp850"):
|
||
self.data = Path(path).read_bytes()
|
||
self.block = struct.unpack(">H", self.data[6:8])[0] or 64
|
||
self.encoding = encoding
|
||
|
||
def get(self, block):
|
||
if not block:
|
||
return ""
|
||
off = block * self.block
|
||
_typ, length = struct.unpack(">II", self.data[off:off + 8])
|
||
raw = self.data[off + 8:off + 8 + length]
|
||
text = raw.decode(self.encoding, "replace").rstrip("\0\x1a ")
|
||
return text.replace("\r\n", "\n").replace("\r", "\n")
|
||
|
||
|
||
def num(s, default=0.0):
|
||
try:
|
||
return float(str(s).replace(",", "."))
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
|
||
def iso(d):
|
||
d = (d or "").strip()
|
||
return f"{d[:4]}-{d[4:6]}-{d[6:8]}" if len(d) >= 8 and d[:8].isdigit() else ""
|
||
|
||
|
||
# --------------------------------------------------------------------------- Laden
|
||
|
||
def load_params(data_dir):
|
||
p = {
|
||
"lohn": {"1": {"10": 43.75, "11": 90.77, "12": 67.22}, "2": {"10": 49.95, "11": 114.72, "12": 87.32}},
|
||
"zuschlag": [20.0, 35.0],
|
||
"mwst": 19.0,
|
||
"bzfakt": 1.0,
|
||
"basis": {"CU": 550.0, "AL": 200.0},
|
||
"stand": "",
|
||
"zeitraum": "",
|
||
}
|
||
std = data_dir / "STDPARAM.DAT"
|
||
if std.exists():
|
||
cp = configparser.ConfigParser()
|
||
cp.read_string(std.read_text("cp1252", "replace"))
|
||
s = cp["PARAMETER"] if cp.has_section("PARAMETER") else {}
|
||
p["lohn"]["1"]["10"] = num(s.get("LOHN1"), 43.75)
|
||
p["lohn"]["2"]["10"] = num(s.get("LOHN2"), 49.95)
|
||
p["zuschlag"] = [num(s.get("ZU1"), 20), num(s.get("ZU2"), 35)]
|
||
p["mwst"] = num(s.get("MWST"), 19)
|
||
p["bzfakt"] = num(s.get("BZFAKT"), 1)
|
||
lstd = data_dir / "LOHNSTD.DBF"
|
||
if lstd.exists():
|
||
rows = read_dbf(lstd)
|
||
if rows:
|
||
r = max(rows, key=lambda r: r["ZEIT"]) # aktuellster Zeitraum, z.B. "1920"
|
||
p["zeitraum"] = f"20{r['ZEIT'][:2]}/{r['ZEIT'][2:]}"
|
||
p["lohn"]["1"]["11"] = num(r["S1_PG11"])
|
||
p["lohn"]["2"]["11"] = num(r["S2_PG11"])
|
||
p["lohn"]["1"]["12"] = num(r["S1_PG12"])
|
||
p["lohn"]["2"]["12"] = num(r["S2_PG12"])
|
||
roh = data_dir / "ROHSTOFF.DBF"
|
||
if roh.exists():
|
||
for r in read_dbf(roh):
|
||
p["basis"][r["ROHART"]] = num(r["ROHBASIS"])
|
||
ver = data_dir / "DVERSION.DAT"
|
||
if ver.exists():
|
||
m = re.search(r"UPDATE=(\d\d)/(\d\d)/(\d\d)", ver.read_text("cp1252", "replace"))
|
||
if m:
|
||
p["stand"] = f"20{m.group(1)}-{m.group(2)}-{m.group(3)}"
|
||
return p
|
||
|
||
|
||
def load_projects(proj_dir):
|
||
projects = []
|
||
if not proj_dir.is_dir():
|
||
return projects
|
||
for f in sorted(proj_dir.glob("*.prj")):
|
||
lines = f.read_text("cp1252", "replace").splitlines()
|
||
items = []
|
||
for line in lines[1:]:
|
||
c = line.split(";")
|
||
if not c or not c[0] or c[0] == "V":
|
||
continue
|
||
if c[0] == "T":
|
||
text = c[1].strip() if len(c) > 1 else ""
|
||
if text:
|
||
items.append({"t": text})
|
||
continue
|
||
if not re.match(r"^\d\d\.\d\d\.\d\d", c[0]) or len(c) < 5:
|
||
continue
|
||
item = {"p": c[0].strip(), "n": num(c[4], 1)}
|
||
if num(c[1]) > 0:
|
||
item["lohn"] = num(c[1])
|
||
if num(c[2]) > 0:
|
||
item["zu"] = num(c[2])
|
||
items.append(item)
|
||
m = re.match(r"^(\d\d)(\d\d)(\d\d)_(.*)$", f.stem)
|
||
nr = f"{m.group(1)}.{m.group(2)}.{m.group(3)}" if m else ""
|
||
name = (m.group(4) if m else f.stem).replace("_", " ")
|
||
projects.append({"file": f.name, "nr": nr, "name": name, "items": items})
|
||
return projects
|
||
|
||
|
||
def load(data_dir):
|
||
t0 = time.time()
|
||
data_dir = Path(data_dir)
|
||
if not (data_dir / "KFE.DBF").exists():
|
||
sys.exit(f"KFE.DBF nicht gefunden in: {data_dir}\nBitte mit --data das Daten-Verzeichnis angeben.")
|
||
|
||
groups = {r["GR"]: r["BEZ"] for r in read_dbf(data_dir / "GRUPPEN.DBF")}
|
||
|
||
vt_memo = Memo(data_dir / "VORTEXT.FPT")
|
||
vortexte = {r["VT_ID"]: vt_memo.get(r["VT_BEZ"]) for r in read_dbf(data_dir / "VORTEXT.DBF")}
|
||
|
||
memo = Memo(data_dir / "KFE.FPT")
|
||
positions = []
|
||
for r in read_dbf(data_dir / "KFE.DBF"):
|
||
roh = []
|
||
for i in range(1, 5):
|
||
art = r.get(f"ROHART{i}", "")
|
||
if art:
|
||
roh.append([art, num(r[f"ROHGEW{i}"]), num(r[f"ROHBASIS{i}"]), num(r[f"ROHKORR{i}"])])
|
||
lg = r.get("LOHNGR", "")
|
||
dates = [d for d in (r["INFO_DAT"], r["KNOTIZ_DAT"], r["PNOTIZ_DAT"]) if d]
|
||
vt = r["METAVTNR"] if r["METAVTNR"] in vortexte else ""
|
||
positions.append({
|
||
"p": r["LPOSNR"],
|
||
"a": r["LPOSNR_A"],
|
||
"me": r["MEH"],
|
||
"ae": num(r["AE"]),
|
||
"mat": num(r["MAT"]),
|
||
"i": r["INFO"],
|
||
"k": memo.get(r["KNOTIZ"]),
|
||
"z": memo.get(r["PNOTIZ"]),
|
||
"v": vt,
|
||
"lg": lg if lg in ("11", "12") else "10",
|
||
"d": iso(max(dates)) if dates else "",
|
||
"r": roh,
|
||
})
|
||
positions.sort(key=lambda x: x["p"])
|
||
|
||
payload = {
|
||
"meta": {"params": load_params(data_dir), "meh": MEH, "quelle": str(data_dir)},
|
||
"groups": groups,
|
||
"vortexte": vortexte,
|
||
"positions": positions,
|
||
"projects": load_projects(data_dir / "Projekt"),
|
||
}
|
||
print(f" {len(positions)} Positionen, {len(vortexte)} Vortexte, {len(groups)} Gruppen, "
|
||
f"{len(payload['projects'])} Projekte geladen ({time.time() - t0:.1f}s)")
|
||
return payload
|
||
|
||
|
||
# --------------------------------------------------------------------------- HTTP
|
||
|
||
def make_handler(index_html, data_json):
|
||
data_gz = gzip.compress(data_json, 6)
|
||
etag = '"' + hashlib.md5(data_json).hexdigest()[:16] + '"'
|
||
|
||
class Handler(BaseHTTPRequestHandler):
|
||
def log_message(self, fmt, *args):
|
||
pass
|
||
|
||
def send(self, code, body, ctype, extra=None):
|
||
self.send_response(code)
|
||
self.send_header("Content-Type", ctype)
|
||
self.send_header("Content-Length", str(len(body)))
|
||
for k, v in (extra or {}).items():
|
||
self.send_header(k, v)
|
||
self.end_headers()
|
||
if self.command != "HEAD":
|
||
self.wfile.write(body)
|
||
|
||
def do_GET(self):
|
||
path = self.path.split("?", 1)[0]
|
||
if path in ("/", "/index.html"):
|
||
# index.html bei jedem Aufruf neu lesen, damit Änderungen ohne Neustart wirken
|
||
html = (HERE / "index.html").read_bytes() if (HERE / "index.html").exists() else index_html
|
||
self.send(200, html, "text/html; charset=utf-8", {"Cache-Control": "no-cache"})
|
||
elif path == "/api/data":
|
||
if self.headers.get("If-None-Match") == etag:
|
||
self.send_response(304)
|
||
self.end_headers()
|
||
return
|
||
hdr = {"ETag": etag, "Cache-Control": "no-cache"}
|
||
if "gzip" in self.headers.get("Accept-Encoding", ""):
|
||
hdr["Content-Encoding"] = "gzip"
|
||
self.send(200, data_gz, "application/json", hdr)
|
||
else:
|
||
self.send(200, data_json, "application/json", hdr)
|
||
else:
|
||
self.send(404, b"Not found", "text/plain")
|
||
|
||
do_HEAD = do_GET
|
||
|
||
return Handler
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(description="Web-Viewer für die ZVEH-Kalkulationshilfe (KFE)")
|
||
env = os.environ.get
|
||
ap.add_argument("--data", default=env("KFE_DATA", str(DEFAULT_DATA)),
|
||
help="Verzeichnis mit KFE.DBF (env KFE_DATA, default: %(default)s)")
|
||
ap.add_argument("--host", default=env("KFE_HOST", "127.0.0.1"),
|
||
help="Bind-Adresse, 0.0.0.0 für Zugriff im LAN (env KFE_HOST)")
|
||
ap.add_argument("--port", type=int, default=int(env("KFE_PORT", "8765")), help="Port (env KFE_PORT)")
|
||
ap.add_argument("--no-browser", action="store_true", default=env("KFE_NO_BROWSER") == "1",
|
||
help="Browser nicht automatisch öffnen (env KFE_NO_BROWSER=1)")
|
||
ap.add_argument("--static", metavar="DATEI", help="Einzelne HTML-Datei mit eingebetteten Daten erzeugen")
|
||
args = ap.parse_args()
|
||
|
||
print("KFE-Viewer: lade Daten aus", args.data)
|
||
payload = load(args.data)
|
||
data_json = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
|
||
index_html = (HERE / "index.html").read_bytes()
|
||
|
||
if args.static:
|
||
# "</" escapen, damit Texte im JSON das <script>-Tag nicht beenden können
|
||
inline = data_json.decode("utf-8").replace("</", "<\\/")
|
||
html = index_html.decode("utf-8").replace(
|
||
"<!--KFE_DATA-->", f"<script>window.KFE_DATA={inline};</script>", 1)
|
||
Path(args.static).write_text(html, "utf-8")
|
||
print(f" Statische Datei geschrieben: {args.static} ({len(html) / 1e6:.1f} MB)")
|
||
return
|
||
|
||
server = ThreadingHTTPServer((args.host, args.port), make_handler(index_html, data_json))
|
||
url = f"http://{'127.0.0.1' if args.host in ('0.0.0.0', '') else args.host}:{args.port}/"
|
||
print(f" Läuft auf {url} (Strg+C zum Beenden)")
|
||
if not args.no_browser:
|
||
webbrowser.open(url)
|
||
# SIGTERM (z.B. "docker stop") wie Strg+C behandeln – als PID 1 würde Python es sonst ignorieren
|
||
def stop(*_):
|
||
raise KeyboardInterrupt
|
||
signal.signal(signal.SIGTERM, stop)
|
||
try:
|
||
server.serve_forever()
|
||
except KeyboardInterrupt:
|
||
print("\nBeendet.")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|