17 Commits
43 changed files with 1642 additions and 358 deletions

No files matched your search

-46
View File
@@ -1,47 +1 @@
# Aufgaben: Tätigkeitsnachweis-Generator (Streamlit App) – PDF-Migration
> **Migration**: Das XLSX-Template war schreibgeschützt und wurde durch ein PDF-Template ersetzt.
> Alle `_Field`-Platzhalter im PDF werden durch echte Daten ersetzt.
## 🔄 Migration: XLSX → PDF Template ✅
- [x] **M1**: `requirements.txt` → `PyMuPDF` hinzufügen
- [x] **M2**: `config/settings.ini` → `filename_pattern` auf `.pdf`-Endung ändern
- [x] **M3**: `config/output-fields.json` → `target_cell` entfernen, `pdf_field` hinzufügen (Mapping source_field → PDF-Platzhalter-Name)
- [x] **M4**: `src/config.py` → `get_available_templates()` auf `.pdf`-Dateien umstellen
- [x] **M5**: `src/validator.py` → `validate_template()` für PDF angepasst (prüft `_Field`-Platzhalter via PyMuPDF)
- [x] **M6**: `src/writer.py` → Komplett neu: PDF-basierte Generierung mit PyMuPDF
- Pre-Scan: Alle `_Field`-Platzhalter im Template-PDF finden (Position, Schriftgröße)
- Für jeden Mitarbeiter: neues PDF, pro Monat eine Seite (Template-Kopie), `_Field`-Texte redacten + ersetzen
- Dateiname: `{nachname}_{vorname}_Tätigkeitsnachweis.pdf`
- [x] **M7**: `app.py` → Template-Validierung für PDF angepasst (Dropdown zeigt `.pdf`, Validierung prüft `_Field`-Marker)
- [x] **M8**: `coldrun.py` → An PDF-Template angepasst
- [x] **M9**: `tests/test_e2e.py` → Tests für PDF-Output umgeschrieben (PyMuPDF statt openpyxl zur Output-Validierung)
- [x] **M10**: `.copilot/PLAN.md` und `.copilot/TASKS.md` aktualisiert
## 🔍 Analyse (abgeschlossen)
- [x] **A1**: Input-Datei `input/2025 JM.xlsx` analysiert
- [x] **A2**: PDF-Template `template/Tätigkeitsnachweise Stellenförderung.pdf` liegt vor (mit `_Field`-Platzhaltern)
## ⚙️ Konfiguration (migriert)
- [x] **K1**: `config/input-fields.json` — unverändert gültig
- [x] **K2**: `config/output-fields.json` — `target_cell` entfernt, `pdf_field` hinzugefügt
- [x] **K3**: `config/settings.ini` — filename_pattern auf `.pdf`
## 🧱 Core-Module
- [x] **C1**: `requirements.txt` — streamlit, openpyxl, PyMuPDF
- [x] **C2**: `src/__init__.py` + `src/models.py` — `OutputMapping` um `pdf_field` erweitert
- [x] **C3**: `src/config.py` — Template-Suche auf `.pdf` umgestellt
- [x] **C4**: `src/parser.py` — unverändert
- [x] **C5**: `src/validator.py` — `validate_template()` für PDF via PyMuPDF
- [x] **C6**: `src/writer.py` — Komplett neu mit PyMuPDF (redact + insert_text)
## 🖥️ UI & Start
- [x] **U1**: `app.py` — PDF-Template-Validierung angepasst
- [x] **U2**: `start.bat` — unverändert
- [x] **U3**: `coldrun.py` — An PDF angepasst, funktioniert
- [x] **U4**: `README.md` — vorhanden
## ✅ Test
- [x] **T1**: E2E-Test mit realer Input-Datei und PDF-Template → 5/5 Tests bestanden
- [x] **T2**: Edge Cases — Umlaute korrekt im PDF, fehlende Marker (Ort/Datum) werden übersprungen
+62 -8
View File
@@ -1,9 +1,63 @@
input/
output/
.venv/
__pycache__/
### macOS
# Finder metadata
.DS_Store
.~lock.*#
*.xlsx#
~*.xlsx
/versions
# Thumbnails
._*
# Custom folder icons
Icon␍
# Volume root files
.DocumentRevisions-V100
.fseventsd
.Spotlight-V100
.TemporaryItems
.Trashes
.VolumeIcon.icns
.com.apple.timemachine.donotpresent
### Windows
# Windows thumbnail cache files
Thumbs.db
# Folder config file
[Dd]esktop.ini
# Recycle Bin used on file shares
$RECYCLE.BIN/
# Windows shortcuts
*.lnk
### Linux
# Backup files
*~
# Temporary files from deleted open files
.fuse_hidden*
# KDE directory preferences
.directory
# Linux trash folder
.Trash-*
# NFS temporary files
.nfs*
### VS Code
# VSCode settings (keep shared configuration)
.vscode/*
!.vscode/settings.json
!.vscode/tasks.json
!.vscode/launch.json
!.vscode/extensions.json
# Local History for Visual Studio Code
.history/
# Built Visual Studio Code Extensions
*.vsix
3_Archiv
2_build
+9
View File
@@ -0,0 +1,9 @@
input/
output/
.venv/
__pycache__/
.DS_Store
.~lock.*#
*.xlsx#
~*.xlsx
/versions
@@ -7,6 +7,74 @@ und dieses Projekt folgt [Semantic Versioning](https://semver.org/lang/de/).
---
## [1.3] – 2026-08-11
### Hinzugefügt
- **Monatsauswahl pro Mitarbeiter**: Tätigkeitsnachweise können nun für individuell gewählte Monate erzeugt werden.
- Die Monatsauswahl erscheint rechts neben der Mitarbeiterliste für die zuletzt angeklickte Person.
- Die Auswahl wird in Kurzform direkt hinter dem Namen angezeigt, z. B. `(Jan, Mär, Dez)`.
- Standardmäßig sind die letzten zwei vorhandenen Monate ausgewählt.
- Schaltflächen zum Auswählen bzw. Abwählen aller Monate ergänzen die Monatsliste.
- **Versionsprüfung**: Die App prüft beim Start im zentralen SMB-Verzeichnis auf neuere
`Tätigkeitsnachweis-Generator vX.Y.Z.zip`- und künftig auch `.exe`-Dateien.
- **Versionsanzeige**: Die aktuell installierte Programmversion wird dezent am unteren rechten Rand der App angezeigt.
### Geändert
- **Mitarbeiterauswahl**: Mitarbeiter sind nach dem Einlesen einer Input-Datei nicht mehr standardmäßig ausgewählt.
Die Buttons zum Auswählen bzw. Abwählen aller Mitarbeiter befinden sich direkt über der Mitarbeiterliste.
- **Update-Hinweis**: Bei einer neueren zentralen Version wird ein schwebender Hinweis oben rechts angezeigt,
einschließlich Schließen- und Ordner-Schaltfläche.
---
## [1.2.1] – 2026-06-30
### Behoben
- **Build-Pipeline**: fix broken server start and page call
---
## [1.2] – 2026-06-30
### Hinzugefügt
- **Build-Pipeline** (`1_build-pipeline/`): Skripte zum Erstellen einer standalone `.exe`:
- `build.bat` / `build.py`: Orchestrierter Build mit **Semantic Versioning** – liest aktuelle Version
aus `CHANGELOG.md`, bump't nach Wahl (major/minor/patch) und aktualisiert das Changelog automatisch
- `build.spec`: PyInstaller-Spec-Datei mit allen Hidden Imports für Streamlit, openpyxl und PyMuPDF
- `run.py`: Einstiegspunkt für die gebundelte `.exe` – setzt Pfade via `APP_BASE_DIR`,
startet den Streamlit-Server und öffnet den Browser
- Ausgabe nach `2_build/` mit fertiger Ordnerstruktur (`config/`, `input/`, `template/`, `output/`)
- **`app.py`** unterstützt jetzt `APP_BASE_DIR`-Umgebungsvariable für Pfadauflösung im gebundelten Modus
### Geändert
- **Statische Schriftgröße** (`src/writer.py`): Schriftgröße wird nicht mehr dynamisch aus dem PDF-Template extrahiert,
sondern statisch aus `settings.ini` (`[font]` → `fontsize`) gelesen, einheitlich für alle Felder.
`AppConfig` um `fontsize` erweitert, `config.py` lädt den Wert.
- **Autoskalierung entfernt** (`src/writer.py`): Die automatische Schriftverkleinerung für lange Texte
(`Qualifikation_Field`, `Funktion_Field`) wurde ersatzlos gestrichen – alle Texte haben exakt dieselbe Größe.
- **Vertikale Textzentrierung** (`src/writer.py`): Einzeilige Felder werden jetzt vertikal mittig im
Marker-Rechteck platziert. Mehrzeilige Felder (`Tätigkeiten_Field`, `Dienstreisen_Field`) bleiben
oben ausgerichtet.
- **Doppelte Felder unterstützt** (`src/writer.py`): `Datum_Field` und andere Marker, die mehrfach im
PDF-Template vorkommen, werden jetzt alle mit demselben Wert befüllt.
- **Stellenanteil-Formatierung** (`src/writer.py`): `Stellenanteil in %` wird mit deutschem Dezimalkomma
und zwei Nachkommastellen ausgegeben (z. B. `75` → `75,00`).
---
## [1.1] – 2026-06-24
### Hinzugefügt
@@ -87,8 +87,26 @@ Mapping von Input-Feldern auf Template-Zellen:
## Tests
End-to-End-Tests parsen die reale Input-Datei (`input/2025 JM.xlsx`) und das reale
Template (`template/Tätigkeitsnachweise Stellenförderung.xlsx`) und validieren
Beim Start über `start.bat` oder die neu gebaute EXE erscheint bei einer bereits
laufenden Programminstanz ein Ja/Nein-Dialog. „Ja“ beendet die bisherige Instanz
und startet die neue; „Nein“ lässt die bisherige Instanz geöffnet und bricht den
neuen Start ab. Aktuelle Auswahlen gehen beim Beenden verloren. Ein fremdes
Programm auf Port 8501 wird nicht beendet.
Unter Windows führt `test.bat` alle Tests im Ordner `tests/` aus. Das Skript
verwendet die Projektumgebung `.venv` und installiert bei Bedarf `pytest`.
Per Doppelklick bleibt das Ergebnisfenster geöffnet. Für automatisierte Aufrufe:
```powershell
.\0_application\test.bat --no-pause
```
Dieser Aufruf erfolgt aus dem Projekt-Hauptordner. Bei fehlgeschlagenen Tests
liefert das Skript einen Fehlercode und führt die übrigen Tests trotzdem aus.
Die Tests laufen nicht automatisch beim Programmstart oder Build.
End-to-End-Tests parsen die feste Testdatei (`tests/fixture/2026 JM.xlsx`) und das
PDF-Template (`template/Tätigkeitsnachweise Stellenförderung.pdf`) und validieren
den gesamten Generierungs-Workflow.
```bash
@@ -96,12 +114,13 @@ den gesamten Generierungs-Workflow.
.venv/bin/pip install pytest
# Tests ausführen
.venv/bin/python -m pytest tests/test_e2e.py -v
.venv/bin/python -m pytest tests -v
```
Die Tests prüfen:
- Vorhandensein der Input- und Template-Dateien
- Korrekte Parsierung (168 Datensätze, 19 Personen)
- Vollständige Pipeline für die erste Person („Draßdo, Cornelia"): 12 Monats-Sheets
mit allen Zellinhalten (Antragsnummer, Name, Stellenanteil, Qualifikation, Monat,
Funktion, Tätigkeit, Ort, Datum)
- Korrekte Parsierung (100 Datensätze, 17 Personen)
- Vollständige PDF-Pipeline für die erste Person („Bardaus, Madalina“): fünf Monate
von Januar bis Mai 2026 mit zehn Seiten, Antragsnummer, Name, Stellenanteil,
Monat, Funktion, Tätigkeit, Ort und korrekten Monatsenddaten
- Neustartlogik unter Windows mit simulierten Prozessen
+57
View File
@@ -0,0 +1,57 @@
# Tätigkeitsnachweis-Generator
Programm zur automatisierten Erstellung von Tätigkeitsnachweisen aus Excel-Daten.
Die aktuellesten Programmversion find sich immer hier:
"j:\Fachbereiche\BO\Produktionsschule\Maßnahmeunterlagen\20_Personal\05_Tätigkeitsnachweise ESF\Nachweise-automatisiert-generieren\"
Um das Programm zu starten: Doppelklick auf `Tätigkeitsnachweis-Generator.exe`
## Bedienung
1. **Input-Datei wählen** – Dropdown zeigt alle `.xlsx`-Dateien aus dem `input/`-Ordner
- Nach Auswahl wird die Datei sofort validiert
- Angezeigt werden: Anzahl Mitarbeiter, Antragsnummern, Einträge pro Monat
2. **Template wählen** – Dropdown zeigt alle `.xlsx`-Dateien aus dem `template/`-Ordner
- Nach Auswahl wird das Template sofort validiert
- Angezeigt wird: Anzahl gefundener Zielfelder
3. **Mitarbeiter auswählen** – Multiselect-Liste aller gefundenen Mitarbeiter
- Standardmäßig ist keiner ausgewählt
- Nicht selektierte Mitarbeiter werden bei der Generierung übersprungen
4. **Mitarbeitermonate auswählen** – Multiselect-Liste aller verfügbaren Monate im Dokument. Vorselektiert sind die aktuell letzten beiden Monate.
5. **"Tätigkeitsnachweise generieren"** klicken
- Die generierten Dateien werden im `output/`-Ordner abgelegt
## Konfiguration
Alle Einstellungen liegen im `config/`-Ordner:
### `settings.ini`
- `[paths]` – Ordner für Input, Template, Output
- `[output]` – Dateinamen-Pattern (`{first_name}`, `{last_name}`)
- `[parsing]` – Monats-Sheet-Erkennung, Leerzeilen-Verhalten
### `input-fields.json`
Definiert die Spalten der Input-Tabelle:
- `column`: Spaltenbuchstabe (A, B, C…)
- `field_name`: Exakte Spaltenüberschrift aus der Excel-Datei
- `type`: `string` oder `number`
- `required`: `true` oder `false`
### `output-fields.json`
Mapping von Input-Feldern auf Template-Zellen:
- `source_field`: Exakte Spaltenüberschrift aus `input-fields.json`
- `target_cell`: Zelladresse im Template (z. B. `"C16"`)
- `format`: Optionaler Format-String (z. B. `"%.2f"` für Prozentwerte)
## Erwartetes Input-Format
- Eine `.xlsx`-Datei mit einem Sheet pro Monat (z. B. "Januar 2025", "Februar 2025", …)
- Jedes Sheet hat in Zeile 1 die Spaltenüberschriften (laut `input-fields.json`)
- Daten beginnen ab Zeile 2
- Jede Zeile entspricht einer Mitarbeiter-Funktion-Kombination für diesen Monat
## Output
- Pro selektiertem Mitarbeiter wird eine `.xlsx`-Datei erstellt
- Jeder Monat wird als eigenes Sheet (Kopie des Templates) eingefügt
- Dateiname: `{Nachname}_{Vorname}_Tätigkeitsnachweis.xlsx`
+183 -9
View File
@@ -8,6 +8,13 @@ from pathlib import Path
import streamlit as st
# Projekt-Root zum Python-Path hinzufügen
# Im gebundelten Modus (PyInstaller) zeigt APP_BASE_DIR auf das Verzeichnis der .exe,
# wo config/, input/, template/, output/ liegen. Im Entwicklungsmodus bleibt es None
# und PROJECT_ROOT wird aus __file__ abgeleitet.
_BASE_DIR = os.environ.get("APP_BASE_DIR")
if _BASE_DIR:
PROJECT_ROOT = Path(_BASE_DIR).resolve()
else:
PROJECT_ROOT = Path(__file__).resolve().parent
sys.path.insert(0, str(PROJECT_ROOT))
@@ -22,6 +29,27 @@ from src.models import AppConfig
from src.parser import parse_input_file, group_by_employee, get_unique_antragsnummern, get_unique_employees
from src.validator import validate_input_file, validate_template
from src.writer import generate_activity_reports
from version_check import UPDATE_DIR, get_latest_version
def set_active_employee(employee_name: str):
"""Merkt die zuletzt angeklickte Person für die Monatsliste."""
st.session_state.active_month_employee = employee_name
def sync_employee_months(employee_name: str, months: list[str]):
"""Übernimmt Monats-Checkboxen vor dem nächsten UI-Rendern."""
st.session_state.employee_months[employee_name] = {
month for month in months
if st.session_state.get(f"month_{employee_name}_{month}", False)
}
def set_all_employee_months(employee_name: str, months: list[str], selected: bool):
"""Setzt alle Monats-Checkboxen einer Person gesammelt."""
st.session_state.employee_months[employee_name] = set(months) if selected else set()
for month in months:
st.session_state[f"month_{employee_name}_{month}"] = selected
# --- Seite konfigurieren ---
st.set_page_config(
@@ -49,6 +77,60 @@ except Exception as e:
st.error(f"Fehler beim Laden der Konfiguration: {e}")
st.stop()
# --- Versionsinformation für die dezente Anzeige am Seitenende ---
current_version_text = ""
try:
local_version, latest_version = get_latest_version()
current_version_text = f"v{'.'.join(map(str, local_version))}"
if latest_version:
available_version, _ = latest_version
version_text = ".".join(map(str, available_version))
st.html(
f'''<style>
.update-notice {{
position: fixed; top: 4rem; right: 1rem; z-index: 10000;
width: 290px; padding: 8px 12px; color: #5f4700;
background: #fff8e1; border: 1px solid #e0b84c; border-radius: 8px;
box-shadow: 0 3px 12px rgba(0, 0, 0, .18); font-size: .85rem;
}}
#update-notice-dismiss {{ display:none; }}
#update-notice-dismiss:checked + .update-notice {{ display:none; }}
body:has(#update-notice-dismiss:checked) div.st-key-update_notice_open {{ display:none!important; }}
.update-notice__actions {{ display:flex; justify-content:flex-end; gap:8px; margin-top:8px; }}
.update-notice__actions--footer {{ min-height:26px; }}
.update-notice__button {{
display:inline-flex; align-items:center; justify-content:center;
width:26px; height:26px; color:#5f4700; background:#fffdf3;
border:1px solid #d9bd66; border-radius:5px; text-decoration:none;
font-size:1rem; line-height:1;
}}
.update-notice__button:hover {{ background:#f6e7b2; }}
div.st-key-update_notice_open {{
position:fixed; top:8.5rem; right:3.75rem; z-index:10001; width:26px;
}}
div.st-key-update_notice_open div[data-testid="stButton"] {{ margin:0; }}
div.st-key-update_notice_open div[data-testid="stButton"] button {{
min-width:200px; min-height:26px; height:26px; padding:0; color:#5f4700;
background:#fffdf3; border:1px solid #d9bd66; border-radius:5px;
}}
</style>
<input id="update-notice-dismiss" type="checkbox">
<aside class="update-notice" role="status">
<div class="update-notice__actions">
<label class="update-notice__button" for="update-notice-dismiss" title="Hinweis schließen">×</label>
</div>
<strong>Neue Version v{version_text} verfügbar</strong>
<div class="update-notice__actions update-notice__actions--footer">
</div>
</aside>'''
)
with st.container(key="update_notice_open"):
if st.button("Update Ordner öffnen 📁", key="open_update_directory", help="Update-Ordner öffnen"):
if sys.platform == "win32":
os.startfile(str(UPDATE_DIR))
except Exception:
pass
# --- Session State initialisieren ---
if "parsed_records" not in st.session_state:
st.session_state.parsed_records = []
@@ -65,7 +147,8 @@ if "cached_val_result" not in st.session_state:
st.session_state.cached_val_result = None
if "cached_tpl_result" not in st.session_state:
st.session_state.cached_tpl_result = None
if "employee_months" not in st.session_state:
st.session_state.employee_months = {}
# --- Layout: zwei Spalten ---
col1, col2 = st.columns(2)
@@ -223,25 +306,95 @@ if st.session_state.input_validated and st.session_state.parsed_records:
employees = get_unique_employees(st.session_state.parsed_records)
st.caption(f"{len(employees)} Mitarbeiter gefunden – wähle aus, für wen Tätigkeitsnachweise erstellt werden sollen:")
# "Alle" / "Keine" Buttons
btn_col1, btn_col2 = st.columns(2)
if btn_col1.button("☑ Alle auswählen", use_container_width=True):
# Monatsbezeichnungen für die kompakte Anzeige hinter dem Namen
month_labels = {
"01": "Jan", "02": "Feb", "03": "Mär", "04": "Apr",
"05": "Mai", "06": "Jun", "07": "Jul", "08": "Aug",
"09": "Sep", "10": "Okt", "11": "Nov", "12": "Dez",
}
all_groups = group_by_employee(st.session_state.parsed_records)
available_months = {}
for emp in employees:
months = sorted({
record.monat.split(".", 1)[0].zfill(2)
for record in all_groups.get(emp, [])
if record.monat and record.monat.split(".", 1)[0].isdigit()
}, key=lambda value: int(value))
available_months[emp] = months
if emp not in st.session_state.employee_months:
st.session_state.employee_months[emp] = set(months[-2:])
# Mitarbeiter- und Monatsliste teilen sich die verfügbare Breite.
employee_col, month_col = st.columns(2)
with employee_col:
employee_btn_col1, employee_btn_col2 = st.columns(2)
if employee_btn_col1.button("☑ Alle auswählen", use_container_width=True):
for emp in employees:
st.session_state[f"sel_{emp}"] = True
if btn_col2.button("☐ Keine auswählen", use_container_width=True):
if employee_btn_col2.button("☐ Keine auswählen", use_container_width=True):
for emp in employees:
st.session_state[f"sel_{emp}"] = False
# Mitarbeiter-Liste mit Rahmen
with st.container(border=True):
selected_employees = []
for emp in employees:
key = f"sel_{emp}"
if key not in st.session_state:
st.session_state[key] = True
if st.checkbox(emp, key=key):
st.session_state[key] = False
selected_months = st.session_state.employee_months.get(emp, set())
short_months = ", ".join(month_labels[m] for m in available_months[emp] if m in selected_months)
label = f"{emp} ({short_months})" if short_months else f"{emp} (**keine Monate**)"
if st.checkbox(
label,
key=key,
on_change=set_active_employee,
args=(emp,),
):
selected_employees.append(emp)
with month_col:
selected_for_months = st.session_state.get("active_month_employee")
if selected_for_months not in employees:
selected_for_months = None
month_btn_col1, month_btn_col2 = st.columns(2)
month_btn_col1.button(
"☑ Alle Monate",
use_container_width=True,
disabled=not selected_for_months,
on_click=set_all_employee_months,
args=(selected_for_months, available_months[selected_for_months], True) if selected_for_months else ("", [], True),
)
month_btn_col2.button(
"☐ Keine Monate",
use_container_width=True,
disabled=not selected_for_months,
on_click=set_all_employee_months,
args=(selected_for_months, available_months[selected_for_months], False) if selected_for_months else ("", [], False),
)
with st.container(border=True):
st.caption("Monate für den angeklickten Mitarbeiter")
if selected_for_months:
st.write(selected_for_months)
month_values = {}
for month in available_months[selected_for_months]:
month_key = f"month_{selected_for_months}_{month}"
if month_key not in st.session_state:
st.session_state[month_key] = month in st.session_state.employee_months[selected_for_months]
month_values[month] = st.checkbox(
f"{month_labels[month]} ({month})",
key=month_key,
on_change=sync_employee_months,
args=(selected_for_months, available_months[selected_for_months]),
)
st.session_state.employee_months[selected_for_months] = {
month for month, checked in month_values.items() if checked
}
else:
st.info("Einen Mitarbeiter links auswählen, um seine Monate festzulegen.")
st.caption(f"{len(selected_employees)} von {len(employees)} Mitarbeitern ausgewählt")
else:
selected_employees = []
@@ -276,7 +429,20 @@ if generate_clicked:
else:
# Filtere Records auf selektierte Mitarbeiter
all_groups = group_by_employee(st.session_state.parsed_records)
selected_groups = {name: all_groups[name] for name in selected_employees if name in all_groups}
selected_groups = {
name: [
record for record in all_groups[name]
if record.monat.split(".", 1)[0].zfill(2)
in st.session_state.employee_months.get(name, set())
]
for name in selected_employees
if name in all_groups
}
selected_groups = {name: records for name, records in selected_groups.items() if records}
if not selected_groups:
st.error("Für die ausgewählten Mitarbeiter sind keine Monate ausgewählt.")
st.stop()
template_path = Path(app_config.template_dir) / selected_template
output_dir = Path(app_config.output_dir)
@@ -289,6 +455,7 @@ if generate_clicked:
mapping=app_config.output_mapping,
output_dir=output_dir,
filename_pattern=app_config.filename_pattern,
fontsize=app_config.fontsize,
)
st.session_state.generated_files = generated
except Exception as e:
@@ -320,3 +487,10 @@ if "generated_files" in st.session_state and st.session_state.generated_files:
help="Ordner im Finder öffnen",
use_container_width=True,
)
if current_version_text:
st.markdown(
f'<div style="text-align:right; color:#888; font-size:0.8em; margin-top:2em;">'
f'Programmversion {current_version_text}</div>',
unsafe_allow_html=True,
)
@@ -1,4 +1,6 @@
@echo off
chcp 65001 >nul
set "PYTHONUTF8=1"
title Tätigkeitsnachweis-Generator – Cold Run
echo ========================================
echo Cold Run – Headless-Generierung
View File
File renamed without changes.
+1
View File
@@ -113,6 +113,7 @@ def main():
mapping=config.output_mapping,
output_dir=output_dir,
filename_pattern=config.filename_pattern,
fontsize=config.fontsize,
)
output_file = generated[0]
File renamed without changes.
File renamed without changes.
@@ -6,6 +6,10 @@ output_dir = output
[output]
filename_pattern = {last_name}_{first_name}_Tätigkeitsnachweis.pdf
[font]
# Statische Schriftgröße (in Punkt) für alle Felder im PDF
fontsize = 9
[parsing]
# Sheets whose names match these patterns are treated as monthly data sheets
# Use * as wildcard. Sheets not matching are ignored (e.g. "Alle Rollen").
File renamed without changes.
+103
View File
@@ -0,0 +1,103 @@
"""Stop an existing instance of this application's Windows server."""
import json
import os
import subprocess
from pathlib import Path
_CONFIRM_SCRIPT = r'''
function Confirm-ServerRestart {
Add-Type -AssemblyName System.Windows.Forms
$answer = [System.Windows.Forms.MessageBox]::Show(
"Es läuft bereits eine Version des Tätigkeitsnachweis-Generators.`n`nSoll das alte Programm beendet und die neue Version gestartet werden?`nDie aktuellen Auswahlen gehen dabei verloren.",
'Tätigkeitsnachweis-Generator',
[System.Windows.Forms.MessageBoxButtons]::YesNo,
[System.Windows.Forms.MessageBoxIcon]::Question,
[System.Windows.Forms.MessageBoxDefaultButton]::Button2
)
return $answer -eq [System.Windows.Forms.DialogResult]::Yes
}
'''
def stop_existing_server(port: int = 8501) -> bool:
"""Ask before stopping this app. Return False if the user declines."""
if os.name != "nt":
raise RuntimeError("Der kontrollierte Server-Neustart unterstützt nur Windows.")
app_path = str(Path(__file__).resolve().with_name("app.py"))
# Pass paths through the environment, never interpolate them into shell code.
env = dict(os.environ, APP_RESTART_PATH=app_path, APP_RESTART_PID=str(os.getpid()),
APP_RESTART_PORT=str(port),
APP_RESTART_VENV=str(Path(app_path).parent / ".venv" / "Scripts"))
script = _CONFIRM_SCRIPT + r'''
[Console]::OutputEncoding = [System.Text.UTF8Encoding]::new()
$ErrorActionPreference = 'Stop'
try {
$port = [int]$env:APP_RESTART_PORT
$listeners = @(Get-NetTCPConnection -State Listen -ErrorAction Stop |
Where-Object { $_.LocalPort -eq $port })
$processIds = @($listeners | Select-Object -ExpandProperty OwningProcess -Unique)
$targets = @()
foreach ($processId in $processIds) {
if ($processId -eq [int]$env:APP_RESTART_PID) {
throw "Der aktuelle Prozess belegt bereits Port $port."
}
$process = Get-CimInstance Win32_Process -Filter "ProcessId = $processId"
if (-not $process) { continue }
$isExe = $process.Name -eq 'Tätigkeitsnachweis-Generator.exe'
$command = $process.CommandLine
$appArgument = '(?i)\bstreamlit(?:\.exe)?"?\s+run\s+"?' +
[regex]::Escape($env:APP_RESTART_PATH) + '"?(?:\s|$)'
$isDev = $process.Name -match '^(python(?:w)?(?:\d+(?:\.\d+)*)?(?:w)?|streamlit)\.exe$' -and
$command -match '(?i)\bstreamlit(?:\.exe)?"?\s+run\s+' -and
($command -match $appArgument -or
($command -match '(?i)\bstreamlit(?:\.exe)?"?\s+run\s+"?app\.py"?(?:\s|$)' -and
[IO.Path]::GetDirectoryName($process.ExecutablePath) -eq $env:APP_RESTART_VENV))
if (-not ($isExe -or $isDev)) {
throw "Port $port wird von einem anderen Programm verwendet (PID $processId, $($process.Name)). Bitte manuell prüfen."
}
$targets += $processId
}
if ($targets.Count -gt 0 -and -not (Confirm-ServerRestart)) {
exit 2
}
foreach ($processId in $targets) {
Stop-Process -Id $processId -Force -ErrorAction Stop
}
$deadline = (Get-Date).AddSeconds(10)
do {
$remaining = @(Get-NetTCPConnection -State Listen -ErrorAction Stop |
Where-Object { $_.LocalPort -eq $port })
if ($remaining.Count -eq 0) { break }
Start-Sleep -Milliseconds 200
} while ((Get-Date) -lt $deadline)
if ($remaining.Count -gt 0) { throw "Port $port wurde nicht rechtzeitig freigegeben." }
ConvertTo-Json -Compress -InputObject @($targets)
} catch {
[Console]::Error.WriteLine($_.Exception.Message)
exit 1
}
'''
result = subprocess.run(
["powershell.exe", "-NoProfile", "-NonInteractive", "-Command", script],
env=env, capture_output=True, text=True, encoding="utf-8", errors="replace",
)
if result.returncode == 2:
return False
if result.returncode:
raise RuntimeError((result.stderr or "").strip() or "Der bestehende Server konnte nicht beendet werden.")
stopped = json.loads(result.stdout.strip() or "[]")
if stopped:
print(f"[INFO] Vorherige Instanz beendet (PID: {stopped}).", flush=True)
return True
if __name__ == "__main__":
try:
if not stop_existing_server():
print("[INFO] Die bisherige Version bleibt geöffnet. Keine neue Instanz gestartet.")
raise SystemExit(2)
except (RuntimeError, OSError, subprocess.SubprocessError) as exc:
print(f"[FEHLER] {exc}")
raise SystemExit(1)
File renamed without changes.
@@ -21,6 +21,7 @@ def load_settings(config_dir: Path, project_root: Path) -> AppConfig:
paths = parser["paths"] if "paths" in parser else {}
output = parser["output"] if "output" in parser else {}
parsing = parser["parsing"] if "parsing" in parser else {}
font = parser["font"] if "font" in parser else {}
return AppConfig(
input_dir=str(project_root / paths.get("input_dir", "input")),
@@ -30,6 +31,7 @@ def load_settings(config_dir: Path, project_root: Path) -> AppConfig:
month_sheet_pattern=parsing.get("month_sheet_pattern", "* 20??"),
skip_empty_name=parsing.getboolean("skip_empty_name", True),
skip_empty_antragsnummer=parsing.getboolean("skip_empty_antragsnummer", True),
fontsize=font.getfloat("fontsize", 11.0),
)
@@ -63,6 +63,7 @@ class AppConfig:
month_sheet_pattern: str = "* 20??"
skip_empty_name: bool = True
skip_empty_antragsnummer: bool = True
fontsize: float = 11.0
@dataclass
File renamed without changes.
File renamed without changes.
+42 -67
View File
@@ -67,7 +67,11 @@ def _get_field_value(record: EmployeeRecord, employee_name: str, source_field: s
if source_field == "Monat (MM.JJ)":
return _format_month_for_template(record.monat)
elif source_field == "Stellenanteil in %":
return record.stellenanteil
try:
value = int(float(str(record.stellenanteil).strip()))
return f"{value},00"
except (ValueError, TypeError):
return str(record.stellenanteil)
elif source_field == "Name":
return employee_name
elif source_field == "Antragsnummer":
@@ -101,10 +105,11 @@ def _get_field_value(record: EmployeeRecord, employee_name: str, source_field: s
def _scan_template_fields(
template_path: Path,
mapping: list[OutputMapping],
) -> dict[str, dict]:
fontsize: float,
) -> dict[str, list[dict]]:
"""
Scannt ALLE Seiten des PDF-Templates nach _Field-Platzhaltern.
Rückgabe: dict[pdf_field] → {source_field, page_num, rect, fontsize, fontname, ...}
Rückgabe: dict[pdf_field] → Liste von {source_field, page_num, rect, fontsize, fontname, ...}
Felder mit pdf_field=None werden übersprungen.
"""
doc = fitz.open(template_path)
@@ -115,43 +120,23 @@ def _scan_template_fields(
if m.pdf_field is not None:
pdf_to_source[m.pdf_field] = m.source_field
field_info: dict[str, dict] = {}
field_info: dict[str, list[dict]] = {}
# 1. Per search_for die Rechtecke aller PDF-Feld-Marker finden (alle Seiten)
# Alle Vorkommen aller PDF-Feld-Marker finden (alle Seiten)
for pdf_field, source_field in pdf_to_source.items():
found_any = False
occurrences: list[dict] = []
for page_num in range(doc.page_count):
page = doc[page_num]
found = page.search_for(pdf_field)
if found:
field_info[pdf_field] = {
for rect in found:
occurrences.append({
"source_field": source_field,
"page_num": page_num,
"rect": found[0],
"fontsize": 11, # Fallback
"rect": rect,
"fontsize": fontsize,
"fontname": "helv",
}
found_any = True
break # erstes Vorkommen (seitenübergreifend)
if not found_any:
field_info[pdf_field] = None # nicht gefunden
# 2. Schriftgrößen aus den Text-Details extrahieren (alle Seiten)
for page_num in range(doc.page_count):
page = doc[page_num]
text_dict = page.get_text("dict")
for block in text_dict.get("blocks", []):
if "lines" not in block:
continue
for line in block["lines"]:
for span in line["spans"]:
text = span["text"].strip()
if text.endswith("_Field") and text in field_info:
info = field_info[text]
if info is not None:
info["fontsize"] = span["size"]
info["fontname"] = span.get("font", "helv")
info["rect"] = fitz.Rect(span["bbox"])
})
field_info[pdf_field] = occurrences
doc.close()
return field_info
@@ -166,69 +151,55 @@ def _fill_pdf_pages(
record: EmployeeRecord,
employee_name: str,
mapping: list[OutputMapping],
field_info: dict[str, dict],
field_info: dict[str, list[dict]],
fontsize: float,
) -> None:
"""Befüllt alle Template-Seiten: redacted _Field-Platzhalter und fügt echte Daten ein."""
# Mehrzeilige Felder, die als Textbox (mit Zeilenumbruch) behandelt werden
MULTILINE_FIELDS = {"Tätigkeiten_Field", "Dienstreisen_Field"}
TEXTBOX_WIDTH_PT = 400 # Breite der Textbox in Punkt
# Felder mit automatischer Schriftgrößen-Anpassung bei langem Text
AUTOSCALE_FIELDS = {"Qualifikation_Field", "Funktion_Field"}
AUTOSCALE_MAX_CHARS = 35 # ab dieser Länge wird verkleinert
AUTOSCALE_MIN_FACTOR = 0.75 # minimale Schriftgröße relativ zum Original
for m in mapping:
if m.pdf_field is None:
continue
info = field_info.get(m.pdf_field)
if info is None:
occurrences = field_info.get(m.pdf_field, [])
if not occurrences:
continue # Feld nicht im Template gefunden
value = _get_field_value(record, employee_name, m.source_field)
formatted = _format_value(value, m.format)
# Jedes Vorkommen des Feldes im PDF befüllen
for info in occurrences:
page_num = info["page_num"]
if page_num >= len(pages):
continue
page = pages[page_num]
rect = info["rect"]
fontsize = info.get("fontsize", 11)
value = _get_field_value(record, employee_name, m.source_field)
formatted = _format_value(value, m.format)
# Platzhalter immer redacten, auch wenn kein Wert zum Einsetzen da ist
# Platzhalter redacten
page.add_redact_annot(rect)
page.apply_redactions()
if not formatted:
continue
# Automatische Schriftgrößen-Anpassung für lange Texte
if m.pdf_field in AUTOSCALE_FIELDS and len(formatted) > AUTOSCALE_MAX_CHARS:
text_width = fitz.get_text_length(formatted, fontname="helv", fontsize=fontsize)
available_width = rect.width
if text_width > available_width:
factor = available_width / text_width
factor = max(factor, AUTOSCALE_MIN_FACTOR)
fontsize = fontsize * factor
# Mehrzeilige Felder: Zeilenweise per insert_text schreiben
if m.pdf_field in MULTILINE_FIELDS:
# Doppelte Leerzeilen entfernen
# ... (unverändert)
clean_text = "\n".join(
line for line in formatted.split("\n") if line.strip()
)
if not clean_text:
continue
# Zeilenweise mit automatischem Umbruch schreiben
lines = clean_text.split("\n")
line_spacing = fontsize * 1.2 # Zeilenabstand
y = rect.y0 + fontsize # Startposition (unterhalb Marker-Oberkante)
line_spacing = fontsize * 1.2
output_lines: list[str] = []
for line in lines:
# Wort-Umbruch: Zeile in Stücke passend zur Textbox-Breite aufteilen
max_width = TEXTBOX_WIDTH_PT
words = line.split(" ")
current_line = ""
@@ -238,21 +209,24 @@ def _fill_pdf_pages(
if text_width <= max_width or not current_line:
current_line = test_line
else:
page.insert_text(
(rect.x0, y), current_line,
fontname="helv", fontsize=fontsize,
)
y += line_spacing
output_lines.append(current_line)
current_line = word
if current_line:
output_lines.append(current_line)
y = rect.y0 + fontsize
for current_line in output_lines:
page.insert_text(
(rect.x0, y), current_line,
fontname="helv", fontsize=fontsize,
)
y += line_spacing
else:
# Einzeilige Felder: vertikal zentriert platzieren
mid_y = (rect.y0 + rect.y1) / 2
baseline_y = mid_y + fontsize * 0.35
page.insert_text(
rect.bl,
(rect.x0, baseline_y),
formatted,
fontname="helv",
fontsize=fontsize,
@@ -269,6 +243,7 @@ def generate_activity_reports(
mapping: list[OutputMapping],
output_dir: Path,
filename_pattern: str,
fontsize: float = 11.0,
) -> list[Path]:
"""
Erzeugt für jeden Mitarbeiter ein PDF.
@@ -278,7 +253,7 @@ def generate_activity_reports(
generated_files: list[Path] = []
# Pre-Scan: Feld-Positionen auf allen Template-Seiten ermitteln
field_info = _scan_template_fields(template_path, mapping)
field_info = _scan_template_fields(template_path, mapping, fontsize)
for employee_name, records in employee_records.items():
if not records:
@@ -301,7 +276,7 @@ def generate_activity_reports(
temp_doc = fitz.open(template_path)
pages = [temp_doc[i] for i in range(temp_doc.page_count)]
_fill_pdf_pages(pages, record, employee_name, mapping, field_info)
_fill_pdf_pages(pages, record, employee_name, mapping, field_info, fontsize)
# Alle modifizierten Seiten ins Ergebnis-PDF einfügen
result_doc.insert_pdf(temp_doc)
+10 -1
View File
@@ -1,4 +1,7 @@
@echo off
chcp 65001 >nul
set "PYTHONUTF8=1"
cd /d "%~dp0"
title Tätigkeitsnachweis-Generator
echo ========================================
echo Tätigkeitsnachweis-Generator
@@ -43,6 +46,12 @@ echo.
echo [INFO] Starte Tätigkeitsnachweis-Generator...
echo [INFO] Die App öffnet sich im Browser unter http://localhost:8501
echo.
streamlit run app.py
".venv\Scripts\python.exe" server_lifecycle.py
if %ERRORLEVEL% equ 2 exit /b 0
if %ERRORLEVEL% neq 0 (
pause
exit /b 1
)
".venv\Scripts\python.exe" -m streamlit run "%~dp0app.py" --server.port=8501 --server.address=localhost
pause
View File
File renamed without changes.
View File
File renamed without changes.
+45
View File
@@ -0,0 +1,45 @@
@echo off
chcp 65001 >nul
setlocal
set "PYTHONUTF8=1"
cd /d "%~dp0"
title Taetigkeitsnachweis-Generator - Tests
echo ========================================
echo Alle Tests ausfuehren
echo ========================================
echo.
if not exist ".venv\Scripts\python.exe" (
echo [FEHLER] Die virtuelle Umgebung fehlt. Bitte zuerst start.bat ausfuehren.
set "TEST_EXIT_CODE=1"
goto finish
)
".venv\Scripts\python.exe" -c "import pytest" >nul 2>nul
if errorlevel 1 (
echo [INFO] Installiere pytest in der Projektumgebung...
".venv\Scripts\python.exe" -m pip install pytest
if errorlevel 1 (
echo [FEHLER] pytest konnte nicht installiert werden.
set "TEST_EXIT_CODE=1"
goto finish
)
)
REM Keine Abbruchoption: Auch nach fehlgeschlagenen Tests laeuft die Suite weiter.
REM Ohne Cache funktioniert der Testlauf auch bei gesperrtem pytest-Cache.
".venv\Scripts\python.exe" -m pytest tests -v -p no:cacheprovider
set "TEST_EXIT_CODE=%ERRORLEVEL%"
echo.
if "%TEST_EXIT_CODE%"=="0" (
echo [OK] Alle Tests bestanden.
) else (
echo [FEHLER] Testlauf nicht erfolgreich. Details stehen oben.
)
:finish
REM Fuer automatisierte Aufrufe: test.bat --no-pause
if /I not "%~1"=="--no-pause" pause
exit /b %TEST_EXIT_CODE%
View File
File renamed without changes.
File renamed without changes.
Binary file not shown.
+121
View File
@@ -0,0 +1,121 @@
"""End-to-end checks using the fixed 2026 workbook and the PDF template."""
import sys
from pathlib import Path
import fitz
import pytest
PROJECT_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
from src.config import load_settings, load_input_schema, load_output_mapping
from src.parser import parse_input_file, group_by_employee, get_unique_employees
from src.writer import generate_activity_reports
CONFIG_DIR = PROJECT_ROOT / "config"
INPUT_FILE = Path(__file__).resolve().parent / "fixture" / "2026 JM.xlsx"
TEMPLATE_FILE = PROJECT_ROOT / "template" / "Tätigkeitsnachweise Stellenförderung.pdf"
@pytest.fixture(scope="module")
def app_config():
config = load_settings(CONFIG_DIR, PROJECT_ROOT)
config.input_schema = load_input_schema(CONFIG_DIR)
config.output_mapping = load_output_mapping(CONFIG_DIR)
return config
@pytest.fixture(scope="module")
def parsed_records(app_config):
assert INPUT_FILE.is_file(), f"Testdatei nicht gefunden: {INPUT_FILE}"
return parse_input_file(
filepath=INPUT_FILE,
schema=app_config.input_schema,
month_sheet_pattern=app_config.month_sheet_pattern,
skip_empty_name=app_config.skip_empty_name,
skip_empty_antragsnummer=app_config.skip_empty_antragsnummer,
)
@pytest.fixture
def output_dir(tmp_path):
out = tmp_path / "output"
out.mkdir()
return out
class TestEndToEndFirstPerson:
def test_input_file_exists(self):
assert INPUT_FILE.is_file(), f"Testdatei nicht gefunden: {INPUT_FILE}"
def test_template_file_exists(self):
assert TEMPLATE_FILE.is_file(), f"PDF-Template nicht gefunden: {TEMPLATE_FILE}"
def test_template_has_field_markers(self, app_config):
with fitz.open(TEMPLATE_FILE) as doc:
missing = [
mapping.pdf_field
for mapping in app_config.output_mapping
if mapping.pdf_field is not None
and not any(page.search_for(mapping.pdf_field) for page in doc)
]
assert not missing, f"Folgende _Field-Platzhalter fehlen im PDF: {missing}"
def test_parse_real_input(self, parsed_records):
# Fixed expectations for tests/fixture/2026 JM.xlsx.
assert len(parsed_records) == 100
employees = get_unique_employees(parsed_records)
assert len(employees) == 17
assert employees[0] == "Bardaus, Madalina"
first_records = group_by_employee(parsed_records)[employees[0]]
assert [record.monat for record in first_records] == [
"01.2026", "02.2026", "03.2026", "04.2026", "05.2026"
]
for record in first_records:
assert record.antragsnummer == "100781516"
assert record.funktion == "Psychologin"
assert record.stellenanteil == "50"
assert record.qualifikation == ""
assert record.taetigkeit.startswith("Psychologische Betreuung")
def test_full_pipeline_first_person(self, app_config, parsed_records, output_dir):
grouped = group_by_employee(parsed_records)
assert len(grouped) == 17
first_person = sorted(grouped)[0]
assert first_person == "Bardaus, Madalina"
assert len(grouped[first_person]) == 5
generated = generate_activity_reports(
employee_records={first_person: grouped[first_person]},
template_path=TEMPLATE_FILE,
mapping=app_config.output_mapping,
output_dir=output_dir,
filename_pattern=app_config.filename_pattern,
fontsize=app_config.fontsize,
)
assert len(generated) == 1
output_file = generated[0]
assert output_file.is_file()
assert output_file.name == "Bardaus_Madalina_Tätigkeitsnachweis.pdf"
with fitz.open(output_file) as doc:
assert doc.page_count == 10 # Five months, two pages per month.
assert "_Field" not in "".join(page.get_text() for page in doc)
# Independently specified month ends include February's 28 days.
expected_months = [
("01.2026", "31.01.2026"),
("02.2026", "28.02.2026"),
("03.2026", "31.03.2026"),
("04.2026", "30.04.2026"),
("05.2026", "31.05.2026"),
]
for index, (month, month_end) in enumerate(expected_months):
data_text = doc[index * 2].get_text()
signature_text = doc[index * 2 + 1].get_text()
for value in [
"100781516", "Bardaus, Madalina", "50,00", month,
"Psychologin", "Psychologische Betreuung",
]:
assert value in data_text, f"{value!r} fehlt für {month}"
assert "100781516" not in signature_text
assert "Dresden" in signature_text
assert month_end in signature_text
@@ -0,0 +1,162 @@
"""Exercise the actual PowerShell logic with simulated Windows processes."""
import os
from pathlib import Path
import subprocess
import sys
import unittest
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import server_lifecycle
@unittest.skipUnless(os.name == "nt", "Windows process management")
class ServerLifecycleTests(unittest.TestCase):
def simulate(self, process, listening=True, stop_error=False, confirm=True):
captured = {}
def capture(command, **kwargs):
captured.update(script=command[-1], env=kwargs["env"])
return subprocess.CompletedProcess(command, 0, "[]", "")
with patch.object(server_lifecycle.subprocess, "run", side_effect=capture):
server_lifecycle.stop_existing_server()
setup = r'''
$script:occupied = LISTENING
function Get-NetTCPConnection {
param($State, $ErrorAction)
if ($script:occupied) { [pscustomobject]@{ LocalPort=8501; OwningProcess=12345 } }
}
function Get-CimInstance {
param($ClassName, $Filter)
PROCESS
}
function Stop-Process {
param($Id, [switch]$Force, $ErrorAction)
[Console]::Error.WriteLine('STOP_CALLED')
STOP
$script:occupied = $false
}
'''.replace("LISTENING", "$true" if listening else "$false")
setup = setup.replace("PROCESS", process).replace(
"\n STOP\n", "\n throw 'simulated access denied'\n" if stop_error else "\n")
confirm_mock = """
function Confirm-ServerRestart {
[Console]::Error.WriteLine('CONFIRM_CALLED')
return ANSWER
}
""".replace("ANSWER", "$true" if confirm else "$false")
script = captured["script"].replace(server_lifecycle._CONFIRM_SCRIPT, confirm_mock)
command = [
"powershell.exe", "-NoProfile", "-NonInteractive", "-Command",
setup + script,
]
# Explicit pipes keep both outputs available independently of pytest's
# own console capture. communicate() drains both pipes concurrently.
with subprocess.Popen(
command, env=captured["env"], stdout=subprocess.PIPE,
stderr=subprocess.PIPE, text=True, errors="replace",
) as child:
try:
stdout, stderr = child.communicate(timeout=30)
except subprocess.TimeoutExpired:
child.kill()
child.communicate()
raise
self.assertIsInstance(stdout, str, "stdout wurde nicht erfasst")
self.assertIsInstance(stderr, str, "stderr wurde nicht erfasst")
return subprocess.CompletedProcess(command, child.returncode, stdout, stderr)
def test_free_port(self):
result = self.simulate("throw 'should not inspect a process'", listening=False)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(result.stdout.strip(), "[]")
self.assertNotIn("CONFIRM_CALLED", result.stderr)
def test_previous_exe_is_stopped(self):
result = self.simulate(
"[pscustomobject]@{ Name='Tätigkeitsnachweis-Generator.exe'; CommandLine=''; ExecutablePath='' }")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(result.stdout.strip(), "[12345]")
self.assertIn("CONFIRM_CALLED", result.stderr)
self.assertIn("STOP_CALLED", result.stderr)
def test_declining_preserves_previous_instance(self):
result = self.simulate(
"[pscustomobject]@{ Name='Tätigkeitsnachweis-Generator.exe'; CommandLine=''; ExecutablePath='' }",
confirm=False)
self.assertEqual(result.returncode, 2, result.stderr)
self.assertIn("CONFIRM_CALLED", result.stderr)
self.assertNotIn("STOP_CALLED", result.stderr)
def test_declining_returns_false_to_launcher(self):
with patch.object(server_lifecycle.subprocess, "run", return_value=
subprocess.CompletedProcess([], 2, "", "")):
self.assertFalse(server_lifecycle.stop_existing_server())
def test_development_absolute_path_is_stopped(self):
result = self.simulate(r'''[pscustomobject]@{
Name='python.exe'; ExecutablePath='C:\Python\python.exe';
CommandLine=('python -m streamlit run "' + $env:APP_RESTART_PATH + '"')
}''')
self.assertEqual(result.returncode, 0, result.stderr)
def test_legacy_development_start_is_stopped(self):
result = self.simulate(r'''[pscustomobject]@{
Name='python.exe'; CommandLine='streamlit.exe run app.py';
ExecutablePath=($env:APP_RESTART_VENV + '\python.exe')
}''')
self.assertEqual(result.returncode, 0, result.stderr)
def test_versioned_python_prompts_before_stopping(self):
# Windows Store Python reports the real interpreter name even when
# launched through the venv's python.exe.
process = r'''[pscustomobject]@{
Name='python3.13.exe';
ExecutablePath='C:\Program Files\WindowsApps\Python\python3.13.exe';
CommandLine=('".venv\Scripts\python.exe" -m streamlit run "' +
$env:APP_RESTART_PATH + '" --server.port=8501 --server.address=localhost')
}'''
with self.subTest(answer="yes"):
result = self.simulate(process, confirm=True)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn("CONFIRM_CALLED", result.stderr)
self.assertIn("STOP_CALLED", result.stderr)
with self.subTest(answer="no"):
result = self.simulate(process, confirm=False)
self.assertEqual(result.returncode, 2, result.stderr)
self.assertIn("CONFIRM_CALLED", result.stderr)
self.assertNotIn("STOP_CALLED", result.stderr)
def test_unrelated_versioned_python_is_preserved(self):
result = self.simulate(r'''[pscustomobject]@{
Name='python3.13.exe'; ExecutablePath='C:\Python\python3.13.exe';
CommandLine='python -m streamlit run "C:\other-project\app.py"'
}''')
self.assertNotEqual(result.returncode, 0)
self.assertIn("PID 12345", result.stderr)
self.assertNotIn("CONFIRM_CALLED", result.stderr)
self.assertNotIn("STOP_CALLED", result.stderr)
def test_unrelated_server_is_preserved(self):
result = self.simulate(r'''[pscustomobject]@{
Name='python.exe'; CommandLine='streamlit run app.py';
ExecutablePath='C:\other-project\.venv\Scripts\python.exe'
}''')
self.assertNotEqual(result.returncode, 0)
self.assertIn("PID 12345", result.stderr)
self.assertNotIn("STOP_CALLED", result.stderr)
self.assertNotIn("CONFIRM_CALLED", result.stderr)
def test_stop_failure_aborts_start(self):
result = self.simulate(
"[pscustomobject]@{ Name='Tätigkeitsnachweis-Generator.exe'; CommandLine=''; ExecutablePath='' }",
stop_error=True)
self.assertNotEqual(result.returncode, 0)
self.assertIn("simulated access denied", result.stderr)
if __name__ == "__main__":
unittest.main()
+49
View File
@@ -0,0 +1,49 @@
"""Prüft auf neuere Programmversionen im zentralen SMB-Verzeichnis."""
import re
from pathlib import Path
UPDATE_DIR = Path(r"j:\Fachbereiche\BO\Produktionsschule\Maßnahmeunterlagen\20_Personal\05_Tätigkeitsnachweise ESF\Nachweise-automatisiert-generieren")
VERSION_PATTERN = re.compile(r"Tätigkeitsnachweis-Generator v(\d+)\.(\d+)\.(\d+)(?:\.\d+)?\.(zip|exe)$", re.IGNORECASE)
CHANGELOG_PATTERN = re.compile(r"^## \[(\d+)\.(\d+)(?:\.(\d+))?\]", re.MULTILINE)
def get_latest_version():
match = CHANGELOG_PATTERN.search(Path(__file__).with_name("CHANGELOG.md").read_text(encoding="utf-8"))
current = tuple(int(part or 0) for part in match.groups()) if match else (0, 0, 0)
try:
available = UPDATE_DIR.exists()
except OSError:
available = False
if not available:
return current, None
newer = []
try:
paths = UPDATE_DIR.iterdir()
for path in paths:
found = VERSION_PATTERN.fullmatch(path.name)
if found:
version = tuple(int(found.group(i)) for i in range(1, 4))
if version > current:
newer.append((version, path))
except OSError:
return current, None
if newer:
return current, max(newer)
return current, None
def main() -> int:
current, latest = get_latest_version()
print(f"[INFO] Lokale Version: v{'.'.join(map(str, current))}")
if latest:
version, path = latest
print(f"[HINWEIS] Neuere Version verfügbar: v{'.'.join(map(str, version))}")
print(f"[HINWEIS] Datei: {path}")
else:
print("[INFO] Keine neuere Version gefunden oder Update-Verzeichnis nicht erreichbar.")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+1
View File
@@ -0,0 +1 @@
1.3
+47
View File
@@ -0,0 +1,47 @@
@echo off
chcp 65001 >nul
setlocal enabledelayedexpansion
set "PYTHONUTF8=1"
cd /d "%~dp0"
cd ..
echo ============================================================
echo Tätigkeitsnachweis-Generator – Build-Pipeline
echo ============================================================
echo.
REM ── Virtuelle Umgebung aktivieren ────────────────────────────
if exist "0_application\.venv\Scripts\python.exe" (
set PYTHON=0_application\.venv\Scripts\python.exe
echo [OK] Virtuelle Umgebung gefunden
) else if exist ".venv\Scripts\python.exe" (
set PYTHON=.venv\Scripts\python.exe
echo [OK] Virtuelle Umgebung gefunden
) else (
echo [WARN] Keine virtuelle Umgebung gefunden, nutze System-Python
set PYTHON=python
)
echo.
REM ── Parameter durchreichen ──────────────────────────────────
set ARGS=%*
if "%ARGS%"=="" (
echo Starte interaktiven Build …
echo.
)
REM ── Build ausführen ─────────────────────────────────────────
%PYTHON% "1_build-pipeline\build.py" %ARGS%
if errorlevel 1 (
echo.
echo [FEHLER] Build fehlgeschlagen!
pause
exit /b 1
)
echo.
echo Build abgeschlossen.
pause
+377
View File
@@ -0,0 +1,377 @@
"""Build-Orchestrierung mit Semantic Versioning.
Liest die aktuelle Version aus CHANGELOG.md, bump't sie gemäß
Benutzereingabe (major/minor/patch), aktualisiert das Changelog
und baut via PyInstaller eine standalone .exe nach 2_build/.
Verwendung:
python build.py # interaktiver Modus
python build.py --bump minor # nicht-interaktiv
python build.py --bump patch --message "Hotfix: PDF-Schriftgröße korrigiert"
"""
import argparse
import os
import re
import shutil
import subprocess
import sys
from datetime import date
from pathlib import Path
from typing import Optional
# ── Pfade ───────────────────────────────────────────────────────
BUILD_PIPELINE_DIR = Path(__file__).resolve().parent
PROJECT_ROOT = BUILD_PIPELINE_DIR.parent
APP_DIR = PROJECT_ROOT / "0_application"
BUILD_OUTPUT_DIR = PROJECT_ROOT / "2_build"
CHANGELOG_PATH = APP_DIR / "CHANGELOG.md"
SPEC_PATH = BUILD_PIPELINE_DIR / "build.spec"
VERSION_FILE = BUILD_PIPELINE_DIR / ".last_version"
# ── Versions-Helfer ─────────────────────────────────────────────
def _read_changelog() -> str:
"""Liest das gesamte CHANGELOG.md."""
if not CHANGELOG_PATH.exists():
return ""
return CHANGELOG_PATH.read_text(encoding="utf-8")
def _write_changelog(content: str) -> None:
"""Schreibt CHANGELOG.md."""
CHANGELOG_PATH.write_text(content, encoding="utf-8")
def get_current_version() -> Optional[str]:
"""Ermittelt die aktuellste Version aus CHANGELOG.md.
Sucht nach dem ersten Vorkommen von '## [X.Y.Z]' oder '## [X.Y]'.
"""
content = _read_changelog()
if not content:
return None
match = re.search(r"##\s*\[(\d+\.\d+(?:\.\d+)?)\]", content)
return match.group(1) if match else None
def bump_version(current: str, bump_type: str) -> str:
"""Erhöht eine Semantic-Version.
Args:
current: Aktuelle Version (z.B. '1.1' oder '1.1.0')
bump_type: 'major', 'minor' oder 'patch'
Returns:
Neue Version als String.
"""
parts = current.split(".")
# Normalisiere auf 3 Teile (MAJOR.MINOR.PATCH)
while len(parts) < 3:
parts.append("0")
major, minor, patch = map(int, parts[:3])
if bump_type == "major":
major += 1
minor = 0
patch = 0
elif bump_type == "minor":
minor += 1
patch = 0
elif bump_type == "patch":
patch += 1
else:
raise ValueError(f"Unbekannter Bump-Typ: {bump_type}")
# Gib kurzes Format zurück wenn Patch=0
if patch == 0:
return f"{major}.{minor}"
return f"{major}.{minor}.{patch}"
def update_changelog(
new_version: str,
bump_type: str,
message: Optional[str] = None,
) -> None:
"""Fügt einen neuen Versionseintrag am Anfang des Changelogs ein.
Args:
new_version: Die neue Version (z.B. '1.2')
bump_type: Art des Bumps (für die Sektions-Vorlage)
message: Optionale Beschreibung der Änderungen.
"""
today = date.today().isoformat()
old_content = _read_changelog()
# Sektionen-Vorlage je nach Bump-Typ
sections = _build_sections(bump_type, message)
header = (
f"## [{new_version}] – {today}\n\n"
f"{sections}\n"
f"---\n\n"
)
# Nach dem ersten '---' einfügen (nach dem Dateikopf von Keep a Changelog)
# oder einfach am Anfang nach der ersten Zeile
lines = old_content.splitlines(keepends=True)
# Finde die erste Version-Header-Zeile (## [X.Y...)
insert_idx = None
for i, line in enumerate(lines):
if re.match(r"##\s*\[\d+\.\d+", line):
insert_idx = i
break
if insert_idx is not None:
new_content = "".join(lines[:insert_idx]) + header + "".join(lines[insert_idx:])
else:
# Fallback: nach dem Dateikopf (nach dem ersten '---')
for i, line in enumerate(lines):
if line.strip() == "---":
insert_idx = i + 1
break
if insert_idx is not None:
new_content = "".join(lines[:insert_idx]) + "\n" + header + "".join(lines[insert_idx:])
else:
new_content = old_content + "\n" + header
_write_changelog(new_content)
print(f" ✓ CHANGELOG.md aktualisiert → [{new_version}]")
def _build_sections(bump_type: str, message: Optional[str]) -> str:
"""Baut die Sektionen für den Changelog-Eintrag."""
if message:
if bump_type == "major":
return f"### Hinzugefügt\n\n- {message}\n"
elif bump_type == "minor":
return f"### Hinzugefügt\n\n- {message}\n"
else:
return f"### Behoben\n\n- {message}\n"
# Standard-Vorlagen
if bump_type == "major":
return (
"### Hinzugefügt\n\n"
"- \n\n"
"### Geändert\n\n"
"- \n\n"
"### Entfernt\n\n"
"- \n"
)
elif bump_type == "minor":
return (
"### Hinzugefügt\n\n"
"- \n\n"
"### Geändert\n\n"
"- \n"
)
else:
return (
"### Behoben\n\n"
"- \n"
)
# ── Build ───────────────────────────────────────────────────────
def _ensure_pyinstaller() -> None:
"""Stellt sicher, dass PyInstaller installiert ist."""
try:
import PyInstaller # noqa: F401
except ImportError:
print("⚙ PyInstaller wird installiert …")
subprocess.check_call([sys.executable, "-m", "pip", "install", "pyinstaller"])
print(" ✓ PyInstaller installiert")
def _prepare_build_dir() -> None:
"""Bereitet das 2_build/-Verzeichnis vor."""
BUILD_OUTPUT_DIR.mkdir(exist_ok=True)
# Erstelle/aktualisiere Unterordner
for folder in ["input", "output", "template", "config"]:
target = BUILD_OUTPUT_DIR / folder
target.mkdir(exist_ok=True)
# Kopiere config-Dateien, falls noch nicht vorhanden
config_src = APP_DIR / "config"
config_dst = BUILD_OUTPUT_DIR / "config"
if config_src.exists():
for f in config_src.iterdir():
if f.is_file() and not (config_dst / f.name).exists():
shutil.copy2(f, config_dst / f.name)
# README für input-Ordner
input_readme = BUILD_OUTPUT_DIR / "input" / "_DATEIEN_HIER_ABLEGEN_.txt"
if not input_readme.exists():
input_readme.write_text(
"Excel-Dateien (.xlsx) in diesen Ordner legen.\n"
"Die App erkennt sie automatisch.\n",
encoding="utf-8",
)
# README für template-Ordner
tpl_readme = BUILD_OUTPUT_DIR / "template" / "_DATEIEN_HIER_ABLEGEN_.txt"
if not tpl_readme.exists():
tpl_readme.write_text(
"PDF-Templates (.pdf) mit _Field-Platzhaltern in diesen Ordner legen.\n"
"Die App erkennt sie automatisch.\n",
encoding="utf-8",
)
# README.md aus 0_application/ kopieren
readme_src = APP_DIR / "README.md"
if readme_src.exists():
shutil.copy2(readme_src, BUILD_OUTPUT_DIR / "README.md")
print(" ✓ README.md nach 2_build/ kopiert")
print(f" ✓ 2_build/ vorbereitet")
def _run_pyinstaller() -> Path:
"""Führt PyInstaller mit der Spec-Datei aus.
Returns:
Pfad zur erstellten .exe.
"""
print("\n🔨 PyInstaller baut .exe …")
print(" (Dies kann einige Minuten dauern)\n")
result = subprocess.run(
[
sys.executable, "-m", "PyInstaller",
str(SPEC_PATH),
"--distpath", str(BUILD_PIPELINE_DIR / "dist"),
"--workpath", str(BUILD_PIPELINE_DIR / "build_temp"),
"--noconfirm",
"--clean",
],
cwd=str(PROJECT_ROOT),
capture_output=False,
)
if result.returncode != 0:
print("\n❌ PyInstaller-Build fehlgeschlagen!")
sys.exit(1)
exe_path = BUILD_PIPELINE_DIR / "dist" / "Tätigkeitsnachweis-Generator.exe"
if not exe_path.exists():
print(f"\n❌ .exe nicht gefunden: {exe_path}")
sys.exit(1)
print(f" ✓ .exe erstellt: {exe_path.name}")
return exe_path
def _copy_to_build(exe_path: Path) -> Path:
"""Kopiert die .exe nach 2_build/."""
dest = BUILD_OUTPUT_DIR / exe_path.name
shutil.copy2(exe_path, dest)
print(f" ✓ .exe nach 2_build/ kopiert")
return dest
def _cleanup() -> None:
"""Räumt temporäre Build-Artefakte auf."""
for d in ["dist", "build_temp"]:
path = BUILD_PIPELINE_DIR / d
if path.exists():
shutil.rmtree(path, ignore_errors=True)
print(" ✓ Temporäre Dateien entfernt")
def _save_version(version: str) -> None:
"""Speichert die Version für spätere Referenz."""
VERSION_FILE.write_text(version)
def _print_summary(version: str, exe_path: Path) -> None:
"""Gibt eine Build-Zusammenfassung aus."""
print(f"\n{'='*60}")
print(f" ✅ Build {version} erfolgreich!")
print(f" 📂 {exe_path}")
print(f"{'='*60}\n")
print("Auslieferungs-Struktur (2_build/):")
print(" ├── Tätigkeitsnachweis-Generator.exe")
print(" ├── config/ ← Konfiguration (editierbar)")
print(" ├── input/ ← Excel-Dateien hier ablegen")
print(" ├── template/ ← PDF-Templates hier ablegen")
print(" └── output/ ← Generierte PDFs erscheinen hier")
# ── CLI ─────────────────────────────────────────────────────────
def main() -> None:
parser = argparse.ArgumentParser(
description="Build mit Semantic Versioning",
)
parser.add_argument(
"--bump",
choices=["major", "minor", "patch"],
help="Version bump-Typ (ohne Angabe: interaktiv)",
)
parser.add_argument(
"--message", "-m",
help="Changelog-Beschreibung für diese Version",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="Nur anzeigen, was passieren würde (kein Build)",
)
args = parser.parse_args()
# ── Version ermitteln ────────────────────────────────────
current = get_current_version()
if not current:
print("❌ Keine Version in CHANGELOG.md gefunden. Beginne mit 0.1.")
current = "0.1"
print(f"\n📋 Aktuelle Version: {current}")
# Bump-Typ bestimmen
bump_type = args.bump
if not bump_type:
print("\nWelche Art von Änderung?")
print(" [1] major – inkompatible API-Änderungen")
print(" [2] minor – neue Funktionalität (abwärtskompatibel)")
print(" [3] patch – Bugfixes")
try:
choice = input(" → ").strip()
bump_type = {"1": "major", "2": "minor", "3": "patch"}.get(choice, "minor")
except (EOFError, KeyboardInterrupt):
print("\nAbgebrochen.")
sys.exit(0)
new_version = bump_version(current, bump_type)
print(f"🚀 Neue Version: {new_version}")
if args.dry_run:
print("\n[Trockenlauf – kein Build]")
print(f" Würde CHANGELOG.md aktualisieren: [{new_version}]")
print(f" Würde PyInstaller-Build starten")
return
# ── CHANGELOG aktualisieren ──────────────────────────────
print("\n📝 Changelog …")
update_changelog(new_version, bump_type, args.message)
# ── Build ────────────────────────────────────────────────
_ensure_pyinstaller()
_prepare_build_dir()
try:
exe_path = _run_pyinstaller()
final_exe = _copy_to_build(exe_path)
finally:
_cleanup()
_save_version(new_version)
_print_summary(new_version, final_exe)
if __name__ == "__main__":
main()
+161
View File
@@ -0,0 +1,161 @@
# -*- mode: python ; coding: utf-8 -*-
"""PyInstaller Spec-Datei für den Tätigkeitsnachweis-Generator.
Baut eine standalone .exe aus run.py, die Streamlit, openpyxl und
PyMuPDF (fitz) enthält.
Verwendung:
pyinstaller build.spec
"""
import os
import sys
from pathlib import Path
# Pfade relativ zum Speicherort dieser Spec-Datei
# PyInstaller stellt SPECPATH bereit (__file__ existiert in exec()-Kontext nicht)
_SPEC_DIR = Path(SPECPATH).resolve()
_PROJECT_ROOT = _SPEC_DIR.parent
_APP_DIR = _PROJECT_ROOT / "0_application"
# ── Streamlit & PyMuPDF Hidden Imports ──────────────────────────
# Streamlit hat viele dynamische Imports; collect_all holt alles.
# PyMuPDF (fitz) braucht die native DLL.
hiddenimports = [
# Streamlit-Komponenten
"streamlit",
"streamlit.web.bootstrap",
"streamlit.runtime",
"streamlit.runtime.scriptrunner",
"streamlit.runtime.state",
"streamlit.runtime.caching",
"streamlit.watcher",
"streamlit.watcher.local_sources_watcher",
"streamlit.watcher.poll_path_watcher",
"streamlit.watcher.event_based_path_watcher",
"streamlit.elements",
"streamlit.elements.arrow",
"streamlit.elements.pyplot",
"streamlit.elements.altair",
"streamlit.elements.plotly",
"streamlit.proto",
# openpyxl
"openpyxl",
"openpyxl.cell",
"openpyxl.styles",
# PyMuPDF
"fitz",
# Standard-Libs die manchmal übersehen werden
"tornado",
"watchdog",
"pandas",
"numpy",
"PIL",
"altair",
"pyarrow",
"git",
"blinker",
"rich",
"pydeck",
]
# ── Binaries (PyMuPDF DLL) ──────────────────────────────────────
binaries = []
# Finde die PyMuPDF-DLL
try:
import fitz
fitz_dir = Path(fitz.__file__).parent
for f in fitz_dir.iterdir():
if f.suffix.lower() in (".dll", ".pyd", ".so", ".dylib"):
binaries.append((str(f), "fitz"))
except ImportError:
pass
# ── Data Files ──────────────────────────────────────────────────
datas = [
(str(_APP_DIR / "app.py"), "."),
(str(_APP_DIR / "version_check.py"), "."),
(str(_APP_DIR / "CHANGELOG.md"), "."),
(str(_APP_DIR / "src"), "src"),
(str(_APP_DIR / "config"), "config"),
]
# ── Alle Streamlit-Abhängigkeiten & Metadaten VOR Analysis() sammeln ─
from PyInstaller.utils.hooks import collect_all, copy_metadata
streamlit_datas, streamlit_binaries, streamlit_hidden = collect_all("streamlit")
datas.extend(streamlit_datas)
binaries.extend(streamlit_binaries)
hiddenimports.extend(streamlit_hidden)
# Package-Metadaten: streamlit & Co. nutzen importlib.metadata.version(),
# das die METADATA-Dateien aus dem dist-info-Ordner benötigt.
for pkg in ["streamlit", "pyarrow", "numpy", "pandas", "altair", "openpyxl"]:
try:
datas += copy_metadata(pkg)
except Exception:
pass
# Deduplizieren (Datas und Hiddenimports)
hiddenimports = list(set(hiddenimports))
seen = set()
_unique_datas = []
for src, dst in datas:
key = (src, dst)
if key not in seen:
seen.add(key)
_unique_datas.append((src, dst))
datas = _unique_datas
# ── Analyse ─────────────────────────────────────────────────────
a = Analysis(
[str(_SPEC_DIR / "run.py")],
pathex=[str(_APP_DIR)],
binaries=binaries,
datas=datas,
hiddenimports=hiddenimports,
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=[
"tkinter",
"matplotlib",
"scipy",
"IPython",
"jupyter",
"notebook",
"sqlalchemy",
],
win_no_prefer_redirects=False,
win_private_assemblies=False,
cipher=None,
noarchive=False,
)
# ── PyZ ─────────────────────────────────────────────────────────
pyz = PYZ(a.pure, a.zipped_data, cipher=None)
# ── EXE ─────────────────────────────────────────────────────────
exe = EXE(
pyz,
a.scripts,
a.binaries,
a.zipfiles,
a.datas,
[],
name="Tätigkeitsnachweis-Generator",
debug=False,
bootloader_ignore_signals=False,
strip=False,
upx=True,
upx_exclude=[],
runtime_tmpdir=None,
console=True, # True = Konsolenfenster sichtbar (zeigt Streamlit-Logs)
disable_windowed_traceback=False,
argv_emulation=False,
target_arch=None,
codesign_identity=None,
entitlements_file=None,
)
+109
View File
@@ -0,0 +1,109 @@
"""Einstiegspunkt für die gebundelte Tätigkeitsnachweis-Generator .exe.
Wird von PyInstaller als Entry-Point verwendet. Im gebundelten Modus
(sys.frozen = True) liegen die Python-Dateien (app.py, src/) in sys._MEIPASS,
während die Daten-Ordner (config/, input/, template/, output/) neben der .exe
erwartet werden.
Dieses Skript:
1. Setzt APP_BASE_DIR auf das Verzeichnis der .exe
2. Fügt sys._MEIPASS zum Python-Pfad hinzu (für src/-Imports)
3. Startet den Streamlit-Server und öffnet den Browser
"""
import os
import subprocess
import sys
import threading
import time
import webbrowser
from pathlib import Path
if not getattr(sys, "frozen", False):
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "0_application"))
from server_lifecycle import stop_existing_server
def _get_base_dir() -> Path:
"""Verzeichnis der .exe (bzw. des Skripts im Entwicklungsmodus)."""
if getattr(sys, 'frozen', False):
return Path(sys.executable).resolve().parent
else:
# Entwicklungsmodus: run.py liegt in 1_build-pipeline/,
# die App in 0_application/
return Path(__file__).resolve().parent.parent / "0_application"
def _get_app_path() -> str:
"""Pfad zur app.py (gebundelt oder im Entwicklungsmodus)."""
if getattr(sys, 'frozen', False):
return os.path.join(sys._MEIPASS, "app.py")
else:
return str(Path(__file__).resolve().parent.parent / "0_application" / "app.py")
def _ensure_dirs(base_dir: Path) -> None:
"""Stellt sicher, dass input/, output/, template/, config/ existieren."""
for folder in ["input", "output", "template", "config"]:
(base_dir / folder).mkdir(exist_ok=True)
def main() -> None:
# Only open the browser/start Streamlit once the previous server has stopped.
try:
if not stop_existing_server():
print("[INFO] Die bisherige Version bleibt geöffnet. Keine neue Instanz gestartet.")
return
except (RuntimeError, OSError, subprocess.SubprocessError) as exc:
print(f"[FEHLER] Server-Neustart fehlgeschlagen: {exc}", file=sys.stderr)
raise SystemExit(1)
base_dir = _get_base_dir()
# Im gebundelten Modus: sys._MEIPASS enthält src/ und app.py
if getattr(sys, 'frozen', False):
if sys._MEIPASS not in sys.path:
sys.path.insert(0, sys._MEIPASS)
# APP_BASE_DIR teilt app.py mit, wo die Daten-Ordner liegen
os.environ["APP_BASE_DIR"] = str(base_dir)
_ensure_dirs(base_dir)
# ── Streamlit-Konfiguration ──────────────────────────
os.environ["STREAMLIT_SERVER_ADDRESS"] = "localhost"
os.environ["STREAMLIT_SERVER_PORT"] = "8501"
os.environ["STREAMLIT_SERVER_HEADLESS"] = "true"
os.environ["STREAMLIT_BROWSER_GATHER_USAGE_STATS"] = "false"
os.environ["STREAMLIT_SERVER_ENABLE_CORS"] = "false"
os.environ["STREAMLIT_SERVER_ENABLE_XSRF_PROTECTION"] = "false"
# Browser manuell öffnen
def _open_browser() -> None:
time.sleep(2)
webbrowser.open("http://localhost:8501")
threading.Thread(target=_open_browser, daemon=True).start()
# ── Streamlit via CLI-Codepfad starten ───────────────
# Dieser Weg ist identisch zu 'streamlit run app.py' und
# funktioniert im PyInstaller-Bundle zuverlässiger als
# bootstrap.run() mit Streamlit >= 1.58.
sys.argv = [
"streamlit", "run", _get_app_path(),
"--server.port=8501",
"--server.address=localhost",
"--server.headless=true",
"--browser.gatherUsageStats=false",
"--server.enableCORS=false",
"--server.enableXsrfProtection=false",
"--global.developmentMode=false",
]
try:
from streamlit.cli import main as cli_main
except ImportError:
from streamlit.web.cli import main as cli_main
cli_main()
if __name__ == "__main__":
main()
Binary file not shown.
-220
View File
@@ -1,220 +0,0 @@
"""End-to-End-Test: Reale Input-Datei + reales PDF-Template → PDF-Output für erste Person."""
import sys
from pathlib import Path
import fitz # PyMuPDF
import pytest
# Projekt-Root zum Python-Path hinzufügen
PROJECT_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
from src.config import load_settings, load_input_schema, load_output_mapping
from src.parser import parse_input_file, group_by_employee, get_unique_employees
from src.writer import generate_activity_reports
# ──────────────────────────────────────────────
# Pfade zu den realen Fixture-Dateien
# ──────────────────────────────────────────────
CONFIG_DIR = PROJECT_ROOT / "config"
INPUT_FILE = PROJECT_ROOT / "input" / "2025 JM.xlsx"
TEMPLATE_FILE = PROJECT_ROOT / "template" / "Tätigkeitsnachweise Stellenförderung.pdf"
def _load_app_config():
"""Lädt die echte App-Konfiguration (Schema, Mapping, Settings)."""
config = load_settings(CONFIG_DIR, PROJECT_ROOT)
config.input_schema = load_input_schema(CONFIG_DIR)
config.output_mapping = load_output_mapping(CONFIG_DIR)
return config
# ──────────────────────────────────────────────
# Fixtures
# ──────────────────────────────────────────────
@pytest.fixture(scope="module")
def app_config():
"""Reale App-Konfiguration aus config/."""
return _load_app_config()
@pytest.fixture(scope="module")
def parsed_records(app_config):
"""Geparste Records aus der realen Input-Datei."""
return parse_input_file(
filepath=INPUT_FILE,
schema=app_config.input_schema,
month_sheet_pattern=app_config.month_sheet_pattern,
skip_empty_name=app_config.skip_empty_name,
skip_empty_antragsnummer=app_config.skip_empty_antragsnummer,
)
@pytest.fixture
def output_dir(tmp_path):
"""Temporäres Output-Verzeichnis."""
out = tmp_path / "output"
out.mkdir()
return out
# ──────────────────────────────────────────────
# Hilfsfunktion: Text aus PDF extrahieren
# ──────────────────────────────────────────────
def _get_pdf_text(pdf_path: Path) -> str:
"""Extrahiert den gesamten Text aus einem PDF."""
doc = fitz.open(pdf_path)
text = ""
for page in doc:
text += page.get_text()
doc.close()
return text
def _get_page_text(pdf_path: Path, page_num: int = 0) -> str:
"""Extrahiert Text von einer bestimmten PDF-Seite."""
doc = fitz.open(pdf_path)
text = doc[page_num].get_text() if page_num < doc.page_count else ""
doc.close()
return text
# ──────────────────────────────────────────────
# End-to-End Tests
# ──────────────────────────────────────────────
class TestEndToEndFirstPerson:
"""End-to-End: Reale Input-Datei + reales PDF-Template → nur erste Person."""
def test_input_file_exists(self):
"""Stellt sicher, dass die reale Input-Datei vorhanden ist."""
assert INPUT_FILE.exists(), f"Input-Datei nicht gefunden: {INPUT_FILE}"
def test_template_file_exists(self):
"""Stellt sicher, dass das reale PDF-Template vorhanden ist."""
assert TEMPLATE_FILE.exists(), f"PDF-Template nicht gefunden: {TEMPLATE_FILE}"
def test_template_has_field_markers(self, app_config):
"""Stellt sicher, dass alle _Field-Platzhalter im PDF-Template vorhanden sind."""
doc = fitz.open(TEMPLATE_FILE)
missing = []
for m in app_config.output_mapping:
if m.pdf_field is None:
continue
# Suche auf allen Seiten
found_any = False
for page_num in range(doc.page_count):
if doc[page_num].search_for(m.pdf_field):
found_any = True
break
if not found_any:
missing.append(m.pdf_field)
doc.close()
assert len(missing) == 0, (
f"Folgende _Field-Platzhalter fehlen im PDF: {missing}"
)
def test_parse_real_input(self, parsed_records):
"""Validiert das Parsen der realen Input-Datei."""
# Sollte 168 Einträge über 12 Monate für 19 Personen ergeben
assert len(parsed_records) == 168, (
f"Erwartet 168 Records, aber {len(parsed_records)} gefunden"
)
employees = get_unique_employees(parsed_records)
assert len(employees) == 19, (
f"Erwartet 19 Mitarbeiter, aber {len(employees)} gefunden"
)
# Erste Person alphabetisch: Draßdo, Cornelia
assert employees[0] == "Draßdo, Cornelia", (
f"Erwartet 'Draßdo, Cornelia' als erste Person, aber '{employees[0]}'"
)
def test_full_pipeline_first_person(
self, app_config, parsed_records, output_dir
):
"""Kompletter Workflow für die erste gefundene Person (Draßdo, Cornelia)."""
# 1. Nach Mitarbeiter gruppieren
grouped = group_by_employee(parsed_records)
assert len(grouped) == 19
# 2. Erste Person alphabetisch
first_person = sorted(grouped.keys())[0]
assert first_person == "Draßdo, Cornelia"
first_records = grouped[first_person]
# Draßdo hat 12 Einträge (einen pro Monat)
assert len(first_records) == 12, (
f"Erwartet 12 Records für {first_person}, aber {len(first_records)}"
)
# 3. Reports NUR für die erste Person generieren
single_person_group = {first_person: first_records}
generated = generate_activity_reports(
employee_records=single_person_group,
template_path=TEMPLATE_FILE,
mapping=app_config.output_mapping,
output_dir=output_dir,
filename_pattern=app_config.filename_pattern,
)
# 4. Ergebnisse prüfen
assert len(generated) == 1, (
f"Erwartet 1 generierte Datei, aber {len(generated)}"
)
output_file = generated[0]
assert output_file.exists(), f"Output-Datei nicht gefunden: {output_file}"
assert output_file.suffix == ".pdf"
erwarteter_name = "Draßdo_Cornelia_Tätigkeitsnachweis.pdf"
assert output_file.name == erwarteter_name, (
f"Erwartet '{erwarteter_name}', aber '{output_file.name}'"
)
# 5. PDF öffnen und Inhalt validieren
doc = fitz.open(output_file)
assert doc.page_count == 24, (
f"Erwartet 24 Seiten (12 Monate × 2 Seiten), aber {doc.page_count}"
)
# 6. Prüfe, dass KEINE _Field-Platzhalter mehr im Output sind
full_text = ""
for page in doc:
full_text += page.get_text()
assert "_Field" not in full_text, (
"Es sind noch _Field-Platzhalter im Output-PDF vorhanden!"
)
# 7. Prüfe Seite 0 (Januar, Seite 1 – Stammdaten)
jan0_text = doc[0].get_text()
assert "100740439" in jan0_text, "Antragsnummer fehlt auf Januar-Seite 1"
assert "Draßdo, Cornelia" in jan0_text, "Name fehlt auf Januar-Seite 1"
assert "50" in jan0_text, "Stellenanteil fehlt auf Januar-Seite 1"
assert "Meisterabschluß Küche" in jan0_text, "Qualifikation fehlt auf Januar-Seite 1"
assert "01.2025" in jan0_text, "Monat fehlt auf Januar-Seite 1"
assert "Anleiterin / Ausbilderin" in jan0_text, "Funktion fehlt auf Januar-Seite 1"
assert "Tätigkeiten im Praxisbereich Garten" in jan0_text, "Tätigkeit fehlt auf Januar-Seite 1"
# 8. Prüfe Seite 1 (Januar, Seite 2 – Unterschriftsblock mit Ort/Datum)
jan1_text = doc[1].get_text()
assert "100740439" not in jan1_text, "Antragsnummer sollte nicht auf Seite 2 sein"
assert "Dresden" in jan1_text, "Ort 'Dresden' fehlt auf Januar-Seite 2"
assert "31.01.2025" in jan1_text, "Datum '31.01.2025' fehlt auf Januar-Seite 2"
# Tätigkeit wiederholt auf Seite 2
assert "Tätigkeiten im Praxisbereich Garten" in jan1_text, "Tätigkeit fehlt auf Januar-Seite 2"
# 9. Prüfe Juli (Seite 12+13) – Stellenanteil wechselt auf 40%
jul0_text = doc[12].get_text()
assert "40" in jul0_text, "Stellenanteil 40% fehlt auf Juli-Seite 1"
assert "07.2025" in jul0_text, "Monat 07.2025 fehlt auf Juli-Seite 1"
jul1_text = doc[13].get_text()
assert "31.07.2025" in jul1_text, "Datum 31.07.2025 fehlt auf Juli-Seite 2"
doc.close()