1370 lines
55 KiB
Python
1370 lines
55 KiB
Python
# -*- coding:utf-8 -*-
|
||
"""
|
||
@Author : xuxingchen
|
||
@Contact : xuxingchen@sinochem.com
|
||
@Desc : 住户信息表 & 住户-房产关联表 增&删&改&查
|
||
"""
|
||
import csv
|
||
import os
|
||
import re
|
||
import shutil
|
||
from typing import Union, Optional, List
|
||
|
||
from pydantic import BaseModel, field_validator
|
||
|
||
from config import PREFIX_PATH, FACES_DIR_PATH, SUB_PATH
|
||
from config import APP_HOST as HOST_IP
|
||
from config import IMAGE_SERVER_PORT as PORT
|
||
from device.call import ServicesCall, DelFaceItem, AddFaceItem
|
||
from models.devices import DevicesTable
|
||
from models.houses import HousesTable
|
||
from utils import logger
|
||
from utils.database import BaseTable, get_table_handler
|
||
from utils.misc import now_datetime_second, InvalidException, is_image_valid, get_file_md5, encrypt_number, \
|
||
decrypt_number
|
||
|
||
ROLE_TYPES = ["业主", "亲属", "租户"]
|
||
|
||
|
||
class HouseholdersInfo(BaseModel):
|
||
householder_id: int
|
||
name: str
|
||
sex: str
|
||
phone: str
|
||
rooms: Union[str, list]
|
||
add_datetime: str
|
||
|
||
@field_validator("rooms")
|
||
def convert_rooms(cls, value):
|
||
return [i.strip() for i in value.split(",")]
|
||
|
||
|
||
class HouseholderDetailInfo(BaseModel):
|
||
householder_id: int
|
||
name: str
|
||
phone: str
|
||
face_url: str
|
||
rooms: list
|
||
authorized_devices: list
|
||
|
||
|
||
class AuthHouseholdersInfo(BaseModel):
|
||
householder_id: int
|
||
name: str
|
||
phone: str
|
||
rooms: Union[str, list]
|
||
|
||
@field_validator("rooms")
|
||
def convert_rooms(cls, value):
|
||
return [i.strip() for i in value.split(",")]
|
||
|
||
|
||
class AuthUsersInfo(BaseModel):
|
||
user_id: int
|
||
name: str
|
||
sex: str
|
||
phone: str
|
||
card_type: str
|
||
card_id: str
|
||
user_type: str
|
||
start_datetime: str
|
||
expire_datetime: str
|
||
|
||
|
||
class AuthUsersDetailInfo(AuthUsersInfo):
|
||
face_url: str
|
||
authorized_devices: List[dict]
|
||
|
||
|
||
class AddAuthUserInfo(BaseModel):
|
||
name: str
|
||
sex: str
|
||
phone: str
|
||
card_type: str
|
||
card_id: str
|
||
user_type: str
|
||
start_datetime: str
|
||
expire_datetime: str
|
||
face_url: str
|
||
authorized_devices: List[int]
|
||
|
||
@field_validator("sex")
|
||
def check_sex(cls, value):
|
||
types = ["男", "女"]
|
||
if value not in types:
|
||
raise InvalidException(f"请提供正确的sex类型 {types}")
|
||
else:
|
||
return value
|
||
|
||
@field_validator('phone')
|
||
def check_phone(cls, value):
|
||
pattern = re.compile(r'^(13[0-9]|14[01456879]|15[0-35-9]|16[2567]|17[0-8]|18[0-9]|19[0-35-9])\d{8}$')
|
||
if pattern.search(value) is None:
|
||
raise InvalidException("请提供正确的手机号码")
|
||
th = get_table_handler()
|
||
if HouseholdersTable.exists_user_phone(th, value):
|
||
raise InvalidException("手机号码已存在，请勿重复提交")
|
||
return value
|
||
|
||
@field_validator("user_type")
|
||
def check_user_type(cls, value):
|
||
types = ["员工", "访客", "其他"]
|
||
if value not in types:
|
||
raise InvalidException(f"请提供正确的user_type类型 {types}")
|
||
else:
|
||
return value
|
||
|
||
@field_validator("face_url")
|
||
def check_face_url(cls, value, values):
|
||
filename = value.split('/')[-1]
|
||
file_path = str(os.path.join(FACES_DIR_PATH, filename))
|
||
if is_image_valid(file_path):
|
||
if filename.startswith('_'):
|
||
new_filename = \
|
||
f"{values.data.get('name')}{encrypt_number(values.data.get('phone'))}.{filename.split('.')[-1]}"
|
||
shutil.copy(file_path, f"{FACES_DIR_PATH}/{new_filename}")
|
||
os.remove(file_path)
|
||
return new_filename
|
||
return filename
|
||
else:
|
||
raise InvalidException(f"地址或图片无效，请提供有效的图片地址 {value}")
|
||
|
||
@field_validator("authorized_devices")
|
||
def check_authorized_devices(cls, value):
|
||
if len(value) > 0:
|
||
return value
|
||
else:
|
||
raise InvalidException("请提供至少一个门禁设备权限")
|
||
|
||
|
||
class UpdateAuthUserInfo(AddAuthUserInfo):
|
||
user_id: int
|
||
phone: str
|
||
|
||
@field_validator("user_id")
|
||
def check_user_id(cls, value):
|
||
th = get_table_handler()
|
||
if HouseholdersTable.exists_user_id(th, value):
|
||
return value
|
||
else:
|
||
raise InvalidException("user_id 不存在")
|
||
|
||
@field_validator('phone')
|
||
def check_phone(cls, value):
|
||
pattern = re.compile(r'^(13[0-9]|14[01456879]|15[0-35-9]|16[2567]|17[0-8]|18[0-9]|19[0-35-9])\d{8}$')
|
||
if pattern.search(value) is None:
|
||
raise InvalidException("请提供正确的手机号码")
|
||
return value
|
||
|
||
|
||
class AddHouseholderInfo(BaseModel):
|
||
name: str
|
||
sex: str
|
||
phone: str
|
||
property_info: List[tuple[str, str]]
|
||
|
||
@field_validator('sex')
|
||
def check_sex(cls, value):
|
||
if value not in ["男", "女"]:
|
||
raise InvalidException("请提供正确的性别类型")
|
||
return value
|
||
|
||
@field_validator('phone')
|
||
def check_phone(cls, value):
|
||
pattern = re.compile(r'^(13[0-9]|14[01456879]|15[0-35-9]|16[2567]|17[0-8]|18[0-9]|19[0-35-9])\d{8}$')
|
||
if pattern.search(value) is None:
|
||
raise InvalidException("请提供正确的手机号码")
|
||
th = get_table_handler()
|
||
if HouseholdersTable.exists_householder_phone(th, value):
|
||
raise InvalidException("住户手机号码已存在，请勿重复提交")
|
||
return value
|
||
|
||
@field_validator('property_info')
|
||
def check_property_info(cls, value):
|
||
th = get_table_handler()
|
||
room_ids = []
|
||
for item in value:
|
||
if not HousesTable.exists(th, item[0]):
|
||
raise InvalidException("对应房产不存在")
|
||
if item[1] not in ROLE_TYPES:
|
||
raise InvalidException("请提供正确的身份类型")
|
||
room_ids.append(item[0])
|
||
if len(set(room_ids)) != len(room_ids):
|
||
raise InvalidException("不能存在重复房产")
|
||
return value
|
||
|
||
|
||
class GetHouseholdersInfo(BaseModel):
|
||
search_type: Optional[str] = None
|
||
search_key: Optional[str] = None
|
||
role_type: str = "全部"
|
||
limit: Optional[int] = None
|
||
page: Optional[int] = None
|
||
|
||
@field_validator("search_type")
|
||
def check_search_type(cls, value):
|
||
types = {
|
||
"姓名": "name",
|
||
"电话号码": "phone"
|
||
}
|
||
if value not in ["姓名", "电话号码"]:
|
||
raise InvalidException("请提供正确的关键字类型")
|
||
else:
|
||
return types[value]
|
||
|
||
@field_validator("search_key")
|
||
def check_search_key(cls, value, values):
|
||
if value and values.data.get("search_type", None):
|
||
return value
|
||
else:
|
||
raise InvalidException("search 参数缺失")
|
||
|
||
@field_validator("page")
|
||
def check_page(cls, value, values):
|
||
if (value is not None and value > 0
|
||
and values.data.get("limit", None) is not None and values.data.get("limit") != 0):
|
||
return value
|
||
else:
|
||
raise InvalidException("page 参数缺失或异常")
|
||
|
||
@field_validator("role_type")
|
||
def check_role_type(cls, value):
|
||
if value not in ROLE_TYPES + ["全部"]:
|
||
raise InvalidException("请提供正确的身份类型")
|
||
else:
|
||
return value
|
||
|
||
|
||
class GetAuthHouseholdersInfo(BaseModel):
|
||
search_type: Optional[str] = None
|
||
search_key: Optional[str] = None
|
||
space_type: Optional[str] = None
|
||
space_id: Optional[str] = None
|
||
limit: Optional[int] = None
|
||
page: Optional[int] = None
|
||
|
||
@field_validator("search_type")
|
||
def check_search_type(cls, value):
|
||
types = {
|
||
"住户姓名": "name",
|
||
"手机号码": "phone"
|
||
}
|
||
if value not in types.keys():
|
||
raise InvalidException(f"请提供正确的关键字类型 {list(types.keys())}")
|
||
else:
|
||
return types[value]
|
||
|
||
@field_validator("search_key")
|
||
def check_search_key(cls, value, values):
|
||
if (value or value.strip() == "") and values.data.get("search_type", None):
|
||
if value.strip() == "":
|
||
value = None
|
||
return value
|
||
else:
|
||
raise InvalidException("search 参数缺失")
|
||
|
||
@field_validator("space_type")
|
||
def check_space_type(cls, value):
|
||
types = {
|
||
"区域": "name_id",
|
||
"楼栋": "building_id",
|
||
"单元": "unit_id",
|
||
"楼层": "floor_id",
|
||
"房间号": "room_id"
|
||
}
|
||
if value not in types.keys():
|
||
raise InvalidException(f"请提供正确的关联房产类型 {list(types.keys())}")
|
||
else:
|
||
return types[value]
|
||
|
||
@field_validator("space_id")
|
||
def check_space_id(cls, value, values):
|
||
if value and values.data.get("space_type", None):
|
||
return value
|
||
else:
|
||
raise InvalidException("space 参数缺失")
|
||
|
||
@field_validator("page")
|
||
def check_page(cls, value, values):
|
||
if (value is not None and value > 0
|
||
and values.data.get("limit", None) is not None and values.data.get("limit") != 0):
|
||
return value
|
||
else:
|
||
raise InvalidException("page 参数缺失或异常")
|
||
|
||
|
||
class GetAuthUsersInfo(BaseModel):
|
||
search_type: Optional[str] = None
|
||
search_key: Optional[str] = None
|
||
user_type: Optional[str] = None
|
||
limit: Optional[int] = None
|
||
page: Optional[int] = None
|
||
|
||
@field_validator("search_type")
|
||
def check_search_type(cls, value):
|
||
types = {
|
||
"姓名": "name",
|
||
"手机号码": "phone"
|
||
}
|
||
if value not in types.keys():
|
||
raise InvalidException(f"请提供正确的关键字类型 {list(types.keys())}")
|
||
else:
|
||
return types[value]
|
||
|
||
@field_validator("search_key")
|
||
def check_search_key(cls, value, values):
|
||
if value and values.data.get("search_type", None):
|
||
return value
|
||
else:
|
||
raise InvalidException("search 参数缺失")
|
||
|
||
@field_validator("user_type")
|
||
def check_space_type(cls, value):
|
||
types = {
|
||
"员工": "员工",
|
||
"访客": "访客",
|
||
"其他": "其他"
|
||
}
|
||
if value not in types.keys():
|
||
raise InvalidException(f"请提供正确的人员类型 {list(types.keys())}")
|
||
else:
|
||
return types[value]
|
||
|
||
@field_validator("page")
|
||
def check_page(cls, value, values):
|
||
if (value is not None and value > 0
|
||
and values.data.get("limit", None) is not None and values.data.get("limit") != 0):
|
||
return value
|
||
else:
|
||
raise InvalidException("page 参数缺失或异常")
|
||
|
||
|
||
class UpdateHouseholderInfo(BaseModel):
|
||
householder_id: int
|
||
property_info: List[tuple[str, str]]
|
||
|
||
@field_validator('property_info')
|
||
def check_property_info(cls, value):
|
||
th = get_table_handler()
|
||
for item in value:
|
||
if not HousesTable.exists(th, item[0]):
|
||
raise InvalidException("对应房产不存在")
|
||
if item[1] not in ROLE_TYPES:
|
||
raise InvalidException("请提供正确的身份类型")
|
||
return value
|
||
|
||
|
||
class UpdateHouseholderFace(BaseModel):
|
||
id: int
|
||
face_url: str
|
||
|
||
@field_validator("id")
|
||
def check_id(cls, value):
|
||
th = get_table_handler()
|
||
if HouseholdersTable.exists_id(th, value):
|
||
return value
|
||
else:
|
||
raise InvalidException(f"无效ID {value}")
|
||
|
||
@field_validator("face_url")
|
||
def check_face_url(cls, value, values):
|
||
filename = value.split('/')[-1]
|
||
file_path = str(os.path.join(FACES_DIR_PATH, filename))
|
||
if is_image_valid(file_path):
|
||
# 若为临时图片，则进行规则命名
|
||
if filename.startswith('_'):
|
||
th = get_table_handler()
|
||
name_phone = HouseholdersTable.get_householder_np_by_id(th, values.data.get('id'))
|
||
new_filename = f"{name_phone['name']}{name_phone['phone']}.{filename.split('.')[-1]}"
|
||
shutil.copy(file_path, f"{FACES_DIR_PATH}/{new_filename}")
|
||
os.remove(file_path)
|
||
return new_filename
|
||
return filename
|
||
else:
|
||
raise InvalidException(f"地址或图片无效，请提供有效的图片地址 {value}")
|
||
|
||
|
||
class HouseholdersTable(BaseTable):
|
||
@staticmethod
|
||
def check(table_handler: BaseTable):
|
||
"""检测是否存在当前表"""
|
||
# householders 住户信息表
|
||
table_handler.query("SELECT name FROM sqlite_master WHERE type='table' AND name='householders'")
|
||
if table_handler.cursor.fetchone() is None:
|
||
table_handler.execute(
|
||
"""
|
||
CREATE TABLE householders (
|
||
householder_id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
name TEXT,
|
||
sex TEXT,
|
||
phone TEXT,
|
||
face_url TEXT DEFAULT '',
|
||
face_md5 TEXT,
|
||
type TEXT,
|
||
user_type TEXT,
|
||
card_type TEXT,
|
||
card_id TEXT,
|
||
auth_start TEXT,
|
||
auth_expire TEXT,
|
||
add_datetime TEXT,
|
||
update_datetime TEXT,
|
||
UNIQUE (phone, type)
|
||
)
|
||
"""
|
||
)
|
||
init_config_path = os.path.join(os.path.dirname(os.path.abspath("__file__")),
|
||
"data/InitialData/householders.csv")
|
||
if os.path.exists(init_config_path):
|
||
with open(init_config_path, newline='', encoding='utf8') as csvfile:
|
||
csvreader = csv.reader(csvfile)
|
||
head = next(csvreader)
|
||
data = []
|
||
if len(head) == 11:
|
||
for row in csvreader:
|
||
householder_id = row[0].strip()
|
||
name = row[1].strip()
|
||
sex = row[2].strip()
|
||
phone = encrypt_number(row[3].strip())
|
||
face_url = row[4].strip()
|
||
if face_url and os.path.exists(f"{FACES_DIR_PATH}/{face_url}"):
|
||
face_md5 = get_file_md5(f"{FACES_DIR_PATH}/{face_url}")
|
||
else:
|
||
face_md5 = None
|
||
_type = row[5].strip()
|
||
if _type != "住户":
|
||
user_type = row[6].strip()
|
||
card_type = row[7].strip()
|
||
card_id = row[8].strip()
|
||
auth_start = row[9].strip()
|
||
auth_expire = row[10].strip()
|
||
else:
|
||
user_type, card_type, card_id, auth_start, auth_expire = None, None, None, None, None
|
||
data.append((householder_id, name, sex, phone, face_url, face_md5,
|
||
_type, user_type, card_type, card_id, auth_start, auth_expire,
|
||
now_datetime_second(), now_datetime_second()))
|
||
table_handler.executemany(
|
||
"""
|
||
INSERT OR IGNORE INTO householders
|
||
(householder_id, name, sex, phone, face_url, face_md5,
|
||
type, user_type, card_type, card_id, auth_start, auth_expire,
|
||
add_datetime, update_datetime)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""", data
|
||
)
|
||
|
||
# householders_type 住户-房产关联表
|
||
table_handler.query("SELECT name FROM sqlite_master WHERE type='table' AND name='householders_type'")
|
||
if table_handler.cursor.fetchone() is None:
|
||
table_handler.execute(
|
||
"""
|
||
CREATE TABLE householders_type (
|
||
householder_id INT,
|
||
room_id TEXT,
|
||
type TEXT,
|
||
update_datetime TEXT,
|
||
UNIQUE (householder_id, room_id)
|
||
)
|
||
"""
|
||
)
|
||
init_config_path = os.path.join(os.path.dirname(os.path.abspath("__file__")),
|
||
"data/InitialData/householders_type.csv")
|
||
if os.path.exists(init_config_path):
|
||
with open(init_config_path, newline='', encoding='utf8') as csvfile:
|
||
csvreader = csv.reader(csvfile)
|
||
head = next(csvreader)
|
||
data = []
|
||
count_info = {}
|
||
if len(head) == 3:
|
||
for row in csvreader:
|
||
householder_id = row[0].strip()
|
||
room_id = row[1].strip()
|
||
_type = row[2].strip()
|
||
if not HouseholdersTable.exists_householder_type(table_handler, householder_id, room_id):
|
||
if room_id in count_info.keys():
|
||
count_info[room_id] += 1
|
||
else:
|
||
count_info[room_id] = 1
|
||
data.append((int(householder_id), room_id, _type, now_datetime_second()))
|
||
table_handler.executemany(
|
||
"""
|
||
INSERT INTO householders_type
|
||
(householder_id, room_id, type, update_datetime)
|
||
VALUES (?, ?, ?, ?)
|
||
ON CONFLICT (householder_id, room_id) DO NOTHING
|
||
""", data
|
||
)
|
||
for _id, _count in count_info.items():
|
||
old_count = HousesTable.get_householder_count(table_handler, _id)
|
||
HousesTable.update_householder_count(table_handler, _id, old_count + _count)
|
||
|
||
@staticmethod
|
||
def insert(table_handler: BaseTable, name: str, sex: str, phone: str, _type: str):
|
||
try:
|
||
table_handler.execute(
|
||
"""
|
||
INSERT OR IGNORE INTO householders
|
||
(name, sex, phone, type, add_datetime, update_datetime)
|
||
VALUES (?, ?, ?, ?, ?, ?)
|
||
""", (name, sex, encrypt_number(phone), _type, now_datetime_second(), now_datetime_second())
|
||
)
|
||
finally:
|
||
table_handler.query(
|
||
"""
|
||
SELECT householder_id
|
||
FROM householders
|
||
WHERE phone = ?
|
||
""", (encrypt_number(phone),)
|
||
)
|
||
res = table_handler.cursor.fetchall()
|
||
if res:
|
||
return res[0][0]
|
||
else:
|
||
return None
|
||
|
||
@staticmethod
|
||
def insert_type(table_handler: BaseTable, room_id: str, householder_id: int, _type: str):
|
||
table_handler.execute(
|
||
"""
|
||
INSERT INTO householders_type
|
||
(householder_id, room_id, type, update_datetime)
|
||
VALUES (?, ?, ?, ?)
|
||
ON CONFLICT (householder_id, room_id) DO UPDATE
|
||
SET type = ?, update_datetime = ?
|
||
""", (householder_id, room_id, _type, now_datetime_second(), _type, now_datetime_second())
|
||
)
|
||
|
||
@staticmethod
|
||
def add_auth_user_info(table_handler: BaseTable, obj: AddAuthUserInfo):
|
||
"""插入通用人员信息 & 进行通用人员设备授权"""
|
||
now_datetime = now_datetime_second()
|
||
face_md5 = get_file_md5(f"{FACES_DIR_PATH}/{obj.face_url}")
|
||
table_handler.execute(
|
||
"""
|
||
INSERT OR IGNORE INTO householders
|
||
(name, sex, phone, face_url, face_md5, type,
|
||
user_type, card_type, card_id,
|
||
auth_start, auth_expire, add_datetime, update_datetime)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""",
|
||
(obj.name, obj.sex, encrypt_number(obj.phone), obj.face_url, face_md5, "非住户",
|
||
obj.user_type, obj.card_type, obj.card_id,
|
||
obj.start_datetime, obj.expire_datetime, now_datetime, now_datetime)
|
||
)
|
||
user_id = HouseholdersTable.get_householder_id(table_handler, obj.phone, '非住户')
|
||
failed_ids = []
|
||
codes = []
|
||
msgs = []
|
||
if user_id:
|
||
try:
|
||
# 人脸下置，设备授权
|
||
sc = ServicesCall()
|
||
face_item = AddFaceItem(
|
||
user_id=user_id,
|
||
name=obj.name,
|
||
phone_number=obj.phone,
|
||
face_url=obj.face_url,
|
||
start_date=obj.start_datetime.replace(' ', 'T') + ':00.000Z',
|
||
expire_date=obj.expire_datetime.replace(' ', 'T') + ':00.000Z',
|
||
device_ids=""
|
||
)
|
||
for device_id in obj.authorized_devices:
|
||
face_item.device_ids = f"[{device_id}]"
|
||
_callback, code, msg = sc.add_face(device_id, face_item)
|
||
if _callback:
|
||
DevicesTable.insert_device_auth_record(table_handler, device_id, face_item.user_id,
|
||
face_item.start_date, face_item.expire_date)
|
||
else:
|
||
failed_ids.append(device_id)
|
||
codes.append(code)
|
||
msgs.append(msg)
|
||
finally:
|
||
if 0 < len(failed_ids):
|
||
table_handler.execute(
|
||
"""
|
||
DELETE FROM householders
|
||
WHERE householder_id = ?
|
||
""", (user_id,)
|
||
)
|
||
return {"status": False, "message": f"部分设备 - {failed_ids} 人脸授权失败，错误码：{codes}，{msgs}"}
|
||
return {"status": True}
|
||
|
||
@staticmethod
|
||
def get_householder_id(table_handler: BaseTable, phone: str, _type: str) -> str:
|
||
table_handler.query("SELECT householder_id FROM householders WHERE phone=? AND type=?",
|
||
(encrypt_number(phone), _type))
|
||
res = table_handler.cursor.fetchall()
|
||
if res:
|
||
return res[0][0]
|
||
else:
|
||
raise None
|
||
|
||
@staticmethod
|
||
def get_householder_info(table_handler: BaseTable, search_type, search_key, role_type,
|
||
page: Optional[int] = None, limit: Optional[int] = None):
|
||
sub_sql_list = []
|
||
if role_type != "全部":
|
||
sub_sql_list.append(f"types like '%{role_type}%'")
|
||
if search_type is not None:
|
||
if search_type == "name":
|
||
sub_sql_list.append(f"name like '%{search_key}%'")
|
||
else:
|
||
sub_sql_list.append(f"{search_type} = '{search_key}'")
|
||
|
||
if sub_sql_list:
|
||
sub_sql = "WHERE " + " AND ".join(sub_sql_list)
|
||
else:
|
||
sub_sql = ""
|
||
if page is not None and limit is not None:
|
||
page_sql = f"LIMIT {limit} OFFSET {(page - 1) * limit}"
|
||
else:
|
||
page_sql = ""
|
||
query_sql = f"""
|
||
SELECT * FROM (
|
||
SELECT h.householder_id,
|
||
h.name,
|
||
h.sex,
|
||
h.phone,
|
||
GROUP_CONCAT(houses.house_name) AS rooms,
|
||
GROUP_CONCAT(hh.type) AS types,
|
||
h.add_datetime
|
||
FROM householders h
|
||
LEFT JOIN householders_type hh ON h.householder_id = hh.householder_id
|
||
LEFT JOIN houses ON hh.room_id = houses.room_id
|
||
WHERE h.type = '住户'
|
||
GROUP BY h.householder_id, h.name, h.sex, h.phone, h.add_datetime
|
||
) {sub_sql}
|
||
"""
|
||
table_handler.query(f"SELECT COUNT(*) FROM ({query_sql})")
|
||
total = table_handler.cursor.fetchall()[0][0]
|
||
table_handler.query(f"{query_sql} {page_sql}")
|
||
res = table_handler.cursor.fetchall()
|
||
if res:
|
||
householder_info = []
|
||
for i in res:
|
||
householder_info.append(HouseholdersInfo(
|
||
householder_id=i[0],
|
||
name=i[1],
|
||
sex=i[2],
|
||
phone=decrypt_number(i[3]),
|
||
rooms=i[4],
|
||
add_datetime=i[6]
|
||
).__dict__)
|
||
return {"info": householder_info, "total": total}
|
||
else:
|
||
return {"info": [], "total": total}
|
||
|
||
@staticmethod
|
||
def get_auth_householder_info(table_handler: BaseTable,
|
||
search_type: Optional[str], search_key: Optional[str],
|
||
space_type: Optional[str], space_id: Optional[str],
|
||
page: Optional[int] = None, limit: Optional[int] = None):
|
||
sub_sql_list = []
|
||
if search_type is not None and search_key is not None:
|
||
if search_type == "name":
|
||
sub_sql_list.append(f"{search_type} like '%{search_key}%'")
|
||
else:
|
||
sub_sql_list.append(f"{search_type} = '{search_key}'")
|
||
if space_type is not None and space_id is not None:
|
||
space_sub_sql = f"""
|
||
AND h.householder_id IN (
|
||
SELECT DISTINCT h1.householder_id
|
||
FROM householders h1
|
||
LEFT JOIN householders_type hh1 ON h1.householder_id = hh1.householder_id
|
||
LEFT JOIN houses houses1 ON hh1.room_id = houses1.room_id
|
||
WHERE houses1.{space_type} = '{space_id}'
|
||
)
|
||
"""
|
||
else:
|
||
space_sub_sql = ""
|
||
if sub_sql_list:
|
||
sub_sql = "WHERE " + " AND ".join(sub_sql_list)
|
||
else:
|
||
sub_sql = ""
|
||
if page is not None and limit is not None:
|
||
page_sql = f"LIMIT {limit} OFFSET {(page - 1) * limit}"
|
||
else:
|
||
page_sql = ""
|
||
query_sql = f"""
|
||
SELECT * FROM (
|
||
SELECT h.householder_id, h.name, h.phone,
|
||
GROUP_CONCAT(houses.house_name || '（' || hh.type || '）') AS rooms
|
||
FROM householders h
|
||
LEFT JOIN householders_type hh ON h.householder_id = hh.householder_id
|
||
LEFT JOIN houses ON hh.room_id = houses.room_id
|
||
WHERE h.type = '住户' {space_sub_sql}
|
||
GROUP BY h.householder_id, h.name, h.phone
|
||
) {sub_sql}
|
||
"""
|
||
table_handler.query(f"SELECT COUNT(*) FROM ({query_sql})")
|
||
total = table_handler.cursor.fetchall()[0][0]
|
||
table_handler.query(f"{query_sql} {page_sql}")
|
||
res = table_handler.cursor.fetchall()
|
||
if res:
|
||
householder_info = []
|
||
for i in res:
|
||
householder_info.append(AuthHouseholdersInfo(
|
||
householder_id=i[0],
|
||
name=i[1],
|
||
phone=decrypt_number(i[2]),
|
||
rooms=i[3]
|
||
).__dict__)
|
||
return {"info": householder_info, "total": total}
|
||
else:
|
||
return {"info": [], "total": total}
|
||
|
||
@staticmethod
|
||
def get_auth_users_info(table_handler: BaseTable,
|
||
search_type: Optional[str], search_key: Optional[str], user_type: Optional[str],
|
||
page: Optional[int] = None, limit: Optional[int] = None):
|
||
sub_sql_list = []
|
||
if search_type is not None:
|
||
if search_type == "name":
|
||
sub_sql_list.append(f"{search_type} like '%{search_key}%'")
|
||
else:
|
||
sub_sql_list.append(f"{search_type} = '{search_key}'")
|
||
if user_type is not None:
|
||
sub_sql_list.append(f"user_type = '{user_type}'")
|
||
if sub_sql_list:
|
||
sub_sql = "AND " + " AND ".join(sub_sql_list)
|
||
else:
|
||
sub_sql = ""
|
||
if page is not None and limit is not None:
|
||
page_sql = f"LIMIT {limit} OFFSET {(page - 1) * limit}"
|
||
else:
|
||
page_sql = ""
|
||
query_sql = f"""
|
||
SELECT householder_id, name, sex, phone, card_type, card_id, user_type, auth_start, auth_expire
|
||
FROM householders
|
||
WHERE type != '住户' {sub_sql}
|
||
"""
|
||
table_handler.query(f"SELECT COUNT(*) FROM ({query_sql})")
|
||
total = table_handler.cursor.fetchall()[0][0]
|
||
table_handler.query(f"{query_sql} {page_sql}")
|
||
res = table_handler.cursor.fetchall()
|
||
if res:
|
||
householder_info = []
|
||
for i in res:
|
||
householder_info.append(AuthUsersInfo(
|
||
user_id=i[0], name=i[1], sex=i[2], phone=decrypt_number(i[3]),
|
||
card_type=i[4], card_id=i[5], user_type=i[6],
|
||
start_datetime=i[7], expire_datetime=i[8]
|
||
).__dict__)
|
||
return {"info": householder_info, "total": total}
|
||
else:
|
||
return {"info": [], "total": total}
|
||
|
||
@staticmethod
|
||
def get_auth_user_info(table_handler: BaseTable, householder_id: int):
|
||
query_sql = """
|
||
SELECT householder_id, name, sex, phone, face_url, card_type, card_id, user_type, auth_start, auth_expire
|
||
FROM householders
|
||
WHERE type != '住户' AND householder_id = ?
|
||
"""
|
||
table_handler.query(query_sql, (householder_id,))
|
||
res = table_handler.cursor.fetchall()
|
||
if res:
|
||
householder_info = AuthUsersDetailInfo(
|
||
user_id=res[0][0], name=res[0][1], sex=res[0][2], phone=decrypt_number(res[0][3]),
|
||
face_url=f"http://{HOST_IP}:{PORT}{PREFIX_PATH}/{SUB_PATH}/{res[0][4]}" if res[0][4] else '',
|
||
card_type=res[0][5], card_id=res[0][6], user_type=res[0][7],
|
||
start_datetime=res[0][8], expire_datetime=res[0][9],
|
||
authorized_devices=DevicesTable.get_auth_device_info(table_handler, str(res[0][0]))
|
||
).__dict__
|
||
return householder_info
|
||
else:
|
||
raise InvalidException(f"无效ID {householder_id}")
|
||
|
||
@staticmethod
|
||
def get_room_ids(table_handler: BaseTable, householder_id):
|
||
"""获取住户的所有房产ID"""
|
||
table_handler.query(
|
||
"""
|
||
SELECT room_id
|
||
FROM householders_type
|
||
WHERE householder_id = ?
|
||
""", (householder_id,)
|
||
)
|
||
res = table_handler.cursor.fetchall()
|
||
if res:
|
||
return [i[0] for i in res]
|
||
else:
|
||
return []
|
||
|
||
@staticmethod
|
||
def get_rooms_by_householder_id(table_handler: BaseTable, householder_id):
|
||
"""获取住户的所有房产ID"""
|
||
table_handler.query(
|
||
"""
|
||
SELECT h.house_name, type, ht.room_id
|
||
FROM householders_type ht
|
||
LEFT JOIN houses h ON h.room_id = ht.room_id
|
||
WHERE householder_id = ?
|
||
""", (householder_id,)
|
||
)
|
||
res = table_handler.cursor.fetchall()
|
||
if res:
|
||
return [list(i) for i in res]
|
||
else:
|
||
return []
|
||
|
||
@staticmethod
|
||
def get_householder_np_by_id(table_handler: BaseTable, householder_id: int):
|
||
table_handler.query(
|
||
"""
|
||
SELECT name, phone
|
||
FROM householders
|
||
WHERE householder_id = ?
|
||
""", (householder_id,)
|
||
)
|
||
res = table_handler.cursor.fetchall()
|
||
if res:
|
||
return {"name": res[0][0], "phone": decrypt_number(res[0][1])}
|
||
else:
|
||
raise InvalidException(f"无效ID {householder_id}")
|
||
|
||
@staticmethod
|
||
def get_face_url(table_handler: BaseTable, householder_id: int):
|
||
table_handler.query(
|
||
"""
|
||
SELECT DISTINCT face_url
|
||
FROM householders
|
||
WHERE householder_id = ?
|
||
""", (householder_id,)
|
||
)
|
||
res = table_handler.cursor.fetchall()
|
||
if res:
|
||
return res[0][0]
|
||
else:
|
||
return ""
|
||
|
||
@staticmethod
|
||
def get_face_md5(table_handler: BaseTable, householder_id: int):
|
||
table_handler.query(
|
||
"""
|
||
SELECT DISTINCT face_md5
|
||
FROM householders
|
||
WHERE householder_id = ?
|
||
""", (householder_id,)
|
||
)
|
||
res = table_handler.cursor.fetchall()
|
||
if res:
|
||
return res[0][0]
|
||
else:
|
||
return ""
|
||
|
||
@staticmethod
|
||
def get_householder_info_by_id(table_handler: BaseTable, householder_id: int):
|
||
table_handler.query(
|
||
"""
|
||
SELECT name, sex, phone
|
||
FROM householders
|
||
WHERE householder_id = ?
|
||
""", (householder_id,)
|
||
)
|
||
res = table_handler.cursor.fetchall()
|
||
if res:
|
||
householder_info = {
|
||
"householder_id": householder_id,
|
||
"name": res[0][0],
|
||
"sex": res[0][1],
|
||
"phone": decrypt_number(res[0][2]),
|
||
"property_info": []
|
||
}
|
||
table_handler.query(
|
||
"""
|
||
SELECT householders_type.room_id, house_name, area_name, building_name, unit_name,
|
||
householders_type.type, householder_count
|
||
FROM householders_type
|
||
LEFT JOIN houses ON houses.room_id = householders_type.room_id
|
||
WHERE householder_id = ?
|
||
""", (householder_id,)
|
||
)
|
||
res = table_handler.cursor.fetchall()
|
||
if res:
|
||
for i in res:
|
||
householder_info["property_info"].append({
|
||
"room_id": i[0],
|
||
"room_name": i[1],
|
||
"area_name": i[2] if i[2] else "",
|
||
"building_name": i[3] if i[3] else "",
|
||
"unit_name": i[4] if i[4] else "",
|
||
"type": i[5],
|
||
"household_count": i[6]
|
||
})
|
||
return householder_info
|
||
return {}
|
||
|
||
@staticmethod
|
||
def get_auth_householder_detail_info(table_handler: BaseTable, householder_id: int):
|
||
info = HouseholdersTable.get_householder_info_by_id(table_handler, householder_id)
|
||
rooms = HouseholdersTable.get_rooms_by_householder_id(table_handler, info["householder_id"])
|
||
face_url = HouseholdersTable.get_face_url(table_handler, householder_id)
|
||
device_ids = DevicesTable.get_auth_device_ids(table_handler, str(householder_id))
|
||
authorized_devices = []
|
||
for device_id in device_ids:
|
||
start_date, expire_date = DevicesTable.get_auth_interval(table_handler, str(householder_id), device_id)
|
||
authorized_devices.append({
|
||
"device_name": DevicesTable.get_device_name(table_handler, device_id),
|
||
"device_type": DevicesTable.get_device_scope_type(table_handler, device_id),
|
||
"open_type": ["人脸"],
|
||
"start_date": start_date,
|
||
"expire_date": expire_date,
|
||
"method": "自动授权",
|
||
"from": "系统"
|
||
})
|
||
|
||
resp = HouseholderDetailInfo(
|
||
householder_id=info["householder_id"],
|
||
name=info["name"],
|
||
phone=info["phone"],
|
||
face_url=f"http://{HOST_IP}:{PORT}{PREFIX_PATH}/{SUB_PATH}/{face_url}",
|
||
rooms=rooms,
|
||
authorized_devices=authorized_devices
|
||
)
|
||
return resp.__dict__
|
||
|
||
@staticmethod
|
||
def update_householder_info(table_handler: BaseTable,
|
||
householder_id: int,
|
||
property_info: List[tuple[str, str]]):
|
||
# 查询对应住户关联的房产信息，比对更新数据中的房产信息
|
||
update_room_ids = [i[0] for i in property_info]
|
||
update_types = [i[1] for i in property_info]
|
||
table_handler.query(
|
||
"""
|
||
SELECT room_id, type
|
||
FROM householders_type
|
||
WHERE householder_id = ?
|
||
""", (householder_id,)
|
||
)
|
||
res = table_handler.cursor.fetchall()
|
||
if res:
|
||
update_data = []
|
||
delete_data = []
|
||
insert_data = [[v, update_types[i]] for i, v in enumerate(update_room_ids)]
|
||
sqls = []
|
||
params = []
|
||
update_count = []
|
||
for i in res:
|
||
try:
|
||
index = update_room_ids.index(i[0])
|
||
if update_types[index] != i[1]:
|
||
update_data.append([i[0], update_types[index]])
|
||
insert_data.remove([i[0], update_types[index]])
|
||
except ValueError:
|
||
delete_data.append(i[0])
|
||
table_handler.query(
|
||
"""
|
||
SELECT name, phone, face_url, auth_start, auth_expire
|
||
FROM householders
|
||
WHERE householder_id = ?
|
||
""", (householder_id,)
|
||
)
|
||
householder_info = table_handler.cursor.fetchall()[0]
|
||
sc = ServicesCall()
|
||
face_item = AddFaceItem(
|
||
user_id=householder_id,
|
||
name=householder_info[0],
|
||
phone_number=householder_info[1],
|
||
face_url=householder_info[2],
|
||
start_date=householder_info[3].replace(' ', 'T') + ':00.000Z',
|
||
expire_date=householder_info[4].replace(' ', 'T') + ':00.000Z',
|
||
device_ids=""
|
||
)
|
||
if update_data:
|
||
for data in update_data:
|
||
sqls.append(
|
||
"""
|
||
UPDATE householders_type
|
||
SET type = ?, update_datetime = ?
|
||
WHERE householder_id = ? AND room_id = ?
|
||
"""
|
||
)
|
||
params.append((data[1], now_datetime_second(), householder_id, data[0]))
|
||
if insert_data:
|
||
for data in insert_data:
|
||
sqls.append(
|
||
"""
|
||
INSERT INTO householders_type
|
||
(room_id, householder_id, type, update_datetime)
|
||
VALUES (?, ?, ?, ?)
|
||
"""
|
||
)
|
||
params.append((data[0], householder_id, data[1], now_datetime_second()))
|
||
old_count = HousesTable.get_householder_count(table_handler, data[0])
|
||
update_count.append([data[0], old_count + 1])
|
||
# TODO 同步增加对应设备中的人脸信息
|
||
unit_id = HousesTable.get_unit_id_by_room_id(table_handler, data[0])
|
||
device_ids = DevicesTable.get_device_ids_by_unit_id(table_handler, unit_id)
|
||
for device_id in device_ids:
|
||
face_item.device_ids = f"[{device_id}]"
|
||
_callback, code, msg = sc.add_face(device_id, face_item)
|
||
if _callback:
|
||
DevicesTable.insert_device_auth_record(table_handler, device_id, face_item.user_id,
|
||
face_item.start_date, face_item.expire_date)
|
||
logger.Logger.error(f"[update_householder_info] "
|
||
f"住户[{householder_id}]人脸增加失败，失败代码：{code}，提示信息：{msg}")
|
||
if delete_data:
|
||
for room_id in delete_data:
|
||
sqls.append(
|
||
"""
|
||
DELETE FROM householders_type
|
||
WHERE householder_id = ? AND room_id = ?
|
||
"""
|
||
)
|
||
params.append((householder_id, room_id))
|
||
old_count = HousesTable.get_householder_count(table_handler, room_id)
|
||
update_count.append([room_id, old_count - 1])
|
||
# TODO 获取房产-单元 同步移除对应设备中的人脸
|
||
unit_id = HousesTable.get_unit_id_by_room_id(table_handler, room_id)
|
||
device_ids = DevicesTable.get_device_ids_by_unit_id(table_handler, unit_id)
|
||
for device_id in device_ids:
|
||
_callback, code, msg = sc.del_face(device_id,
|
||
DelFaceItem(user_id=householder_id, device_ids=f"[{device_id}]"))
|
||
if _callback:
|
||
DevicesTable.delete_invalid_auth_record(table_handler, device_id, str(householder_id))
|
||
logger.Logger.error(f"[update_householder_info] "
|
||
f"住户[{householder_id}]人脸移除失败，失败代码：{code}，提示信息：{msg}")
|
||
if len(update_room_ids) == 0:
|
||
sqls.append(
|
||
"""
|
||
DELETE FROM householders
|
||
WHERE householder_id = ?
|
||
"""
|
||
)
|
||
params.append((householder_id,))
|
||
table_handler.execute(sqls, params)
|
||
for i in update_count:
|
||
HousesTable.update_householder_count(table_handler, i[0], i[1])
|
||
return {"status": True}
|
||
else:
|
||
return {"status": False, "message": "不存在对应住户或住户名下无关联房产"}
|
||
|
||
@staticmethod
|
||
def update_auth_user_info(table_handler: BaseTable, obj: UpdateAuthUserInfo):
|
||
# 通用人员授权信息更新
|
||
nd = now_datetime_second().replace(' ', 'T') + '.000Z'
|
||
sd = obj.start_datetime.replace(' ', 'T') + ':00.000Z'
|
||
ed = obj.expire_datetime.replace(' ', 'T') + ':00.000Z'
|
||
sc = ServicesCall()
|
||
face_item = AddFaceItem(
|
||
user_id=obj.user_id,
|
||
name=obj.name,
|
||
phone_number=obj.phone,
|
||
face_url=obj.face_url,
|
||
start_date=sd,
|
||
expire_date=ed,
|
||
device_ids=""
|
||
)
|
||
old_face_url = HouseholdersTable.get_face_url(table_handler, obj.user_id)
|
||
# 查询当前人员授权情况
|
||
table_handler.query(
|
||
"""
|
||
SELECT device_id, start_date, expire_date
|
||
FROM devices_auth
|
||
WHERE _id = ?
|
||
""", (obj.user_id,)
|
||
)
|
||
res = table_handler.cursor.fetchall()
|
||
if res:
|
||
delete_data = [] # 移除权限
|
||
insert_data = [[device_id, sd, ed] for device_id in obj.authorized_devices] # 新增权限
|
||
for i in res:
|
||
if old_face_url == obj.face_url: # 对比人脸图片是否变更
|
||
# 人脸无实际变更，根据权限变更重新赋权
|
||
logger.Logger.debug("新旧人脸一致，按需赋权")
|
||
if [i[0], sd, ed] in insert_data and ed == i[2] and (sd < nd or sd == i[1]):
|
||
insert_data.remove([i[0], sd, ed])
|
||
else:
|
||
delete_data.append(i[0])
|
||
else:
|
||
# 若人脸变更直接移除所有人脸授权，重新赋权
|
||
delete_data.append(i[0])
|
||
delete_failed_ids = []
|
||
delete_failed_codes = []
|
||
delete_failed_msgs = []
|
||
message = ""
|
||
if delete_data:
|
||
for device_id in delete_data:
|
||
_callback, code, msg = sc.del_face(device_id, DelFaceItem(
|
||
user_id=obj.user_id, device_ids=f"[{device_id}]"))
|
||
if _callback:
|
||
DevicesTable.delete_invalid_auth_record(table_handler, device_id, face_item.user_id)
|
||
else:
|
||
delete_failed_ids.append(device_id)
|
||
delete_failed_codes.append(code)
|
||
delete_failed_msgs.append(msg)
|
||
if 0 < len(delete_failed_ids) < len(delete_data):
|
||
message += f"部分设备 - {delete_failed_ids} 人脸移除失败，错误码：{delete_failed_codes}, {delete_failed_msgs}"
|
||
elif len(delete_failed_ids) == len(delete_data) != 0:
|
||
raise InvalidException("设备授权无法更新：人脸移除失败")
|
||
insert_failed_ids = []
|
||
insert_failed_codes = []
|
||
if insert_data:
|
||
for data in insert_data:
|
||
face_item.device_ids = f"[{data[0]}]"
|
||
_callback, code, msg = sc.add_face(data[0], face_item)
|
||
if _callback:
|
||
DevicesTable.insert_device_auth_record(table_handler, data[0], face_item.user_id,
|
||
face_item.start_date, face_item.expire_date)
|
||
else:
|
||
insert_failed_ids.append(data[0])
|
||
insert_failed_codes.append(code)
|
||
if 0 < len(insert_failed_ids) < len(insert_data):
|
||
message += f"部分设备 - {delete_failed_ids} 人脸授权失败，错误码：{delete_failed_codes}"
|
||
elif len(delete_failed_ids) == len(delete_data) != 0:
|
||
raise InvalidException("设备授权无法更新：人脸下发失败")
|
||
if message != '':
|
||
return InvalidException(message)
|
||
else:
|
||
table_handler.execute(
|
||
"""
|
||
UPDATE householders
|
||
SET name = ?, sex = ?, phone= ?,
|
||
face_url = ?, face_md5= ?,
|
||
user_type = ?, card_type = ?, card_id = ?,
|
||
auth_start = ?, auth_expire = ?, update_datetime = ?
|
||
WHERE householder_id = ?
|
||
""", (obj.name, obj.sex, encrypt_number(obj.phone),
|
||
obj.face_url, get_file_md5(f"{FACES_DIR_PATH}/{obj.face_url}"),
|
||
obj.user_type, obj.card_type, obj.card_id,
|
||
sd, ed, now_datetime_second(), obj.user_id)
|
||
)
|
||
return {"status": True}
|
||
else:
|
||
return {"status": False, "message": "不存在对应ID的授权"}
|
||
|
||
@staticmethod
|
||
def update_householder_face(table_handler: BaseTable, householder_id: int, face_url: str):
|
||
"""更新住户信息中的人脸地址 & 人脸地址同步到权限内的设备"""
|
||
new_md5 = get_file_md5(f"{FACES_DIR_PATH}/{face_url}")
|
||
if new_md5 != HouseholdersTable.get_face_md5(table_handler, householder_id):
|
||
table_handler.execute(
|
||
"""
|
||
UPDATE householders
|
||
SET face_url = ?, update_datetime = ? WHERE householder_id = ?
|
||
""", (face_url, now_datetime_second(), householder_id)
|
||
)
|
||
# 查询住户关联单元，获取单元权限内的设备，下发人脸
|
||
table_handler.query(
|
||
"""
|
||
SELECT h.householder_id, name, phone, GROUP_CONCAT(unit_id) as unit_ids
|
||
FROM householders h
|
||
LEFT JOIN householders_type ht ON ht.householder_id = h.householder_id
|
||
LEFT JOIN houses ON ht.room_id = houses.room_id
|
||
WHERE h.type = '住户' AND h.householder_id = ?
|
||
""", (householder_id,)
|
||
)
|
||
res = table_handler.cursor.fetchall()
|
||
if res:
|
||
sc = ServicesCall()
|
||
face_item = AddFaceItem(
|
||
user_id=res[0][0],
|
||
name=res[0][1],
|
||
phone_number=decrypt_number(res[0][2]),
|
||
face_url=face_url,
|
||
device_ids=""
|
||
)
|
||
for unit_id in res[0][3].split(','):
|
||
device_ids = DevicesTable.get_device_ids_by_unit_id(table_handler, unit_id)
|
||
for device_id in device_ids:
|
||
face_item.device_ids = f"[{device_id}]"
|
||
_callback, code, msg = sc.add_face(device_id, face_item)
|
||
if _callback:
|
||
DevicesTable.insert_device_auth_record(table_handler, device_id, face_item.user_id,
|
||
face_item.start_date, face_item.expire_date)
|
||
return {"status": True}
|
||
else:
|
||
logger.Logger.debug("新旧人脸一致，不做任何变更")
|
||
return {"status": True}
|
||
|
||
@staticmethod
|
||
def update_user_info(table_handler: BaseTable, householder_id: int, face_url: str):
|
||
"""更新通用人员信息中的人脸地址 & 人脸地址同步到下发过的设备中"""
|
||
table_handler.execute(
|
||
"""
|
||
UPDATE householders
|
||
SET face_url = ? WHERE householder_id = ?
|
||
""", (face_url, householder_id)
|
||
)
|
||
# 查询人员授权过的设备，下发人脸
|
||
device_ids = DevicesTable.get_auth_device_ids(table_handler, str(householder_id))
|
||
name_phone = HouseholdersTable.get_householder_np_by_id(table_handler, householder_id)
|
||
sc = ServicesCall()
|
||
face_item = AddFaceItem(
|
||
user_id=householder_id,
|
||
name=name_phone["name"],
|
||
phone_number=name_phone["phone"],
|
||
face_url=face_url,
|
||
device_ids=""
|
||
)
|
||
for device_id in device_ids:
|
||
face_item.device_ids = f"[{device_id}]"
|
||
_callback, code, _ = sc.add_face(device_id, face_item)
|
||
if _callback:
|
||
DevicesTable.insert_device_auth_record(table_handler, device_id, face_item.user_id,
|
||
face_item.start_date, face_item.expire_date)
|
||
return {"status": True}
|
||
|
||
@staticmethod
|
||
def delete_householder_info(table_handler: BaseTable, householder_id: int):
|
||
# 确认住户ID有效
|
||
if not HouseholdersTable.exists_householder_id(table_handler, householder_id):
|
||
raise InvalidException("不存在对应住户ID")
|
||
# 1. 查询当前住户授权的授权的设备ID
|
||
device_ids = DevicesTable.get_auth_device_ids(table_handler, str(householder_id))
|
||
# 2. 对所有授权设备进行授权移除操作
|
||
if device_ids:
|
||
sc = ServicesCall()
|
||
_callback, code, msg = sc.del_face(device_ids[0], DelFaceItem(
|
||
user_id=str(householder_id),
|
||
device_ids=str(device_ids)
|
||
))
|
||
if not _callback:
|
||
raise InvalidException(f"住户人脸授权移除失败，错误码{code}，{msg}")
|
||
DevicesTable.delete_householder_all_auth(table_handler, str(householder_id))
|
||
# 3. 同步更新住户名下房产入住人数，并移除住户记录
|
||
room_ids = HouseholdersTable.get_room_ids(table_handler, householder_id)
|
||
for room_id in room_ids:
|
||
old_count = HousesTable.get_householder_count(table_handler, room_id)
|
||
HousesTable.update_householder_count(table_handler, room_id, old_count - 1)
|
||
sqls = [
|
||
"DELETE FROM householders WHERE householder_id = ?",
|
||
"DELETE FROM householders_type WHERE householder_id = ?"
|
||
]
|
||
table_handler.execute(sqls, [(householder_id,), (householder_id,)])
|
||
return {"status": True}
|
||
|
||
@staticmethod
|
||
def delete_auth_user_info(table_handler: BaseTable, user_id: int):
|
||
# 确认通用用户ID有效
|
||
if not HouseholdersTable.exists_user_id(table_handler, user_id):
|
||
raise InvalidException("不存在对应通用用户ID")
|
||
# 1. 查询当前住户授权的授权的设备ID
|
||
device_ids = DevicesTable.get_auth_device_ids(table_handler, str(user_id))
|
||
# 2. 对所有授权设备进行授权移除操作
|
||
if device_ids:
|
||
sc = ServicesCall()
|
||
_callback, code, msg = sc.del_face(device_ids[0], DelFaceItem(
|
||
user_id=str(user_id),
|
||
device_ids=str(device_ids)
|
||
))
|
||
if not _callback:
|
||
raise InvalidException(f"住户人脸授权移除失败，错误码{code}，{msg}")
|
||
DevicesTable.delete_householder_all_auth(table_handler, str(user_id))
|
||
sqls = [
|
||
"DELETE FROM householders WHERE householder_id = ?",
|
||
"DELETE FROM householders_type WHERE householder_id = ?"
|
||
]
|
||
table_handler.execute(sqls, [(user_id,), (user_id,)])
|
||
return {"status": True}
|
||
|
||
@staticmethod
|
||
def delete_type(table_handler: BaseTable, householder_id: int):
|
||
table_handler.execute(
|
||
"""
|
||
DELETE FROM householders_type
|
||
WHERE householder_id = ?
|
||
""", (householder_id,)
|
||
)
|
||
|
||
@staticmethod
|
||
def exists_householder_phone(table_handler: BaseTable, phone: str) -> bool:
|
||
"""查询住户手机号码是否存在"""
|
||
table_handler.query(
|
||
"""
|
||
SELECT householder_id
|
||
FROM householders
|
||
WHERE type = '住户' AND phone = ?
|
||
""", (encrypt_number(phone),)
|
||
)
|
||
res = table_handler.cursor.fetchall()
|
||
if res:
|
||
return True
|
||
else:
|
||
return False
|
||
|
||
@staticmethod
|
||
def exists_user_phone(table_handler: BaseTable, phone: str):
|
||
"""查询非住户手机号码是否存在"""
|
||
table_handler.query(
|
||
"""
|
||
SELECT householder_id
|
||
FROM householders
|
||
WHERE type != '住户' AND phone = ?
|
||
""", (encrypt_number(phone),)
|
||
)
|
||
res = table_handler.cursor.fetchall()
|
||
if res:
|
||
return True
|
||
else:
|
||
return False
|
||
|
||
@staticmethod
|
||
def exists_id(table_handler: BaseTable, householder_id: int) -> bool:
|
||
"""查询ID是否存在"""
|
||
table_handler.query(
|
||
"""
|
||
SELECT householder_id
|
||
FROM householders
|
||
WHERE householder_id = ?
|
||
""",
|
||
(householder_id,)
|
||
)
|
||
res = table_handler.cursor.fetchall()
|
||
if res:
|
||
return True
|
||
else:
|
||
return False
|
||
|
||
@staticmethod
|
||
def exists_householder_id(table_handler: BaseTable, householder_id: int) -> bool:
|
||
"""查询住户ID是否存在"""
|
||
table_handler.query(
|
||
"""
|
||
SELECT householder_id
|
||
FROM householders
|
||
WHERE type = '住户' AND householder_id = ?
|
||
""",
|
||
(householder_id,)
|
||
)
|
||
res = table_handler.cursor.fetchall()
|
||
if res:
|
||
return True
|
||
else:
|
||
return False
|
||
|
||
@staticmethod
|
||
def exists_user_id(table_handler: BaseTable, user_id: int) -> bool:
|
||
"""查询非住户ID是否存在"""
|
||
table_handler.query(
|
||
"""
|
||
SELECT householder_id
|
||
FROM householders
|
||
WHERE type != '住户' AND householder_id = ?
|
||
""",
|
||
(user_id,)
|
||
)
|
||
res = table_handler.cursor.fetchall()
|
||
if res:
|
||
return True
|
||
else:
|
||
return False
|
||
|
||
@staticmethod
|
||
def exists_householder_type(table_handler: BaseTable, householder_id: str, room_id: str) -> bool:
|
||
"""查询住户-房产关联信息是否存在"""
|
||
table_handler.query(
|
||
"""
|
||
SELECT householder_id
|
||
FROM householders_type
|
||
WHERE householder_id = ? AND room_id = ?
|
||
""",
|
||
(householder_id, room_id)
|
||
)
|
||
res = table_handler.cursor.fetchall()
|
||
if res:
|
||
return True
|
||
else:
|
||
return False
|