from typing import Optional, List, Dict, Any from alipay.aop.api.domain.DepartmentInfoDTO import DepartmentInfoDTO from alipay.aop.api.domain.AlipayCommerceEcDepartmentCreateModel import AlipayCommerceEcDepartmentCreateModel from alipay.aop.api.domain.AlipayCommerceEcDepartmentDeleteModel import AlipayCommerceEcDepartmentDeleteModel from alipay.aop.api.domain.AlipayCommerceEcDepartmentInfoQueryModel import AlipayCommerceEcDepartmentInfoQueryModel from alipay.aop.api.response.AlipayCommerceEcDepartmentCreateResponse import AlipayCommerceEcDepartmentCreateResponse from alipay.aop.api.response.AlipayCommerceEcDepartmentDeleteResponse import AlipayCommerceEcDepartmentDeleteResponse from alipay.aop.api.response.AlipayCommerceEcDepartmentInfoQueryResponse import AlipayCommerceEcDepartmentInfoQueryResponse from alipay.aop.api.response.AlipayCommerceEcDepartmentSublistQueryResponse import AlipayCommerceEcDepartmentSublistQueryResponse from alipay.aop.api.request.AlipayCommerceEcDepartmentCreateRequest import AlipayCommerceEcDepartmentCreateRequest from alipay.aop.api.request.AlipayCommerceEcDepartmentDeleteRequest import AlipayCommerceEcDepartmentDeleteRequest from alipay.aop.api.request.AlipayCommerceEcDepartmentInfoModifyRequest import AlipayCommerceEcDepartmentInfoModifyRequest from alipay.aop.api.request.AlipayCommerceEcDepartmentInfoQueryRequest import AlipayCommerceEcDepartmentInfoQueryRequest from alipay.aop.api.request.AlipayCommerceEcDepartmentSublistQueryRequest import AlipayCommerceEcDepartmentSublistQueryRequest from app.api.v1.module_system.auth.schema import AuthSchema from app.core.alipay import AlipayClient from app.core.logger import log from app.core.exceptions import CustomException from .crud import DepartmentCRUD from .schema import ( DepartmentCreateSchema, DepartmentUpdateSchema, DepartmentDetailOutSchema, DepartmentOperationOutSchema, DepartmentTreeOutSchema, DepartmentListOutSchema, DepartmentOutSchema ) class DepartmentService: """部门服务""" @classmethod async def create_department_service( cls, auth: AuthSchema, data: DepartmentCreateSchema ) -> DepartmentOperationOutSchema: """创建部门""" # 调用支付宝接口创建部门 department_create_model = AlipayCommerceEcDepartmentCreateModel() department_create_model.enterprise_id = data.enterprise_id department_create_model.department_name = data.department_name if data.department_code: department_create_model.department_code = data.department_code if data.parent_department_id: department_create_model.parent_department_id = data.parent_department_id or "-1" client = AlipayClient.get_client() request = AlipayCommerceEcDepartmentCreateRequest() request.biz_model = department_create_model response = client.execute(request) if not response: raise CustomException(msg="创建部门失败: 无响应") result = AlipayCommerceEcDepartmentCreateResponse() result.parse_response_content(response) if not result.is_success(): log.error(f"支付宝接口调用失败: {result.code} - {result.msg}") raise CustomException(msg=f"创建部门失败: {result.sub_code or result.msg or result.code}") # 保存到数据库 crud = DepartmentCRUD(auth) department_data = { "department_id": result.department_id or "", "department_name": data.department_name, "department_code": data.department_code, "parent_department_id": data.parent_department_id or "-1", "enterprise_id": data.enterprise_id, "sort_order": data.sort_order, "leader_employee_id": data.leader_employee_id, "leader_employee_name": data.leader_employee_name } await crud.create(department_data) return DepartmentOperationOutSchema( department_id=result.department_id or "", department_name=data.department_name ) @classmethod async def delete_department_service( cls, auth: AuthSchema, department_id: str, enterprise_id: str ) -> DepartmentOperationOutSchema: return DepartmentOperationOutSchema( department_id=department_id, department_name="" ) @classmethod async def update_department_service( cls, auth: AuthSchema, department_id: str, enterprise_id: str, data: DepartmentUpdateSchema ) -> DepartmentOperationOutSchema: """更新部门""" return DepartmentOperationOutSchema( department_id=department_id, department_name="" ) @classmethod async def get_department_service( cls, auth: AuthSchema, department_id: str, enterprise_id: str ) -> DepartmentDetailOutSchema: """查询部门详情 (alipay.commerce.ec.department.info.query)""" # 调用支付宝接口查询部门详情 model = AlipayCommerceEcDepartmentInfoQueryModel() model.enterprise_id = enterprise_id model.department_id = department_id request = AlipayCommerceEcDepartmentInfoQueryRequest() request.biz_model = model client = AlipayClient.get_client() response = client.execute(request) if not response: raise CustomException(msg="查询部门详情失败: 无响应") result = AlipayCommerceEcDepartmentInfoQueryResponse() result.parse_response_content(response) if not result.is_success(): log.error(f"支付宝接口调用失败: {result.code} - {result.msg}") raise CustomException(msg=f"查询部门详情失败: {result.sub_code or result.msg or result.code}") # 解析部门信息 department_info = result.department_info if not department_info: raise CustomException(msg="部门信息不存在") # 转换为响应模型 department_out = DepartmentOutSchema( department_id=department_info.department_id or "", department_name=department_info.department_name or "", department_code=department_info.department_code, parent_department_id=department_info.parent_department_id, enterprise_id=enterprise_id, status="NORMAL", ) return DepartmentDetailOutSchema( department=department_out, sub_departments=[] ) @classmethod async def get_sub_departments_service( cls, auth: AuthSchema, parent_department_id: str, enterprise_id: str ) -> List[str]: """查询子部门列表""" return [] @classmethod async def list_service( cls, auth: AuthSchema, page_no: int = 1, page_size: int = 20, search: dict | None = None, ) -> dict: """分页查询部门列表""" log.info(f"查询部门列表: {page_no}, {page_size}, {search}") crud = DepartmentCRUD(auth) offset = (page_no - 1) * page_size return await crud.page( offset=offset, limit=page_size, order_by=[{"id": "desc"}], search=search or {}, out_schema=DepartmentListOutSchema, ) @classmethod async def get_department_tree_service( cls, auth: AuthSchema, enterprise_id: str ) -> List[DepartmentTreeOutSchema]: """获取部门树形结构""" crud = DepartmentCRUD(auth) # 查询指定企业下的所有部门 departments = await crud.list({"enterprise_id": enterprise_id}) # 转换为字典，方便查找 department_dict = {dept.department_id: dept for dept in departments} # 构建树形结构 tree = [] for dept in departments: # 找到父部门 parent_id = dept.parent_department_id if not parent_id or parent_id == "-1" or parent_id not in department_dict: # 没有父部门，作为根节点 tree.append(dept) else: # 添加到父部门的子节点 if not hasattr(department_dict[parent_id], 'children'): department_dict[parent_id].children = [] department_dict[parent_id].children.append(dept) # 排序子节点 def sort_children(node): if hasattr(node, 'children') and node.children: node.children.sort(key=lambda x: (x.sort_order or 0, x.department_name)) for child in node.children: sort_children(child) for node in tree: sort_children(node) # 转换为响应模型 return [DepartmentTreeOutSchema.model_validate(node) for node in tree] @classmethod async def get_all_departments( cls, auth: AuthSchema, enterprise_id: str ) -> List[DepartmentOutSchema]: """获取所有部门列表（不分页）""" crud = DepartmentCRUD(auth) departments = await crud.list({"enterprise_id": enterprise_id}) return [DepartmentOutSchema.model_validate(dept) for dept in departments]