rgonsal/comunidadhll
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

sistema A2S

Browse Source
This commit is contained in:
devRaGonSa
2026-03-20 14:38:57 +01:00
parent a82a90a1b4
commit 87c1f4e8c3
75 changed files with 7625 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
backend/app/__init__.py
View File

@@ -1 +1,59 @@
"""Base package for the future HLL Vietnam Python backend."""
"""Minimal bootstrap package for the HLL Vietnam Python backend."""
from .config import get_allowed_origins, get_bind_address
from .main import create_server, run
from .normalizers import normalize_a2s_server_info, normalize_server_record
from .payloads import build_health_payload
from .routes import resolve_get_payload
from .snapshots import build_server_snapshot, build_snapshot_batch, utc_now
from .storage import initialize_storage, persist_snapshot_batch
def collect_server_snapshots(*args: object, **kwargs: object) -> dict[str, object]:
"""Proxy collector access without importing the module during package init."""
from .collector import collect_server_snapshots as _collect_server_snapshots
return _collect_server_snapshots(*args, **kwargs)
def fetch_a2s_probe(*args: object, **kwargs: object) -> dict[str, object]:
"""Proxy A2S probe access without importing the collector during package init."""
from .collector import fetch_a2s_probe as _fetch_a2s_probe
return _fetch_a2s_probe(*args, **kwargs)
def query_server_info(*args: object, **kwargs: object) -> object:
"""Proxy A2S info queries without importing the module during package init."""
from .a2s_client import query_server_info as _query_server_info
return _query_server_info(*args, **kwargs)
def fetch_controlled_server_source() -> tuple[dict[str, object], ...]:
"""Proxy the controlled source without importing the module during package init."""
from .collector import (
fetch_controlled_server_source as _fetch_controlled_server_source,
)
return tuple(_fetch_controlled_server_source())
__all__ = [
"build_health_payload",
"build_server_snapshot",
"build_snapshot_batch",
"collect_server_snapshots",
"create_server",
"fetch_a2s_probe",
"fetch_controlled_server_source",
"get_allowed_origins",
"get_bind_address",
"initialize_storage",
"normalize_a2s_server_info",
"normalize_server_record",
"persist_snapshot_batch",
"query_server_info",
"resolve_get_payload",
"run",
"utc_now",
]

176
backend/app/a2s_client.py Normal file
View File

@@ -0,0 +1,176 @@
"""Minimal Steam A2S info client for development-time HLL server probes."""
from __future__ import annotations
import argparse
import json
import socket
import struct
from dataclasses import asdict, dataclass
DEFAULT_A2S_TIMEOUT = 6.0
_A2S_PREFIX = b"\xFF\xFF\xFF\xFF"
_A2S_INFO_REQUEST = _A2S_PREFIX + b"\x54Source Engine Query\x00"
_A2S_CHALLENGE_RESPONSE = 0x41
_A2S_INFO_RESPONSE = 0x49
class A2SError(RuntimeError):
"""Base error for A2S query failures."""
class A2STimeoutError(A2SError):
"""Raised when an A2S query does not complete before the timeout."""
class A2SProtocolError(A2SError):
"""Raised when an A2S server returns an unexpected payload."""
@dataclass(frozen=True, slots=True)
class A2SServerInfo:
"""Minimal metadata returned by an A2S info query."""
host: str
query_port: int
server_name: str
map_name: str | None
players: int
max_players: int
protocol: int
folder: str | None = None
game: str | None = None
version: str | None = None
def query_server_info(
host: str,
query_port: int,
*,
timeout: float = DEFAULT_A2S_TIMEOUT,
) -> A2SServerInfo:
"""Query one server using A2S_INFO and return minimal reusable metadata."""
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as udp_socket:
udp_socket.settimeout(timeout)
address = (host, query_port)
try:
udp_socket.sendto(_A2S_INFO_REQUEST, address)
payload = _receive_packet(udp_socket)
if _is_challenge_packet(payload):
challenge = payload[5:9]
udp_socket.sendto(_A2S_INFO_REQUEST + challenge, address)
payload = _receive_packet(udp_socket)
except socket.timeout as error:
raise A2STimeoutError(
f"A2S query to {host}:{query_port} timed out after {timeout:.1f}s."
) from error
except OSError as error:
raise A2SError(
f"A2S query to {host}:{query_port} failed: {error}."
) from error
return _parse_info_payload(payload, host=host, query_port=query_port)
def main() -> None:
"""Allow a direct development-time probe of one A2S target."""
parser = argparse.ArgumentParser(description="Probe one server with A2S_INFO.")
parser.add_argument("host", help="Server hostname or IPv4 address.")
parser.add_argument("query_port", type=int, help="Server Steam query port.")
parser.add_argument(
"--timeout",
type=float,
default=DEFAULT_A2S_TIMEOUT,
help="Socket timeout in seconds.",
)
args = parser.parse_args()
payload = asdict(
query_server_info(args.host, args.query_port, timeout=args.timeout)
)
print(json.dumps(payload, indent=2))
def _receive_packet(udp_socket: socket.socket) -> bytes:
payload, _ = udp_socket.recvfrom(4096)
return payload
def _is_challenge_packet(payload: bytes) -> bool:
return (
len(payload) >= 9
and payload.startswith(_A2S_PREFIX)
and payload[4] == _A2S_CHALLENGE_RESPONSE
)
def _parse_info_payload(
payload: bytes,
*,
host: str,
query_port: int,
) -> A2SServerInfo:
if len(payload) < 6 or not payload.startswith(_A2S_PREFIX):
raise A2SProtocolError("A2S response did not include the expected packet header.")
if payload[4] != _A2S_INFO_RESPONSE:
raise A2SProtocolError(
f"A2S response type {payload[4]!r} is not an info response."
)
protocol = payload[5]
offset = 6
server_name, offset = _read_c_string(payload, offset)
map_name, offset = _read_c_string(payload, offset)
folder, offset = _read_c_string(payload, offset)
game, offset = _read_c_string(payload, offset)
offset += 2 # app id
players = _read_byte(payload, offset)
max_players = _read_byte(payload, offset + 1)
offset += 6 # players, max, bots, server type, environment, visibility
offset += 1 # vac
version, offset = _read_c_string(payload, offset)
if offset < len(payload):
extra_data_flag = payload[offset]
offset += 1
if extra_data_flag & 0x80:
offset += 2
if extra_data_flag & 0x10:
_, offset = _read_c_string(payload, offset)
if extra_data_flag & 0x40:
offset += 2
offset += 8
if extra_data_flag & 0x20:
offset += 8
return A2SServerInfo(
host=host,
query_port=query_port,
server_name=server_name or "Unknown server",
map_name=map_name or None,
players=players,
max_players=max_players,
protocol=protocol,
folder=folder or None,
game=game or None,
version=version or None,
)
def _read_c_string(payload: bytes, offset: int) -> tuple[str, int]:
end = payload.find(b"\x00", offset)
if end == -1:
raise A2SProtocolError("A2S response ended before a null-terminated string.")
return payload[offset:end].decode("utf-8", errors="replace"), end + 1
def _read_byte(payload: bytes, offset: int) -> int:
if offset >= len(payload):
raise A2SProtocolError("A2S response ended before expected integer fields.")
return struct.unpack_from("<B", payload, offset)[0]
if __name__ == "__main__":
main()

