ucvl-home-vision/households.py

"""Household ownership and request scope. A site is a location within one household."""
import datetime as dt
import json
import re
import secrets
import hashlib
from appearance import png_bytes


class Households:
    TABLES = ('sites', 'assets', 'recorders', 'cameras', 'devices', 'people', 'family_links', 'family_events', 'users')

    def __init__(self, app):
        self.a = app

    @staticmethod
    def manager(user):
        return bool(user and user.get('role') in ('admin', 'family_admin'))

    def initialize(self):
        a = self.a
        with a.LOCK, a.DB:
            a.DB.execute('CREATE TABLE IF NOT EXISTS households (id TEXT PRIMARY KEY, body TEXT NOT NULL)')
            a.DB.execute('CREATE TABLE IF NOT EXISTS devices (id TEXT PRIMARY KEY, body TEXT NOT NULL)')
            default = a.setting('defaultHousehold')
            if not default:
                default = secrets.token_hex(8)
                a.DB.execute('INSERT INTO settings VALUES (?,?)', ('defaultHousehold', json.dumps(default)))
                item = dict(id=default, name='赵府', note='原有家庭资料', createdAt=dt.datetime.now(dt.timezone.utc).isoformat())
                a.DB.execute('INSERT INTO households VALUES (?,?)', (default, json.dumps(item)))
            # One-time ownership backfill preserves identifiers, credentials and device paths.
            for table in self.TABLES:
                for row in a.DB.execute('SELECT id,body FROM ' + table).fetchall():
                    item = json.loads(row['body'])
                    if 'familyId' not in item:
                        item['familyId'] = default
                        a.DB.execute('UPDATE ' + table + ' SET body=? WHERE id=?', (json.dumps(item), row['id']))
            for row in a.DB.execute('SELECT revision,before_json,after_json FROM family_changes').fetchall():
                for column in ('before_json', 'after_json'):
                    item = json.loads(row[column]) if row[column] else None
                    if item is not None and 'familyId' not in item:
                        item['familyId'] = default
                        a.DB.execute('UPDATE family_changes SET ' + column + '=? WHERE revision=?',
                                     (json.dumps(item), row['revision']))

    def family_of(self, item):
        # Legacy in-process import helpers remain in the migrated default household.
        return item.get('familyId', self.a.setting('defaultHousehold')) if item is not None else None

    def scope(self, user):
        family = (user or {}).get('_familyId') or self.family_of(user or {})
        if not family or not self.a.get_object('households', family):
            raise self.a.Problem('账号尚未分配有效家庭，请联系平台管理员', 403)
        return family

    def context(self, user, requested=None):
        if not user:
            return None
        own = self.family_of(user)
        if requested and user['role'] != 'admin' and requested != own:
            raise self.a.Problem('没有这个家庭的访问权限', 403)
        result = dict(user, _familyId=requested or own)
        self.scope(result)
        return result

    def owns(self, user, item):
        if not item or self.family_of(item) != self.scope(user):
            raise self.a.Problem('记录不存在或不属于当前家庭', 404)
        return item

    def objects(self, user, table):
        family = self.scope(user)
        return [x for x in self.a.objects(table) if self.family_of(x) == family]

    def site_ids(self, user):
        """Optional branch grants, always restricted to the account's own organization."""
        sites = self.objects(user, 'sites')
        ids = {s['id'] for s in sites if s['id'] in user.get('siteIds', [])}
        if user.get('includeSubspaces', False):
            children = {}
            for site in sites:
                children.setdefault(site.get('parentId'), []).append(site['id'])
            queue = list(ids)
            for parent in queue:
                for child in children.get(parent, []):
                    if child not in ids:
                        ids.add(child)
                        queue.append(child)
        return ids

    def prepare(self, user, table, body, references=()):
        family = self.scope(user)
        if body.get('familyId', family) != family:
            raise self.a.Problem('不能修改记录的家庭归属', 403)
        if body.get('id'):
            if not isinstance(body['id'], str) or not re.fullmatch('[a-f0-9]{16}', body['id']):
                raise self.a.Problem('对象编号不正确')
            self.owns(user, self.a.get_object(table, body['id']))
        for target, key in references:
            if body.get(key):
                self.owns(user, self.a.get_object(target, body[key]))
        return family

    def endpoint(self, user, host):
        family = self.scope(user)
        if any(item.get('host') == host and self.family_of(item) != family
               for table in ('cameras', 'recorders') for item in self.a.objects(table)):
            raise self.a.Problem('该设备地址已归属其他家庭，不能重复接入', 403)

    def list(self, user):
        rows = self.a.objects('households') if user['role'] == 'admin' else [self.a.get_object('households', self.scope(user))]
        def label(uid):
            account = self.a.get_object('users', uid)
            return account.get('name', '') if account else ''
        return [dict({k: v for k, v in row.items() if k != 'coverData' and (k != 'leadershipHistory' or self.manager(user))}, ownerName=label(row.get('ownerId')), successorName=label(row.get('successorId')),
                     coverUrl=('/api/households/cover?id='+row['id']+'&v='+row.get('coverRevision','0')) if row.get('coverData') else '',
                     canAssignOwner=user['role'] == 'admin' or row.get('ownerId') == user['id'],
                     members=sum(self.family_of(u) == row['id'] for u in self.a.objects('users')),
                     sites=sum(self.family_of(s) == row['id'] for s in self.a.objects('sites'))) for row in rows]

    def leadership(self, data, user):
        """Explicit handover; a backup already has management access during an emergency."""
        a = self.a
        with a.LOCK:
            home = a.get_object('households', data.get('id'))
            if not home or home['id'] != self.scope(user):
                raise a.Problem('请先进入要管理的家庭', 404)
            if user['role'] != 'admin' and user['id'] != home.get('ownerId'):
                raise a.Problem('只有一家之主或超级管理员可以交接管理权', 403)
            if data.get('revision') != home.get('leadershipRevision', 0):
                raise a.Problem('家庭负责人已变更，请重新载入', 409)
            owner_id = a.clean_text(data.get('ownerId', ''), 32, True)
            successor_id = a.clean_text(data.get('successorId', ''), 32)
            reason = a.clean_text(data.get('reason', ''), 300, True)
            if owner_id == successor_id:
                raise a.Problem('备用负责人必须是另一位家庭管理员')
            for uid in filter(None, (owner_id, successor_id)):
                account = self.owns(user, a.get_object('users', uid))
                if account['role'] != 'family_admin' or account['disabled']:
                    raise a.Problem('负责人须为本家庭已启用的家庭管理员')
            record = dict(at=dt.datetime.now(dt.timezone.utc).isoformat(), actor=user['username'], reason=reason,
                          previousOwnerId=home.get('ownerId', ''), ownerId=owner_id, successorId=successor_id)
            home.update(ownerId=owner_id, successorId=successor_id,
                        leadershipRevision=home.get('leadershipRevision', 0) + 1,
                        leadershipHistory=(home.get('leadershipHistory', []) + [record])[-50:])
            a.save_object('households', home)
            a.audit('家庭负责人交接', user['username'] + ' / ' + home['name'])
            return {'ok': True}

    def save(self, data, user):
        a = self.a
        if not self.manager(user):
            raise a.Problem('需要家庭管理员权限', 403)
        with a.LOCK:
            old = a.get_object('households', data.get('id'))
            if data.get('id') and not old:
                raise a.Problem('家庭不存在', 404)
            if user['role'] != 'admin' and (not old or old['id'] != self.scope(user)):
                raise a.Problem('只能管理自己所属的家庭', 403)
            kind = data.get('kind', old.get('kind', 'home') if old else 'home')
            if kind not in ('home', 'office', 'factory'):
                raise a.Problem('请选择家庭、办公场所或工厂')
            item = dict(old or {}, id=old['id'] if old else secrets.token_hex(8),
                        name=a.clean_text(data.get('name', ''), 80, True),
                        kind=kind,
                        note=a.clean_text(data.get('note', ''), 300),
                        createdAt=old['createdAt'] if old else dt.datetime.now(dt.timezone.utc).isoformat())
            for field in ('province','city','district'):
                item[field] = a.clean_text(data.get(field, (old or {}).get(field, '')), 60)
            if 'coverData' in data:
                cover = data['coverData']
                if not isinstance(cover, str): raise a.Problem('封面图片格式不正确')
                if cover:
                    try: png_bytes(cover, 2*1024*1024)
                    except ValueError as error: raise a.Problem(str(error).replace('Logo', '封面'))
                item['coverData'] = cover
                item['coverRevision'] = hashlib.sha256(cover.encode()).hexdigest()[:16]
            a.save_object('households', item)
            a.audit('保存家庭', item['name'])
            return item

    def revision_key(self, user):
        family = self.scope(user)
        return 'familyRevision' if family == self.a.setting('defaultHousehold') else 'familyRevision:' + family

    def save_view(self, data, user):
        if not self.manager(user):
            raise self.a.Problem('需要场所管理员权限', 403)
        groups = data.get('groups')
        if (not isinstance(groups, list) or len(groups) > 5 or
                any(g not in ('type', 'space', 'manager', 'owner', 'tag') for g in groups) or len(set(groups)) != len(groups)):
            raise self.a.Problem('分组维度不可重复，请选择类型、区域、负责人、所有人或标签')
        with self.a.LOCK:
            home = self.a.get_object('households', self.scope(user))
            home['deviceGrouping'] = groups
            self.a.save_object('households', home)
            self.a.audit('更新设备分组方式', home['name'])
            return {'ok': True}