forked from LengKundee/NUNA
77 lines
2.1 KiB
Python
77 lines
2.1 KiB
Python
|
"""
|
||
|
|
Common utility functions for the Google Drive cleanup and trading data tools.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import datetime as dt
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Any, Dict, Optional
|
||
|
|
|
||
|
|
|
||
|
|
def eprint(*args: object) -> None:
|
||
|
|
"""Print to stderr."""
|
||
|
|
print(*args, file=sys.stderr)
|
||
|
|
|
||
|
|
|
||
|
|
def human_bytes(n: Optional[int]) -> str:
|
||
|
|
"""Format bytes into a human-readable string (KB, MB, GB, etc.)."""
|
||
|
|
if n is None:
|
||
|
|
return "—"
|
||
|
|
units = ["B", "KB", "MB", "GB", "TB", "PB"]
|
||
|
|
x = float(n)
|
||
|
|
for u in units:
|
||
|
|
if x < 1024 or u == units[-1]:
|
||
|
|
return f"{x:.1f}{u}" if u != "B" else f"{int(x)}B"
|
||
|
|
x /= 1024
|
||
|
|
return f"{n}B"
|
||
|
|
|
||
|
|
|
||
|
|
def now_stamp() -> str:
|
||
|
|
"""Get a UTC timestamp string (filesystem-safe)."""
|
||
|
|
return dt.datetime.now(dt.timezone.utc).strftime("%Y%m%d-%H%M%SZ")
|
||
|
|
|
||
|
|
|
||
|
|
def now_local_stamp() -> str:
|
||
|
|
"""Get a local timestamp string including microseconds (filesystem-safe)."""
|
||
|
|
return dt.datetime.now().strftime("%Y%m%d-%H%M%S-%f")
|
||
|
|
|
||
|
|
|
||
|
|
def load_json_config(path: Path) -> Dict[str, Any]:
|
||
|
|
"""Load a JSON configuration file."""
|
||
|
|
if not path.exists():
|
||
|
|
return {}
|
||
|
|
with path.open("r", encoding="utf-8") as f:
|
||
|
|
payload = json.load(f)
|
||
|
|
if not isinstance(payload, dict):
|
||
|
|
raise ValueError(f"Config must be a JSON object: {path}")
|
||
|
|
return payload
|
||
|
|
|
||
|
|
|
||
|
|
def deep_merge(base: Dict[str, Any], override: Dict[str, Any]) -> Dict[str, Any]:
|
||
|
|
"""Recursively merge two dictionaries."""
|
||
|
|
out: Dict[str, Any] = dict(base)
|
||
|
|
for k, v in override.items():
|
||
|
|
if (
|
||
|
|
k in out
|
||
|
|
and isinstance(out[k], dict)
|
||
|
|
and isinstance(v, dict)
|
||
|
|
):
|
||
|
|
out[k] = deep_merge(out[k], v)
|
||
|
|
else:
|
||
|
|
out[k] = v
|
||
|
|
return out
|
||
|
|
|
||
|
|
|
||
|
|
def mkdirp(p: Path) -> None:
|
||
|
|
"""Create directory and its parents if they don't exist."""
|
||
|
|
p.mkdir(parents=True, exist_ok=True)
|
||
|
|
|
||
|
|
|
||
|
|
def write_json(path: str | Path, payload: Any) -> None:
|
||
|
|
"""Write a payload to a JSON file."""
|
||
|
|
with open(path, "w", encoding="utf-8") as f:
|
||
|
|
json.dump(payload, f, indent=2, sort_keys=True)
|