269
backend/app/collector.py Normal file
View File

@@ -0,0 +1,269 @@
"""Minimal collector bootstrap for provisional server snapshots."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
from typing import Callable, Mapping, Sequence
from .a2s_client import DEFAULT_A2S_TIMEOUT, query_server_info
from .normalizers import normalize_a2s_server_info, normalize_server_record
from .server_targets import A2SServerTarget, load_a2s_targets
from .snapshots import build_snapshot_batch, utc_now
from .storage import persist_snapshot_batch
RawSourceFetcher = Callable[[], Sequence[Mapping[str, object]]]
TargetProbe = Callable[[A2SServerTarget, float], Mapping[str, object]]
CONTROLLED_RAW_SERVER_SOURCE: tuple[dict[str, object], ...] = (
{
"external_server_id": "hll-esp-tactical-rotation",
"server_name": "HLL ESP Tactical Rotation",
"status": "online",
"players": 74,
"max_players": 100,
"current_map": "Sainte-Marie-du-Mont",
"region": "EU",
},
{
"external_server_id": "hll-latam-night-offensive",
"server_name": "HLL LATAM Night Offensive",
"status": "online",
"players": 51,
"max_players": 100,
"current_map": "Carentan",
"region": "LATAM",
},
{
"external_server_id": "hll-community-reserve",
"server_name": "HLL Community Reserve",
"status": "offline",
"players": 0,
"max_players": 100,
"current_map": None,
"region": "EU",
},
)
def fetch_controlled_server_source() -> Sequence[Mapping[str, object]]:
"""Return the controlled development source used by the collector bootstrap."""
return CONTROLLED_RAW_SERVER_SOURCE
def fetch_a2s_probe(
host: str,
query_port: int,
*,
timeout: float = DEFAULT_A2S_TIMEOUT,
source_name: str = "a2s-info",
external_server_id: str | None = None,
region: str | None = None,
) -> dict[str, object]:
"""Probe one A2S target and normalize its metadata for the collector model."""
server_info = query_server_info(host, query_port, timeout=timeout)
return normalize_a2s_server_info(
server_info,
source_name=source_name,
external_server_id=external_server_id,
region=region,
)
def fetch_configured_a2s_probes(
*,
timeout: float = DEFAULT_A2S_TIMEOUT,
probe_target: TargetProbe | None = None,
) -> tuple[dict[str, object], ...]:
"""Probe the configured A2S targets without hardcoding them in collector logic."""
probe = probe_target or _probe_configured_target
return tuple(
dict(probe(target, timeout))
for target in load_a2s_targets()
)
def collect_server_snapshots(
*,
fetch_raw_source: RawSourceFetcher = fetch_controlled_server_source,
source_name: str = "controlled-placeholder",
source_mode: str = "controlled",
timeout: float = DEFAULT_A2S_TIMEOUT,
allow_controlled_fallback: bool = True,
probe_target: TargetProbe | None = None,
persist: bool = False,
db_path: Path | None = None,
) -> dict[str, object]:
"""Collect snapshot batches from controlled data, A2S, or auto mode."""
normalized_records, collection_details = _collect_normalized_records(
fetch_raw_source=fetch_raw_source,
source_name=source_name,
source_mode=source_mode,
timeout=timeout,
allow_controlled_fallback=allow_controlled_fallback,
probe_target=probe_target,
)
captured_at = utc_now()
payload = {
"source_name": collection_details["source_name"],
"collection_mode": collection_details["collection_mode"],
"fallback_used": collection_details["fallback_used"],
"target_count": collection_details["target_count"],
"success_count": collection_details["success_count"],
"errors": collection_details["errors"],
"captured_at": captured_at.isoformat().replace("+00:00", "Z"),
"snapshots": build_snapshot_batch(
normalized_records,
captured_at=captured_at,
),
}
if persist:
payload["storage"] = persist_snapshot_batch(
payload["snapshots"],
source_name=payload["source_name"],
captured_at=payload["captured_at"],
db_path=db_path,
)
return payload
def main() -> None:
"""Allow manual collector execution during development."""
parser = argparse.ArgumentParser(description="Collect development server snapshots.")
parser.add_argument(
"--source",
choices=("controlled", "a2s", "auto"),
default="auto",
help="Choose controlled data, configured A2S targets, or auto with fallback.",
)
parser.add_argument(
"--timeout",
type=float,
default=DEFAULT_A2S_TIMEOUT,
help="Socket timeout in seconds for A2S probes.",
)
parser.add_argument(
"--no-fallback",
action="store_true",
help="Disable fallback to controlled data when A2S fails.",
)
args = parser.parse_args()
payload = collect_server_snapshots(
source_mode=args.source,
timeout=args.timeout,
allow_controlled_fallback=not args.no_fallback,
persist=True,
)
print(json.dumps(payload, indent=2))
def _collect_normalized_records(
*,
fetch_raw_source: RawSourceFetcher,
source_name: str,
source_mode: str,
timeout: float,
allow_controlled_fallback: bool,
probe_target: TargetProbe | None,
) -> tuple[list[dict[str, object]], dict[str, object]]:
if source_mode == "controlled":
raw_records = fetch_raw_source()
return (
[
normalize_server_record(record, source_name=source_name)
for record in raw_records
],
{
"source_name": source_name,
"collection_mode": "controlled",
"fallback_used": False,
"target_count": 0,
"success_count": 0,
"errors": [],
},
)
configured_targets = load_a2s_targets()
records: list[dict[str, object]] = []
errors: list[dict[str, object]] = []
probe = probe_target or _probe_configured_target
for target in configured_targets:
try:
records.append(dict(probe(target, timeout)))
except Exception as error: # noqa: BLE001 - keep collector failures controlled
errors.append(
{
"target": target.name,
"host": target.host,
"query_port": target.query_port,
"message": str(error),
}
)
if records:
return (
records,
{
"source_name": "a2s-info",
"collection_mode": "a2s",
"fallback_used": False,
"target_count": len(configured_targets),
"success_count": len(records),
"errors": errors,
},
)
if source_mode == "a2s" or not allow_controlled_fallback:
return (
[],
{
"source_name": "a2s-info",
"collection_mode": "a2s",
"fallback_used": False,
"target_count": len(configured_targets),
"success_count": 0,
"errors": errors,
},
)
raw_records = fetch_raw_source()
normalized_records = [
normalize_server_record(record, source_name=source_name)
for record in raw_records
]
return (
normalized_records,
{
"source_name": source_name,
"collection_mode": "controlled-fallback",
"fallback_used": True,
"target_count": len(configured_targets),
"success_count": 0,
"errors": errors,
},
)
def _probe_configured_target(
target: A2SServerTarget,
timeout: float,
) -> dict[str, object]:
return fetch_a2s_probe(
target.host,
target.query_port,
timeout=timeout,
source_name=target.source_name,
external_server_id=target.external_server_id,
region=target.region,
)
if __name__ == "__main__":
main()

77
backend/app/config.py Normal file
View File

@@ -0,0 +1,77 @@
"""Local development configuration for the HLL Vietnam backend bootstrap."""
from __future__ import annotations
import os
from pathlib import Path
DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 8000
DEFAULT_STORAGE_FILENAME = "hll_vietnam_dev.sqlite3"
DEFAULT_REFRESH_INTERVAL_SECONDS = 300
DEFAULT_ALLOWED_ORIGINS = (
"null",
"http://127.0.0.1:5500",
"http://127.0.0.1:8080",
"http://localhost:5500",
"http://localhost:8080",
)
DEFAULT_A2S_TARGETS_ENV_VAR = "HLL_BACKEND_A2S_TARGETS"
DEFAULT_A2S_SOURCE_NAME = "community-hispana-a2s"
def get_bind_address() -> tuple[str, int]:
"""Return the host and port used by the local backend bootstrap."""
host = os.getenv("HLL_BACKEND_HOST", DEFAULT_HOST)
port = int(os.getenv("HLL_BACKEND_PORT", str(DEFAULT_PORT)))
return host, port
def get_allowed_origins() -> tuple[str, ...]:
"""Return the small allowlist used for local frontend development."""
raw_origins = os.getenv(
"HLL_BACKEND_ALLOWED_ORIGINS",
",".join(DEFAULT_ALLOWED_ORIGINS),
)
origins = []
for origin in raw_origins.split(","):
normalized_origin = _normalize_origin(origin)
if normalized_origin:
origins.append(normalized_origin)
return tuple(origins) or DEFAULT_ALLOWED_ORIGINS
def _normalize_origin(origin: str) -> str:
"""Normalize configured origins so env overrides match browser Origin values."""
return origin.strip().rstrip("/")
def get_storage_path() -> Path:
"""Return the local SQLite path used for development snapshot persistence."""
default_path = Path(__file__).resolve().parent.parent / "data" / DEFAULT_STORAGE_FILENAME
configured_path = os.getenv("HLL_BACKEND_STORAGE_PATH")
return Path(configured_path) if configured_path else default_path
def get_refresh_interval_seconds() -> int:
"""Return the default interval used by the local refresh loop."""
configured_value = os.getenv(
"HLL_BACKEND_REFRESH_INTERVAL_SECONDS",
str(DEFAULT_REFRESH_INTERVAL_SECONDS),
)
interval_seconds = int(configured_value)
if interval_seconds <= 0:
raise ValueError("HLL_BACKEND_REFRESH_INTERVAL_SECONDS must be positive.")
return interval_seconds
def get_a2s_targets_payload() -> str | None:
"""Return the optional JSON payload that overrides local A2S targets."""
raw_payload = os.getenv(DEFAULT_A2S_TARGETS_ENV_VAR)
if raw_payload is None:
return None
normalized = raw_payload.strip()
return normalized or None

73
backend/app/main.py Normal file
View File

@@ -0,0 +1,73 @@
"""Minimal HTTP entrypoint for the HLL Vietnam backend bootstrap."""
from __future__ import annotations
import json
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from .config import get_allowed_origins, get_bind_address
from .routes import resolve_get_payload
class HealthHandler(BaseHTTPRequestHandler):
"""Serve the minimal routes required for the backend bootstrap."""
server_version = "HLLVietnamBackend/0.1"
def do_OPTIONS(self) -> None: # noqa: N802 - BaseHTTPRequestHandler interface
self.send_response(HTTPStatus.NO_CONTENT)
self._send_default_headers()
self.send_header("Access-Control-Allow-Methods", "GET, OPTIONS")
self.send_header("Access-Control-Allow-Headers", "Content-Type")
self.end_headers()
def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler interface
status, payload = resolve_get_payload(self.path)
if status is None:
self._write_json(
HTTPStatus.NOT_FOUND,
{"status": "error", "message": "Route not found"},
)
return
self._write_json(status, payload)
def log_message(self, format: str, *args: object) -> None:
# Keep local startup output clean unless future tasks need request logging.
return
def _write_json(self, status: HTTPStatus, payload: dict[str, object]) -> None:
body = json.dumps(payload).encode("utf-8")
self.send_response(status)
self._send_default_headers(content_length=len(body))
self.end_headers()
self.wfile.write(body)
def _send_default_headers(self, content_length: int | None = None) -> None:
origin = self.headers.get("Origin")
if origin in get_allowed_origins():
self.send_header("Access-Control-Allow-Origin", origin)
self.send_header("Vary", "Origin")
self.send_header("Content-Type", "application/json; charset=utf-8")
if content_length is not None:
self.send_header("Content-Length", str(content_length))
def create_server() -> ThreadingHTTPServer:
"""Build the HTTP server using the package-supported handler and bind settings."""
host, port = get_bind_address()
return ThreadingHTTPServer((host, port), HealthHandler)
def run() -> None:
"""Start the local bootstrap server."""
host, port = get_bind_address()
server = create_server()
print(f"HLL Vietnam backend bootstrap listening on http://{host}:{port}")
server.serve_forever()
if __name__ == "__main__":
run()

89
backend/app/normalizers.py Normal file
View File

@@ -0,0 +1,89 @@
"""Normalization helpers for provisional server collection flows."""
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import Mapping
if TYPE_CHECKING:
from .a2s_client import A2SServerInfo
def normalize_server_record(
raw_record: Mapping[str, object],
*,
source_name: str,
) -> dict[str, object]:
"""Normalize a raw server record into the collector's internal shape."""
external_server_id = _string_or_none(raw_record.get("external_server_id"))
return {
"external_server_id": external_server_id,
"server_name": _string_or_default(raw_record.get("server_name"), "Unknown server"),
"status": _normalize_status(raw_record.get("status")),
"players": _coerce_int(raw_record.get("players")),
"max_players": _coerce_int(raw_record.get("max_players")),
"current_map": _string_or_none(raw_record.get("current_map")),
"region": _string_or_none(raw_record.get("region")),
"source_name": source_name,
"snapshot_origin": "controlled-fallback",
"source_ref": external_server_id or source_name,
}
def normalize_a2s_server_info(
server_info: "A2SServerInfo",
*,
source_name: str,
external_server_id: str | None = None,
region: str | None = None,
) -> dict[str, object]:
"""Normalize a probed A2S payload into the collector's internal shape."""
resolved_external_id = external_server_id or (
f"a2s:{server_info.host}:{server_info.query_port}"
)
return {
"external_server_id": resolved_external_id,
"server_name": server_info.server_name or "Unknown server",
"status": "online",
"players": server_info.players,
"max_players": server_info.max_players,
"current_map": server_info.map_name,
"region": region,
"source_name": source_name,
"snapshot_origin": "real-a2s",
"source_ref": f"a2s://{server_info.host}:{server_info.query_port}",
}
def _normalize_status(value: object) -> str:
if not isinstance(value, str):
return "unknown"
normalized = value.strip().lower()
if normalized in {"online", "offline", "unknown"}:
return normalized
return "unknown"
def _coerce_int(value: object) -> int | None:
if value is None:
return None
try:
return int(value)
except (TypeError, ValueError):
return None
def _string_or_none(value: object) -> str | None:
if not isinstance(value, str):
return None
stripped = value.strip()
return stripped or None
def _string_or_default(value: object, default: str) -> str:
normalized = _string_or_none(value)
return normalized or default

