394c46ebfcOptional metadata or restamp builds the same commit history as node-red. A call that omits both stays digest-only with restamp false. Local and shared stamps reject metadata.
207 lines
6.8 KiB
Python
207 lines
6.8 KiB
Python
"""verae-meta-git/1 documents, matching node-red-contrib-verae lib/metadata-history.js."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
SCHEMA = "verae-meta-git/1"
|
|
|
|
|
|
def is_plain_object(value: Any) -> bool:
|
|
return isinstance(value, dict)
|
|
|
|
|
|
def is_blank(value: Any) -> bool:
|
|
if value is None:
|
|
return True
|
|
if isinstance(value, str):
|
|
return value.strip() == ""
|
|
if isinstance(value, list):
|
|
return len(value) == 0
|
|
if isinstance(value, dict):
|
|
return len(value) == 0
|
|
return False
|
|
|
|
|
|
def _clone(value: Any) -> Any:
|
|
if isinstance(value, list):
|
|
return [_clone(item) for item in value]
|
|
if isinstance(value, dict):
|
|
return {key: _clone(item) for key, item in value.items()}
|
|
return value
|
|
|
|
|
|
def _scalar(value: Any) -> str:
|
|
if isinstance(value, bool) or value is None or isinstance(value, str):
|
|
return json.dumps(value, ensure_ascii=False, separators=(",", ":"))
|
|
if isinstance(value, int) and not isinstance(value, bool):
|
|
return json.dumps(value)
|
|
if isinstance(value, float):
|
|
if value.is_integer():
|
|
return json.dumps(int(value))
|
|
return json.dumps(value, ensure_ascii=False, separators=(",", ":"))
|
|
return json.dumps(value, ensure_ascii=False, separators=(",", ":"))
|
|
|
|
|
|
def canonical(value: Any) -> str:
|
|
if isinstance(value, list):
|
|
return "[" + ",".join(canonical(item) for item in value) + "]"
|
|
if isinstance(value, dict):
|
|
keys = sorted(value)
|
|
return "{" + ",".join(_scalar(key) + ":" + canonical(value[key]) for key in keys) + "}"
|
|
return _scalar(value)
|
|
|
|
|
|
def _same(a: Any, b: Any) -> bool:
|
|
return canonical(a) == canonical(b)
|
|
|
|
|
|
def extend_snapshot(base: dict | None, patch: dict | None) -> dict:
|
|
out = _clone(base) if isinstance(base, dict) else {}
|
|
if not isinstance(patch, dict):
|
|
return out
|
|
for key, nxt in patch.items():
|
|
if is_blank(nxt):
|
|
continue
|
|
prev = out.get(key, None)
|
|
has = key in out
|
|
if has and isinstance(prev, dict) and isinstance(nxt, dict):
|
|
out[key] = extend_snapshot(prev, nxt)
|
|
elif isinstance(prev, list) and isinstance(nxt, list):
|
|
extra = [_clone(item) for item in nxt if not is_blank(item)]
|
|
if extra:
|
|
out[key] = list(prev) + extra
|
|
elif isinstance(nxt, list):
|
|
extra = [_clone(item) for item in nxt if not is_blank(item)]
|
|
if not extra:
|
|
continue
|
|
out[key] = (list(prev) + extra) if isinstance(prev, list) else extra
|
|
elif isinstance(nxt, dict):
|
|
nested = extend_snapshot(prev if isinstance(prev, dict) else {}, nxt)
|
|
if not is_blank(nested):
|
|
out[key] = nested
|
|
else:
|
|
out[key] = _clone(nxt)
|
|
return out
|
|
|
|
|
|
def changes_between(before: dict | None, after: dict | None) -> dict:
|
|
added: dict[str, Any] = {}
|
|
extended: dict[str, Any] = {}
|
|
updated: dict[str, Any] = {}
|
|
prior = before if isinstance(before, dict) else {}
|
|
nxt = after if isinstance(after, dict) else {}
|
|
for key in sorted(set(prior) | set(nxt)):
|
|
if key not in nxt:
|
|
continue
|
|
if key not in prior:
|
|
added[key] = _clone(nxt[key])
|
|
continue
|
|
if _same(prior[key], nxt[key]):
|
|
continue
|
|
if isinstance(prior[key], dict) and isinstance(nxt[key], dict):
|
|
nested = changes_between(prior[key], nxt[key])
|
|
if not is_blank(nested["added"]) or not is_blank(nested["extended"]) or not is_blank(nested["updated"]):
|
|
extended[key] = nested
|
|
elif (
|
|
isinstance(prior[key], list)
|
|
and isinstance(nxt[key], list)
|
|
and len(nxt[key]) >= len(prior[key])
|
|
):
|
|
appended = nxt[key][len(prior[key]) :]
|
|
if appended:
|
|
extended[key] = {"appended": _clone(appended)}
|
|
else:
|
|
updated[key] = {"from": _clone(prior[key]), "to": _clone(nxt[key])}
|
|
return {"added": added, "extended": extended, "updated": updated}
|
|
|
|
|
|
def _commit_id(commit: dict) -> str:
|
|
material = canonical(
|
|
{
|
|
"parent": commit["parent"],
|
|
"at": commit["at"],
|
|
"message": commit["message"],
|
|
"snapshot": commit["snapshot"],
|
|
}
|
|
)
|
|
return hashlib.sha256(material.encode()).hexdigest()
|
|
|
|
|
|
def is_history(value: Any) -> bool:
|
|
return (
|
|
isinstance(value, dict)
|
|
and value.get("schema") == SCHEMA
|
|
and isinstance(value.get("commits"), list)
|
|
and isinstance(value.get("snapshot"), dict)
|
|
)
|
|
|
|
|
|
def _history_from_snapshot(snapshot: dict, at: str) -> dict:
|
|
stored = _clone(snapshot)
|
|
commit = {
|
|
"parent": None,
|
|
"at": at,
|
|
"message": "stamp",
|
|
"snapshot": stored,
|
|
"changes": changes_between({}, stored),
|
|
}
|
|
commit["id"] = _commit_id(commit)
|
|
return {
|
|
"schema": SCHEMA,
|
|
"head": commit["id"],
|
|
"snapshot": _clone(stored),
|
|
"commits": [commit],
|
|
}
|
|
|
|
|
|
def _coerce(existing: Any, at: str) -> dict:
|
|
if is_history(existing):
|
|
return _clone(existing)
|
|
if isinstance(existing, dict) and not is_blank(existing):
|
|
return _history_from_snapshot(existing, at)
|
|
return {"schema": SCHEMA, "head": None, "snapshot": {}, "commits": []}
|
|
|
|
|
|
def _has_changes(changes: dict) -> bool:
|
|
return not is_blank(changes["added"]) or not is_blank(changes["extended"]) or not is_blank(changes["updated"])
|
|
|
|
|
|
def append_metadata(
|
|
existing: Any,
|
|
patch: dict | None,
|
|
*,
|
|
restamp: bool = False,
|
|
at: str | None = None,
|
|
prior_at: str | None = None,
|
|
message: str | None = None,
|
|
) -> dict | None:
|
|
if not at:
|
|
at = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
|
|
blank_patch = not isinstance(patch, dict) or is_blank(patch) or is_blank(extend_snapshot({}, patch))
|
|
if not restamp and blank_patch and not is_history(existing) and (existing is None or is_blank(existing)):
|
|
return None
|
|
doc = _coerce(existing, prior_at or at)
|
|
head = doc.get("head")
|
|
if not restamp and blank_patch:
|
|
return doc if head else None
|
|
snapshot = extend_snapshot(doc.get("snapshot") or {}, patch if isinstance(patch, dict) else {})
|
|
changes = changes_between(doc.get("snapshot") or {}, snapshot)
|
|
if not restamp and not _has_changes(changes) and head:
|
|
return doc
|
|
msg = message or ("restamp" if head else "stamp")
|
|
commit = {
|
|
"parent": head,
|
|
"at": at,
|
|
"message": msg,
|
|
"snapshot": snapshot,
|
|
"changes": changes,
|
|
}
|
|
commit["id"] = _commit_id(commit)
|
|
doc["commits"] = list(doc.get("commits") or []) + [commit]
|
|
doc["head"] = commit["id"]
|
|
doc["snapshot"] = snapshot
|
|
return doc
|