from typing import Annotated import uuid from fastapi import APIRouter, Depends, Path, Query from fastapi.responses import JSONResponse from app.api.v1.module_system.auth.schema import AuthSchema from app.common.response import ResponseSchema, SuccessResponse from app.core.dependencies import AuthPermission from app.core.logger import log from app.core.router_class import OperationLogRoute from app.plugin.module_payment.expense.institution.schema import InstitutionListOutSchema from .service import InstitutionService, InstitutionScopeService, IssueruleService from alipay.aop.api.domain.AlipayEbppInvoiceInstitutionCreateModel import ( AlipayEbppInvoiceInstitutionCreateModel, ) from alipay.aop.api.response.AlipayEbppInvoiceInstitutionCreateResponse import ( AlipayEbppInvoiceInstitutionCreateResponse, ) from alipay.aop.api.domain.AlipayEbppInvoiceInstitutionDeleteModel import ( AlipayEbppInvoiceInstitutionDeleteModel, ) from alipay.aop.api.response.AlipayEbppInvoiceInstitutionDeleteResponse import ( AlipayEbppInvoiceInstitutionDeleteResponse, ) from alipay.aop.api.domain.AlipayEbppInvoiceInstitutionModifyModel import ( AlipayEbppInvoiceInstitutionModifyModel, ) from alipay.aop.api.response.AlipayEbppInvoiceInstitutionModifyResponse import ( AlipayEbppInvoiceInstitutionModifyResponse, ) InstitutionRouter = APIRouter( route_class=OperationLogRoute, prefix="/institution", tags=["费控制度"], ) @InstitutionRouter.post( "", summary="创建费控制度", description="创建费控制度。支持串联调用：创建制度→设置成员→创建发放规则", ) async def create_institution_controller( data: dict, auth: Annotated[AuthSchema, Depends(AuthPermission(["module_payment:expense:institution:create"]))], ) -> JSONResponse: """创建费控制度（含完整串联流程）""" enterprise_id = data.get("enterprise_id", "") if not enterprise_id: from app.plugin.module_payment.enterprise.model import EnterpriseModel from sqlalchemy import select tenant_id = auth.user.tenant_id if auth.user and auth.user.tenant_id else auth.tenant_id log.info(f"推导 enterprise_id: tenant_id={tenant_id}, user_tenant_id={getattr(auth.user, 'tenant_id', None)}") stmt = select(EnterpriseModel).where(EnterpriseModel.tenant_id == tenant_id).limit(1) result = await auth.db.execute(stmt) enterprise = result.scalar_one_or_none() log.info(f"查询 enterprise 结果: {enterprise.enterprise_id if enterprise else 'None'}") enterprise_id = enterprise.enterprise_id if enterprise else "" if enterprise_id: data["enterprise_id"] = enterprise_id # 字段映射：前端 name → Alipay institution_name if data.get("name") and not data.get("institution_name"): data["institution_name"] = data["name"] # Alipay 必填：商户外部单据号（唯一标识） if not data.get("outer_source_id"): data["outer_source_id"] = str(uuid.uuid4()).replace("-", "") # expense_type 映射：前端值 → 支付宝标准值 EXPENSE_TYPE_MAP = {"GENERAL": "DEFAULT", "DEFAULT": "DEFAULT"} if data.get("expense_type") in EXPENSE_TYPE_MAP: data["expense_type"] = EXPENSE_TYPE_MAP[data["expense_type"]] # 时间格式补全：YYYY-MM-DD → YYYY-MM-DD HH:mm:ss if data.get("effective_start_date") and len(data["effective_start_date"]) == 10: data["effective_start_date"] = data["effective_start_date"] + " 00:00:00" if data.get("effective_end_date") and len(data["effective_end_date"]) == 10: data["effective_end_date"] = data["effective_end_date"] + " 23:59:59" elif not data.get("effective_end_date") and data.get("effective_time_type") == "unlimited": # 长期有效：设为2099年底 data["effective_end_date"] = "2099-12-31 23:59:59" # 默认使用规则（支付宝必填） if not data.get("standard_info_list"): single_limit = data.get("single_limit", 0) period_type = data.get("period_type", "") amount = data.get("amount", 0) standard_info = { "standard_name": data.get("institution_name", "默认规则"), "standard_desc": f"单笔限额{single_limit}元" if single_limit else "通用规则", "consume_mode": "DEFAULT", "payment_policy": "PERSONAL", "personal_qrcode_mode": 0, "outer_source_id": str(uuid.uuid4()).replace("-", ""), } condition_list = [] if single_limit: condition_list.append({"rule_factor": "QUOTA_TOTAL", "rule_name": "单次消费金额", "rule_value": str(single_limit)}) # 定额发放时，将周期限额写入使用规则条件 PERIOD_FACTOR_MAP = { "daily": "QUOTA_DAY", "weekly": "QUOTA_WEEK", "monthly": "QUOTA_MONTH", "quarterly": "QUOTA_QUARTER", "yearly": "QUOTA_YEAR", } if data.get("grant_mode") == "period" and period_type in PERIOD_FACTOR_MAP and amount: condition_list.append({ "rule_factor": PERIOD_FACTOR_MAP[period_type], "rule_name": f"{period_type}限额", "rule_value": str(amount), }) # 至少保证有一条条件规则（支付宝必填） if not condition_list: condition_list.append({ "rule_factor": "QUOTA_TOTAL", "rule_name": "单次消费金额", "rule_value": "0", }) standard_info["standard_condition_info_list"] = condition_list data["standard_info_list"] = [standard_info] institution_create_model = AlipayEbppInvoiceInstitutionCreateModel.from_alipay_dict(data) # 解析适用成员数据 scope_data = None adapter_type = data.get("applicable_scope") if adapter_type and adapter_type not in ("NONE", "none"): ADAPTER_TYPE_MAP = {"all": "EMPLOYEE_ALL", "employee": "EMPLOYEE_SELECT", "department": "EMPLOYEE_DEPARTMENT"} mapped_adapter = ADAPTER_TYPE_MAP.get(adapter_type, adapter_type) scope_data = { "adapter_type": mapped_adapter, "owner_type": data.get("scope_owner_type", "EMPLOYEE"), "add_owner_id_list": data.get("scope_owner_id_list"), } # 全体员工时把 scope 写入创建请求（避免默认无scope导致支付宝后台不可操作） if adapter_type == "all": data["institution_scope_info"] = { "adapter_type": "ALL", "owner_type": "EMPLOYEE", } # 解析发放规则数据 issuerule_data = None if data.get("grant_mode") == "period": period_type_raw = data.get("period_type", "monthly") # 映射前端period_type到支付宝枚举 ISSUE_TYPE_MAP = { "daily": "ISSUE_DAY", "weekly": "ISSUE_WEEK", "monthly": "ISSUE_MONTH", "quarterly": "ISSUE_QUARTER", "yearly": "ISSUE_YEAR", } issue_type = ISSUE_TYPE_MAP.get(period_type_raw, "ISSUE_MONTH") amount = data.get("amount", 0) # 有效时间配置 effective_time_type = data.get("effective_time_type", "unlimited") if effective_time_type == "unlimited": effective_period = '{"all": true}' elif effective_time_type == "workday": workday_start = data.get("workday_start_time", "00:00") workday_end = data.get("workday_end_time", "23:59") effective_period = f'{{"regular":{{"workday":[["{workday_start}","{workday_end}"]]}}}}' else: effective_period = '{"all": true}' issuerule_data = { "quota_type": "CAP", "issue_type": issue_type, "issue_amount_value": str(amount), "issue_rule_name": data.get("name", "") + "-发放规则", "effective_period": effective_period, "invalid_mode": 1 if data.get("effective_time_type") == "unlimited" else 0, "share_mode": 0, "outer_source_id": data.get("outer_source_id") or str(uuid.uuid4()), } result = await InstitutionService.create_institution_full_flow( auth=auth, institution_model=institution_create_model, enterprise_id=enterprise_id, scope_data=scope_data, issuerule_data=issuerule_data, raw_data=data, ) log.info(f"创建费控制度成功: institution_id={result.get('institution_id')}") return SuccessResponse(data=result, msg="创建费控制度成功") @InstitutionRouter.get( "", summary="查询费控制度列表", description="分页查询费控制度列表", response_model=ResponseSchema[InstitutionListOutSchema], ) async def list_institution_controller( auth: Annotated[AuthSchema, Depends(AuthPermission(["module_payment:expense:institution:list"]))], page_no: Annotated[int, Query(description="页码")] = 1, page_size: Annotated[int, Query(description="每页数量")] = 20, enterprise_id: Annotated[str | None, Query(description="企业ID")] = None, name: Annotated[str | None, Query(description="制度名称")] = None, expense_type: Annotated[str | None, Query(description="费用类型")] = None, status: Annotated[str | None, Query(description="状态")] = None, ) -> JSONResponse: """查询费控制度列表""" search = {} if enterprise_id: search["enterprise_id"] = enterprise_id if name: search["name"] = name if expense_type: search["expense_type"] = expense_type if status: search["status"] = status result = await InstitutionService.list_service( auth=auth, page_no=page_no, page_size=page_size, search=search ) return SuccessResponse(data=result, msg="查询费控制度列表成功") @InstitutionRouter.get( "/{institution_id}", summary="查询费控制度详情", description="查询费控制度详情 (alipay.ebpp.invoice.institution.detailinfo.query)，失败时降级到本地DB", ) async def detail_institution_controller( institution_id: Annotated[str, Path(description="制度ID")], auth: Annotated[AuthSchema, Depends(AuthPermission(["module_payment:expense:institution:detail"]))], enterprise_id: Annotated[str | None, Query(description="企业ID")] = None, ) -> JSONResponse: """查询费控制度详情""" if not enterprise_id: from app.plugin.module_payment.enterprise.model import EnterpriseModel from sqlalchemy import select tenant_id = auth.user.tenant_id if auth.user and auth.user.tenant_id else auth.tenant_id stmt = select(EnterpriseModel).where(EnterpriseModel.tenant_id == tenant_id).limit(1) result = await auth.db.execute(stmt) ent = result.scalar_one_or_none() enterprise_id = ent.enterprise_id if ent else "" result = await InstitutionService.detailinfo_query_service( auth=auth, institution_id=institution_id, enterprise_id=enterprise_id, ) if result is None: return SuccessResponse(data=None, msg="制度不存在") return SuccessResponse(data=result, msg="查询费控制度详情成功") @InstitutionRouter.delete( "", summary="删除费控制度", description="删除费控制度 (alipay.ebpp.invoice.institution.delete)", ) async def delete_institution_controller( data: dict, auth: Annotated[AuthSchema, Depends(AuthPermission(["module_payment:expense:institution:delete"]))], ) -> JSONResponse: """删除费控制度""" institution_delete_model = AlipayEbppInvoiceInstitutionDeleteModel.from_alipay_dict(data) result = await InstitutionService.delete_institution_service(auth=auth, data=institution_delete_model) log.info(f"删除费控制度成功: institution_id={institution_delete_model.institution_id}, enterprise_id={institution_delete_model.enterprise_id}") return SuccessResponse(data=result, msg="删除费控制度成功") @InstitutionRouter.post( "/modify", summary="编辑费控制度", description="编辑费控制度 (alipay.ebpp.invoice.institution.modify)", ) async def modify_institution_controller( data: dict, auth: Annotated[AuthSchema, Depends(AuthPermission(["module_payment:expense:institution:modify"]))], ) -> JSONResponse: """编辑费控制度""" institution_id = data.get("institution_id", "") # enterprise_id 推导 if not data.get("enterprise_id"): from app.plugin.module_payment.enterprise.model import EnterpriseModel from sqlalchemy import select tenant_id = auth.user.tenant_id if auth.user and auth.user.tenant_id else auth.tenant_id stmt = select(EnterpriseModel).where(EnterpriseModel.tenant_id == tenant_id).limit(1) result = await auth.db.execute(stmt) enterprise = result.scalar_one_or_none() if enterprise: data["enterprise_id"] = enterprise.enterprise_id # name → institution_name if data.get("name") and not data.get("institution_name"): data["institution_name"] = data["name"] # 时间格式 if data.get("effective_start_date") and len(data["effective_start_date"]) == 10: data["effective_start_date"] = data["effective_start_date"] + " 00:00:00" if data.get("effective_end_date") and len(data["effective_end_date"]) == 10: data["effective_end_date"] = data["effective_end_date"] + " 23:59:59" elif not data.get("effective_end_date") and data.get("effective_time_type") == "unlimited": data["effective_end_date"] = "2099-12-31 23:59:59" # expense_type 映射 EXPENSE_TYPE_MAP = {"GENERAL": "DEFAULT", "DEFAULT": "DEFAULT"} if data.get("expense_type") in EXPENSE_TYPE_MAP: data["expense_type"] = EXPENSE_TYPE_MAP[data["expense_type"]] # 提取 scope 变更数据（需与基础修改分两次请求） applicable_scope = data.get("applicable_scope", "") scope_info = None enterprise_id = data.get("enterprise_id", "") if applicable_scope and applicable_scope not in ("NONE", "none"): ADAPTER_MAP = {"all": "EMPLOYEE_ALL", "employee": "EMPLOYEE_SELECT", "department": "EMPLOYEE_DEPARTMENT"} new_adapter = ADAPTER_MAP.get(applicable_scope, applicable_scope) # 查询当前scope：计算旧→新的差异 old_ids = [] try: scope_old = await InstitutionScopeService.scopepageinfo_query_service( auth=auth, institution_id=institution_id, enterprise_id=enterprise_id, page_num=1, page_size=100, ) old_ids = [str(i) for i in (scope_old.get("owner_id_list") or []) if i] except Exception: log.warning(f"查询旧scope失败，将全量覆盖: institution_id={data.get('institution_id', '')}") new_ids_raw = data.get("scope_owner_id_list") or [] new_ids = [str(i) for i in new_ids_raw if i is not None and str(i).strip()] # 计算差异 old_set, new_set = set(old_ids), set(new_ids) add_ids = list(new_set - old_set) delete_ids = list(old_set - new_set) scope_info = { "enterprise_id": enterprise_id, "adapter_type": new_adapter, "owner_type": "EMPLOYEE", } if new_adapter == "EMPLOYEE_ALL": # 全体员工模式：不传员工ID给支付宝（但本地同步仍然需要） log.info(f"全体员工模式，跳过员工ID传参") else: if add_ids: scope_info["add_owner_id_list"] = add_ids if delete_ids: scope_info["delete_owner_id_list"] = delete_ids if not add_ids and not delete_ids: scope_info = None log.info("scope 无变化，跳过") elif applicable_scope in ("NONE", "none"): # 暂不设置：清空支付宝范围（调用 scope.modify 删除所有成员） try: scope_old = await InstitutionScopeService.scopepageinfo_query_service( auth=auth, institution_id=institution_id, enterprise_id=enterprise_id, page_num=1, page_size=100, ) old_adapter = scope_old.get("adapter_type", "") old_ids = [str(i) for i in (scope_old.get("owner_id_list") or []) if i] # 对非NONE的旧范围，清空所有成员 if old_adapter and old_adapter not in ("NONE",) and old_ids: scope_info = { "enterprise_id": enterprise_id, "adapter_type": old_adapter, "owner_type": "EMPLOYEE", "delete_owner_id_list": old_ids, } log.info(f"暂不设置 - 清空支付宝旧成员: {old_adapter}, count={len(old_ids)}") except Exception: log.warning(f"查询旧scope失败，跳过清空支付宝成员") # 从请求中移除 scope 数据，避免与基础修改冲突 data.pop("modify_scope_info", None) # 构建金额/限额变更的标准规则信息（同步到支付宝） new_amount = data.get("amount") new_single_limit = data.get("single_limit") period_type = data.get("period_type", "") grant_mode = data.get("grant_mode", "") if institution_id and enterprise_id and (new_amount is not None or new_single_limit is not None): try: def _to_dict(obj): """Alipay SDK 对象转 dict，已经是 dict 则直接返回""" if isinstance(obj, dict): return obj if hasattr(obj, 'to_alipay_dict'): return obj.to_alipay_dict() return {} # 从支付宝查询当前制度详情 detail_dict = await InstitutionService.detailinfo_query_service( auth=auth, institution_id=institution_id, enterprise_id=enterprise_id ) # 构建 modify_standard_detail_info modify_standard_list = [] if detail_dict and detail_dict.get("standard_info_list"): std_list = detail_dict["standard_info_list"] if not isinstance(std_list, list): std_list = [std_list] for std_obj in std_list: std = _to_dict(std_obj) std_id = std.get("standard_id") if not std_id: continue conditions = std.get("standard_condition_info_list") or [] if not isinstance(conditions, list): conditions = [conditions] modify_condition_list = [] for cond_obj in conditions: cond = _to_dict(cond_obj) cond_factor = cond.get("rule_factor", "") cond_id = cond.get("rule_id", "") # 周期限额变更 if grant_mode == "period" and period_type: PERIOD_FACTOR_MAP = { "daily": "QUOTA_DAY", "weekly": "QUOTA_WEEK", "monthly": "QUOTA_MONTH", "quarterly": "QUOTA_QUARTER", "yearly": "QUOTA_YEAR", } target_factor = PERIOD_FACTOR_MAP.get(period_type) if target_factor and cond_factor == target_factor and new_amount is not None: modify_condition_list.append({ "rule_id": cond_id, "rule_factor": cond_factor, "rule_value": str(new_amount), }) # 单笔限额变更 if cond_factor == "QUOTA_TOTAL" and new_single_limit is not None: modify_condition_list.append({ "rule_id": cond_id, "rule_factor": "QUOTA_TOTAL", "rule_value": str(new_single_limit), }) if modify_condition_list: modify_standard_list.append({ "standard_id": std_id, "modify_condition_list": modify_condition_list, }) if modify_standard_list: data["modify_standard_detail_info"] = { "modify_standard_list": modify_standard_list } log.info(f"已构建金额变更信息: amount={new_amount}, single_limit={new_single_limit}, rules={len(modify_standard_list)}") else: log.warning(f"未找到可修改的标准规则: institution_id={institution_id}") # 周期发放制度：更新发放规则金额 if grant_mode == "period" and new_amount is not None and detail_dict: issue_rule_list = detail_dict.get("issue_rule_info_list") or [] if not isinstance(issue_rule_list, list): issue_rule_list = [issue_rule_list] for rule in issue_rule_list: rule_dict = _to_dict(rule) issue_rule_id = rule_dict.get("issue_rule_id", "") if issue_rule_id: data["modify_issue_rule_detail_info"] = { "modify_issue_rule_list": { "issue_rule_id": issue_rule_id, "issue_amount_value": str(new_amount), } } log.info(f"已构建发放规则金额变更: issue_rule_id={issue_rule_id}, amount={new_amount}") break else: log.warning(f"未查询到制度详情，跳过金额同步: institution_id={institution_id}") except Exception as e: log.warning(f"构建标准规则变更信息失败(将跳过金额同步): {e}") # 第1次请求：仅修改制度基础信息（不含 scope） base_data = {k: v for k, v in data.items() if k != "modify_scope_info"} institution_modify_model = AlipayEbppInvoiceInstitutionModifyModel.from_alipay_dict(base_data) try: result = await InstitutionService.modify_institution_service( auth=auth, data=institution_modify_model, raw_data=base_data, scope_info=scope_info ) except Exception as e: err_msg = str(e) if "consult" in err_msg.lower() or "咨询" in err_msg or "发" in err_msg: raise CustomException(msg="制度下存在发放规则，咨询模式不允许修改为外部服务商，请先删除发放规则后再试") raise log.info(f"编辑费控制度成功: institution_id={institution_modify_model.institution_id}") return SuccessResponse(data=result, msg="编辑费控制度成功") # ========== 制度成员范围管理 ========== @InstitutionRouter.get( "/{institution_id}/scope", summary="查询制度成员范围", description="查询制度下成员范围 (alipay.ebpp.invoice.institution.scopepageinfo.query)", ) async def list_scope_controller( institution_id: Annotated[str, Path(description="制度ID")], auth: Annotated[AuthSchema, Depends(AuthPermission(["module_payment:expense:institution:scope:list"]))], enterprise_id: Annotated[str | None, Query(description="企业ID")] = None, owner_type: Annotated[str | None, Query(description="适配ID类型")] = None, page_num: Annotated[int, Query(description="页码")] = 1, page_size: Annotated[int, Query(description="每页条数")] = 20, ) -> JSONResponse: """查询制度成员""" result = await InstitutionScopeService.scopepageinfo_query_service( auth=auth, institution_id=institution_id, enterprise_id=enterprise_id, page_num=page_num, page_size=page_size, owner_type=owner_type, ) # 如果本地库标记为暂不设置，覆盖支付宝返回值 try: from .crud import InstitutionCRUD crud = InstitutionCRUD(auth) local_inst = await crud.get(institution_id=institution_id, enterprise_id=enterprise_id) if local_inst and getattr(local_inst, 'applicable_scope', '') in ('none', 'NONE'): result["adapter_type"] = "NONE" result["owner_id_list"] = [] except Exception: pass return SuccessResponse(data=result, msg="查询成功") @InstitutionRouter.post( "/{institution_id}/scope", summary="设置制度成员范围", description="设置/修改制度成员范围 (alipay.ebpp.invoice.institution.scope.modify)", ) async def modify_scope_controller( institution_id: Annotated[str, Path(description="制度ID")], data: dict, auth: Annotated[AuthSchema, Depends(AuthPermission(["module_payment:expense:institution:scope:modify"]))], ) -> JSONResponse: """设置制度成员""" from app.plugin.module_payment.employee.model import EmployeeModel from sqlalchemy import select enterprise_id = data.get("enterprise_id", "") new_adapter = data.get("adapter_type", "EMPLOYEE_ALL") # ====== 1. 查询旧适配类型和老员工ID列表 ====== old_adapter = "NONE" old_employee_ids: set[str] = set() try: scope_old = await InstitutionScopeService.scopepageinfo_query_service( auth=auth, institution_id=institution_id, enterprise_id=enterprise_id, page_num=1, page_size=100, ) old_adapter = scope_old.get("adapter_type", old_adapter) raw_old = [str(i) for i in (scope_old.get("owner_id_list") or []) if i] if old_adapter == "EMPLOYEE_ALL": # 全体员工模式 → 所有已签约员工 stmt = select(EmployeeModel.employee_id).where( EmployeeModel.enterprise_id == enterprise_id, EmployeeModel.status == "ACTIVATED", ) result = await auth.db.execute(stmt) old_employee_ids = {row[0] for row in result.fetchall() if row[0]} elif old_adapter == "EMPLOYEE_DEPARTMENT": # 按部门模式 → 部门下的所有已签约员工 from app.plugin.module_payment.employee.model import EmployeeModel stmt = select(EmployeeModel.employee_id, EmployeeModel.department_ids).where( EmployeeModel.enterprise_id == enterprise_id, EmployeeModel.status == "ACTIVATED", ) result = await auth.db.execute(stmt) dept_set = set(raw_old) for row in result.fetchall(): if row[1] and dept_set.intersection(set(row[1])): if row[0]: old_employee_ids.add(row[0]) else: old_employee_ids = set(raw_old) except Exception: old_adapter = "NONE" # 查不到旧范围时标记为NONE，确保走分步调用的逻辑 log.warning(f"查询旧scope失败(将全量处理): institution_id={institution_id}") # ====== 2. 计算新员工ID列表 ====== new_employee_ids: set[str] = set() if new_adapter == "EMPLOYEE_ALL": stmt = select(EmployeeModel.employee_id).where( EmployeeModel.enterprise_id == enterprise_id, EmployeeModel.status == "ACTIVATED", ) result = await auth.db.execute(stmt) new_employee_ids = {row[0] for row in result.fetchall() if row[0]} elif new_adapter == "EMPLOYEE_DEPARTMENT": dept_ids = data.get("add_owner_id_list") or [] dept_set = set(str(d) for d in dept_ids if d) stmt = select(EmployeeModel.employee_id, EmployeeModel.department_ids).where( EmployeeModel.enterprise_id == enterprise_id, EmployeeModel.status == "ACTIVATED", ) result = await auth.db.execute(stmt) for row in result.fetchall(): if row[1] and dept_set.intersection(set(row[1])): if row[0]: new_employee_ids.add(row[0]) else: raw_new = data.get("add_owner_id_list") or [] # EMPLOYEE_SELECT 模式：验证员工属于当前企业 from sqlalchemy import select as _sa_select emp_verify = await auth.db.execute( _sa_select(EmployeeModel.employee_id).where( EmployeeModel.enterprise_id == enterprise_id, EmployeeModel.employee_id.in_([str(i) for i in raw_new if i]), ) ) validated = [row[0] for row in emp_verify.fetchall() if row[0]] new_employee_ids = set(validated) # ====== 3. 计算员工级差异（用于配额联动） ====== add_emp_ids = list(new_employee_ids - old_employee_ids) delete_emp_ids = list(old_employee_ids - new_employee_ids) # 提取传给支付宝的原始ID（部门模式传部门ID，员工模式传员工ID） ====== alipay_add_ids: list[str] = [] alipay_delete_ids: list[str] = [] if new_adapter == "EMPLOYEE_DEPARTMENT": dept_ids = data.get("add_owner_id_list") or [] alipay_add_ids = [str(d) for d in dept_ids if d] elif new_adapter == "EMPLOYEE_SELECT": alipay_add_ids = list(new_employee_ids - old_employee_ids) alipay_delete_ids = list(old_employee_ids - new_employee_ids) # ====== 4. 调用支付宝 scope.modify ====== # 部门模式：需一次性传部门ID，不支持分步 if new_adapter == "EMPLOYEE_DEPARTMENT": scope_data = { "enterprise_id": enterprise_id, "adapter_type": "EMPLOYEE_DEPARTMENT", "owner_type": data.get("owner_type", "EMPLOYEE"), } if alipay_add_ids: scope_data["add_owner_id_list"] = alipay_add_ids result = await InstitutionScopeService.scope_modify_service( auth=auth, institution_id=institution_id, data=scope_data, ) elif old_adapter != new_adapter and new_adapter != "EMPLOYEE_ALL": # SELECT模式 + 适配类型变更：先改类型，再加员工 await InstitutionScopeService.scope_modify_service( auth=auth, institution_id=institution_id, data={ "enterprise_id": enterprise_id, "adapter_type": new_adapter, "owner_type": data.get("owner_type", "EMPLOYEE"), }, ) if alipay_add_ids or alipay_delete_ids: scope_data = { "enterprise_id": enterprise_id, "adapter_type": new_adapter, "owner_type": data.get("owner_type", "EMPLOYEE"), } if alipay_add_ids: scope_data["add_owner_id_list"] = alipay_add_ids if alipay_delete_ids: scope_data["delete_owner_id_list"] = alipay_delete_ids result = await InstitutionScopeService.scope_modify_service( auth=auth, institution_id=institution_id, data=scope_data, ) else: result = {"result": True} else: # ALL模式 或 类型不变 if alipay_add_ids or alipay_delete_ids: scope_data = { "enterprise_id": enterprise_id, "adapter_type": new_adapter, "owner_type": data.get("owner_type", "EMPLOYEE"), } if alipay_add_ids: scope_data["add_owner_id_list"] = alipay_add_ids if alipay_delete_ids: scope_data["delete_owner_id_list"] = alipay_delete_ids result = await InstitutionScopeService.scope_modify_service( auth=auth, institution_id=institution_id, data=scope_data, ) else: result = {"result": True} # ====== 5. 更新本地库 + 额度联动 ====== try: from .crud import InstitutionCRUD crud = InstitutionCRUD(auth) scope_map = {"EMPLOYEE_ALL": "all", "EMPLOYEE_SELECT": "employee", "EMPLOYEE_DEPARTMENT": "department"} applicable_scope = scope_map.get(new_adapter, "all") update_data = {"applicable_scope": applicable_scope} if new_adapter == "EMPLOYEE_DEPARTMENT": dept_ids = data.get("add_owner_id_list") or [] if dept_ids: update_data["department_id"] = str(dept_ids[0]) await crud.update_by_institution_id(institution_id, update_data) if enterprise_id: from .service import InstitutionService scope_info = { "adapter_type": new_adapter, "owner_type": "EMPLOYEE", "add_owner_id_list": add_emp_ids, "delete_owner_id_list": delete_emp_ids, } await InstitutionService._sync_modify_quotas_by_scope( auth=auth, institution_id=institution_id, enterprise_id=enterprise_id, scope_info=scope_info, raw_data={}, ) except Exception as e: log.warning(f"本地scope同步失败(不影响支付宝侧): {e}") log.info(f"设置成员成功: {old_adapter}→{new_adapter}, " f"加{len(add_ids)}人, 减{len(delete_ids)}人") return SuccessResponse(data=result, msg="设置成功") # ========== 自动额度发放规则管理 ========== @InstitutionRouter.post( "/{institution_id}/issuerule", summary="创建自动发放规则", description="创建自动额度发放规则 (alipay.ebpp.invoice.issuerule.create)", ) async def create_issuerule_controller( institution_id: Annotated[str, Path(description="制度ID")], data: dict, auth: Annotated[AuthSchema, Depends(AuthPermission(["module_payment:expense:institution:issuerule:create"]))], ) -> JSONResponse: """创建自动发放规则""" result = await IssueruleService.create_issuerule_service( auth=auth, institution_id=institution_id, enterprise_id=data.get("enterprise_id", ""), quota_type=data.get("quota_type", "CAP"), issue_type=data.get("issue_type", "ISSUE_MONTH"), issue_amount_value=data.get("issue_amount_value", "0"), outer_source_id=data.get("outer_source_id"), issue_rule_name=data.get("issue_rule_name"), effective_period=data.get("effective_period"), invalid_mode=data.get("invalid_mode"), share_mode=data.get("share_mode"), ) log.info(f"创建自动发放规则成功: institution_id={institution_id}") return SuccessResponse(data=result, msg="创建自动发放规则成功") @InstitutionRouter.put( "/{institution_id}/issuerule/{issue_rule_id}", summary="编辑自动发放规则", description="编辑自动额度发放规则 (alipay.ebpp.invoice.issuerule.modify)", ) async def modify_issuerule_controller( institution_id: Annotated[str, Path(description="制度ID")], issue_rule_id: Annotated[str, Path(description="发放规则ID")], data: dict, auth: Annotated[AuthSchema, Depends(AuthPermission(["module_payment:expense:institution:issuerule:modify"]))], ) -> JSONResponse: result = await IssueruleService.modify_issuerule_service( auth=auth, institution_id=institution_id, issue_rule_id=issue_rule_id, enterprise_id=data.get("enterprise_id", ""), quota_type=data.get("quota_type"), issue_type=data.get("issue_type"), issue_amount_value=data.get("issue_amount_value"), issue_rule_name=data.get("issue_rule_name"), effective=data.get("effective"), effective_period=data.get("effective_period"), invalid_mode=data.get("invalid_mode"), share_mode=data.get("share_mode"), ) log.info(f"编辑自动发放规则成功: issue_rule_id={issue_rule_id}") return SuccessResponse(data=result, msg="编辑自动发放规则成功") @InstitutionRouter.delete( "/{institution_id}/issuerule", summary="删除自动发放规则", description="删除自动额度发放规则 (alipay.ebpp.invoice.issuerule.delete)", ) async def delete_issuerule_controller( institution_id: Annotated[str, Path(description="制度ID")], data: dict, auth: Annotated[AuthSchema, Depends(AuthPermission(["module_payment:expense:institution:issuerule:delete"]))], ) -> JSONResponse: result = await IssueruleService.delete_issuerule_service( auth=auth, institution_id=institution_id, issue_rule_id_list=data.get("issue_rule_id_list", []), enterprise_id=data.get("enterprise_id", ""), ) log.info(f"删除自动发放规则成功: institution_id={institution_id}") return SuccessResponse(data=result, msg="删除自动发放规则成功")