feat: organize places and device workspaces with scoped heritage records

heritage.py

@@ -0,0 +1,78 @@
+"""Scoped cultural and property records; records never transfer access or ownership."""
+import datetime as dt
+from urllib.parse import urlsplit
+
+
+class Heritage:
+    def __init__(self, app):
+        self.a = app
+
+    def initialize(self):
+        with self.a.LOCK, self.a.DB:
+            self.a.DB.execute('CREATE TABLE IF NOT EXISTS heritage_entries (id TEXT PRIMARY KEY, body TEXT NOT NULL)')
+
+    def require(self, user, edit=False):
+        a = self.a
+        home = a.get_object('households', a.HOUSEHOLDS.scope(user))
+        if home.get('kind', 'home') != 'home':
+            raise a.Problem('家庭传承资料属于家庭，请先进入对应家庭', 403)
+        if not a.HOUSEHOLDS.manager(user) and user.get('heritageAccess', 'none') not in (('edit',) if edit else ('read', 'edit')):
+            raise a.Problem('尚未获得文化与财产档案权限，请联系家庭管理员', 403)
+
+    def snapshot(self, user):
+        self.require(user)
+        return dict(entries=self.a.HOUSEHOLDS.objects(user, 'heritage_entries'),
+                    canEdit=self.a.HOUSEHOLDS.manager(user) or user.get('heritageAccess') == 'edit')
+
+    def save(self, data, user):
+        a = self.a
+        with a.LOCK:
+            self.require(user, True)
+            family = a.HOUSEHOLDS.prepare(user, 'heritage_entries', data)
+            key = a.object_key('heritage_entries', data)
+            old = a.get_object('heritage_entries', key) or {}
+            if data.get('revision', 0) != old.get('revision', 0):
+                raise a.Problem('档案已被更新，请重新打开后再保存', 409)
+            kind = data.get('kind')
+            if kind not in ('culture', 'asset') or old and kind != old['kind']:
+                raise a.Problem('请选择文化资料或财产档案；已有档案不能变更类别')
+            archived = data.get('archived', False)
+            if not isinstance(archived, bool):
+                raise a.Problem('归档状态不正确')
+            name = a.clean_text(data.get('name', ''), 120, True)
+            date = a.clean_text(data.get('date', ''), 10)
+            if date:
+                try:
+                    if dt.date.fromisoformat(date).isoformat() != date:
+                        raise ValueError()
+                except ValueError:
+                    raise a.Problem('日期格式不正确')
+            link = a.clean_text(data.get('sourceUrl', ''), 1500)
+            if link:
+                try:
+                    parts = urlsplit(link)
+                    if parts.scheme not in ('http', 'https') or not parts.hostname or parts.username or parts.password:
+                        raise ValueError()
+                except ValueError:
+                    raise a.Problem('资料链接须为不含账号密码的 HTTP 或 HTTPS 地址')
+            related = a.clean_text(data.get('deviceId', ''), 32)
+            source = data.get('deviceSource', '')
+            if related:
+                table = {'camera': 'assets', 'recorder': 'recorders', 'device': 'devices'}.get(source)
+                if not table:
+                    raise a.Problem('关联设备类型不正确')
+                a.HOUSEHOLDS.owns(user, a.get_object(table, related))
+                # Ordinary record editors may reference only their own authorized areas.
+                device = a.get_object(table, related)
+                if not a.HOUSEHOLDS.manager(user) and device.get('siteId') not in a.HOUSEHOLDS.site_ids(user):
+                    raise a.Problem('未获授权关联这台设备', 403)
+            fields = {k: a.clean_text(data.get(k, ''), 80) for k in ('ownerName', 'custodianName', 'successorName', 'contributor')}
+            item = dict(id=key, familyId=family, kind=kind, name=name, date=date, sourceUrl=link,
+                        content=a.clean_text(data.get('content', ''), 8000),
+                        handoverNote=a.clean_text(data.get('handoverNote', ''), 2000),
+                        deviceId=related, deviceSource=source if related else '', archived=archived,
+                        revision=old.get('revision', 0)+1, **fields,
+                        updatedAt=dt.datetime.now(dt.timezone.utc).isoformat(), updatedBy=user['name'])
+            a.save_object('heritage_entries', item)
+            a.audit('保存传承资料', user['username']+' / '+kind)
+            return item