"""Checkout maintenance; standard library only, so broken dependencies cannot block rollback.""" from __future__ import annotations import argparse from contextlib import contextmanager from datetime import datetime, timezone import fcntl import hashlib import json import logging import os from pathlib import Path import re import shutil import subprocess import sys import tempfile import uuid from .setup import get_latest_mustang_version, install_mustang def point(link: Path, target: Path) -> None: temporary = link.with_name(link.name + '.next') temporary.unlink(missing_ok=True) temporary.symlink_to(target) os.replace(temporary, link) @contextmanager def locked(state: Path): state.mkdir(parents=True, exist_ok=True, mode=0o700) with (state / 'update.lock').open('a') as handle: try: fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError as exc: raise RuntimeError('Ein anderer Update-/Rollback-Lauf ist aktiv.') from exc yield class Maintenance: def __init__(self, root: Path, log): self.root = root.resolve() self.state = self.root / '.runtime' self.log = log def run(self, args, *, cwd=None, env=None, timeout=1200): self.log.info('Aufruf: %s', ' '.join(map(str, args))) result = subprocess.run(list(map(str, args)), cwd=cwd or self.root, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, timeout=timeout) self.log.info('%s', result.stdout.rstrip()) if result.returncode: raise RuntimeError(f'Befehl fehlgeschlagen (Exit {result.returncode}): {args[0]}') return result.stdout def health(self, release: Path): python = release / 'venv/bin/python' env = os.environ.copy() for key in ('PYTHONPATH', 'JR_EINVOICE_PRESETS', 'JR_EINVOICE_HOME'): env.pop(key, None) env['JR_EINVOICE_MUSTANG_JAR'] = str(release / 'Mustang-CLI.jar') env['JR_EINVOICE_INTEGRATION'] = '1' self.log.info('PRÜFUNG: pip check, WeasyPrint, CLI und echte Rechnungs-/Negativtests') self.run([python, '-I', '-m', 'pip', 'check'], env=env) with tempfile.TemporaryDirectory(prefix='einvoice-health-') as folder: env['JR_EINVOICE_HOME'] = folder self.run([python, '-I', '-c', 'from weasyprint import HTML; HTML(string="

PDF-Test

").write_pdf()'], cwd=folder, env=env) for module in ('jr_einvoice.quick_cli', 'jr_einvoice.cli'): code = ('from jr_einvoice.cli import app; app()' if module.endswith('.cli') else 'from jr_einvoice.quick_cli import main; main()') self.run([python, '-I', '-c', code, '--help'], cwd=folder, env=env) self.run([python, '-I', '-m', 'unittest', 'discover', '-s', self.root / 'tests', '-v'], cwd=folder, env=env) def baseline(self): baseline = self.state / 'releases' / ('baseline-' + uuid.uuid4().hex[:8]) baseline.mkdir(parents=True) venv = self.root / '.venv' jar = Path(os.environ.get('JR_EINVOICE_MUSTANG_JAR', self.root / 'tools/mustang/Mustang-CLI.jar')).resolve() if not (venv / 'bin/python').exists() or not jar.is_file(): raise RuntimeError('Erstinstallation nötig: .venv und Mustang fehlen; siehe README.') (baseline / 'venv').symlink_to(venv) shutil.copy2(jar, baseline / 'Mustang-CLI.jar') return baseline def activate(self, target: Path): active = self.state / 'current' old = active.resolve() if active.is_symlink() else self.baseline() # Persist rollback target before changing the single runtime pointer. point(self.state / 'previous', old) self.log.info('AKTIVIERUNG: %s', target) point(active, target) try: self.health(active.resolve()) except BaseException: point(active, old) self.log.error('ROLLBACK: bisherige Laufzeit wieder aktiv: %s', old) raise self.log.info('ERFOLG: Laufzeit aktiv. Rollback verfügbar: %s', old) def update(self, apply=False, version=None): version = version or get_latest_mustang_version(self.log) if not re.fullmatch(r'\d+(?:\.\d+)+', version): raise ValueError(f'Keine stabile Mustang-Version: {version!r}; --mustang-version verwenden.') release = self.state / 'releases' / (datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ-') + uuid.uuid4().hex[:8]) release.mkdir(parents=True) self.log.info('KANDIDAT: %s; Mustang %s', release, version) self.run(['java', '-version']) self.run([sys.executable, '-m', 'venv', release / 'venv']) python = release / 'venv/bin/python' self.run([python, '-m', 'pip', 'install', '--upgrade', 'pip']) self.run([python, '-m', 'pip', 'install', '--upgrade', '--upgrade-strategy', 'eager', '--report', release / 'install-report.json', f'{self.root}[pdf]']) install_mustang(version, release / 'Mustang-CLI.jar', self.log) frozen = self.run([python, '-m', 'pip', 'freeze', '--all']) (release / 'requirements.freeze.txt').write_text(frozen) packages = json.loads(self.run([python, '-m', 'pip', 'list', '--format=json'])) old_python = (self.state / 'current/venv/bin/python') if (self.state / 'current').exists() else self.root / '.venv/bin/python' old = {p['name'].lower(): p['version'] for p in json.loads(self.run([old_python, '-m', 'pip', 'list', '--format=json']))} for package in packages: before = old.get(package['name'].lower(), '(neu)') if before != package['version']: self.log.info('VERSION: %s: %s -> %s', package['name'], before, package['version']) manifest = {'mustang': version, 'jar_sha256': hashlib.sha256((release / 'Mustang-CLI.jar').read_bytes()).hexdigest(), 'packages': packages, 'source_commit': self.run(['git', 'rev-parse', 'HEAD']).strip()} (release / 'manifest.json').write_text(json.dumps(manifest, indent=2) + '\n') self.health(release) if apply: self.activate(release) else: self.log.info('PRÜFUNG ERFOLGREICH: keine Umschaltung. Zum Übernehmen erneut mit --apply starten.') def rollback(self): previous = self.state / 'previous' if not previous.exists(): raise RuntimeError('Kein vorheriger Laufzeitstand vorhanden.') target = previous.resolve() self.health(target) self.activate(target) def main(argv=None): parser = argparse.ArgumentParser(description='Python-Abhängigkeiten und Mustang isoliert aktualisieren und testen.') mode = parser.add_mutually_exclusive_group() mode.add_argument('--apply', action='store_true', help='Nach erfolgreichen Tests aktivieren') mode.add_argument('--rollback', action='store_true', help='Vorherigen Stand prüfen und aktivieren') mode.add_argument('--status', action='store_true', help='Aktiven und vorherigen Stand anzeigen, ohne Netzwerk') parser.add_argument('--mustang-version', help='Explizite stabile Version statt Maven-Release') args = parser.parse_args(argv) root = Path(__file__).resolve().parent.parent state = root / '.runtime' try: with locked(state): logs = state / 'logs' logs.mkdir(exist_ok=True) logfile = logs / (datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ-') + uuid.uuid4().hex[:8] + '.log') logger = logging.getLogger('maintenance') logger.setLevel(logging.INFO) logger.handlers.clear() for handler in (logging.StreamHandler(sys.stdout), logging.FileHandler(logfile)): handler.setFormatter(logging.Formatter('%(asctime)s %(levelname)s %(message)s')) logger.addHandler(handler) logger.info('PROTOKOLL: %s', logfile) maintenance = Maintenance(root, logger) if args.status: for name in ('current', 'previous'): link = state / name logger.info('%s: %s', name, link.resolve() if link.exists() else '(noch nicht verwaltet)') elif args.rollback: maintenance.rollback() else: maintenance.update(args.apply, args.mustang_version) except (Exception, KeyboardInterrupt) as exc: logging.getLogger('maintenance').error('FEHLER: %s; nicht aktivierte Kandidaten bleiben zur Diagnose erhalten.', exc) return 1 return 0