180
backend/app/payloads.py Normal file
View File

@@ -0,0 +1,180 @@
"""Placeholder payload builders for the HLL Vietnam backend."""
from __future__ import annotations
from .server_targets import load_a2s_targets
from .storage import list_latest_snapshots, list_server_history, list_snapshot_history
def build_health_payload() -> dict[str, str]:
"""Return a small status payload without committing to business contracts."""
return {
"status": "ok",
"service": "hll-vietnam-backend",
"phase": "bootstrap",
}
def build_community_payload() -> dict[str, object]:
"""Return placeholder community content aligned with the documented contract."""
return {
"status": "ok",
"data": {
"title": "Comunidad Hispana HLL Vietnam",
"summary": "Punto de encuentro para jugadores, escuadras y comunidad.",
"discord_invite_url": "https://discord.com/invite/PedEqZ2Xsa",
},
}
def build_trailer_payload() -> dict[str, object]:
"""Return placeholder trailer metadata for future frontend consumption."""
return {
"status": "ok",
"data": {
"video_url": "https://www.youtube.com/embed/JzYzYNVWZ_A",
"title": "Trailer HLL Vietnam",
"provider": "youtube",
},
}
def build_discord_payload() -> dict[str, object]:
"""Return public Discord placeholder data without real integration."""
return {
"status": "ok",
"data": {
"invite_url": "https://discord.com/invite/PedEqZ2Xsa",
"label": "Unirse al Discord",
"availability": "manual",
},
}
def build_servers_payload() -> dict[str, object]:
"""Return a controlled placeholder for current Hell Let Loose servers."""
return {
"status": "ok",
"data": {
"title": "Servidores actuales de Hell Let Loose",
"context": "current-hll-reference",
"source": "controlled-placeholder",
"items": [
{
"server_name": "HLL ESP Tactical Rotation",
"status": "online",
"players": 74,
"max_players": 100,
"current_map": "Sainte-Marie-du-Mont",
"region": "EU",
},
{
"server_name": "HLL LATAM Night Offensive",
"status": "online",
"players": 51,
"max_players": 100,
"current_map": "Carentan",
"region": "LATAM",
},
{
"server_name": "HLL Community Reserve",
"status": "offline",
"players": 0,
"max_players": 100,
"current_map": None,
"region": "EU",
},
],
},
}
def build_server_latest_payload() -> dict[str, object]:
"""Return the latest persisted snapshot for each known server."""
items = _enrich_server_items(list_latest_snapshots())
return {
"status": "ok",
"data": {
"title": "Ultimo estado conocido de servidores",
"context": "current-hll-history",
"source": "local-snapshot-storage",
"summary_window_size": 6,
"items": items,
},
}
def build_server_history_payload(*, limit: int = 20) -> dict[str, object]:
"""Return recent persisted snapshots across all known servers."""
items = _enrich_server_items(list_snapshot_history(limit=limit))
return {
"status": "ok",
"data": {
"title": "Historial reciente de servidores",
"context": "current-hll-history",
"source": "local-snapshot-storage",
"limit": limit,
"items": items,
},
}
def build_server_detail_history_payload(
server_id: str,
*,
limit: int = 20,
) -> dict[str, object]:
"""Return recent persisted snapshots for one server."""
items = _enrich_server_items(list_server_history(server_id, limit=limit))
return {
"status": "ok",
"data": {
"title": "Historial por servidor",
"context": "current-hll-history",
"source": "local-snapshot-storage",
"server_id": server_id,
"limit": limit,
"items": items,
},
}
def build_error_payload(message: str) -> dict[str, str]:
"""Return the shared error payload shape used by the backend bootstrap."""
return {
"status": "error",
"message": message,
}
def _enrich_server_items(items: list[dict[str, object]]) -> list[dict[str, object]]:
target_index = {
target.external_server_id: target
for target in load_a2s_targets()
if target.external_server_id
}
enriched_items: list[dict[str, object]] = []
for item in items:
enriched_items.append(_enrich_server_item(item, target_index))
return enriched_items
def _enrich_server_item(
item: dict[str, object],
target_index: dict[str, object],
) -> dict[str, object]:
enriched = dict(item)
external_server_id = enriched.get("external_server_id")
snapshot_origin = enriched.get("snapshot_origin")
target = target_index.get(external_server_id)
if not target or snapshot_origin != "real-a2s":
enriched["host"] = None
enriched["query_port"] = None
enriched["game_port"] = None
return enriched
enriched["host"] = target.host
enriched["query_port"] = target.query_port
enriched["game_port"] = target.game_port
return enriched

