Initialize data freshness from Git history

This commit is contained in:
liuji
2026-10-10 18:03:47 +08:00
parent 11ba68882b
commit 4e262060dd
+32 -2
View File
@@ -186,6 +186,31 @@ def read_sync_log() -> dict[str, object]:
return payload if isinstance(payload, dict) else {} return payload if isinstance(payload, dict) else {}
def git_data_update_times() -> tuple[str, str] | None:
"""Return the latest committed data time in UTC and Shanghai time."""
try:
result = subprocess.run(
["git", "log", "-1", "--format=%cI", "--", str(PUBLISH_PATH)],
cwd=ROOT,
check=True,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
committed_at = result.stdout.strip()
if not committed_at:
return None
timestamp = datetime.fromisoformat(committed_at)
if timestamp.tzinfo is None:
return None
return (
timestamp.astimezone(timezone.utc).isoformat().replace("+00:00", "Z"),
timestamp.astimezone(ZoneInfo("Asia/Shanghai")).isoformat(),
)
except (OSError, subprocess.SubprocessError, ValueError):
return None
def publish(content: bytes, sync_log: dict[str, object]) -> None: def publish(content: bytes, sync_log: dict[str, object]) -> None:
log_content = (json.dumps(sync_log, ensure_ascii=False, indent=2) + "\n").encode("utf-8") log_content = (json.dumps(sync_log, ensure_ascii=False, indent=2) + "\n").encode("utf-8")
targets = { targets = {
@@ -270,8 +295,13 @@ def sync() -> None:
last_data_update_at = previous_sync_log.get("last_data_update_at") last_data_update_at = previous_sync_log.get("last_data_update_at")
last_data_update_at_shanghai = previous_sync_log.get("last_data_update_at_shanghai") last_data_update_at_shanghai = previous_sync_log.get("last_data_update_at_shanghai")
if not isinstance(last_data_update_at, str) or not last_data_update_at: if not isinstance(last_data_update_at, str) or not last_data_update_at:
last_data_update_at = None historical_times = git_data_update_times()
if not isinstance(last_data_update_at_shanghai, str) or not last_data_update_at_shanghai: if historical_times is None:
last_data_update_at = None
last_data_update_at_shanghai = None
else:
last_data_update_at, last_data_update_at_shanghai = historical_times
elif not isinstance(last_data_update_at_shanghai, str) or not last_data_update_at_shanghai:
last_data_update_at_shanghai = None last_data_update_at_shanghai = None
sync_log = { sync_log = {
"last_success_at": success_at.isoformat().replace("+00:00", "Z"), "last_success_at": success_at.isoformat().replace("+00:00", "Z"),