# Copyright 2025–2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""
Local documentation server management for MAFw.
This module provides the :class:`~.ServerStatusEnum`, helper functions, and
orchestration logic for managing a ``python -m http.server`` subprocess used
to serve locally built versioned documentation.
"""
from __future__ import annotations
import json
import subprocess
import sys
from datetime import datetime, timezone
from enum import StrEnum
from pathlib import Path
from typing import Any, List, Tuple, cast
from urllib.error import HTTPError, URLError
from urllib.request import urlopen
import psutil
from mafw.devtools import DevtoolsError
SERVER_DEFAULT_DIRECTORY = Path('docs') / 'build'
"""Default directory served by the local documentation server."""
SERVER_METADATA_FILENAME = '.multiversion-doc-server.json'
"""Filename for server metadata stored inside the served directory."""
class ServerStatusEnum(StrEnum):
"""Possible status values for the multiversion-doc local documentation server.
:cvar running: The server process exists, matches the expected signature and responds over HTTP.
:cvar stale: The server process exists and matches the expected signature, but does not respond over HTTP.
:cvar stopped: The recorded PID does not exist or does not match the expected signature.
:cvar unknown: The metadata file is missing, so no status can be determined.
"""
running = 'running'
stale = 'stale'
stopped = 'stopped'
unknown = 'unknown'
def server_default_directory() -> Path:
"""Return the default directory served by the local documentation server.
:return: Default documentation build directory
:rtype: Path
"""
return SERVER_DEFAULT_DIRECTORY.resolve()
def server_metadata_path(directory: Path) -> Path:
"""Return the metadata path associated to a served directory.
:param directory: Served directory
:type directory: Path
:return: Metadata JSON file path
:rtype: Path
"""
return Path(directory) / SERVER_METADATA_FILENAME
def read_server_metadata(directory: Path) -> dict[str, Any] | None:
"""Read local documentation server metadata if available.
:param directory: Served directory to locate metadata under
:type directory: Path
:return: Parsed metadata, or None when the metadata file is missing
:rtype: dict[str, Any] | None
:raises DevtoolsError: If the metadata file exists but cannot be parsed
"""
meta_path = server_metadata_path(directory)
if not meta_path.exists():
return None
try:
loaded = json.loads(meta_path.read_text(encoding='utf-8'))
return cast(dict[str, Any], loaded)
except (OSError, json.JSONDecodeError) as e:
raise DevtoolsError(f'Invalid server metadata file: {meta_path}') from e
def write_server_metadata(directory: Path, metadata: dict[str, Any]) -> None:
"""Write local documentation server metadata.
:param directory: Served directory where metadata is stored
:type directory: Path
:param metadata: Metadata dictionary to serialize as JSON
:type metadata: dict[str, Any]
:raises DevtoolsError: If the metadata file cannot be written
"""
meta_path = server_metadata_path(directory)
try:
meta_path.write_text(json.dumps(metadata, indent=2, sort_keys=True) + '\n', encoding='utf-8')
except OSError as e:
raise DevtoolsError(f'Cannot write server metadata: {meta_path}') from e
def process_matches_http_server(proc: Any, address: str, port: int, directory: Path) -> bool:
"""Check whether a process resembles the expected http.server invocation.
:param proc: psutil Process instance
:type proc: Any
:param address: Expected bind address
:type address: str
:param port: Expected port
:type port: int
:param directory: Expected served directory
:type directory: Path
:return: True when the process signature matches
:rtype: bool
"""
try:
cmdline: List[str] = list(proc.cmdline())
except Exception:
return False
if '-m' not in cmdline:
return False
m_index = cmdline.index('-m')
if m_index + 1 >= len(cmdline) or cmdline[m_index + 1] != 'http.server':
return False
directory_str = str(Path(directory).resolve())
checks = [
('-b', address),
('-d', directory_str),
]
for flag, expected in checks:
if flag not in cmdline:
return False
i = cmdline.index(flag)
if i + 1 >= len(cmdline) or cmdline[i + 1] != expected:
return False
if str(port) not in cmdline:
return False
return True
def http_probe(address: str, port: int, timeout_s: float = 1.0) -> bool:
"""Probe whether the server responds at the given address:port.
:param address: Server address
:type address: str
:param port: Server port
:type port: int
:param timeout_s: Timeout in seconds
:type timeout_s: float
:return: True if a GET to the server root returns HTTP 200
:rtype: bool
"""
url = f'http://{address}:{port}/'
try:
with urlopen(url, timeout=timeout_s) as resp:
return getattr(resp, 'status', None) == 200
except (HTTPError, URLError, OSError):
return False
def format_started_at_utc(now: datetime) -> str:
"""Format a timestamp for server metadata.
:param now: A datetime instance (timezone-aware is preferred)
:type now: datetime
:return: UTC ISO-8601 timestamp with ``Z`` suffix
:rtype: str
"""
utc = now.astimezone(timezone.utc).replace(microsecond=0)
return utc.isoformat().replace('+00:00', 'Z')
[docs]
def get_server_status(directory: Path) -> Tuple[ServerStatusEnum, dict[str, Any] | None]:
"""Determine the current server status from metadata and process/HTTP checks."""
metadata = read_server_metadata(directory)
if metadata is None:
return ServerStatusEnum.unknown, None
pid = metadata.get('pid')
address = metadata.get('address')
port = metadata.get('port')
dir_value = metadata.get('directory')
if (
not isinstance(pid, int)
or not isinstance(address, str)
or not isinstance(port, int)
or not isinstance(dir_value, str)
):
raise DevtoolsError(f'Invalid server metadata contents: {server_metadata_path(directory)}')
if not psutil.pid_exists(pid):
return ServerStatusEnum.stopped, metadata
try:
proc = psutil.Process(pid)
except Exception:
return ServerStatusEnum.stopped, metadata
if not process_matches_http_server(proc, address=address, port=port, directory=Path(dir_value)):
return ServerStatusEnum.stopped, metadata
if http_probe(address=address, port=port):
return ServerStatusEnum.running, metadata
return ServerStatusEnum.stale, metadata
[docs]
def server_start_impl(address: str, port: int, directory: Path, log_file: Path, force_restart: bool) -> None:
"""Implementation for starting a local documentation server."""
directory = Path(directory).resolve()
log_file = Path(log_file).resolve()
if not directory.exists():
raise DevtoolsError(f'Served directory does not exist: {directory}')
status, metadata = get_server_status(directory)
if status in (ServerStatusEnum.running, ServerStatusEnum.stale):
assert metadata is not None
url = f'http://{metadata["address"]}:{metadata["port"]}'
print(f'ℹ️ Documentation server already present ({status}) at {url}')
if not force_restart:
return
server_stop_by_metadata(directory, require_metadata=False)
log_file.parent.mkdir(parents=True, exist_ok=True)
with open(log_file, 'a', encoding='utf-8') as log_handle:
process = subprocess.Popen(
[
sys.executable,
'-m',
'http.server',
'-b',
address,
'-d',
str(directory),
str(port),
],
stdout=log_handle,
stderr=subprocess.STDOUT,
start_new_session=True,
)
metadata_to_write = {
'pid': int(process.pid),
'address': address,
'port': int(port),
'directory': str(directory),
'log_file': str(log_file),
'started_at': format_started_at_utc(datetime.now(timezone.utc)),
}
write_server_metadata(directory, metadata_to_write)
print(f'✅ Documentation server launched and responding at http://{address}:{port}')