70
backend/app/routes.py Normal file
View File

@@ -0,0 +1,70 @@
"""Route resolution helpers for the HLL Vietnam backend bootstrap."""
from __future__ import annotations
from http import HTTPStatus
from urllib.parse import parse_qs, urlparse
from .payloads import (
build_community_payload,
build_discord_payload,
build_error_payload,
build_health_payload,
build_server_detail_history_payload,
build_server_history_payload,
build_server_latest_payload,
build_servers_payload,
build_trailer_payload,
)
GET_ROUTES = {
"/health": build_health_payload,
"/api/community": build_community_payload,
"/api/trailer": build_trailer_payload,
"/api/discord": build_discord_payload,
"/api/servers": build_servers_payload,
}
def resolve_get_payload(path: str) -> tuple[HTTPStatus | None, dict[str, object]]:
"""Resolve the JSON payload for a supported GET route."""
parsed = urlparse(path)
if parsed.path == "/api/servers/latest":
return HTTPStatus.OK, build_server_latest_payload()
if parsed.path == "/api/servers/history":
limit = _parse_limit(parsed.query)
if limit is None:
return HTTPStatus.BAD_REQUEST, build_error_payload("Invalid limit parameter")
return HTTPStatus.OK, build_server_history_payload(limit=limit)
builder = GET_ROUTES.get(parsed.path)
if builder is None:
if parsed.path.startswith("/api/servers/") and parsed.path.endswith("/history"):
server_id = parsed.path.removeprefix("/api/servers/").removesuffix("/history")
server_id = server_id.strip("/")
if not server_id:
return HTTPStatus.BAD_REQUEST, build_error_payload("Server id is required")
limit = _parse_limit(parsed.query)
if limit is None:
return HTTPStatus.BAD_REQUEST, build_error_payload("Invalid limit parameter")
return HTTPStatus.OK, build_server_detail_history_payload(server_id, limit=limit)
return None, {}
return HTTPStatus.OK, builder()
def _parse_limit(query: str) -> int | None:
raw_limit = parse_qs(query).get("limit", ["20"])[0]
try:
limit = int(raw_limit)
except ValueError:
return None
if limit < 1 or limit > 100:
return None
return limit

