ucvl-home-vision/inventory.py

"""Common device responsibility fields and non-video device records."""
import ipaddress
import re
class Inventory:
    def __init__(self, app):
        self.a = app

    def candidates(self, user):
        if not self.a.HOUSEHOLDS.manager(user):
            return []
        a = self.a
        users = a.HOUSEHOLDS.objects(user, 'users')
        if user['role'] == 'admin' and not any(u['id'] == user['id'] for u in users):
            users.append(user)
        result = [dict(ref='user:'+u['id'], name=u['name'], label=u['name']+' / '+u['username'], kind='账号成员')
                  for u in users if not u.get('disabled')]
        linked = {u.get('personId') for u in users if not u.get('disabled')}
        result += [dict(ref='person:'+p['id'], name=p['name'], label=p['name']+' · '+p['id'][-4:], kind='家谱人物')
                   for p in a.HOUSEHOLDS.objects(user, 'people') if not p.get('archived') and p['id'] not in linked]
        return result

    def display(self, item):
        result = dict(item)
        for prefix in ('owner', 'manager'):
            ref = item.get(prefix+'Ref', '')
            kind, _, key = ref.partition(':')
            target = self.a.get_object({'user':'users','person':'people'}.get(kind, 'users'), key) if key else None
            if target and (self.a.HOUSEHOLDS.family_of(target) == self.a.HOUSEHOLDS.family_of(item) or
                           kind == 'user' and target.get('role') == 'admin'):
                result[prefix+'Name'] = target['name']
        return result

    def metadata(self, data, old=None, user=None):
        old = old or {}
        result = {k: self.a.clean_text(data.get(k, old.get(k, '')), 80)
                  for k in ('ownerName', 'managerName')}
        candidates = {p['ref']:p for p in self.candidates(user)}
        for prefix in ('owner', 'manager'):
            field = prefix+'Ref'
            ref = self.a.clean_text(data.get(field, old.get(field, '')), 64)
            if ref:
                if ref in candidates:
                    result[prefix+'Name'] = candidates[ref]['name']
                elif ref == old.get(field):
                    result[prefix+'Name'] = self.display(old).get(prefix+'Name', '')
                else:
                    raise self.a.Problem('请选择当前家庭／场所的人员，或选择手动填写')
            result[field] = ref
        for field in ('managementHost','tailscaleHost'):
            value = self.a.clean_text(data.get(field,old.get(field,'')),253)
            if value:
                try: ipaddress.ip_address(value)
                except ValueError:
                    if re.fullmatch(r'[0-9.]+',value) or not all(re.fullmatch(r'[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?',part) for part in value.split('.')):
                        raise self.a.Problem('设备地址请填写 IP 或域名，不含账号、路径及端口')
            result[field]=value
        for field,choices in [('networkAccess',('unknown','lan','tailscale','public')),('connectionType',('unknown','wired','wifi'))]:
            result[field]=data.get(field,old.get(field,'unknown'))
            if result[field] not in choices:raise self.a.Problem('网络接入类型不正确')
        cidr=self.a.clean_text(data.get('networkCidr',old.get('networkCidr','')),64)
        if cidr:
            try:cidr=str(ipaddress.ip_network(cidr,strict=True))
            except ValueError:raise self.a.Problem('网段请使用 CIDR 格式，例如 192.168.8.0/24')
        result['networkCidr']=cidr
        uplink=self.a.clean_text(data.get('uplinkId',old.get('uplinkId','')),32)
        if uplink:
            records={x['id']:x for table in ('assets','recorders','devices') for x in self.a.HOUSEHOLDS.objects(user,table)}
            if uplink not in records:raise self.a.Problem('上级设备不属于当前家庭／场所')
            seen={data.get('id')};node=uplink
            while node:
                if node in seen:raise self.a.Problem('上级设备不能指向自己或形成循环')
                seen.add(node);node=records.get(node,{}).get('uplinkId')
        result['uplinkId']=uplink
        tags = data.get('tags', old.get('tags', []))
        if not isinstance(tags, list) or len(tags) > 12:
            raise self.a.Problem('自定义标签最多 12 个')
        result['tags'] = sorted({self.a.clean_text(tag, 40, True) for tag in tags})
        return result

    def save(self, data, user):
        a = self.a
        with a.LOCK:
            family = a.HOUSEHOLDS.prepare(user, 'devices', data, (('sites', 'siteId'),))
            key = a.object_key('devices', data)
            old = a.get_object('devices', key)
            item = {k: a.clean_text(data.get(k, ''), 300 if k == 'note' else 120, k in ('name', 'deviceType'))
                    for k in ('name', 'deviceType', 'brand', 'model', 'serial', 'note')}
            if item['deviceType'] in ('摄像头', '摄像机', '录像机'):
                raise a.Problem('视频设备请使用摄像机或录像机档案，以保留取流与通道配置')
            item.update(id=key, familyId=family, siteId=a.clean_text(data.get('siteId', ''), 32),
                        **self.metadata(data, old, user))
            a.save_object('devices', item)
            a.audit('保存设备档案', item['name'])
            return item