"""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)