100
backend/app/scheduler.py Normal file
View File

@@ -0,0 +1,100 @@
"""Local development loop for periodic snapshot refreshes."""
from __future__ import annotations
import argparse
import json
import time
from .a2s_client import DEFAULT_A2S_TIMEOUT
from .collector import collect_server_snapshots
from .config import get_refresh_interval_seconds
def run_local_refresh_loop(
*,
interval_seconds: int,
source_mode: str,
timeout: float,
allow_controlled_fallback: bool,
max_runs: int | None = None,
) -> None:
"""Run the collector periodically until interrupted or the run limit is reached."""
completed_runs = 0
print(
"Starting local snapshot refresh loop "
f"(interval={interval_seconds}s, source={source_mode}, persist=true)."
)
print("Press Ctrl+C to stop.")
try:
while max_runs is None or completed_runs < max_runs:
completed_runs += 1
payload = collect_server_snapshots(
source_mode=source_mode,
timeout=timeout,
allow_controlled_fallback=allow_controlled_fallback,
persist=True,
)
print(json.dumps({"run": completed_runs, **payload}, indent=2))
if max_runs is not None and completed_runs >= max_runs:
break
time.sleep(interval_seconds)
except KeyboardInterrupt:
print("\nLocal snapshot refresh loop stopped by user.")
def main() -> None:
"""Allow local scheduled refresh execution without adding external infrastructure."""
parser = argparse.ArgumentParser(
description="Run periodic local snapshot refreshes for development.",
)
parser.add_argument(
"--interval",
type=int,
default=get_refresh_interval_seconds(),
help="Seconds to wait between persisted refresh runs.",
)
parser.add_argument(
"--source",
choices=("controlled", "a2s", "auto"),
default="auto",
help="Choose controlled data, configured A2S targets, or auto with fallback.",
)
parser.add_argument(
"--timeout",
type=float,
default=DEFAULT_A2S_TIMEOUT,
help="Socket timeout in seconds for A2S probes.",
)
parser.add_argument(
"--no-fallback",
action="store_true",
help="Disable fallback to controlled data when A2S fails.",
)
parser.add_argument(
"--max-runs",
type=int,
default=None,
help="Optional safety limit for the number of refresh cycles to execute.",
)
args = parser.parse_args()
if args.interval <= 0:
raise ValueError("--interval must be a positive integer.")
if args.max_runs is not None and args.max_runs <= 0:
raise ValueError("--max-runs must be positive when provided.")
run_local_refresh_loop(
interval_seconds=args.interval,
source_mode=args.source,
timeout=args.timeout,
allow_controlled_fallback=not args.no_fallback,
max_runs=args.max_runs,
)
if __name__ == "__main__":
main()

106
backend/app/server_targets.py Normal file
View File

