"""保险公司及险种价格管理 API（保险公司 + 开票信息 + 险种价格）""" import math import uuid from datetime import datetime from typing import Any, List, Optional from fastapi import APIRouter, Depends, Query from pydantic import BaseModel, ConfigDict, Field from sqlalchemy import or_ from sqlalchemy.orm import Session from app.core.database import get_db from app.core.menu_guard import require_admin_or_super_admin from app.core.response import error_response, success_response from app.models.models import ( BizInsuranceCompany, BizInsuranceCompanyInvoice, BizInsuranceProjectPrice, SysUser, ) router = APIRouter(prefix="/web/admin", tags=["保险公司及险种价格管理"]) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _get_tenant_no(db: Session, user: dict) -> str: user_id = user.get("user_id") if not user_id: raise PermissionError("用户身份信息缺失，无法确定租户") sys_user = db.query(SysUser).filter(SysUser.id == user_id).first() if not sys_user or not sys_user.tenant_no: raise PermissionError("用户租户信息缺失") return sys_user.tenant_no def _generate_uuid() -> str: return str(uuid.uuid4()).replace("-", "") def _fmt_dt(v) -> Optional[str]: if v is None: return None if isinstance(v, datetime): return v.strftime("%Y-%m-%d %H:%M:%S") return str(v) def _total_pages(total: int, page_size: int) -> int: return math.ceil(total / page_size) if page_size > 0 else 0 # --------------------------------------------------------------------------- # Row-to-dict helpers (camelCase JSON) # --------------------------------------------------------------------------- def _row_to_dict_company(row: BizInsuranceCompany, project_prices: Optional[List[dict]] = None) -> dict: data = { "id": row.id, "companyName": row.company_name, "companyAddress": row.company_address, "contactName": row.contact_name, "contactPhone": row.contact_phone, "companyLogo": row.company_logo or "", "status": row.status, "tenantNo": row.tenant_no, "orderNum": row.order_num, "remark": row.remark, "createBy": row.create_by, "createDeptId": row.create_dept_id, "createTime": _fmt_dt(row.create_time), "updateBy": row.update_by, "updateTime": _fmt_dt(row.update_time), } if project_prices is not None: data["projectPrices"] = project_prices return data def _row_to_dict_invoice(row: BizInsuranceCompanyInvoice) -> dict: return { "id": row.id, "insuranceCompanyId": row.insurance_company_id, "companyName": row.company_name, "taxNo": row.tax_no, "companyAddress": row.company_address, "companyPhone": row.company_phone, "bankName": row.bank_name, "bankAccount": row.bank_account, "tenantNo": row.tenant_no, "orderNum": row.order_num, "remark": row.remark, "createBy": row.create_by, "createDeptId": row.create_dept_id, "createTime": _fmt_dt(row.create_time), "updateBy": row.update_by, "updateTime": _fmt_dt(row.update_time), } def _row_to_dict_project_price(row: BizInsuranceProjectPrice) -> dict: return { "id": row.id, "insuranceCompanyId": row.insurance_company_id, "projectName": row.project_name, "projectFee": float(row.project_fee) if row.project_fee is not None else None, "status": row.status, "tenantNo": row.tenant_no, "orderNum": row.order_num, "remark": row.remark, "createBy": row.create_by, "createDeptId": row.create_dept_id, "createTime": _fmt_dt(row.create_time), "updateBy": row.update_by, "updateTime": _fmt_dt(row.update_time), } # --------------------------------------------------------------------------- # Inline CRUD — 保险公司主表 # --------------------------------------------------------------------------- def _list_insurance_companies( db: Session, *, page: int, page_size: int, keyword: Optional[str], status: Optional[str], tenant_no: str, user_info=None, ): q = db.query(BizInsuranceCompany).filter( BizInsuranceCompany.tenant_no == tenant_no, BizInsuranceCompany.del_flag == 0, ) kw = (keyword or "").strip() if kw: like = f"%{kw}%" q = q.filter( or_( BizInsuranceCompany.company_name.like(like), BizInsuranceCompany.contact_name.like(like), ) ) st = (status or "").strip() if st in ("0", "1"): q = q.filter(BizInsuranceCompany.status == st) if user_info: from app.core.data_scope import apply_data_scope q = apply_data_scope(q, BizInsuranceCompany, user_info, db, tenant_no) total = q.count() rows = ( q.order_by(BizInsuranceCompany.create_time.desc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) company_ids = [r.id for r in rows] prices_by_company: dict[str, list] = {cid: [] for cid in company_ids} if company_ids: prices_q = db.query(BizInsuranceProjectPrice).filter( BizInsuranceProjectPrice.insurance_company_id.in_(company_ids), BizInsuranceProjectPrice.tenant_no == tenant_no, BizInsuranceProjectPrice.del_flag == 0, ) if user_info: from app.core.data_scope import apply_data_scope prices_q = apply_data_scope(prices_q, BizInsuranceProjectPrice, user_info, db, tenant_no) for price_row in prices_q.order_by(BizInsuranceProjectPrice.create_time.desc()).all(): prices_by_company.setdefault(price_row.insurance_company_id, []).append( _row_to_dict_project_price(price_row) ) return [ _row_to_dict_company(r, prices_by_company.get(r.id, [])) for r in rows ], total def _create_insurance_company( db: Session, *, company_name: str, company_address: Optional[str], contact_name: Optional[str], contact_phone: Optional[str], company_logo: Optional[str], status: Optional[str], tenant_no: str, order_num: Optional[int], remark: Optional[str], user_id: str, ) -> BizInsuranceCompany: obj = BizInsuranceCompany( id=_generate_uuid(), company_name=company_name, company_address=company_address, contact_name=contact_name, contact_phone=contact_phone, company_logo=(company_logo or "").strip() or None, status=status or "1", tenant_no=tenant_no, order_num=order_num or 0, remark=remark, create_by=user_id, create_dept_id="", create_time=datetime.now(), update_by=user_id, update_time=datetime.now(), del_flag=0, ) db.add(obj) db.commit() db.refresh(obj) return obj def _update_insurance_company( db: Session, row_id: str, *, company_name: Optional[str], company_address: Optional[str], contact_name: Optional[str], contact_phone: Optional[str], company_logo: Optional[str], status: Optional[str], order_num: Optional[int], remark: Optional[str], user_id: str, tenant_no: str, ) -> BizInsuranceCompany: obj = ( db.query(BizInsuranceCompany) .filter( BizInsuranceCompany.id == row_id, BizInsuranceCompany.tenant_no == tenant_no, BizInsuranceCompany.del_flag == 0, ) .first() ) if not obj: raise ValueError("记录不存在") if company_name is not None: obj.company_name = company_name if company_address is not None: obj.company_address = company_address if contact_name is not None: obj.contact_name = contact_name if contact_phone is not None: obj.contact_phone = contact_phone if company_logo is not None: obj.company_logo = (company_logo or "").strip() or None if status is not None: obj.status = status if order_num is not None: obj.order_num = order_num if remark is not None: obj.remark = remark obj.update_by = user_id obj.update_time = datetime.now() db.commit() db.refresh(obj) return obj def _soft_delete_insurance_company(db: Session, row_id: str, *, tenant_no: str): obj = ( db.query(BizInsuranceCompany) .filter( BizInsuranceCompany.id == row_id, BizInsuranceCompany.tenant_no == tenant_no, BizInsuranceCompany.del_flag == 0, ) .first() ) if not obj: raise ValueError("记录不存在") obj.del_flag = 1 db.commit() # --------------------------------------------------------------------------- # Inline CRUD — 险种开票信息 # --------------------------------------------------------------------------- def _list_insurance_company_invoices( db: Session, insurance_company_id: str, *, tenant_no: str, user_info=None ): # Verify parent exists parent = ( db.query(BizInsuranceCompany) .filter( BizInsuranceCompany.id == insurance_company_id, BizInsuranceCompany.tenant_no == tenant_no, BizInsuranceCompany.del_flag == 0, ) .first() ) if not parent: raise ValueError("保险公司不存在") q = db.query(BizInsuranceCompanyInvoice).filter( BizInsuranceCompanyInvoice.insurance_company_id == insurance_company_id, BizInsuranceCompanyInvoice.tenant_no == tenant_no, BizInsuranceCompanyInvoice.del_flag == 0, ) if user_info: from app.core.data_scope import apply_data_scope q = apply_data_scope(q, BizInsuranceCompanyInvoice, user_info, db, tenant_no) rows = q.order_by(BizInsuranceCompanyInvoice.create_time.desc()).all() return [_row_to_dict_invoice(r) for r in rows] def _create_insurance_company_invoice( db: Session, *, insurance_company_id: str, company_name: str, tax_no: Optional[str], company_address: Optional[str], company_phone: Optional[str], bank_name: Optional[str], bank_account: Optional[str], tenant_no: str, order_num: Optional[int], remark: Optional[str], user_id: str, ) -> BizInsuranceCompanyInvoice: obj = BizInsuranceCompanyInvoice( id=_generate_uuid(), insurance_company_id=insurance_company_id, company_name=company_name, tax_no=tax_no, company_address=company_address, company_phone=company_phone, bank_name=bank_name, bank_account=bank_account, tenant_no=tenant_no, order_num=order_num or 0, remark=remark, create_by=user_id, create_dept_id="", create_time=datetime.now(), update_by=user_id, update_time=datetime.now(), del_flag=0, ) db.add(obj) db.commit() db.refresh(obj) return obj def _update_insurance_company_invoice( db: Session, row_id: str, *, company_name: Optional[str], tax_no: Optional[str], company_address: Optional[str], company_phone: Optional[str], bank_name: Optional[str], bank_account: Optional[str], order_num: Optional[int], remark: Optional[str], user_id: str, tenant_no: str, ) -> BizInsuranceCompanyInvoice: obj = ( db.query(BizInsuranceCompanyInvoice) .filter( BizInsuranceCompanyInvoice.id == row_id, BizInsuranceCompanyInvoice.tenant_no == tenant_no, BizInsuranceCompanyInvoice.del_flag == 0, ) .first() ) if not obj: raise ValueError("记录不存在") if company_name is not None: obj.company_name = company_name if tax_no is not None: obj.tax_no = tax_no if company_address is not None: obj.company_address = company_address if company_phone is not None: obj.company_phone = company_phone if bank_name is not None: obj.bank_name = bank_name if bank_account is not None: obj.bank_account = bank_account if order_num is not None: obj.order_num = order_num if remark is not None: obj.remark = remark obj.update_by = user_id obj.update_time = datetime.now() db.commit() db.refresh(obj) return obj def _soft_delete_insurance_company_invoice(db: Session, row_id: str, *, tenant_no: str): obj = ( db.query(BizInsuranceCompanyInvoice) .filter( BizInsuranceCompanyInvoice.id == row_id, BizInsuranceCompanyInvoice.tenant_no == tenant_no, BizInsuranceCompanyInvoice.del_flag == 0, ) .first() ) if not obj: raise ValueError("记录不存在") obj.del_flag = 1 db.commit() # --------------------------------------------------------------------------- # Inline CRUD — 险种价格 # --------------------------------------------------------------------------- def _list_insurance_project_prices( db: Session, insurance_company_id: str, *, tenant_no: str, user_info=None ): parent = ( db.query(BizInsuranceCompany) .filter( BizInsuranceCompany.id == insurance_company_id, BizInsuranceCompany.tenant_no == tenant_no, BizInsuranceCompany.del_flag == 0, ) .first() ) if not parent: raise ValueError("保险公司不存在") q = db.query(BizInsuranceProjectPrice).filter( BizInsuranceProjectPrice.insurance_company_id == insurance_company_id, BizInsuranceProjectPrice.tenant_no == tenant_no, BizInsuranceProjectPrice.del_flag == 0, ) if user_info: from app.core.data_scope import apply_data_scope q = apply_data_scope(q, BizInsuranceProjectPrice, user_info, db, tenant_no) rows = q.order_by(BizInsuranceProjectPrice.create_time.desc()).all() return [_row_to_dict_project_price(r) for r in rows] def _create_insurance_project_price( db: Session, *, insurance_company_id: str, project_name: str, project_fee: Optional[float], status: Optional[str], tenant_no: str, order_num: Optional[int], remark: Optional[str], user_id: str, ) -> BizInsuranceProjectPrice: obj = BizInsuranceProjectPrice( id=_generate_uuid(), insurance_company_id=insurance_company_id, project_name=project_name, project_fee=project_fee, status=status or "1", tenant_no=tenant_no, order_num=order_num or 0, remark=remark, create_by=user_id, create_dept_id="", create_time=datetime.now(), update_by=user_id, update_time=datetime.now(), del_flag=0, ) db.add(obj) db.commit() db.refresh(obj) return obj def _update_insurance_project_price( db: Session, row_id: str, *, project_name: Optional[str], project_fee: Optional[float], status: Optional[str], order_num: Optional[int], remark: Optional[str], user_id: str, tenant_no: str, ) -> BizInsuranceProjectPrice: obj = ( db.query(BizInsuranceProjectPrice) .filter( BizInsuranceProjectPrice.id == row_id, BizInsuranceProjectPrice.tenant_no == tenant_no, BizInsuranceProjectPrice.del_flag == 0, ) .first() ) if not obj: raise ValueError("记录不存在") if project_name is not None: obj.project_name = project_name if project_fee is not None: obj.project_fee = project_fee if status is not None: obj.status = status if order_num is not None: obj.order_num = order_num if remark is not None: obj.remark = remark obj.update_by = user_id obj.update_time = datetime.now() db.commit() db.refresh(obj) return obj def _soft_delete_insurance_project_price(db: Session, row_id: str, *, tenant_no: str): obj = ( db.query(BizInsuranceProjectPrice) .filter( BizInsuranceProjectPrice.id == row_id, BizInsuranceProjectPrice.tenant_no == tenant_no, BizInsuranceProjectPrice.del_flag == 0, ) .first() ) if not obj: raise ValueError("记录不存在") obj.del_flag = 1 db.commit() # --------------------------------------------------------------------------- # 1. 保险公司（主表） # --------------------------------------------------------------------------- class InsuranceCompanyCreateBody(BaseModel): model_config = ConfigDict(populate_by_name=True) company_name: str = Field(..., alias="companyName") company_address: Optional[str] = Field(None, alias="companyAddress") contact_name: Optional[str] = Field(None, alias="contactName") contact_phone: Optional[str] = Field(None, alias="contactPhone") company_logo: Optional[str] = Field(None, alias="companyLogo") status: Optional[str] = Field("1", alias="status") order_num: Optional[int] = Field(0, alias="orderNum") remark: Optional[str] = None class InsuranceCompanyUpdateBody(BaseModel): model_config = ConfigDict(populate_by_name=True) company_name: Optional[str] = Field(None, alias="companyName") company_address: Optional[str] = Field(None, alias="companyAddress") contact_name: Optional[str] = Field(None, alias="contactName") contact_phone: Optional[str] = Field(None, alias="contactPhone") company_logo: Optional[str] = Field(None, alias="companyLogo") status: Optional[str] = Field(None, alias="status") order_num: Optional[int] = Field(None, alias="orderNum") remark: Optional[str] = None @router.get("/insurance-companies") async def list_insurance_companies( page: int = Query(1, ge=1), pageSize: int = Query(10, ge=1, le=100, alias="pageSize"), keyword: Optional[str] = None, status: Optional[str] = None, user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): tenant_no = _get_tenant_no(db, user) items, total = _list_insurance_companies( db, page=page, page_size=pageSize, keyword=keyword, status=status, tenant_no=tenant_no, user_info=user, ) tp = _total_pages(total, pageSize) return success_response( {"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp} ) @router.post("/insurance-companies") async def create_insurance_company( body: InsuranceCompanyCreateBody, user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): try: tenant_no = _get_tenant_no(db, user) obj = _create_insurance_company( db, company_name=body.company_name, company_address=body.company_address, contact_name=body.contact_name, contact_phone=body.contact_phone, company_logo=body.company_logo, status=body.status, tenant_no=tenant_no, order_num=body.order_num, remark=body.remark, user_id=user.get("user_id"), ) return success_response(_row_to_dict_company(obj), message="创建成功") except ValueError as e: return error_response(400, str(e)) @router.put("/insurance-companies/{row_id}") async def update_insurance_company( row_id: str, body: InsuranceCompanyUpdateBody, user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): try: tenant_no = _get_tenant_no(db, user) payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False) obj = _update_insurance_company( db, row_id, company_name=payload.get("company_name"), company_address=payload.get("company_address"), contact_name=payload.get("contact_name"), contact_phone=payload.get("contact_phone"), company_logo=payload.get("company_logo"), status=payload.get("status"), order_num=payload.get("order_num"), remark=payload.get("remark"), user_id=user.get("user_id"), tenant_no=tenant_no, ) return success_response(_row_to_dict_company(obj), message="更新成功") except ValueError as e: return error_response(400, str(e)) @router.delete("/insurance-companies/{row_id}") async def delete_insurance_company( row_id: str, user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): try: tenant_no = _get_tenant_no(db, user) _soft_delete_insurance_company(db, row_id, tenant_no=tenant_no) return success_response(None, message="删除成功") except ValueError as e: return error_response(400, str(e)) # --------------------------------------------------------------------------- # 2. 保险公司开票信息（子表） # --------------------------------------------------------------------------- class InsuranceCompanyInvoiceCreateBody(BaseModel): model_config = ConfigDict(populate_by_name=True) company_name: str = Field(..., alias="companyName") tax_no: Optional[str] = Field(None, alias="taxNo") company_address: Optional[str] = Field(None, alias="companyAddress") company_phone: Optional[str] = Field(None, alias="companyPhone") bank_name: Optional[str] = Field(None, alias="bankName") bank_account: Optional[str] = Field(None, alias="bankAccount") order_num: Optional[int] = Field(0, alias="orderNum") remark: Optional[str] = None class InsuranceCompanyInvoiceUpdateBody(BaseModel): model_config = ConfigDict(populate_by_name=True) company_name: Optional[str] = Field(None, alias="companyName") tax_no: Optional[str] = Field(None, alias="taxNo") company_address: Optional[str] = Field(None, alias="companyAddress") company_phone: Optional[str] = Field(None, alias="companyPhone") bank_name: Optional[str] = Field(None, alias="bankName") bank_account: Optional[str] = Field(None, alias="bankAccount") order_num: Optional[int] = Field(None, alias="orderNum") remark: Optional[str] = None @router.get("/insurance-companies/{company_id}/invoices") async def list_insurance_company_invoices( company_id: str, user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): try: tenant_no = _get_tenant_no(db, user) items = _list_insurance_company_invoices(db, company_id, tenant_no=tenant_no, user_info=user) return success_response(items) except ValueError as e: return error_response(400, str(e)) @router.post("/insurance-companies/{company_id}/invoices") async def create_insurance_company_invoice( company_id: str, body: InsuranceCompanyInvoiceCreateBody, user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): try: tenant_no = _get_tenant_no(db, user) obj = _create_insurance_company_invoice( db, insurance_company_id=company_id, company_name=body.company_name, tax_no=body.tax_no, company_address=body.company_address, company_phone=body.company_phone, bank_name=body.bank_name, bank_account=body.bank_account, tenant_no=tenant_no, order_num=body.order_num, remark=body.remark, user_id=user.get("user_id"), ) return success_response(_row_to_dict_invoice(obj), message="创建成功") except ValueError as e: return error_response(400, str(e)) @router.put("/insurance-company-invoices/{invoice_id}") async def update_insurance_company_invoice( invoice_id: str, body: InsuranceCompanyInvoiceUpdateBody, user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): try: tenant_no = _get_tenant_no(db, user) payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False) obj = _update_insurance_company_invoice( db, invoice_id, company_name=payload.get("company_name"), tax_no=payload.get("tax_no"), company_address=payload.get("company_address"), company_phone=payload.get("company_phone"), bank_name=payload.get("bank_name"), bank_account=payload.get("bank_account"), order_num=payload.get("order_num"), remark=payload.get("remark"), user_id=user.get("user_id"), tenant_no=tenant_no, ) return success_response(_row_to_dict_invoice(obj), message="更新成功") except ValueError as e: return error_response(400, str(e)) @router.delete("/insurance-company-invoices/{invoice_id}") async def delete_insurance_company_invoice( invoice_id: str, user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): try: tenant_no = _get_tenant_no(db, user) _soft_delete_insurance_company_invoice(db, invoice_id, tenant_no=tenant_no) return success_response(None, message="删除成功") except ValueError as e: return error_response(400, str(e)) # --------------------------------------------------------------------------- # 3. 险种价格（子表） # --------------------------------------------------------------------------- class InsuranceProjectPriceCreateBody(BaseModel): model_config = ConfigDict(populate_by_name=True) project_name: str = Field(..., alias="projectName") project_fee: Optional[float] = Field(None, alias="projectFee") status: Optional[str] = Field("1", alias="status") order_num: Optional[int] = Field(0, alias="orderNum") remark: Optional[str] = None class InsuranceProjectPriceUpdateBody(BaseModel): model_config = ConfigDict(populate_by_name=True) project_name: Optional[str] = Field(None, alias="projectName") project_fee: Optional[float] = Field(None, alias="projectFee") status: Optional[str] = Field(None, alias="status") order_num: Optional[int] = Field(None, alias="orderNum") remark: Optional[str] = None @router.get("/insurance-companies/{company_id}/project-prices") async def list_insurance_project_prices( company_id: str, user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): try: tenant_no = _get_tenant_no(db, user) items = _list_insurance_project_prices(db, company_id, tenant_no=tenant_no, user_info=user) return success_response(items) except ValueError as e: return error_response(400, str(e)) @router.post("/insurance-companies/{company_id}/project-prices") async def create_insurance_project_price( company_id: str, body: InsuranceProjectPriceCreateBody, user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): try: tenant_no = _get_tenant_no(db, user) obj = _create_insurance_project_price( db, insurance_company_id=company_id, project_name=body.project_name, project_fee=body.project_fee, status=body.status, tenant_no=tenant_no, order_num=body.order_num, remark=body.remark, user_id=user.get("user_id"), ) return success_response(_row_to_dict_project_price(obj), message="创建成功") except ValueError as e: return error_response(400, str(e)) @router.put("/insurance-project-prices/{price_id}") async def update_insurance_project_price( price_id: str, body: InsuranceProjectPriceUpdateBody, user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): try: tenant_no = _get_tenant_no(db, user) payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False) obj = _update_insurance_project_price( db, price_id, project_name=payload.get("project_name"), project_fee=payload.get("project_fee"), status=payload.get("status"), order_num=payload.get("order_num"), remark=payload.get("remark"), user_id=user.get("user_id"), tenant_no=tenant_no, ) return success_response(_row_to_dict_project_price(obj), message="更新成功") except ValueError as e: return error_response(400, str(e)) @router.delete("/insurance-project-prices/{price_id}") async def delete_insurance_project_price( price_id: str, user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): try: tenant_no = _get_tenant_no(db, user) _soft_delete_insurance_project_price(db, price_id, tenant_no=tenant_no) return success_response(None, message="删除成功") except ValueError as e: return error_response(400, str(e))