from __future__ import annotations import json import math import re from hashlib import sha1 from pathlib import Path from typing import Any from zoneinfo import ZoneInfo from datetime import datetime ROOT = Path(__file__).resolve().parent.parent CONFIG_DIR = ROOT / "config" DATA_DIR = ROOT / "data" DASHBOARD_DIR = ROOT / "dashboard" def now_iso() -> str: return datetime.utcnow().replace(microsecond=0).isoformat() + "Z" def current_run_date(timezone: str = "Asia/Shanghai") -> str: return datetime.now(ZoneInfo(timezone)).strftime("%Y-%m-%d") def read_json(path: Path, default: Any = None) -> Any: try: return json.loads(path.read_text(encoding="utf-8")) except Exception: return default def write_json(path: Path, payload: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def num(value: Any) -> float | None: if value in (None, "", False): return None try: parsed = float(value) except (TypeError, ValueError): return None if math.isnan(parsed) or math.isinf(parsed): return None return parsed def clamp(value: Any, minimum: float, maximum: float) -> float: parsed = num(value) if parsed is None: return minimum return max(minimum, min(maximum, parsed)) def clean_bool(value: Any, fallback: bool = False) -> bool: if isinstance(value, bool): return value return fallback def one_line(value: Any, limit: int = 220) -> str: text = re.sub(r"\s+", " ", str(value or "")).strip() if len(text) <= limit: return text return text[: max(0, limit - 3)].rstrip() + "..." def safe_last(items: list[Any] | None) -> Any: if not items: return None return items[-1] def sha1_hex(value: str) -> str: return sha1(value.encode("utf-8")).hexdigest() def parse_json_loose(raw_text: str) -> Any: text = str(raw_text or "").strip() if not text: raise ValueError("empty JSON payload") try: return json.loads(text) except json.JSONDecodeError: pass fence_match = re.search(r"```(?:json)?\s*(\{.*\}|\[.*\])\s*```", text, re.S) if fence_match: return json.loads(fence_match.group(1)) brace_match = re.search(r"(\{.*\}|\[.*\])", text, re.S) if brace_match: return json.loads(brace_match.group(1)) raise ValueError("could not find JSON object in response")