feat(snapshots): sort list output by started_at across kinds
This commit is contained in:
@@ -1,7 +1,8 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from typing import Any, Optional, Tuple
|
||||
|
||||
from ..config.load import load_global_config, load_host_config
|
||||
from ..config.merge import build_effective_config
|
||||
@@ -11,6 +12,38 @@ from ..snapshot_meta import iter_snapshot_dirs, normalize_kind, read_snapshot_me
|
||||
from ..util import sanitize_host
|
||||
|
||||
|
||||
def _parse_iso_z(s: Any) -> Optional[datetime]:
|
||||
"""
|
||||
Parse timestamps like '2026-02-02T22:38:07Z' into aware UTC datetime.
|
||||
Returns None if invalid.
|
||||
"""
|
||||
if not isinstance(s, str) or not s:
|
||||
return None
|
||||
# Strictly support trailing 'Z' (UTC) to avoid locale/timezone ambiguity.
|
||||
if not s.endswith("Z"):
|
||||
return None
|
||||
try:
|
||||
dt = datetime.strptime(s, "%Y-%m-%dT%H:%M:%SZ")
|
||||
return dt.replace(tzinfo=timezone.utc)
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def _sort_key(item: dict[str, Any]) -> Tuple[int, datetime, str]:
|
||||
"""
|
||||
Sort by:
|
||||
1) Has started_at meta (1) before missing (0)
|
||||
2) started_at descending
|
||||
3) dirname descending (lexicographic)
|
||||
"""
|
||||
dt = _parse_iso_z(item.get("started_at"))
|
||||
has_dt = 1 if dt is not None else 0
|
||||
# Use epoch for missing to keep key comparable; has_dt separates them anyway.
|
||||
dt2 = dt if dt is not None else datetime.fromtimestamp(0, tz=timezone.utc)
|
||||
dirname = item.get("dirname") or ""
|
||||
return (has_dt, dt2, dirname)
|
||||
|
||||
|
||||
def run_snapshots_list(prefix: Path, host: str, kind: str, limit: int, include_incomplete: bool) -> dict[str, Any]:
|
||||
host = sanitize_host(host)
|
||||
k = normalize_kind(kind)
|
||||
@@ -38,20 +71,13 @@ def run_snapshots_list(prefix: Path, host: str, kind: str, limit: int, include_i
|
||||
else:
|
||||
kinds = [k]
|
||||
|
||||
out: list[dict[str, Any]] = []
|
||||
remaining = limit
|
||||
items: list[dict[str, Any]] = []
|
||||
|
||||
for kk in kinds:
|
||||
if remaining <= 0:
|
||||
break
|
||||
|
||||
for d in iter_snapshot_dirs(host_root, kk):
|
||||
if remaining <= 0:
|
||||
break
|
||||
|
||||
meta = read_snapshot_meta(d)
|
||||
|
||||
out.append(
|
||||
items.append(
|
||||
{
|
||||
"kind": kk,
|
||||
"dirname": d.name,
|
||||
@@ -64,7 +90,12 @@ def run_snapshots_list(prefix: Path, host: str, kind: str, limit: int, include_i
|
||||
"id": meta.get("id"),
|
||||
}
|
||||
)
|
||||
remaining -= 1
|
||||
|
||||
# Global sort: newest first
|
||||
items.sort(key=_sort_key, reverse=True)
|
||||
|
||||
# Apply limit after sorting
|
||||
out = items[:limit]
|
||||
|
||||
return {
|
||||
"ok": True,
|
||||
|
||||
Reference in New Issue
Block a user