"""DOSBox processing helpers for quote bridge. This module provides a lightweight wrapper around DOSBox invocation so the listener can process quote files using a DOS-side script. """ from __future__ import annotations import json import os import subprocess from pathlib import Path from typing import Any, Callable, Dict, List Logger = Callable[[str], None] def _default_path(*parts: str) -> str: base_dir = Path(__file__).resolve().parent return str(base_dir.joinpath(*parts)) def _env_bool(name: str, default: bool = False) -> bool: value = os.environ.get(name) if value is None: return default return value.strip().lower() in {"1", "true", "yes", "on"} def _split_commands(raw: str) -> List[str]: return [cmd.strip() for cmd in raw.split(";") if cmd.strip()] def _load_quote_payload(quote_file: Path) -> Dict[str, Any]: with open(quote_file, 'r', encoding='utf-8') as f: return json.load(f) def _profile_dir() -> Path: return Path(os.environ.get("QUOTE_PROFILE_DIR", _default_path("profiles"))) def _profile_candidates(location: str) -> List[Path]: profile_dir = _profile_dir() candidates: List[Path] = [] if location: candidates.append(profile_dir / f"{location.upper()}.json") candidates.append(profile_dir / f"{location.lower()}.json") candidates.append(profile_dir / "default.json") return candidates def _load_profile_for_quote(quote_file: Path, logger: Logger) -> Dict[str, Any]: payload = _load_quote_payload(quote_file) location = str(payload.get('createdBy', {}).get('location', '') or '').strip() for candidate in _profile_candidates(location): if candidate.exists(): with open(candidate, 'r', encoding='utf-8') as f: profile = json.load(f) profile['_profileFile'] = str(candidate) profile['_quoteLocation'] = location logger(f"Using DOSBox profile: {candidate}") return profile logger("No DOSBox profile file found; using environment defaults") return { '_profileFile': None, '_quoteLocation': location, } def _profile_bool(profile: Dict[str, Any], key: str, env_name: str, default: bool = False) -> bool: value = profile.get(key) if value is None: return _env_bool(env_name, default=default) if isinstance(value, bool): return value return str(value).strip().lower() in {"1", "true", "yes", "on"} def _profile_string(profile: Dict[str, Any], key: str, env_name: str, default: str = "") -> str: value = profile.get(key) if value is None or value == "": return os.environ.get(env_name, default) return str(value) def _profile_float(profile: Dict[str, Any], key: str, env_name: str, default: float) -> float: value = profile.get(key) if value is None or value == "": return float(os.environ.get(env_name, str(default))) return float(value) def _profile_commands(profile: Dict[str, Any], key: str, env_name: str) -> List[str]: value = profile.get(key) if isinstance(value, list): return [str(item).strip() for item in value if str(item).strip()] if isinstance(value, str) and value.strip(): return _split_commands(value) return _split_commands(os.environ.get(env_name, "")) def _quote_path_for_dos(path: Path, mount_path: Path, drive: str) -> str: drive_letter = (drive or "C").strip().upper()[:1] try: rel = path.resolve().relative_to(mount_path.resolve()) rel_str = str(rel).replace("/", "\\") return f"{drive_letter}:\\{rel_str}" except Exception: # Fallback to absolute path transformed to backslashes. return str(path.resolve()).replace("/", "\\") def run_dosbox_for_quote(quote_file: Path, logger: Logger) -> None: """Run DOSBox and execute bridge command for one quote file. Required environment values are provided with sensible defaults, but you'll typically set these in deployment: - DOSBOX_BIN - DOSBOX_MOUNT_PATH - DOSBOX_BRIDGE_COMMAND """ profile = _load_profile_for_quote(quote_file, logger) dosbox_bin = _profile_string(profile, 'dosboxBin', 'DOSBOX_BIN', 'dosbox') mount_path = Path(_profile_string(profile, 'mountPath', 'DOSBOX_MOUNT_PATH', str(quote_file.parent.parent))) mount_drive = _profile_string(profile, 'mountDrive', 'DOSBOX_MOUNT_DRIVE', 'C') working_directory = _profile_string(profile, 'workingDirectory', 'DOSBOX_WORKING_DIRECTORY', '') bridge_command = _profile_string(profile, 'bridgeCommand', 'DOSBOX_BRIDGE_COMMAND', 'BRIDGE.BAT') dosbox_conf = _profile_string(profile, 'dosboxConf', 'DOSBOX_CONF', '') or None dosbox_timeout = _profile_float(profile, 'timeoutSeconds', 'DOSBOX_TIMEOUT_SECONDS', 120.0) dosbox_no_console = _profile_bool(profile, 'noConsole', 'DOSBOX_NOCONSOLE', default=False) dosbox_extra_commands = _profile_commands(profile, 'extraCommands', 'DOSBOX_EXTRA_COMMANDS') quote_path_for_dos = _quote_path_for_dos(quote_file, mount_path, mount_drive) # Build DOS command script. dos_commands = [ f"mount {mount_drive} \"{mount_path}\"", f"{mount_drive}", ] if working_directory: dos_commands.append(f"cd {working_directory}") dos_commands.extend(dosbox_extra_commands) dos_commands.append(f"{bridge_command} \"{quote_path_for_dos}\"") dos_commands.append("exit") cmd = [dosbox_bin] if dosbox_conf: cmd.extend(["-conf", dosbox_conf]) if dosbox_no_console: cmd.append("-noconsole") for line in dos_commands: cmd.extend(["-c", line]) logger(f"DOSBox command: {' '.join(cmd)}") try: result = subprocess.run( cmd, check=False, text=True, capture_output=True, timeout=dosbox_timeout, ) except FileNotFoundError as exc: raise RuntimeError(f"DOSBox executable not found: {dosbox_bin}") from exc except subprocess.TimeoutExpired as exc: raise RuntimeError( f"DOSBox timed out after {dosbox_timeout} seconds" ) from exc if result.stdout: logger(f"DOSBox stdout:\n{result.stdout.strip()}") if result.stderr: logger(f"DOSBox stderr:\n{result.stderr.strip()}") if result.returncode != 0: raise RuntimeError(f"DOSBox exited with code {result.returncode}")