from __future__ import annotations import gzip import re from datetime import date, datetime from pathlib import Path from .config import TIMEZONE UUID_RE = re.compile(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}") def as_int(value: object) -> int: if value is None: return 0 if isinstance(value, bool): return int(value) if isinstance(value, (int, float)): return int(value) return 0 def as_float(value: object) -> float: if value is None: return 0.0 if isinstance(value, bool): return float(value) if isinstance(value, (int, float)): return float(value) return 0.0 def today_in_tz() -> date: return datetime.now(TIMEZONE).date() def now_in_tz() -> datetime: return datetime.now(TIMEZONE) def sanitize_label(value: object, default: str = "unknown") -> str: if value is None: return default text = str(value).strip() return text or default def prom_value(value: str) -> str: return value.replace("\\", r"\\").replace("\n", r"\n").replace('"', r'\"') def metric_line(name: str, value: object, **labels: object) -> str: if labels: rendered = ",".join(f'{key}="{prom_value(sanitize_label(val))}"' for key, val in sorted(labels.items())) return f"{name}{{{rendered}}} {value}" return f"{name} {value}" def open_session_file(path: Path): if path.suffix == ".gz": return gzip.open(path, "rt", encoding="utf-8") return path.open("r", encoding="utf-8") def session_id_from_path(path: Path) -> str | None: match = UUID_RE.search(path.name) if match: return match.group(0).lower() return None