aboutsummaryrefslogtreecommitdiffstats
path: root/xembu.py
diff options
context:
space:
mode:
authorLeonard Kugis <leonard@kug.is>2025-12-23 00:08:47 +0100
committerLeonard Kugis <leonard@kug.is>2025-12-23 00:08:47 +0100
commitec7598f568ff59ecc1eb51572f84d866b0180501 (patch)
tree775944e30a140cc20857a316397d9538e9d1eff6 /xembu.py
parent78f4448a21614ed01b7c4e60eb496889bc58076d (diff)
downloadxembu-ec7598f568ff59ecc1eb51572f84d866b0180501.tar.gz
Removed unneccessary overhead
Diffstat (limited to 'xembu.py')
-rw-r--r--xembu.py131
1 files changed, 46 insertions, 85 deletions
diff --git a/xembu.py b/xembu.py
index 0ae60e6..02f1132 100644
--- a/xembu.py
+++ b/xembu.py
@@ -16,18 +16,17 @@ from modules.general import GeneralModule
from datetime import datetime
CSV_COLUMNS = [
- "Datum",
- "Nutzer",
- "Distributionsgruppe",
- "Distributionsflag",
- "Positionsbezeichnung",
- "Positionswert",
- "Modules",
- "Parameters",
- "Beleg",
+ "date",
+ "debitor",
+ "group",
+ "group_flag",
+ "position",
+ "value",
+ "modules",
+ "parameters",
+ "receipt",
]
-
def _pick_mono_font(size: int = 8) -> font_manager.FontProperties:
for fam in ["Inconsolata", "DejaVu Sans Mono", "monospace"]:
try:
@@ -37,13 +36,11 @@ def _pick_mono_font(size: int = 8) -> font_manager.FontProperties:
return font_manager.FontProperties(size=size)
def _decorate_figure(fig, mono_font, title: str, generated_at: str, page: int, total_pages: int):
- # Margins: links/rechts 2cm, oben/unten 1cm
margin_lr_cm = 2.0
margin_tb_cm = 1.0
- # Zusätzlicher Abstand (Bänder) zwischen Header/Footer und Content
- header_gap_cm = 1.3 # mehr Abstand nach unten
- footer_gap_cm = 2.0 # mehr Abstand nach oben (2-zeiliger Footer)
+ header_gap_cm = 1.3
+ footer_gap_cm = 2.0
cm_to_in = 1 / 2.54
margin_lr_in = margin_lr_cm * cm_to_in
@@ -58,27 +55,22 @@ def _decorate_figure(fig, mono_font, title: str, generated_at: str, page: int, t
header_gap = header_gap_in / h_in
footer_gap = footer_gap_in / h_in
- # Content-Bereich: innerhalb der Margins + zusätzlich Platz für Header/Footer
top = 1 - my - header_gap
bottom = my + footer_gap
if top <= bottom:
- # Fallback, falls es zu eng wird
top = 1 - my
bottom = my
fig.subplots_adjust(left=mx, right=1 - mx, top=top, bottom=bottom)
- # Header/Footer Positionen: jeweils an der inneren Kante der Margins
left_x = mx
right_x = 1 - mx
header_y = 1 - my
footer_y = my
- # Kopfzeile
fig.text(left_x, header_y, title, ha="left", va="top", fontproperties=mono_font, fontsize=9)
fig.text(right_x, header_y, generated_at, ha="right", va="top", fontproperties=mono_font, fontsize=9)
- # Fußzeile links (zweizeilig)
footer_left = (
"xembu - eXtensible Event-based Multiuser Bookkeeping Utility\n"
"Copyright (C) 2024 Leonard Kugis\n"
@@ -87,7 +79,6 @@ def _decorate_figure(fig, mono_font, title: str, generated_at: str, page: int, t
fig.text(left_x, footer_y, footer_left, ha="left", va="bottom",
fontproperties=mono_font, fontsize=7, linespacing=1.1)
- # Fußzeile rechts
fig.text(right_x, footer_y, f"{page} / {total_pages}", ha="right", va="bottom",
fontproperties=mono_font, fontsize=8)
@@ -112,21 +103,18 @@ def parse_value_unit(s: str):
num_str = " ".join(parts[:-1]).strip().replace(",", ".").replace("€", "").strip()
return float(num_str), unit
-
def parse_modules_list(s: str) -> List[str]:
if s is None or (isinstance(s, float) and pd.isna(s)):
return []
mods = [m.strip() for m in str(s).split(",")]
return [m for m in mods if m]
-
def parse_groups_list(s: str) -> List[str]:
if s is None or (isinstance(s, float) and pd.isna(s)):
return []
gs = [g.strip() for g in str(s).split(",")]
return [g for g in gs if g]
-
def parse_parameters_list(s: str) -> List[tuple]:
if s is None or (isinstance(s, float) and pd.isna(s)):
return []
@@ -157,24 +145,23 @@ def parse_parameters_list(s: str) -> List[tuple]:
tuples.append(tuple(vals))
return tuples
-
def parse_csv(path: str) -> pd.DataFrame:
df = _read_csv_flexible(path)
- df["Datum"] = pd.to_datetime(df["Datum"], format="%Y-%m-%d-%H-%M-%S", errors="coerce")
- df["Nutzer"] = df["Nutzer"].astype(str).str.strip()
- df["Distributionsflag"] = df["Distributionsflag"].astype(str).str.strip().str.upper()
- df["Positionsbezeichnung"] = df["Positionsbezeichnung"].astype(str).str.strip()
+ df["date"] = pd.to_datetime(df["date"], format="%Y-%m-%d-%H-%M-%S", errors="coerce")
+ df["debitor"] = df["debitor"].astype(str).str.strip()
+ df["group_flag"] = df["group_flag"].astype(str).str.strip().str.upper()
+ df["position"] = df["position"].astype(str).str.strip()
- df["dist_groups"] = df["Distributionsgruppe"].apply(parse_groups_list)
- df["modules_list"] = df["Modules"].apply(parse_modules_list)
- df["params_list"] = df["Parameters"].apply(parse_parameters_list)
+ df["dist_groups"] = df["group"].apply(parse_groups_list)
+ df["modules_list"] = df["modules"].apply(parse_modules_list)
+ df["params_list"] = df["parameters"].apply(parse_parameters_list)
- vals_units = df["Positionswert"].apply(parse_value_unit)
- df["value"] = vals_units.apply(lambda x: x[0])
+ vals_units = df["value"].apply(parse_value_unit)
+ df["val"] = vals_units.apply(lambda x: x[0])
df["unit"] = vals_units.apply(lambda x: x[1])
- df["Beleg"] = df["Beleg"].where(df["Beleg"].notna(), "")
+ df["receipt"] = df["receipt"].where(df["receipt"].notna(), "")
return df
@@ -193,32 +180,32 @@ def _build_positions_table_figs(df: pd.DataFrame, base_dir: str, mono_font):
figures = []
columns = [
- "Datum", "Nutzer", "Distributionsgruppe", "Flag",
- "Positionsbezeichnung", "Positionswert",
- "Modules", "Parameters", "Beleg", "SHA1",
+ "Date", "Debitor", "Group", "Flag",
+ "Position", "Value",
+ "Modules", "Parameters", "Receipt", "SHA1",
]
table_data = []
- for _, row in df.sort_values("Datum").iterrows():
- sha1 = compute_hash(str(row["Beleg"]), base_dir=base_dir) if row["Beleg"] else None
+ for _, row in df.sort_values("date").iterrows():
+ sha1 = compute_hash(str(row["receipt"]), base_dir=base_dir) if row["receipt"] else None
sha1_fmt = ""
if sha1:
sha1_fmt = sha1[: len(sha1) // 2] + "\n" + sha1[len(sha1) // 2 :]
- groups_str = ", ".join(row["dist_groups"]) if isinstance(row["dist_groups"], list) else str(row["Distributionsgruppe"])
- mods_str = ", ".join(row["modules_list"]) if isinstance(row["modules_list"], list) else str(row["Modules"])
- params_str = str(row["params_list"]) if isinstance(row["params_list"], list) else str(row["Parameters"])
+ groups_str = ", ".join(row["dist_groups"]) if isinstance(row["dist_groups"], list) else str(row["group"])
+ mods_str = ", ".join(row["modules_list"]) if isinstance(row["modules_list"], list) else str(row["modules"])
+ params_str = str(row["params_list"]) if isinstance(row["params_list"], list) else str(row["parameters"])
table_data.append([
- row["Datum"].strftime("%Y-%m-%d %H:%M:%S") if pd.notna(row["Datum"]) else "INVALID",
- row["Nutzer"],
+ row["date"].strftime("%Y-%m-%d %H:%M:%S") if pd.notna(row["date"]) else "INVALID",
+ row["debitor"],
groups_str,
- row["Distributionsflag"],
- row["Positionsbezeichnung"],
- f"{row['value']:.4f} {row['unit']}".strip(),
+ row["group_flag"],
+ row["position"],
+ f"{row['val']:.4f} {row['unit']}".strip(),
mods_str,
params_str,
- str(row["Beleg"]) if row["Beleg"] else "",
+ str(row["receipt"]) if row["receipt"] else "",
sha1_fmt,
])
@@ -265,7 +252,6 @@ def _build_positions_table_figs(df: pd.DataFrame, base_dir: str, mono_font):
return figures
-
def _separator_page(pdf: PdfPages, title: str, mono_font):
fig, ax = plt.subplots(figsize=(8.27, 11.69))
ax.axis("off")
@@ -338,16 +324,14 @@ def create_pdf(
generated_at = datetime.now().strftime("%Y-%m-%d-%H-%M-%S")
- # 1) Alle Seiten als Figures sammeln (damit wir total_pages kennen)
figs: List[plt.Figure] = []
figs.extend(_build_positions_table_figs(df, base_dir=base_dir, mono_font=mono_font))
figs.extend(_build_frame_figs(module_frames, mono_font=mono_font))
figs.extend(_build_bigframe_figs(module_bigframes, mono_font=mono_font))
- figs.extend(module_pages) # bereits fertige Figures aus Modulen
+ figs.extend(module_pages)
total_pages = len(figs)
- # 2) Speichern mit Header/Footer + Seitenzählung
with PdfPages(pdf_path) as pdf:
for i, fig in enumerate(figs, start=1):
_decorate_figure(fig, mono_font, title=title, generated_at=generated_at, page=i, total_pages=total_pages)
@@ -355,50 +339,37 @@ def create_pdf(
plt.close(fig)
def create_bundle(archive_path: str, csv_path: str, df: pd.DataFrame, base_dir: str, pdf_path: Optional[str] = None):
- """
- Bundle enthält: CSV, optional PDF, und alle Belege (relative Pfade aus 'Beleg' relativ zu base_dir).
- Ausgabe: .tar.zst (über externes zstd).
- """
os.makedirs(os.path.dirname(os.path.abspath(archive_path)) or ".", exist_ok=True)
- # Wir bauen ein temporäres .tar daneben und komprimieren danach.
tar_path = archive_path
if tar_path.endswith(".zst"):
- tar_path = tar_path[:-4] # strip ".zst"
+ tar_path = tar_path[:-4]
if not tar_path.endswith(".tar"):
tar_path = tar_path + ".tar"
- # Sammle Belege
beleg_paths = []
- for p in df["Beleg"].astype(str).tolist():
+ for p in df["receipt"].astype(str).tolist():
p = p.strip()
if p:
beleg_paths.append(p)
with tarfile.open(tar_path, "w") as tar:
- # CSV
tar.add(csv_path, arcname=os.path.basename(csv_path))
- # PDF optional
if pdf_path and os.path.exists(pdf_path):
tar.add(pdf_path, arcname=os.path.basename(pdf_path))
- # Belege
missing = []
for rel in sorted(set(beleg_paths)):
abs_path = rel if os.path.isabs(rel) else os.path.join(base_dir, rel)
if os.path.exists(abs_path):
- # arcname: möglichst den relativen Pfad behalten
arcname = os.path.basename(rel) if os.path.isabs(rel) else rel
tar.add(abs_path, arcname=arcname)
else:
missing.append(rel)
- # zstd komprimieren → archive_path
- # zstd -o <archive> <tar>
subprocess.run(["zstd", "-T0", "-o", archive_path, tar_path], check=True)
- # tar löschen (zstd bekommt eine Kopie)
try:
os.remove(tar_path)
except Exception:
@@ -409,13 +380,12 @@ def create_bundle(archive_path: str, csv_path: str, df: pd.DataFrame, base_dir:
for m in missing:
print(f" - {m}")
-
def main():
parser = argparse.ArgumentParser()
- parser.add_argument("csv", help="Pfad zur CSV-Datei")
- parser.add_argument("--title", "-t", help="Titel für PDF-Kopfzeile (optional)")
- parser.add_argument("--pdf", "-p", help="Pfad zur Ziel-PDF (optional)")
- parser.add_argument("--bundle", "-b", help="Pfad zum Bundle (.tar.zst), enthält CSV, PDF (falls erzeugt) und Belege (optional)")
+ parser.add_argument("csv", help="CSV path")
+ parser.add_argument("--title", "-t", help="PDF header title (optional)")
+ parser.add_argument("--pdf", "-p", help="PDF path (optional)")
+ parser.add_argument("--bundle", "-b", help="Path to bundle (.tar.zst), containing CSV, PDF and receipts (optional)")
args = parser.parse_args()
csv_path = os.path.abspath(args.csv)
@@ -424,20 +394,17 @@ def main():
title = args.title if args.title else os.path.basename(csv_path)
df = parse_csv(csv_path)
- if df["Datum"].isna().any():
- bad = df[df["Datum"].isna()][CSV_COLUMNS]
+ if df["date"].isna().any():
+ bad = df[df["date"].isna()][CSV_COLUMNS]
raise ValueError(f"Ungültige Datumsangaben in folgenden Zeilen:\n{bad}")
want_pdf = bool(args.pdf)
mono_font = _pick_mono_font(size=8)
- # Module-Registry
modules: Dict[str, Module] = {
"general": GeneralModule(),
- # weitere Module später hier registrieren
}
- # Modulzuordnung aus CSV
rows_for_module: Dict[str, List[int]] = {}
for idx, row in df.iterrows():
for m in row["modules_list"]:
@@ -445,10 +412,8 @@ def main():
results: List[ModuleResult] = []
- # General immer
results.append(modules["general"].process(df, context={"base_dir": base_dir, "want_pdf": want_pdf, "mono_font": mono_font}))
- # weitere Module optional
for mod_name, indices in rows_for_module.items():
if mod_name == "general":
continue
@@ -459,31 +424,27 @@ def main():
subdf = df.loc[indices].copy()
results.append(mod.process(subdf, context={"base_dir": base_dir, "want_pdf": want_pdf, "mono_font": mono_font}))
- # ---- NEU: Konsolen-Auswertung je Modul
print("\n===== Auswertung =====")
for r in results:
print(r.summary_text)
print("")
- # PDF optional
if args.pdf:
module_frames: List[Frame] = []
- module_bigframes: List[BigFrame] = [] # NEU
+ module_bigframes: List[BigFrame] = []
module_pages: List[plt.Figure] = []
for r in results:
module_frames.extend(r.frames)
- module_bigframes.extend(r.bigframes) # NEU
+ module_bigframes.extend(r.bigframes)
module_pages.extend(r.pages)
create_pdf(df, module_frames, module_bigframes, module_pages, args.pdf, mono_font, base_dir=base_dir, title=title)
print(f"[OK] PDF geschrieben: {args.pdf}")
- # Bundle optional (enthält CSV + ggf. PDF + Belege)
if args.bundle:
create_bundle(args.bundle, csv_path, df, base_dir=base_dir, pdf_path=args.pdf if args.pdf else None)
print(f"[OK] Bundle geschrieben: {args.bundle}")
-
if __name__ == "__main__":
main()