@@ -0,0 +1,106 @@
"""Registry helpers for development-time A2S probe targets."""
from __future__ import annotations
import json
from dataclasses import dataclass
from .config import DEFAULT_A2S_SOURCE_NAME, get_a2s_targets_payload
DEFAULT_A2S_TARGETS = (
{
"name": "Comunidad Hispana #01",
"host": "152.114.195.174",
"query_port": 7778,
"game_port": 7777,
"source_name": DEFAULT_A2S_SOURCE_NAME,
"external_server_id": "comunidad-hispana-01",
"region": "ES",
},
{
"name": "Comunidad Hispana #02",
"host": "152.114.195.150",
"query_port": 7878,
"game_port": 7877,
"source_name": DEFAULT_A2S_SOURCE_NAME,
"external_server_id": "comunidad-hispana-02",
"region": "ES",
},
)
@dataclass(frozen=True, slots=True)
class A2SServerTarget:
"""Minimal configuration needed to query one A2S target."""
name: str
host: str
query_port: int
game_port: int | None
source_name: str
external_server_id: str | None = None
region: str | None = None
def load_a2s_targets() -> tuple[A2SServerTarget, ...]:
"""Load configured A2S targets from env JSON or the local default registry."""
raw_payload = get_a2s_targets_payload()
raw_targets = DEFAULT_A2S_TARGETS if raw_payload is None else _parse_targets(raw_payload)
return tuple(_coerce_target(item) for item in raw_targets)
def _parse_targets(raw_payload: str) -> list[dict[str, object]]:
try:
parsed = json.loads(raw_payload)
except json.JSONDecodeError as error:
raise ValueError("HLL_BACKEND_A2S_TARGETS must be valid JSON.") from error
if not isinstance(parsed, list):
raise ValueError("HLL_BACKEND_A2S_TARGETS must be a JSON array.")
return [item for item in parsed if isinstance(item, dict)]
def _coerce_target(raw_target: dict[str, object]) -> A2SServerTarget:
name = str(raw_target.get("name") or "Unnamed target").strip()
host = str(raw_target.get("host") or "").strip()
source_name = str(raw_target.get("source_name") or DEFAULT_A2S_SOURCE_NAME).strip()
query_port = int(raw_target.get("query_port") or 0)
game_port = _coerce_optional_positive_int(raw_target.get("game_port"))
external_server_id = _string_or_none(raw_target.get("external_server_id"))
region = _string_or_none(raw_target.get("region"))
if not host:
raise ValueError("Each A2S target must define a non-empty host.")
if query_port <= 0:
raise ValueError("Each A2S target must define a valid query_port.")
return A2SServerTarget(
name=name,
host=host,
query_port=query_port,
game_port=game_port,
source_name=source_name or DEFAULT_A2S_SOURCE_NAME,
external_server_id=external_server_id,
region=region,
)
def _string_or_none(value: object) -> str | None:
if not isinstance(value, str):
return None
normalized = value.strip()
return normalized or None
def _coerce_optional_positive_int(value: object) -> int | None:
if value is None:
return None
coerced = int(value)
if coerced <= 0:
raise ValueError("Each A2S target game_port must be positive when defined.")
return coerced

54
backend/app/snapshots.py Normal file
View File

@@ -0,0 +1,54 @@
"""Snapshot builders for normalized provisional server data."""
from __future__ import annotations
from datetime import datetime, timezone
from typing import Iterable, Mapping
def build_server_snapshot(
normalized_record: Mapping[str, object],
*,
captured_at: datetime,
) -> dict[str, object]:
"""Build a consistent snapshot payload for one normalized server."""
timestamp = _as_utc_timestamp(captured_at)
return {
"external_server_id": normalized_record.get("external_server_id"),
"server_name": normalized_record.get("server_name"),
"status": normalized_record.get("status"),
"players": normalized_record.get("players"),
"max_players": normalized_record.get("max_players"),
"current_map": normalized_record.get("current_map"),
"region": normalized_record.get("region"),
"source_name": normalized_record.get("source_name"),
"snapshot_origin": normalized_record.get("snapshot_origin"),
"source_ref": normalized_record.get("source_ref"),
"captured_at": timestamp,
}
def build_snapshot_batch(
normalized_records: Iterable[Mapping[str, object]],
*,
captured_at: datetime,
) -> list[dict[str, object]]:
"""Build snapshots for a batch captured at the same timestamp."""
return [
build_server_snapshot(record, captured_at=captured_at)
for record in normalized_records
]
def utc_now() -> datetime:
"""Return the current UTC timestamp for snapshot capture."""
return datetime.now(timezone.utc)
def _as_utc_timestamp(value: datetime) -> str:
if value.tzinfo is None:
value = value.replace(tzinfo=timezone.utc)
else:
value = value.astimezone(timezone.utc)
return value.isoformat().replace("+00:00", "Z")

514
backend/app/storage.py Normal file
View File

