87 lines
3.2 KiB
Python
87 lines
3.2 KiB
Python
# -*- coding:utf-8 -*-
|
||
"""
|
||
@Author : xuxingchen
|
||
@Contact : xuxingchen@sinochem.com
|
||
@Desc : 房产信息控制逻辑
|
||
"""
|
||
from typing import Optional
|
||
|
||
from fastapi import APIRouter, Header, Request, Query
|
||
from pydantic import BaseModel, field_validator, Field
|
||
|
||
from routers.login import authenticate_token
|
||
from utils import logger
|
||
from utils.logger import TOKEN_ERROR
|
||
from utils.database import get_table_handler
|
||
from models.houses import HousesTable
|
||
from utils.misc import InvalidException
|
||
|
||
router = APIRouter()
|
||
|
||
|
||
class ListsReq(BaseModel):
|
||
project: str = Field(description="必填项，用于限定项目编号")
|
||
area: Optional[str] = Field(None, description="可选项，用于限定区域")
|
||
building: Optional[str] = Field(None, description="可选项，用于限定楼栋")
|
||
unit: Optional[str] = Field(None, description="可选项，用于限定单元")
|
||
floor: Optional[str] = Field(None, description="可选项，用于限定楼层")
|
||
target: str = Field(description="必填项，用于控制输出结果，取值范围：area, building, unit, floor, room")
|
||
|
||
def dict(self, *args, **kwargs):
|
||
_dict = self.__dict__.copy()
|
||
for key, value in self.__dict__.items():
|
||
if value is None:
|
||
_dict.pop(key)
|
||
_dict.pop("target")
|
||
return _dict
|
||
|
||
@field_validator("target")
|
||
def validate_target(cls, value):
|
||
if value not in ["area", "building", "unit", "floor", "room"]:
|
||
raise InvalidException("请提供正确的请求target [area, building, unit, floor, room]")
|
||
else:
|
||
return value
|
||
|
||
|
||
@router.get("/projects", summary="获取所有项目列表")
|
||
async def get_projects_list(request: Request, token: str = Header(...)):
|
||
"""获取项目列表"""
|
||
if not authenticate_token(token):
|
||
raise InvalidException(TOKEN_ERROR)
|
||
th = get_table_handler()
|
||
projects_list = HousesTable.get_items_by_dynamic_item(th, "project", {})
|
||
resp = {"projects": projects_list}
|
||
logger.Logger.debug(f"{request.url.path} {resp}")
|
||
return resp
|
||
|
||
|
||
@router.post("/project/lists", summary="动态获取房产信息")
|
||
async def get_items(request: Request,
|
||
item: ListsReq,
|
||
token: str = Header(...)):
|
||
"""根据请求获取列表"""
|
||
if not authenticate_token(token):
|
||
raise InvalidException(TOKEN_ERROR)
|
||
th = get_table_handler()
|
||
item_list = HousesTable.get_items_by_dynamic_item(th, item.target, item.dict())
|
||
resp = {f"{item.target.split('_')[0]}s": item_list}
|
||
logger.Logger.debug(f"{request.url.path} {resp}")
|
||
return resp
|
||
|
||
|
||
@router.get("/getHouseholderCount", summary="获取房产住户数量")
|
||
async def get_householder_count(request: Request,
|
||
room_id: str = Query(...),
|
||
token: str = Header(...)):
|
||
"""根据请求获取对应房屋中当前住户数量"""
|
||
if not authenticate_token(token):
|
||
raise InvalidException(TOKEN_ERROR)
|
||
th = get_table_handler()
|
||
count = HousesTable.get_householder_count(th, room_id)
|
||
if count is not None:
|
||
resp = {"count": count}
|
||
else:
|
||
raise InvalidException("room_id 不存在")
|
||
logger.Logger.debug(f"{request.url.path} {resp}")
|
||
return resp
|