36054ac329
Server-authoritative attribution: on each scene PUT the server diffs incoming annotation groups and stamps createdBy/editedBy (+ timestamps) from request.user, ignoring client-sent stamps so authorship can't be forged. Each save also writes a Revision (user, time, +N ~N −N summary, snapshot of the annotation layer), visible in the admin and via /revisions/. Projects gain collaborators so several users can edit one project and get distinct attribution. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
62 lines
2.5 KiB
Python
62 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
|