87 lines
2.5 KiB
Python
87 lines
2.5 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
|
||
|
|
def ensure_utc(value: datetime) -> datetime:
|
||
|
|
if value.tzinfo is None:
|
||
|
|
return value.replace(tzinfo=timezone.utc)
|
||
|
|
return value.astimezone(timezone.utc)
|
||
|
|
|
||
|
|
|
||
|
|
def utc_now() -> datetime:
|
||
|
|
return datetime.now(timezone.utc)
|
||
|
|
|
||
|
|
|
||
|
|
def parse_datetime(value: Any) -> datetime | None:
|
||
|
|
if value in (None, ""):
|
||
|
|
return None
|
||
|
|
if isinstance(value, datetime):
|
||
|
|
return ensure_utc(value)
|
||
|
|
text = str(value).strip()
|
||
|
|
candidates = [
|
||
|
|
text,
|
||
|
|
text.replace("Z", "+00:00"),
|
||
|
|
text.replace(" ", "T"),
|
||
|
|
text.replace(" ", "T").replace("Z", "+00:00"),
|
||
|
|
]
|
||
|
|
for candidate in candidates:
|
||
|
|
try:
|
||
|
|
return ensure_utc(datetime.fromisoformat(candidate))
|
||
|
|
except ValueError:
|
||
|
|
continue
|
||
|
|
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
|
||
|
|
try:
|
||
|
|
parsed = datetime.strptime(text, fmt)
|
||
|
|
return parsed.replace(tzinfo=timezone.utc)
|
||
|
|
except ValueError:
|
||
|
|
continue
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def isoformat_z(value: datetime) -> str:
|
||
|
|
return ensure_utc(value).isoformat().replace("+00:00", "Z")
|
||
|
|
|
||
|
|
|
||
|
|
def format_modelhub_datetime(value: datetime) -> str:
|
||
|
|
return ensure_utc(value).strftime("%Y-%m-%d %H:%M:%S")
|
||
|
|
|
||
|
|
|
||
|
|
def json_dumps(value: Any) -> str:
|
||
|
|
return json.dumps(value, ensure_ascii=False, sort_keys=True)
|
||
|
|
|
||
|
|
|
||
|
|
def read_json(path: Path) -> Any:
|
||
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
||
|
|
|
||
|
|
|
||
|
|
def write_json(path: Path, value: Any) -> None:
|
||
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
||
|
|
path.write_text(json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
||
|
|
|
||
|
|
|
||
|
|
def read_jsonl(path: Path) -> list[dict[str, Any]]:
|
||
|
|
if not path.exists():
|
||
|
|
return []
|
||
|
|
rows: list[dict[str, Any]] = []
|
||
|
|
for line in path.read_text(encoding="utf-8").splitlines():
|
||
|
|
if not line.strip():
|
||
|
|
continue
|
||
|
|
rows.append(json.loads(line))
|
||
|
|
return rows
|
||
|
|
|
||
|
|
|
||
|
|
def write_jsonl(path: Path, rows: list[dict[str, Any]]) -> None:
|
||
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
||
|
|
content = "".join(json.dumps(row, ensure_ascii=False, sort_keys=True) + "\n" for row in rows)
|
||
|
|
path.write_text(content, encoding="utf-8")
|
||
|
|
|
||
|
|
|
||
|
|
def append_jsonl(path: Path, row: dict[str, Any]) -> None:
|
||
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
||
|
|
with path.open("a", encoding="utf-8") as handle:
|
||
|
|
handle.write(json.dumps(row, ensure_ascii=False, sort_keys=True) + "\n")
|