@@ -0,0 +1,514 @@
"""Local SQLite persistence for provisional server snapshots."""
from __future__ import annotations
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable, Mapping
from .config import get_storage_path
DEFAULT_GAME_SOURCE = {
"slug": "current-hll",
"display_name": "Current Hell Let Loose",
"provider_kind": "development",
}
SUMMARY_SNAPSHOT_LIMIT = 6
def initialize_storage(*, db_path: Path | None = None) -> Path:
"""Create the local database file and minimal schema when missing."""
resolved_path = db_path or get_storage_path()
resolved_path.parent.mkdir(parents=True, exist_ok=True)
with _connect(resolved_path) as connection:
connection.executescript(
"""
CREATE TABLE IF NOT EXISTS game_sources (
id INTEGER PRIMARY KEY AUTOINCREMENT,
slug TEXT NOT NULL UNIQUE,
display_name TEXT NOT NULL,
provider_kind TEXT NOT NULL,
is_active INTEGER NOT NULL DEFAULT 1,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS servers (
id INTEGER PRIMARY KEY AUTOINCREMENT,
game_source_id INTEGER NOT NULL,
external_server_id TEXT,
server_name TEXT NOT NULL,
region TEXT,
first_seen_at TEXT NOT NULL,
last_seen_at TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
UNIQUE (game_source_id, external_server_id),
FOREIGN KEY (game_source_id) REFERENCES game_sources(id)
);
CREATE TABLE IF NOT EXISTS server_snapshots (
id INTEGER PRIMARY KEY AUTOINCREMENT,
server_id INTEGER NOT NULL,
captured_at TEXT NOT NULL,
status TEXT NOT NULL,
players INTEGER,
max_players INTEGER,
current_map TEXT,
source_name TEXT NOT NULL,
snapshot_origin TEXT,
source_ref TEXT,
raw_payload_ref TEXT,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (server_id) REFERENCES servers(id)
);
CREATE INDEX IF NOT EXISTS idx_server_snapshots_server_time
ON server_snapshots(server_id, captured_at);
"""
)
_ensure_server_snapshot_columns(connection)
return resolved_path
def persist_snapshot_batch(
snapshots: Iterable[Mapping[str, object]],
*,
source_name: str,
captured_at: str,
game_source: Mapping[str, str] | None = None,
db_path: Path | None = None,
) -> dict[str, object]:
"""Persist a batch of normalized snapshots into local SQLite storage."""
resolved_path = initialize_storage(db_path=db_path)
source_definition = dict(DEFAULT_GAME_SOURCE)
if game_source is not None:
source_definition.update(game_source)
persisted = 0
with _connect(resolved_path) as connection:
game_source_id = _upsert_game_source(connection, source_definition)
for snapshot in snapshots:
server_id = _upsert_server(
connection,
game_source_id=game_source_id,
snapshot=snapshot,
captured_at=captured_at,
)
connection.execute(
"""
INSERT INTO server_snapshots (
server_id,
captured_at,
status,
players,
max_players,
current_map,
source_name,
snapshot_origin,
source_ref,
raw_payload_ref
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
server_id,
captured_at,
snapshot.get("status"),
snapshot.get("players"),
snapshot.get("max_players"),
snapshot.get("current_map"),
snapshot.get("source_name") or source_name,
snapshot.get("snapshot_origin"),
snapshot.get("source_ref"),
None,
),
)
persisted += 1
return {
"db_path": str(resolved_path),
"captured_at": captured_at,
"persisted_snapshots": persisted,
"game_source_slug": source_definition["slug"],
}
def list_latest_snapshots(*, db_path: Path | None = None) -> list[dict[str, object]]:
"""Return the latest persisted snapshot for each known server."""
resolved_path = initialize_storage(db_path=db_path)
with _connect(resolved_path) as connection:
rows = connection.execute(
"""
SELECT
servers.id AS server_id,
servers.external_server_id,
servers.server_name,
servers.region,
game_sources.slug AS context,
server_snapshots.source_name,
server_snapshots.snapshot_origin,
server_snapshots.source_ref,
server_snapshots.captured_at,
server_snapshots.status,
server_snapshots.players,
server_snapshots.max_players,
server_snapshots.current_map
FROM servers
INNER JOIN game_sources
ON game_sources.id = servers.game_source_id
INNER JOIN server_snapshots
ON server_snapshots.server_id = servers.id
INNER JOIN (
SELECT server_id, MAX(captured_at) AS latest_captured_at
FROM server_snapshots
GROUP BY server_id
) AS latest
ON latest.server_id = server_snapshots.server_id
AND latest.latest_captured_at = server_snapshots.captured_at
ORDER BY servers.server_name ASC
"""
).fetchall()
items = [_serialize_snapshot_row(row) for row in rows]
return _attach_history_summaries(connection, items)
def list_snapshot_history(
*,
db_path: Path | None = None,
limit: int = 20,
) -> list[dict[str, object]]:
"""Return recent persisted snapshots across all servers."""
resolved_path = initialize_storage(db_path=db_path)
with _connect(resolved_path) as connection:
rows = connection.execute(
"""
SELECT
servers.id AS server_id,
servers.external_server_id,
servers.server_name,
servers.region,
game_sources.slug AS context,
server_snapshots.source_name,
server_snapshots.snapshot_origin,
server_snapshots.source_ref,
server_snapshots.captured_at,
server_snapshots.status,
server_snapshots.players,
server_snapshots.max_players,
server_snapshots.current_map
FROM server_snapshots
INNER JOIN servers
ON servers.id = server_snapshots.server_id
INNER JOIN game_sources
ON game_sources.id = servers.game_source_id
ORDER BY server_snapshots.captured_at DESC, servers.server_name ASC
LIMIT ?
""",
(limit,),
).fetchall()
return [_serialize_snapshot_row(row) for row in rows]
def list_server_history(
server_id: str,
*,
db_path: Path | None = None,
limit: int = 20,
) -> list[dict[str, object]]:
"""Return recent history for one server by numeric id or external id."""
resolved_path = initialize_storage(db_path=db_path)
server_filter, server_value = _build_server_filter(server_id)
with _connect(resolved_path) as connection:
rows = connection.execute(
f"""
SELECT
servers.id AS server_id,
servers.external_server_id,
servers.server_name,
servers.region,
game_sources.slug AS context,
server_snapshots.source_name,
server_snapshots.snapshot_origin,
server_snapshots.source_ref,
server_snapshots.captured_at,
server_snapshots.status,
server_snapshots.players,
server_snapshots.max_players,
server_snapshots.current_map
FROM server_snapshots
INNER JOIN servers
ON servers.id = server_snapshots.server_id
INNER JOIN game_sources
ON game_sources.id = servers.game_source_id
WHERE {server_filter} = ?
ORDER BY server_snapshots.captured_at DESC
LIMIT ?
""",
(server_value, limit),
).fetchall()
return [_serialize_snapshot_row(row) for row in rows]
def _connect(db_path: Path) -> sqlite3.Connection:
connection = sqlite3.connect(db_path)
connection.row_factory = sqlite3.Row
return connection
def _upsert_game_source(
connection: sqlite3.Connection,
game_source: Mapping[str, str],
) -> int:
connection.execute(
"""
INSERT INTO game_sources (slug, display_name, provider_kind, is_active)
VALUES (?, ?, ?, 1)
ON CONFLICT(slug) DO UPDATE SET
display_name = excluded.display_name,
provider_kind = excluded.provider_kind,
is_active = 1,
updated_at = CURRENT_TIMESTAMP
""",
(
game_source["slug"],
game_source["display_name"],
game_source["provider_kind"],
),
)
row = connection.execute(
"SELECT id FROM game_sources WHERE slug = ?",
(game_source["slug"],),
).fetchone()
if row is None:
raise RuntimeError("Failed to resolve game source during snapshot persistence.")
return int(row["id"])
def _upsert_server(
connection: sqlite3.Connection,
*,
game_source_id: int,
snapshot: Mapping[str, object],
captured_at: str,
) -> int:
external_server_id = snapshot.get("external_server_id")
if not isinstance(external_server_id, str) or not external_server_id.strip():
external_server_id = _build_fallback_external_id(snapshot)
server_name = str(snapshot.get("server_name") or "Unknown server")
region = snapshot.get("region")
connection.execute(
"""
INSERT INTO servers (
game_source_id,
external_server_id,
server_name,
region,
first_seen_at,
last_seen_at
) VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(game_source_id, external_server_id) DO UPDATE SET
server_name = excluded.server_name,
region = excluded.region,
last_seen_at = excluded.last_seen_at,
updated_at = CURRENT_TIMESTAMP
""",
(
game_source_id,
external_server_id,
server_name,
region,
captured_at,
captured_at,
),
)
row = connection.execute(
"""
SELECT id
FROM servers
WHERE game_source_id = ? AND external_server_id = ?
""",
(game_source_id, external_server_id),
).fetchone()
if row is None:
raise RuntimeError("Failed to resolve server during snapshot persistence.")
return int(row["id"])
def _build_fallback_external_id(snapshot: Mapping[str, object]) -> str:
server_name = str(snapshot.get("server_name") or "unknown-server")
normalized = "".join(
character.lower() if character.isalnum() else "-"
for character in server_name
)
compact = "-".join(part for part in normalized.split("-") if part)
return compact or "unknown-server"
def _ensure_server_snapshot_columns(connection: sqlite3.Connection) -> None:
columns = {
str(row["name"])
for row in connection.execute("PRAGMA table_info(server_snapshots)").fetchall()
}
if "snapshot_origin" not in columns:
connection.execute("ALTER TABLE server_snapshots ADD COLUMN snapshot_origin TEXT")
if "source_ref" not in columns:
connection.execute("ALTER TABLE server_snapshots ADD COLUMN source_ref TEXT")
connection.execute(
"""
UPDATE server_snapshots
SET snapshot_origin = CASE
WHEN source_name = 'controlled-placeholder' THEN 'controlled-fallback'
WHEN source_name LIKE '%a2s%' THEN 'real-a2s'
ELSE 'unknown'
END
WHERE snapshot_origin IS NULL OR snapshot_origin = ''
"""
)
connection.execute(
"""
UPDATE server_snapshots
SET source_ref = source_name
WHERE source_ref IS NULL OR source_ref = ''
"""
)
_backfill_registered_a2s_source_refs(connection)
def _backfill_registered_a2s_source_refs(connection: sqlite3.Connection) -> None:
from .server_targets import load_a2s_targets
for target in load_a2s_targets():
if not target.external_server_id:
continue
connection.execute(
"""
UPDATE server_snapshots
SET source_ref = ?
WHERE snapshot_origin = 'real-a2s'
AND source_ref = source_name
AND server_id IN (
SELECT id
FROM servers
WHERE external_server_id = ?
)
""",
(
f"a2s://{target.host}:{target.query_port}",
target.external_server_id,
),
)
def _serialize_snapshot_row(row: sqlite3.Row) -> dict[str, object]:
return {
"server_id": row["server_id"],
"external_server_id": row["external_server_id"],
"server_name": row["server_name"],
"region": row["region"],
"context": row["context"],
"source_name": row["source_name"],
"snapshot_origin": row["snapshot_origin"],
"source_ref": row["source_ref"],
"captured_at": row["captured_at"],
"status": row["status"],
"players": row["players"],
"max_players": row["max_players"],
"current_map": row["current_map"],
}
def _attach_history_summaries(
connection: sqlite3.Connection,
items: list[dict[str, object]],
) -> list[dict[str, object]]:
enriched_items: list[dict[str, object]] = []
for item in items:
enriched = dict(item)
enriched["history_summary"] = _build_history_summary(
connection,
int(item["server_id"]),
)
enriched_items.append(enriched)
return enriched_items
def _build_history_summary(
connection: sqlite3.Connection,
server_id: int,
) -> dict[str, object]:
rows = connection.execute(
"""
SELECT
captured_at,
status,
players
FROM server_snapshots
WHERE server_id = ?
ORDER BY captured_at DESC
LIMIT ?
""",
(server_id, SUMMARY_SNAPSHOT_LIMIT),
).fetchall()
return _summarize_history_rows(rows)
def _summarize_history_rows(rows: list[sqlite3.Row]) -> dict[str, object]:
capture_count = len(rows)
player_values = [
int(row["players"])
for row in rows
if row["players"] is not None
]
online_rows = [row for row in rows if row["status"] == "online"]
latest_captured_at = str(rows[0]["captured_at"]) if rows else None
last_seen_online_at = str(online_rows[0]["captured_at"]) if online_rows else None
return {
"window_size": SUMMARY_SNAPSHOT_LIMIT,
"recent_capture_count": capture_count,
"recent_online_count": len(online_rows),
"recent_average_players": _round_average(player_values),
"recent_peak_players": max(player_values, default=None),
"last_seen_online_at": last_seen_online_at,
"minutes_since_last_capture": _minutes_since_timestamp(latest_captured_at),
}
def _round_average(values: list[int]) -> float | None:
if not values:
return None
return round(sum(values) / len(values), 1)
def _minutes_since_timestamp(timestamp: str | None) -> int | None:
if not timestamp:
return None
normalized = timestamp.replace("Z", "+00:00")
captured_at = datetime.fromisoformat(normalized)
if captured_at.tzinfo is None:
captured_at = captured_at.replace(tzinfo=timezone.utc)
delta = datetime.now(timezone.utc) - captured_at.astimezone(timezone.utc)
return max(0, int(delta.total_seconds() // 60))
def _build_server_filter(server_id: str) -> tuple[str, object]:
normalized = server_id.strip()
if normalized.isdigit():
return "servers.id", int(normalized)
return "servers.external_server_id", normalized