184 lines
6.3 KiB
Python
184 lines
6.3 KiB
Python
"""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}")
|