70 lines
1.7 KiB
Python
70 lines
1.7 KiB
Python
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
|