63 lines
2.5 KiB
Python
63 lines
2.5 KiB
Python
|
"""Server-authoritative attribution for the timeline scene.
|
|||
|
|
|
|||
|
|
Annotations are the only authored thing in a scene (clips/tracks/root come from
|
|||
|
|
the OTIO), so that's what we attribute. On each save we diff the incoming
|
|||
|
|
annotation groups against the stored ones and stamp who/when — always from the
|
|||
|
|
trusted server-side identity, never from values the client sent, so authorship
|
|||
|
|
can't be forged.
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
STAMP_KEYS = ("createdBy", "createdAt", "editedBy", "editedAt")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _is_annotation(group):
|
|||
|
|
return isinstance(group, dict) and group.get("type") == "annotation"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _content(group):
|
|||
|
|
"""The group minus its attribution stamps — what we compare for changes."""
|
|||
|
|
return {k: v for k, v in group.items() if k not in STAMP_KEYS}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def annotation_layer(scene):
|
|||
|
|
"""Just the annotation groups of a scene, as {gid: group}."""
|
|||
|
|
return {gid: g for gid, g in (scene or {}).get("groups", {}).items() if _is_annotation(g)}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def apply_attribution(prev_scene, incoming_scene, who, now):
|
|||
|
|
"""Return (new_scene, summary, counts) with annotation stamps reconciled.
|
|||
|
|
|
|||
|
|
`who` is the authenticated username, `now` an ISO timestamp string. New
|
|||
|
|
annotations get created/edited stamps; changed ones get a fresh edited stamp
|
|||
|
|
(preserving the original creator); unchanged ones keep the server's existing
|
|||
|
|
stamps. Client-supplied stamp values are ignored throughout.
|
|||
|
|
"""
|
|||
|
|
prev = annotation_layer(prev_scene)
|
|||
|
|
groups = dict((incoming_scene or {}).get("groups", {}))
|
|||
|
|
created = edited = 0
|
|||
|
|
|
|||
|
|
for gid, g in groups.items():
|
|||
|
|
if not _is_annotation(g):
|
|||
|
|
continue
|
|||
|
|
g = {k: v for k, v in g.items() if k not in STAMP_KEYS} # drop client stamps
|
|||
|
|
old = prev.get(gid)
|
|||
|
|
if old is None:
|
|||
|
|
g.update(createdBy=who, createdAt=now, editedBy=who, editedAt=now)
|
|||
|
|
created += 1
|
|||
|
|
elif _content(g) != _content(old):
|
|||
|
|
g["createdBy"] = old.get("createdBy", who)
|
|||
|
|
g["createdAt"] = old.get("createdAt", now)
|
|||
|
|
g["editedBy"], g["editedAt"] = who, now
|
|||
|
|
edited += 1
|
|||
|
|
else: # unchanged → restore the server's stamps verbatim
|
|||
|
|
for k in STAMP_KEYS:
|
|||
|
|
if k in old:
|
|||
|
|
g[k] = old[k]
|
|||
|
|
groups[gid] = g
|
|||
|
|
|
|||
|
|
deleted = sum(1 for gid in prev if gid not in groups)
|
|||
|
|
new_scene = dict(incoming_scene or {}, groups=groups)
|
|||
|
|
counts = {"created": created, "edited": edited, "deleted": deleted}
|
|||
|
|
summary = f"+{created} ~{edited} −{deleted} annotations"
|
|||
|
|
return new_scene, summary, counts
|