strepsiades
132 lines
6.0 KiB
Python
132 lines
6.0 KiB
Python
from fastapi import HTTPException
|
|
from pydantic import BaseModel, Field, model_validator, field_validator
|
|
from typing import List, Optional
|
|
from enum import Enum
|
|
from ..common.base import BaseResponseWithID
|
|
|
|
|
|
class SupplierType(str, Enum):
|
|
"""类型枚举"""
|
|
PROJECT = "project" # 项目类型
|
|
OTHER = "other" # 其他类型
|
|
|
|
@classmethod
|
|
def supplier_list(cls):
|
|
return [cls.PROJECT.value, cls.OTHER.value]
|
|
|
|
class SupplierNature(str, Enum):
|
|
"""
|
|
性质枚举
|
|
"""
|
|
CONSTRUCTION = "construction" # 施工单位
|
|
DESIGN = "design" # 设计单位
|
|
SUPERVISORY = "supervisory" # 监理单位
|
|
SUPPLIER = "supplier" # 供应商
|
|
SURVEY = "survey" # 2026.3.11 推送系统测试时发现与项目侧 org_role 不一致，已补充勘察单位
|
|
GOVERMENT = "goverment" # 政府
|
|
|
|
@classmethod
|
|
def supplier_type_list(cls):
|
|
return [member.value for member in cls]
|
|
|
|
class SupplierStatus(str, Enum):
|
|
"""状态枚举"""
|
|
ACTIVE = "active" # 正常
|
|
INACTIVE = "inactive" # 禁用
|
|
|
|
class SupplierDetailResponse(BaseResponseWithID):
|
|
"""
|
|
供应商详情模型
|
|
"""
|
|
id: str = Field(..., description="供应商ID", examples=["company:001"])
|
|
name: str = Field(..., description="供应商名称", examples=["腾讯集团"])
|
|
short_name: str = Field(..., description="供应商简称", examples=["腾讯"])
|
|
code: str = Field(..., description="供应商编码", examples=["TX001"])
|
|
company_type: str = Field(..., description="供应商类型", examples=["other"])
|
|
parent_id: Optional[str] = Field(None, description="上级供应商ID", examples=[None])
|
|
path: List[str] = Field(..., description="供应商所处路径", examples=[["company:001"]])
|
|
status: str = Field(..., description="供应商状态", examples=["active"])
|
|
sort: int = Field(..., description="排序", examples=[1])
|
|
credit_code: Optional[str] = Field(None, description="信用代码", examples=["914403007109310121"])
|
|
province: Optional[str] = Field(None, description="省份编码", examples=[None])
|
|
city: Optional[str] = Field(None, description="城市编码", examples=[None])
|
|
company_nature: List[Optional[str]] = Field([], description="供应商性质数组", examples=[['supplier']])
|
|
city_name: Optional[str] = Field(None, description="城市名称", examples=[None])
|
|
province_name: Optional[str] = Field(None, description="省份名称", examples=[None])
|
|
region_path: Optional[list] = Field(default_factory=list, description="供应商国家城市地址编码列表")
|
|
country: Optional[str] = Field(None, description="国家编码", examples=[None])
|
|
country_name: Optional[str] = Field(None, description="国家名称", examples=[None])
|
|
|
|
|
|
class SupplierListResponse(BaseModel):
|
|
"""
|
|
供应商list
|
|
"""
|
|
page: int
|
|
page_size: int
|
|
total: int
|
|
list: List[SupplierDetailResponse] = []
|
|
|
|
|
|
class SupplierCreateRequest(BaseModel):
|
|
"""
|
|
供应商推送创建请求模型
|
|
"""
|
|
name: str = Field(..., description="供应商名称", examples=["腾讯集团"])
|
|
short_name: str = Field(..., description="供应商简称", examples=["腾讯"])
|
|
code: str = Field(..., description="供应商编码", examples=["TX001"])
|
|
company_type: SupplierType = Field(..., description="供应商类型", examples=["project"])
|
|
parent_id: Optional[str] = Field(None, description="上级供应商ID", examples=[None])
|
|
status: SupplierStatus = Field(..., description="供应商状态", examples=["active"])
|
|
path: Optional[List[str]] = Field(default_factory=list, description="供应商路径 [..., 上级id, 本身id]")
|
|
path_name: Optional[List[str]] = Field(default_factory=list, description="供应商路径名称 [..., 上级名称, 本身名称]")
|
|
credit_code: str = Field(..., description="供应商信用代码", examples=["914403007109310121"])
|
|
region_path: Optional[list] = Field(default_factory=list, description="供应商国家城市地址编码列表, 需对接集团侧地图组件, 没有对接可以不传递", examples=[[
|
|
"100000",
|
|
"420000",
|
|
"420800"
|
|
]])
|
|
company_nature: List[SupplierNature] = Field(..., description="供应商性质数组 可多选",
|
|
examples=[['construction',
|
|
'supplier',
|
|
'design',
|
|
'survey',
|
|
'supervisory',
|
|
'goverment']])
|
|
|
|
@model_validator(mode='after')
|
|
def validate_company_type_and_nature(self):
|
|
data = self.model_dump()
|
|
|
|
company_nature = data.get("company_nature")
|
|
if not company_nature:
|
|
raise HTTPException(detail=f"供应商:company_nature为空", status_code=400)
|
|
for cn in company_nature:
|
|
if cn not in SupplierNature.supplier_type_list():
|
|
raise HTTPException(detail=f"供应商:company_nature范围:{SupplierNature.supplier_type_list()}",
|
|
status_code=400)
|
|
|
|
company_type = data.get("company_type")
|
|
if not company_type:
|
|
raise HTTPException(detail=f"供应商:company_type为空", status_code=400)
|
|
if company_type not in SupplierType.supplier_list():
|
|
raise HTTPException(
|
|
detail=f"供应商:company_type范围({SupplierType.supplier_list()})", status_code=400)
|
|
|
|
for field in ["name", "short_name", "code", "credit_code", "status"]:
|
|
field_value = data.get(field)
|
|
if not field_value:
|
|
raise HTTPException(detail=f"字段{field}为空", status_code=400)
|
|
return self
|
|
|
|
@field_validator(
|
|
"credit_code",
|
|
mode="after"
|
|
)
|
|
def clean_credit_code(cls, field_value):
|
|
if field_value is None:
|
|
return None
|
|
import re
|
|
return re.sub(r'\s+', '', str(field_value))
|
|
|