KFE-Viewer: Web-Viewer für die ZVEH-Kalkulationshilfe
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>
This commit is contained in:
commit
a19558fc6c
10 changed files with 1404 additions and 0 deletions
322
server.py
Normal file
322
server.py
Normal file
|
|
@ -0,0 +1,322 @@
|
|||
#!/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()
|
||||
Loading…
Add table
Add a link
Reference in a new issue