5 Commits
19 changed files with 963 additions and 188 deletions

No files matched your search

+4
View File
@@ -61,3 +61,7 @@ $RECYCLE.BIN/
*.vsix
3_Archiv
2_build
# Python-Test- und Build-Artefakte
__pycache__/
.pytest_cache/
+16
View File
@@ -7,6 +7,22 @@ und dieses Projekt folgt [Semantic Versioning](https://semver.org/lang/de/).
---
## [1.3.2] – 2026-10-07
### Behoben
- EXE erkennt laufende Entwicklungsinstanz und fragt vor dem Neustart
---
## [1.3.1] – 2026-10-07
### Behoben
- Python script erkennt jetzt laufende Entwicklungsinstanz und fragt vor dem Neustart
---
## [1.3] – 2026-08-11
### Hinzugefügt
+81 -7
View File
@@ -87,8 +87,27 @@ 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.
Beim Build laufen die Tests automatisch vor der Versionierung und dem Bauen.
Beim Programmstart werden sie nicht ausgeführt.
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 +115,67 @@ 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
## Build und Deployment unter Windows
Alle folgenden Befehle werden im Projekt-Hauptordner ausgeführt:
```powershell
# Interaktiv: Versionsänderung auswählen
.\1_build-pipeline\build.bat
# Bugfix-Version mit Changelog-Beschreibung
.\1_build-pipeline\build.bat --bump patch --message "Fehler beim Programmstart behoben"
# Ablauf anzeigen, ohne Tests, Build oder Deployment auszuführen
.\1_build-pipeline\build.bat --bump patch --dry-run
# Aktuelle Version mit aktualisiertem Changelog erneut als EXE bauen
.\1_build-pipeline\build.bat --no-bump --no-deploy
# Nur den letzten erfolgreichen Build bereitstellen (kein neuer Build)
.\1_build-pipeline\deploy.bat
# Deployment ohne abschließende Tasteneingabe
.\1_build-pipeline\deploy.bat --no-pause
```
Die Pipeline führt zuerst alle Tests aus. Bei einem Testfehler bleiben Version
und Changelog unverändert; Build und Deployment werden nicht gestartet.
Nach erfolgreichen Tests aktualisiert sie die Version, erstellt die EXE und
legt sie mit Konfiguration, PDF-Template, README und Changelog in `2_build/` ab.
`release.json` hält die Version und Prüfsummen des erfolgreichen Builds fest.
Bei einem Buildfehler wird der Changelog zurückgesetzt.
Danach erzeugt das Deployment `Tätigkeitsnachweis-Generator vX.Y.Z.zip` und
überträgt es an:
```text
j:\Fachbereiche\BO\Produktionsschule\Maßnahmeunterlagen\20_Personal\05_Tätigkeitsnachweise ESF\Nachweise-automatisiert-generieren\
```
Das Ziel muss bereits vorhanden und über VPN/SMB erreichbar sein. Ist es nicht
erreichbar oder fehlen Schreibrechte, bricht das Deployment verständlich ab.
Der lokale Build bleibt erhalten. Nach Wiederherstellung der VPN-Verbindung
kann `deploy.bat` die Bereitstellung ohne erneutes Testen, Bauen oder Erhöhen
der Versionsnummer wiederholen. Das Skript akzeptiert nur einen bestätigten,
unveränderten Build mit `release.json`; ältere Builds müssen einmal neu gebaut
werden.
Das ZIP enthält leere `input/`- und `output/`-Ordner; Arbeitsdateien und
Testdaten werden nicht mit ausgeliefert. Es wird zunächst als temporäre Datei
übertragen, per Prüfsumme geprüft und erst dann unter dem Versionsnamen
sichtbar gemacht. Bestehende Versionen mit anderem Inhalt werden nicht
überschrieben. Ein identisches Paket kann erneut bereitgestellt werden.
Die Neustarttests simulieren Prozesse und Dialogantworten; ein manueller
Starttest der fertigen EXE bleibt vor der Weitergabe sinnvoll.
+2
View File
@@ -1,4 +1,6 @@
@echo off
chcp 65001 >nul
set "PYTHONUTF8=1"
title Tätigkeitsnachweis-Generator – Cold Run
echo ========================================
echo Cold Run – Headless-Generierung
+117
View File
@@ -0,0 +1,117 @@
"""Stop an existing instance of this application's Windows server."""
import json
import os
import subprocess
import sys
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 = Path(__file__).resolve().with_name("app.py")
app_paths = [str(app_path)]
if getattr(sys, "frozen", False):
# A locally built EXE in 2_build/ must also recognize the sibling
# development application, whose path differs from _MEIPASS/app.py.
exe_dir = Path(sys.executable).resolve().parent
development_app = exe_dir.parent / "0_application" / "app.py"
if exe_dir.name.lower() == "2_build" and development_app.is_file():
app_paths.append(str(development_app))
# Pass paths through the environment, never interpolate them into shell code.
env = dict(os.environ, APP_RESTART_PATH=str(app_path),
APP_RESTART_PATHS=json.dumps(app_paths), 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
$matchesApp = $false
foreach ($appPath in ($env:APP_RESTART_PATHS | ConvertFrom-Json)) {
$appArgument = '(?i)\bstreamlit(?:\.exe)?"?\s+run\s+"?' +
[regex]::Escape($appPath) + '"?(?:\s|$)'
if ($command -match $appArgument) { $matchesApp = $true }
}
$isDev = $process.Name -match '^(python(?:w)?(?:\d+(?:\.\d+)*)?(?:w)?|streamlit)\.exe$' -and
$command -match '(?i)\bstreamlit(?:\.exe)?"?\s+run\s+' -and
($matchesApp -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)
+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
+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%
+1 -1
View File
@@ -27,7 +27,7 @@ echo ""
echo "[INFO] Führe End-to-End-Tests aus…"
echo ""
"$VENV_DIR/bin/python" -m pytest tests/test_e2e.py -v
"$VENV_DIR/bin/python" -m pytest tests -v
echo ""
read -p "Drücke Enter zum Beenden..."
Binary file not shown.
+185
View File
@@ -0,0 +1,185 @@
"""Verify pipeline gates and deployment against temporary local directories."""
import importlib.util
from pathlib import Path
import sys
from unittest.mock import patch, Mock
import zipfile
import pytest
PIPELINE_DIR = Path(__file__).resolve().parents[2] / "1_build-pipeline"
sys.path.insert(0, str(PIPELINE_DIR))
import deploy
spec = importlib.util.spec_from_file_location("build_pipeline", PIPELINE_DIR / "build.py")
build = importlib.util.module_from_spec(spec)
spec.loader.exec_module(build)
@pytest.fixture
def release(tmp_path):
directory = tmp_path / "build"
directory.mkdir()
for name in deploy.PACKAGE_FILES:
path = directory / name
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(b"test release payload")
deploy.write_release_manifest(directory, "1.3")
target = tmp_path / "share"
target.mkdir()
return directory, target
def test_deployment_packages_release_without_work_files(release):
directory, target = release
for folder in ["input", "output"]:
(directory / folder).mkdir()
(directory / folder / "private.xlsx").write_bytes(b"private")
result = deploy.deploy(directory, target)
assert result.name == "Tätigkeitsnachweis-Generator v1.3.0.zip"
with zipfile.ZipFile(result) as archive:
assert set(archive.namelist()) == set(deploy.PACKAGE_FILES + ["input/", "output/"])
assert archive.testzip() is None
assert not list(target.glob(".upload-*"))
# An unchanged release can be retried without overwriting the published ZIP.
original_time = result.stat().st_mtime_ns
assert deploy.deploy(directory, target) == result
assert result.stat().st_mtime_ns == original_time
def test_missing_vpn_target_is_not_created(release):
directory, target = release
missing = target / "missing-share"
with pytest.raises(OSError, match="VPN"):
deploy.deploy(directory, missing)
assert not missing.exists()
assert not list(directory.glob("*.zip"))
def test_changed_payload_is_not_deployed(release):
directory, target = release
(directory / deploy.EXE_NAME).write_bytes(b"different executable")
with pytest.raises(ValueError, match="verändert"):
deploy.deploy(directory, target)
assert not list(target.iterdir())
def test_unconfirmed_old_build_is_not_deployed(release):
directory, target = release
(directory / deploy.MANIFEST_NAME).unlink()
with pytest.raises(ValueError, match="bestätigter Build"):
deploy.deploy(directory, target)
def test_existing_different_release_is_preserved(release):
directory, target = release
existing = target / "Tätigkeitsnachweis-Generator v1.3.0.zip"
existing.write_bytes(b"already published")
with pytest.raises(FileExistsError, match="anderem Inhalt"):
deploy.deploy(directory, target)
assert existing.read_bytes() == b"already published"
def test_interrupted_upload_is_not_published(release):
directory, target = release
with patch.object(deploy.shutil, "copyfile", side_effect=OSError("VPN disconnected")):
with pytest.raises(OSError, match="lokale Build bleibt erhalten"):
deploy.deploy(directory, target)
assert not list(target.iterdir())
assert list(directory.glob("*.zip"))
assert deploy.validate_release(directory) == "1.3.0"
def test_corrupt_upload_is_not_published(release):
directory, target = release
def corrupt_copy(source, destination):
Path(destination).write_bytes(b"corrupted")
with patch.object(deploy.shutil, "copyfile", side_effect=corrupt_copy):
with pytest.raises(OSError):
deploy.deploy(directory, target)
assert not list(target.iterdir())
@pytest.fixture
def pipeline(tmp_path, monkeypatch):
changelog = tmp_path / "CHANGELOG.md"
changelog.write_text("## [1.3]\n\nOriginal release\n", encoding="utf-8")
version_file = tmp_path / ".last_version"
version_file.write_text("1.3", encoding="utf-8")
monkeypatch.setattr(build, "CHANGELOG_PATH", changelog)
monkeypatch.setattr(build, "VERSION_FILE", version_file)
monkeypatch.setattr(build, "BUILD_OUTPUT_DIR", tmp_path)
monkeypatch.setattr(sys, "argv", ["build.py", "--bump", "patch", "--message", "Fix"])
events = []
names = ["_run_tests", "_ensure_pyinstaller", "_prepare_build_dir", "_run_pyinstaller",
"_copy_to_build", "write_release_manifest", "_cleanup", "_print_summary", "deploy"]
mocks = {}
for name in names:
mocks[name] = Mock(side_effect=lambda *args, _name=name: events.append(_name))
monkeypatch.setattr(build, name, mocks[name])
return changelog, version_file, events, mocks
def test_pipeline_tests_build_then_deploy(pipeline):
changelog, version_file, events, mocks = pipeline
build.main()
assert events == ["_run_tests", "_ensure_pyinstaller", "_prepare_build_dir", "_run_pyinstaller",
"_copy_to_build", "write_release_manifest", "_cleanup", "_print_summary", "deploy"]
assert version_file.read_text() == "1.3.1"
assert "## [1.3.1]" in changelog.read_text(encoding="utf-8")
def test_failing_tests_prevent_version_change_build_and_deploy(pipeline):
changelog, version_file, events, mocks = pipeline
original = changelog.read_bytes()
mocks["_run_tests"].side_effect = SystemExit(1)
with pytest.raises(SystemExit):
build.main()
assert changelog.read_bytes() == original
assert version_file.read_text() == "1.3"
mocks["_run_pyinstaller"].assert_not_called()
mocks["deploy"].assert_not_called()
def test_build_failure_rolls_back_changelog_and_prevents_deployment(pipeline):
changelog, version_file, events, mocks = pipeline
original = changelog.read_text(encoding="utf-8")
mocks["_run_pyinstaller"].side_effect = SystemExit(1)
with pytest.raises(SystemExit):
build.main()
assert changelog.read_text(encoding="utf-8") == original
assert version_file.read_text() == "1.3"
mocks["deploy"].assert_not_called()
mocks["_cleanup"].assert_called_once()
def test_deploy_failure_keeps_successful_build_version(pipeline):
changelog, version_file, events, mocks = pipeline
mocks["deploy"].side_effect = OSError("VPN unavailable")
with pytest.raises(SystemExit):
build.main()
assert version_file.read_text() == "1.3.1"
assert "## [1.3.1]" in changelog.read_text(encoding="utf-8")
mocks["write_release_manifest"].assert_called_once()
def test_dry_run_does_not_test_build_or_deploy(pipeline, monkeypatch):
changelog, version_file, events, mocks = pipeline
monkeypatch.setattr(sys, "argv", ["build.py", "--bump", "patch", "--dry-run"])
build.main()
assert not events
assert version_file.read_text() == "1.3"
def test_rebuild_preserves_changelog_and_version(pipeline, monkeypatch):
changelog, version_file, events, mocks = pipeline
original = changelog.read_bytes()
monkeypatch.setattr(sys, "argv", ["build.py", "--no-bump", "--no-deploy"])
build.main()
assert changelog.read_bytes() == original
assert version_file.read_text() == "1.3"
mocks["write_release_manifest"].assert_called_once_with(build.BUILD_OUTPUT_DIR, "1.3")
mocks["_run_tests"].assert_called_once()
mocks["_run_pyinstaller"].assert_called_once()
mocks["deploy"].assert_not_called()
+62 -161
View File
@@ -1,12 +1,11 @@
"""End-to-End-Test: Reale Input-Datei + reales PDF-Template → PDF-Output für erste Person."""
"""End-to-end checks using the fixed 2026 workbook and the PDF template."""
import sys
from pathlib import Path
import fitz # PyMuPDF
import fitz
import pytest
# Projekt-Root zum Python-Path hinzufügen
PROJECT_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
@@ -14,37 +13,22 @@ 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"
INPUT_FILE = Path(__file__).resolve().parent / "fixture" / "2026 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)."""
@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
# ──────────────────────────────────────────────
# 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."""
assert INPUT_FILE.is_file(), f"Testdatei nicht gefunden: {INPUT_FILE}"
return parse_input_file(
filepath=INPUT_FILE,
schema=app_config.input_schema,
@@ -56,165 +40,82 @@ def parsed_records(app_config):
@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}"
assert INPUT_FILE.is_file(), f"Testdatei 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}"
assert TEMPLATE_FILE.is_file(), 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}"
)
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):
"""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"
)
# Fixed expectations for tests/fixture/2026 JM.xlsx.
assert len(parsed_records) == 100
employees = get_unique_employees(parsed_records)
assert len(employees) == 19, (
f"Erwartet 19 Mitarbeiter, aber {len(employees)} gefunden"
)
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")
# 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
def test_full_pipeline_first_person(self, app_config, parsed_records, output_dir):
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}
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=single_person_group,
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,
)
# 4. Ergebnisse prüfen
assert len(generated) == 1, (
f"Erwartet 1 generierte Datei, aber {len(generated)}"
)
assert len(generated) == 1
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()
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,192 @@
"""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, frozen=False):
captured = {}
def capture(command, **kwargs):
captured.update(script=command[-1], env=kwargs["env"])
return subprocess.CompletedProcess(command, 0, "[]", "")
root = Path(__file__).resolve().parents[2]
with patch.object(server_lifecycle.subprocess, "run", side_effect=capture), \
patch.object(server_lifecycle.sys, "frozen", frozen, create=True), \
patch.object(server_lifecycle.sys, "executable", str(root / "2_build" / "Tätigkeitsnachweis-Generator.exe")), \
patch.object(server_lifecycle, "__file__", str(root / "_MEI-test" / "server_lifecycle.py") if frozen else server_lifecycle.__file__):
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_exe_recognizes_sibling_development_server(self):
process = r'''[pscustomobject]@{
Name='python3.13.exe'; ExecutablePath='C:\Python\python3.13.exe';
CommandLine=('".venv\Scripts\python.exe" -m streamlit run "' +
(($env:APP_RESTART_PATHS | ConvertFrom-Json)[1]) +
'" --server.port=8501 --server.address=localhost')
}'''
for answer, expected_code in [(True, 0), (False, 2)]:
with self.subTest(confirm=answer):
result = self.simulate(process, frozen=True, confirm=answer)
self.assertEqual(result.returncode, expected_code, result.stderr)
self.assertIn("CONFIRM_CALLED", result.stderr)
if answer:
self.assertIn("STOP_CALLED", result.stderr)
else:
self.assertNotIn("STOP_CALLED", result.stderr)
def test_exe_preserves_unrelated_development_server(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"'
}''', frozen=True)
self.assertNotEqual(result.returncode, 0)
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()
+1 -1
View File
@@ -1 +1 @@
1.3
1.3.2
+3 -2
View File
@@ -1,6 +1,7 @@
@echo off
chcp 65001 >nul
setlocal enabledelayedexpansion
set "PYTHONUTF8=1"
cd /d "%~dp0"
cd ..
@@ -36,11 +37,11 @@ REM ── Build ausführen ─────────────────
if errorlevel 1 (
echo.
echo [FEHLER] Build fehlgeschlagen!
echo [FEHLER] Pipeline fehlgeschlagen. Details stehen oben.
pause
exit /b 1
)
echo.
echo Build abgeschlossen.
echo Tests, Build und Deployment abgeschlossen.
pause
+68 -15
View File
@@ -2,7 +2,8 @@
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/.
führt zuerst alle Tests aus, baut via PyInstaller eine standalone .exe
nach 2_build/ und stellt sie anschließend als ZIP auf dem SMB-Laufwerk bereit.
Verwendung:
python build.py # interaktiver Modus
@@ -20,6 +21,8 @@ from datetime import date
from pathlib import Path
from typing import Optional
from deploy import deploy, write_release_manifest
# ── Pfade ───────────────────────────────────────────────────────
BUILD_PIPELINE_DIR = Path(__file__).resolve().parent
PROJECT_ROOT = BUILD_PIPELINE_DIR.parent
@@ -177,6 +180,21 @@ def _build_sections(bump_type: str, message: Optional[str]) -> str:
# ── Build ───────────────────────────────────────────────────────
def _run_tests() -> None:
"""Test with the same interpreter used to build the application."""
try:
import pytest # noqa: F401
except ImportError:
subprocess.check_call([sys.executable, "-m", "pip", "install", "pytest"])
print("\n[INFO] Führe alle Tests aus ...", flush=True)
result = subprocess.run(
[sys.executable, "-m", "pytest", str(APP_DIR / "tests"), "-v", "-p", "no:cacheprovider"],
cwd=str(PROJECT_ROOT),
)
if result.returncode != 0:
print("[FEHLER] Tests fehlgeschlagen. Kein Build und kein Deployment.")
raise SystemExit(1)
def _ensure_pyinstaller() -> None:
"""Stellt sicher, dass PyInstaller installiert ist."""
try:
@@ -196,14 +214,20 @@ def _prepare_build_dir() -> None:
target = BUILD_OUTPUT_DIR / folder
target.mkdir(exist_ok=True)
# Kopiere config-Dateien, falls noch nicht vorhanden
# Liefere die zum aktuellen Quellstand gehörende Konfiguration aus.
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():
if f.is_file():
shutil.copy2(f, config_dst / f.name)
shutil.copy2(CHANGELOG_PATH, BUILD_OUTPUT_DIR / "CHANGELOG.md")
shutil.copy2(
APP_DIR / "template" / "Tätigkeitsnachweise Stellenförderung.pdf",
BUILD_OUTPUT_DIR / "template" / "Tätigkeitsnachweise Stellenförderung.pdf",
)
# README für input-Ordner
input_readme = BUILD_OUTPUT_DIR / "input" / "_DATEIEN_HIER_ABLEGEN_.txt"
if not input_readme.exists():
@@ -308,11 +332,16 @@ def main() -> None:
parser = argparse.ArgumentParser(
description="Build mit Semantic Versioning",
)
parser.add_argument(
version_options = parser.add_mutually_exclusive_group()
version_options.add_argument(
"--bump",
choices=["major", "minor", "patch"],
help="Version bump-Typ (ohne Angabe: interaktiv)",
)
version_options.add_argument("--no-bump", action="store_true",
help="Aktuelle Version erneut bauen; Changelog unverändert übernehmen")
parser.add_argument("--no-deploy", action="store_true",
help="Nur lokal bauen, ohne anschließendes Deployment")
parser.add_argument(
"--message", "-m",
help="Changelog-Beschreibung für diese Version",
@@ -327,6 +356,8 @@ def main() -> None:
# ── Version ermitteln ────────────────────────────────────
current = get_current_version()
if not current:
if args.no_bump:
parser.error("Für --no-bump muss eine Version im CHANGELOG.md vorhanden sein.")
print("❌ Keine Version in CHANGELOG.md gefunden. Beginne mit 0.1.")
current = "0.1"
@@ -334,7 +365,7 @@ def main() -> None:
# Bump-Typ bestimmen
bump_type = args.bump
if not bump_type:
if not bump_type and not args.no_bump:
print("\nWelche Art von Änderung?")
print(" [1] major – inkompatible API-Änderungen")
print(" [2] minor – neue Funktionalität (abwärtskompatibel)")
@@ -346,32 +377,54 @@ def main() -> None:
print("\nAbgebrochen.")
sys.exit(0)
new_version = bump_version(current, bump_type)
new_version = current if args.no_bump else 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(" Würde zuerst alle Tests ausführen")
if not args.no_bump:
print(f" Würde CHANGELOG.md aktualisieren: [{new_version}]")
print(f" Würde PyInstaller-Build starten")
if not args.no_deploy:
print(" Würde anschließend ein versioniertes ZIP auf dem SMB-Laufwerk bereitstellen")
return
# ── CHANGELOG aktualisieren ──────────────────────────────
print("\n📝 Changelog …")
update_changelog(new_version, bump_type, args.message)
# ── Build ────────────────────────────────────────────────
_ensure_pyinstaller()
_prepare_build_dir()
_run_tests()
previous_changelog = _read_changelog()
try:
_ensure_pyinstaller()
(BUILD_OUTPUT_DIR / "release.json").unlink(missing_ok=True)
if not args.no_bump:
print("\n📝 Changelog …")
update_changelog(new_version, bump_type, args.message)
_prepare_build_dir()
exe_path = _run_pyinstaller()
final_exe = _copy_to_build(exe_path)
write_release_manifest(BUILD_OUTPUT_DIR, new_version)
except (Exception, SystemExit, KeyboardInterrupt):
_write_changelog(previous_changelog)
raise
finally:
_cleanup()
_save_version(new_version)
_print_summary(new_version, final_exe)
if args.no_deploy:
print("[INFO] Lokaler Build abgeschlossen; Deployment nicht angefordert.")
return
try:
deploy()
except (OSError, ValueError, KeyError) as exc:
print(f"[FEHLER] {exc}")
print("[INFO] Der Build ist fertig. Deployment später mit 1_build-pipeline\\deploy.bat wiederholen.")
raise SystemExit(1)
if __name__ == "__main__":
main()
try:
main()
except (OSError, ValueError, subprocess.CalledProcessError) as exc:
print(f"[FEHLER] Pipeline abgebrochen: {exc}")
raise SystemExit(1)
+2
View File
@@ -76,6 +76,8 @@ except ImportError:
# ── 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"),
]
+18
View File
@@ -0,0 +1,18 @@
@echo off
chcp 65001 >nul
setlocal
set "PYTHONUTF8=1"
cd /d "%~dp0.."
if exist "0_application\.venv\Scripts\python.exe" (
set "DEPLOY_PYTHON=0_application\.venv\Scripts\python.exe"
) else if exist ".venv\Scripts\python.exe" (
set "DEPLOY_PYTHON=.venv\Scripts\python.exe"
) else (
set "DEPLOY_PYTHON=python"
)
"%DEPLOY_PYTHON%" "1_build-pipeline\deploy.py"
set "DEPLOY_EXIT_CODE=%ERRORLEVEL%"
if /I not "%~1"=="--no-pause" pause
exit /b %DEPLOY_EXIT_CODE%
+143
View File
@@ -0,0 +1,143 @@
"""Package and publish the last successful build to the SMB update directory."""
import argparse
import hashlib
import json
import os
from pathlib import Path
import re
import shutil
import tempfile
import zipfile
PROJECT_ROOT = Path(__file__).resolve().parent.parent
BUILD_DIR = PROJECT_ROOT / "2_build"
DEPLOY_DIR = Path(r"j:\Fachbereiche\BO\Produktionsschule\Maßnahmeunterlagen\20_Personal\05_Tätigkeitsnachweise ESF\Nachweise-automatisiert-generieren")
EXE_NAME = "Tätigkeitsnachweis-Generator.exe"
MANIFEST_NAME = "release.json"
PACKAGE_FILES = [
EXE_NAME, "README.md", "CHANGELOG.md",
"config/settings.ini", "config/input-fields.json", "config/output-fields.json",
"template/Tätigkeitsnachweise Stellenförderung.pdf",
]
def normalize_version(version: str) -> str:
if not re.fullmatch(r"\d+\.\d+(?:\.\d+)?", version):
raise ValueError(f"Ungültige Versionsnummer: {version}")
return ".".join(str(int(part)) for part in (version.split(".") + ["0"])[:3])
def file_hash(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as source:
for block in iter(lambda: source.read(1024 * 1024), b""):
digest.update(block)
return digest.hexdigest()
def write_release_manifest(build_dir: Path, version: str) -> None:
"""Record the version and all payload hashes after a successful build."""
manifest = {
"version": normalize_version(version),
"files": {name: file_hash(build_dir / name) for name in PACKAGE_FILES},
}
(build_dir / MANIFEST_NAME).write_text(
json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
def validate_release(build_dir: Path) -> str:
if not (build_dir / MANIFEST_NAME).is_file():
raise ValueError("Kein bestätigter Build vorhanden. Bitte zuerst build.bat ausführen.")
manifest = json.loads((build_dir / MANIFEST_NAME).read_text(encoding="utf-8"))
version = normalize_version(manifest["version"])
for name in PACKAGE_FILES:
path = build_dir / name
if not path.is_file() or file_hash(path) != manifest.get("files", {}).get(name):
raise ValueError(f"Build-Datei fehlt oder wurde nach dem Build verändert: {name}. Bitte neu bauen.")
return version
def check_deploy_target(target_dir: Path) -> None:
"""Never create the SMB destination: a missing mount must be an error."""
try:
if not target_dir.is_dir():
raise OSError("Verzeichnis ist nicht verfügbar")
# Actually access the directory; exists() alone does not check read access.
with os.scandir(target_dir) as entries:
next(entries, None)
except OSError as exc:
raise OSError(
f"Deployment-Ziel nicht erreichbar: {target_dir}\n"
"Bitte VPN-Verbindung und Netzlaufwerk J: prüfen und deploy.bat erneut starten."
) from exc
def deploy(build_dir: Path = BUILD_DIR, target_dir: Path = DEPLOY_DIR) -> Path:
version = validate_release(build_dir)
check_deploy_target(target_dir)
archive_name = f"Tätigkeitsnachweis-Generator v{version}.zip"
archive = build_dir / archive_name
descriptor, temporary = tempfile.mkstemp(prefix="release-", suffix=".zip", dir=build_dir)
os.close(descriptor)
temporary = Path(temporary)
try:
with zipfile.ZipFile(temporary, "w", zipfile.ZIP_DEFLATED) as package:
for name in PACKAGE_FILES:
package.write(build_dir / name, name)
for directory in ["input/", "output/"]:
# Fixed timestamps keep repeated deployments byte-identical.
package.writestr(zipfile.ZipInfo(directory, date_time=(2020, 1, 1, 0, 0, 0)), b"")
with zipfile.ZipFile(temporary) as package:
if package.testzip() is not None:
raise ValueError("Das erzeugte ZIP-Paket ist beschädigt.")
temporary.replace(archive)
finally:
temporary.unlink(missing_ok=True)
destination = target_dir / archive_name
if destination.exists():
if file_hash(destination) == file_hash(archive):
print(f"[OK] Diese Version ist bereits unverändert bereitgestellt: {destination}")
return destination
raise FileExistsError(f"{destination} existiert bereits mit anderem Inhalt. Bitte eine neue Version bauen.")
# Write access is checked by creating the staging file on the actual share.
remote_temporary = None
try:
descriptor, name = tempfile.mkstemp(prefix=".upload-", suffix=".tmp", dir=target_dir)
os.close(descriptor)
remote_temporary = Path(name)
shutil.copyfile(archive, remote_temporary)
if file_hash(remote_temporary) != file_hash(archive):
raise OSError("Die Prüfsumme nach der SMB-Übertragung stimmt nicht überein.")
# On Windows rename refuses to overwrite another published release.
remote_temporary.rename(destination)
except OSError as exc:
raise OSError(
"Deployment fehlgeschlagen. VPN-Verbindung und Schreibrechte prüfen. "
"Der erfolgreiche lokale Build bleibt erhalten; erneut mit deploy.bat starten."
) from exc
finally:
if remote_temporary is not None:
try:
remote_temporary.unlink(missing_ok=True)
except OSError:
pass # The VPN may have disconnected during transfer.
print(f"[OK] Deployment abgeschlossen: {destination}")
return destination
def main() -> int:
parser = argparse.ArgumentParser(description="Letzten erfolgreichen Build auf dem SMB-Laufwerk bereitstellen")
parser.parse_args()
try:
deploy()
except (OSError, ValueError, KeyError) as exc:
print(f"[FEHLER] {exc}")
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())
+13
View File
@@ -12,12 +12,17 @@ Dieses Skript:
"""
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)."""
@@ -44,6 +49,14 @@ def _ensure_dirs(base_dir: Path) -> None:
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