backend/README.md

@@ -114,6 +114,21 @@ app/
 **登录失败一律 400 或 403，绝不使用 401** —— 前端把 401 一律当作「清 token + 跳登录页」，
 若登录接口自身返回 401，用户输错一次密码就会触发「清空 + 跳转 + 报错」的风暴。
 
+### 报表统计接口（M4 ⑤，限管理员/财务/领导）
+
+| 方法 | 路径 | 说明 |
+| --- | --- | --- |
+| GET | `/api/v1/reports/personal-detail` | 个人绩效工资明细表（FR-029：人员/月份/客户类型/产品类别/岗位筛选；响应带 `summary` 全筛选合计） |
+| GET | `/api/v1/reports/department-summary` | 部门绩效工资汇总表（FR-030：月份必填；团队基数与月结聚合行同源；12.5% 与 5% 分列；含成员内联明细、未分配桶、月结冻结对照额） |
+| GET | `/api/v1/reports/business-stats` | 业务统计表（FR-031：发货/回款/绩效为业务累计值，不随业务日期筛选变化） |
+| GET | `/api/v1/reports/payment-stats` | 回款情况统计表（FR-031：8 档天数区间恒在，空桶为 0） |
+| GET | `/api/v1/reports/payment-detail` | 回款区间下钻（`bucket` 必填，未注册编码 400） |
+| GET | `/api/v1/reports/payment-day-buckets` | 区间档位元数据（前端下拉唯一来源，禁止前端硬编码） |
+| GET | `/api/v1/reports/{code}/export` | Excel 导出（文件流非统一信封；含公司页眉/财务复核栏/领导签字栏；超 2 万行 400；行为入操作日志） |
+
+准入 `require_roles(admin, finance, leader)`（FR-030/031；FR-029 的「营销后勤授权范围内」
+待业务权限清单 #6）。前端菜单同层过滤（路由 `allowedRoles`）。
+
 ## 常用命令
 
 ```bash
@@ -196,6 +211,16 @@ bcrypt 5.0 起对超过 72 字节的密码**直接抛 ValueError**（此前是
 - **留痕表的固化设计**：`ownership_change_logs` 与 `business_status_logs` 对客户/业务/人员 ID 均**无外键**（冗余固化，被引用方删除后记录仍完整可读）；RESTRICT 外键反而会把删除逼成 500。删除保护全部由服务层检查承担（人员→团队成员/客户归属引用；客户→业务引用；业务→发货/回款记录）。两张留痕表 append-only、不继承 TimestampMixin。
 - **金额契约**：schema 入参出参一律 `str`，小数超两位直接拒绝（不静默舍入）；服务层 `Decimal`，比率 `Fraction`；毛利润率出参为分数原样字符串（`"997/1000"`，同 M1 计算预览口径），前端原样展示禁止转 float。
 
+### 报表口径（M4 ⑤，改动前先读）
+
+- **数据源固定单笔生效行**：四张报表的明细一律 `performance_ledgers.is_active=1 AND calc_type=1`；经理聚合行（`calc_type=2`）**只在部门汇总作冻结对照**（`frozen_amount`）单读。聚合行 `receipt_id` 为 NULL、`payment_days` 为 NULL，混入明细会让合同编号/天数列失真，业务统计的绩效也会被整团队聚合额污染。
+- **调整单并入**：仅 `status=2`（生效），按 `(settlement_month, personnel_id)` 并入；`direction=1` 补发记 +、`2` 扣减记 −（金额恒正）。待审批/驳回/撤回的金额绝不能进报表——`report_repository.list_month_adjustments` 与调整单管理页的 `list_adjustments` 是两个口径。
+- **回款金额按回款单去重**：二次分摊拆行使同一回款在台账出现多行（负责人 + 参与人），回款统计的金额侧必须 `DISTINCT receipt_id` 后再求和，否则按拆行数翻倍。个人明细页不放「回款金额」列（同一金额重复展示会让合计无法自洽），金额下沉到快照下钻。
+- **部门口径** = `personnel.manager_id` 实时上溯（无独立部门表）；离职人员不按 `status` 过滤，历史业绩留在原团队；无成员经理也出 0 值行（在职经理枚举 ∪ 数据中出现的经理）。团队基数与月结聚合行同源（`services/manager_share.py` 共享模块），**12.5% 与 5% 分列**——两项之和才是聚合行金额。
+- **回款天数区间**是 8 档代码常量（`core/report_meta.py`，含「≤0 天先款后货」与「无天数」兜底桶），服务端下发元数据、前端禁止硬编码；空桶恒在且为 0。进度登记回款无台账行，不在回款统计口径内（页面注明）。
+- **导出**：openpyxl 普通 Workbook + `Decimal` 原值 + `number_format`（不写 Excel 公式）；超 `REPORT_EXPORT_MAX_ROWS`（20000）行返回 400；文本列防公式注入（`= + - @` 前缀加 `'`）；服务端文件名为 ASCII（响应头 latin-1 限制），中文名由前端 `link.download` 指定。
+- **汇总 = 明细合计**：团队/公司基数由台账分桶结果直接得出，报表侧不另算一遍；集成测试逐项断言。
+
 ## 常见问题
 
 **起服务报 `APP_SECRET_KEY` 相关错误** —— `backend/.env` 未填或不足 32 字符。这是刻意的 fail-fast：空密钥签发 JWT 是安全问题，必须启动即失败。

backend/app/api/deps.py

@@ -119,20 +119,23 @@ AdminUser = Annotated[User, Depends(require_admin)]
 LoginUser = Annotated[User, Depends(get_current_user)]
 
 
-def require_roles(*role_codes: str):
+def require_roles(*role_codes: str, message: str = "无权访问审批管理"):
     """要求当前用户具备任一指定角色（依赖工厂）。
 
     与 require_admin 的差别：admin 不是天然放行——传入集合里显式含
     ROLE_ADMIN 才可见。用于审批等模块的粗粒度准入；节点级准入仍由
     service 按单据状态二次校验（双层：模块可见 ≠ 可操作）。
 
+    ``message`` 允许各模块替换 403 文案（默认值保持审批模块原文案，
+    避免既有调用与测试变动）。
+
     :raises ForbiddenError: 已登录但不具备任一指定角色（403）
     """
 
     def _checker(current_user: Annotated[User, Depends(get_current_user)]) -> User:
         codes = set(user_repository.get_role_codes(current_user))
         if not codes & set(role_codes):
-            raise ForbiddenError("无权访问审批管理")
+            raise ForbiddenError(message)
         return current_user
 
     return _checker

backend/app/api/router.py

@@ -16,6 +16,7 @@ from app.api.v1.business import ratings as business_ratings
 from app.api.v1.business import receipts as business_receipts
 from app.api.v1.business import shipments as business_shipments
 from app.api.v1.calc import router as calc_router
+from app.api.v1.reports import router as reports_router
 from app.api.v1.settlements import router as settlements_router
 from app.api.v1.system import approvals as system_approvals
 from app.api.v1.system import business_roles as system_business_roles
@@ -65,5 +66,7 @@ v1_router.include_router(business_router)
 v1_router.include_router(calc_router)
 # 月度结算（M3 ④）：/api/v1/settlements/*，读登录即可、写限管理员与营销后勤
 v1_router.include_router(settlements_router)
+# 报表统计（M4 ⑤）：/api/v1/reports/*，只读，限管理员/财务/领导
+v1_router.include_router(reports_router)
 
 api_router.include_router(v1_router)

backend/app/api/v1/reports.py

@@ -0,0 +1,178 @@
+"""报表统计接口（M4 ⑤）：四张报表只读查询 + Excel 导出。
+
+准入：管理员/财务/领导（FR-030/031 明确；FR-029 的营销后勤「授权范围内」
+待权限清单 #6，暂不开放）。前后端双层：前端 allowedRoles 控菜单，本层
+require_roles 兜底。
+
+导出返回文件流（非统一信封），沿用参数导出先例：服务端文件名为 ASCII
+（HTTP 头按 latin-1 编码，中文名会 UnicodeEncodeError），中文名由前端
+``link.download`` 指定。
+"""
+
+from datetime import date
+from typing import Annotated, Literal
+
+from fastapi import APIRouter, Depends, Path, Query
+from fastapi.responses import StreamingResponse
+from pydantic import BaseModel
+
+from app.api.deps import DbSession, RequestMetaDep, require_roles
+from app.core.constants import ROLE_ADMIN, ROLE_FINANCE, ROLE_LEADER
+from app.core.response import ApiResponse, ok
+from app.models import User
+from app.schemas.report import (
+    BusinessStatsQuery,
+    BusinessStatsRow,
+    DepartmentSummaryOut,
+    DepartmentSummaryQuery,
+    PaymentDetailQuery,
+    PaymentDetailRow,
+    PaymentStatsOut,
+    PaymentStatsQuery,
+    PersonalDetailQuery,
+    PersonalDetailRow,
+    ReportPage,
+)
+from app.services import report_service
+
+router = APIRouter(prefix="/reports", tags=["绩效工资 · 报表统计"])
+
+_ReportViewer = Annotated[
+    User,
+    Depends(require_roles(ROLE_ADMIN, ROLE_FINANCE, ROLE_LEADER, message="无权访问报表统计")),
+]
+
+
+@router.get(
+    "/personal-detail",
+    response_model=ApiResponse[ReportPage[PersonalDetailRow]],
+    summary="个人绩效工资明细表",
+)
+def personal_detail(
+    session: DbSession,
+    _: _ReportViewer,
+    query: Annotated[PersonalDetailQuery, Query()],
+) -> ApiResponse[ReportPage[PersonalDetailRow]]:
+    """按人员/月份/客户类型/产品类别/岗位查明细（口径与台账一致）。"""
+    return ok(report_service.list_personal_detail(session, query))
+
+
+@router.get(
+    "/department-summary",
+    response_model=ApiResponse[DepartmentSummaryOut],
+    summary="部门绩效工资汇总表",
+)
+def department_summary(
+    session: DbSession,
+    _: _ReportViewer,
+    query: Annotated[DepartmentSummaryQuery, Query()],
+) -> ApiResponse[DepartmentSummaryOut]:
+    """按经理团队/月份汇总（团队基数与月结聚合行同源）。"""
+    return ok(report_service.department_summary(session, query))
+
+
+@router.get(
+    "/business-stats",
+    response_model=ApiResponse[ReportPage[BusinessStatsRow]],
+    summary="业务统计表",
+)
+def business_stats(
+    session: DbSession,
+    _: _ReportViewer,
+    query: Annotated[BusinessStatsQuery, Query()],
+) -> ApiResponse[ReportPage[BusinessStatsRow]]:
+    """按业务查询发货、回款、绩效（回款/绩效为业务累计值）。"""
+    return ok(report_service.business_stats(session, query))
+
+
+@router.get(
+    "/payment-stats",
+    response_model=ApiResponse[PaymentStatsOut],
+    summary="回款情况统计表",
+)
+def payment_stats(
+    session: DbSession,
+    _: _ReportViewer,
+    query: Annotated[PaymentStatsQuery, Query()],
+) -> ApiResponse[PaymentStatsOut]:
+    """按回款天数区间/客户类型/产品类别统计（空桶恒在且为 0）。"""
+    return ok(report_service.payment_stats(session, query))
+
+
+@router.get(
+    "/payment-detail",
+    response_model=ApiResponse[ReportPage[PaymentDetailRow]],
+    summary="回款区间下钻",
+)
+def payment_detail(
+    session: DbSession,
+    _: _ReportViewer,
+    query: Annotated[PaymentDetailQuery, Query()],
+) -> ApiResponse[ReportPage[PaymentDetailRow]]:
+    """区间桶下钻到单笔台账行（含回款金额与批次号）。"""
+    return ok(report_service.payment_detail(session, query))
+
+
+@router.get(
+    "/payment-day-buckets", response_model=ApiResponse[list[dict]], summary="回款天数区间档位"
+)
+def payment_day_buckets(_: _ReportViewer) -> ApiResponse[list[dict]]:
+    """区间档位元数据（前端渲染下拉/表头，禁止前端硬编码档位）。"""
+    return ok(report_service.get_payment_day_buckets())
+
+
+class ExportQuery(BaseModel):
+    """导出通用筛选参数（按报表编码取用相关字段，未用字段忽略）。"""
+
+    settlement_month: str | None = None
+    personnel_id: int | None = None
+    customer_type: str | None = None
+    product_category: str | None = None
+    role_code: str | None = None
+    manager_id: int | None = None
+    keyword: str | None = None
+    customer_id: int | None = None
+    owner_personnel_id: int | None = None
+    status: int | None = None
+    date_from: date | None = None
+    date_to: date | None = None
+
+
+@router.get("/{code}/export", summary="导出报表 Excel")
+def export_report(
+    code: Annotated[
+        Literal["personal-detail", "department-summary", "business-stats", "payment-stats"],
+        Path(description="报表编码"),
+    ],
+    session: DbSession,
+    current_user: _ReportViewer,
+    meta: RequestMetaDep,
+    query: Annotated[ExportQuery, Query()],
+) -> StreamingResponse:
+    """导出 xlsx（含公司页眉/财务复核栏/领导签字栏），行为入操作日志。"""
+    content, filename = report_service.export_report(
+        session,
+        code,
+        _build_export_query(code, query),
+        operator=current_user,
+        meta=meta,
+    )
+    return StreamingResponse(
+        iter([content]),
+        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+        headers={
+            "Content-Disposition": (f'attachment; filename="{filename}_{date.today():%Y%m%d}.xlsx"')
+        },
+    )
+
+
+def _build_export_query(code: str, query: ExportQuery):
+    """导出参数 → 各报表自己的查询模型（department-summary 月份必填由模型校验）。"""
+    data = query.model_dump()
+    if code == "personal-detail":
+        return PersonalDetailQuery(**data)
+    if code == "department-summary":
+        return DepartmentSummaryQuery(**data)
+    if code == "business-stats":
+        return BusinessStatsQuery(**data)
+    return PaymentStatsQuery(**data)

backend/app/repositories/report_repository.py

@@ -15,6 +15,7 @@
 from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
+from app.core.report_meta import DayBucket
 from app.models import (
     AdjustmentOrder,
     Business,
@@ -36,6 +37,7 @@ _ACTIVE_SINGLE = (PerformanceLedger.calc_type == 1, PerformanceLedger.is_active
 
 # ==== 个人绩效明细（FR-029）====
 
+
 def _personal_filters(query: PersonalDetailQuery) -> tuple:
     filters = [*_ACTIVE_SINGLE]
     if query.settlement_month:
@@ -65,29 +67,34 @@ def count_personal_detail(session: Session, query: PersonalDetailQuery) -> int:
     """个人绩效明细计数（与 list_personal_detail 同筛选）。"""
     return int(
         session.execute(
-            select(func.count())
-            .select_from(PerformanceLedger)
-            .where(*_personal_filters(query))
+            select(func.count()).select_from(PerformanceLedger).where(*_personal_filters(query))
         ).scalar_one()
     )
 
 
 def sum_personal_detail(session: Session, query: PersonalDetailQuery) -> dict:
     """全筛选口径合计（分页与合计同条件，避免「页脚合计≠全部结果」）。"""
-    return session.execute(
-        select(
-            func.count().label("row_count"),
-            func.coalesce(func.sum(PerformanceLedger.performance_amount), 0).label("performance_total"),
-            func.coalesce(func.sum(PerformanceLedger.payable_amount), 0).label("payable_total"),
-            func.coalesce(func.sum(PerformanceLedger.converted_profit), 0).label(
-                "converted_profit_total"
-            ),
-        ).where(*_personal_filters(query))
-    ).mappings().one()
+    return (
+        session.execute(
+            select(
+                func.count().label("row_count"),
+                func.coalesce(func.sum(PerformanceLedger.performance_amount), 0).label(
+                    "performance_total"
+                ),
+                func.coalesce(func.sum(PerformanceLedger.payable_amount), 0).label("payable_total"),
+                func.coalesce(func.sum(PerformanceLedger.converted_profit), 0).label(
+                    "converted_profit_total"
+                ),
+            ).where(*_personal_filters(query))
+        )
+        .mappings()
+        .one()
+    )
 
 
 # ==== 部门汇总取数（FR-030）====
 
+
 def list_month_receipt_rows(session: Session, month: str) -> list[PerformanceLedger]:
     """同月全量生效单笔台账行（部门汇总基数与成员明细的取数）。"""
     return list(
@@ -137,6 +144,7 @@ def list_month_settlement(session: Session, month: str) -> MonthlySettlement | N
 
 # ==== 业务统计（FR-031 之一）====
 
+
 def _business_perf_subquery():
     """业务维度绩效子查询（排除聚合行，防 12.5%/5% 聚合额污染单业务绩效）。"""
     return (
@@ -212,23 +220,28 @@ def count_business_stats(session: Session, query: BusinessStatsQuery) -> int:
 def sum_business_stats(session: Session, query: BusinessStatsQuery) -> dict:
     """业务统计全筛选合计（业务数/发货/回款/销售额/绩效/应付）。"""
     perf_sq = _business_perf_subquery()
-    return session.execute(
-        select(
-            func.count().label("row_count"),
-            func.coalesce(func.sum(Business.total_ship_amount), 0).label("ship_total"),
-            func.coalesce(func.sum(Business.total_received_amount), 0).label("received_total"),
-            func.coalesce(func.sum(Business.total_sale_amount), 0).label("sale_total"),
-            func.coalesce(func.sum(perf_sq.c.perf), 0).label("performance_total"),
-            func.coalesce(func.sum(perf_sq.c.payable), 0).label("payable_total"),
+    return (
+        session.execute(
+            select(
+                func.count().label("row_count"),
+                func.coalesce(func.sum(Business.total_ship_amount), 0).label("ship_total"),
+                func.coalesce(func.sum(Business.total_received_amount), 0).label("received_total"),
+                func.coalesce(func.sum(Business.total_sale_amount), 0).label("sale_total"),
+                func.coalesce(func.sum(perf_sq.c.perf), 0).label("performance_total"),
+                func.coalesce(func.sum(perf_sq.c.payable), 0).label("payable_total"),
+            )
+            .outerjoin(Customer, Customer.id == Business.customer_id)
+            .outerjoin(perf_sq, perf_sq.c.business_id == Business.id)
+            .where(*_business_filters(query))
         )
-        .outerjoin(Customer, Customer.id == Business.customer_id)
-        .outerjoin(perf_sq, perf_sq.c.business_id == Business.id)
-        .where(*_business_filters(query))
-    ).mappings().one()
+        .mappings()
+        .one()
+    )
 
 
 # ==== 回款统计（FR-031 之二）====
 
+
 def _payment_filters(query: PaymentStatsQuery | PaymentDetailQuery) -> tuple:
     filters = [*_ACTIVE_SINGLE]
     if query.settlement_month:
@@ -255,9 +268,7 @@ def group_payment_by_days(session: Session, query: PaymentStatsQuery) -> list:
                 func.coalesce(func.sum(PerformanceLedger.performance_amount), 0).label(
                     "performance_total"
                 ),
-                func.coalesce(func.sum(PerformanceLedger.payable_amount), 0).label(
-                    "payable_total"
-                ),
+                func.coalesce(func.sum(PerformanceLedger.payable_amount), 0).label("payable_total"),
                 func.count().label("ledger_count"),
             )
             .where(*_payment_filters(query))
@@ -326,7 +337,9 @@ def _group_by_dimension(session: Session, query: PaymentStatsQuery, dim_column)
     dim_perf = (
         select(
             dim_column.label("dim"),
-            func.coalesce(func.sum(PerformanceLedger.performance_amount), 0).label("performance_total"),
+            func.coalesce(func.sum(PerformanceLedger.performance_amount), 0).label(
+                "performance_total"
+            ),
             func.coalesce(func.sum(PerformanceLedger.payable_amount), 0).label("payable_total"),
             func.count().label("ledger_count"),
         )
@@ -342,10 +355,7 @@ def _group_by_dimension(session: Session, query: PaymentStatsQuery, dim_column)
         .distinct()
         .subquery()
     )
-    perf_rows = {
-        row.dim: row
-        for row in session.execute(dim_perf).all()
-    }
+    perf_rows = {row.dim: row for row in session.execute(dim_perf).all()}
     receipt_rows = session.execute(
         select(
             dim_receipts.c.dim,
@@ -356,15 +366,36 @@ def _group_by_dimension(session: Session, query: PaymentStatsQuery, dim_column)
     ).all()
     receipt_amounts = {row.dim: row.receipt_amount for row in receipt_rows}
     return [
-        (row.dim, receipt_amounts.get(row.dim), row.performance_total, row.payable_total, row.ledger_count)
+        (
+            row.dim,
+            receipt_amounts.get(row.dim),
+            row.performance_total,
+            row.payable_total,
+            row.ledger_count,
+        )
         for row in perf_rows.values()
     ]
 
 
+def _bucket_condition(bucket: DayBucket) -> tuple:
+    """区间桶的 days 过滤条件：none 桶 IS NULL，其余按 min/max 闭区间。"""
+    if bucket.code == "none":
+        return (PerformanceLedger.payment_days.is_(None),)
+    conds = []
+    if bucket.min_days is not None:
+        conds.append(PerformanceLedger.payment_days >= bucket.min_days)
+    if bucket.max_days is not None:
+        conds.append(PerformanceLedger.payment_days <= bucket.max_days)
+    return tuple(conds)
+
+
 def list_payment_detail(
-    session: Session, query: PaymentDetailQuery, *, offset: int, limit: int
+    session: Session, query: PaymentDetailQuery, bucket: DayBucket, *, offset: int, limit: int
 ) -> list:
-    """区间下钻：单笔生效行联回款表（行级展示回款金额与批次号）。"""
+    """区间下钻：单笔生效行联回款表（行级展示回款金额与批次号）。
+
+    bucket 由服务层解析（未注册编码 400），仓储只按区间拼条件。
+    """
     from app.models import Receipt
 
     stmt = (
@@ -374,19 +405,19 @@ def list_payment_detail(
             Receipt.receipt_batch_no.label("receipt_batch_no"),
         )
         .outerjoin(Receipt, Receipt.id == PerformanceLedger.receipt_id)
-        .where(*_payment_filters(query))
+        .where(*_payment_filters(query), *_bucket_condition(bucket))
     )
     stmt = stmt.order_by(PerformanceLedger.receipt_date.desc(), PerformanceLedger.id.desc())
     stmt = stmt.offset(offset).limit(limit)
     return list(session.execute(stmt).all())
 
 
-def count_payment_detail(session: Session, query: PaymentDetailQuery) -> int:
+def count_payment_detail(session: Session, query: PaymentDetailQuery, bucket: DayBucket) -> int:
     """区间下钻计数（与 list_payment_detail 同筛选）。"""
     return int(
         session.execute(
             select(func.count())
             .select_from(PerformanceLedger)
-            .where(*_payment_filters(query))
+            .where(*_payment_filters(query), *_bucket_condition(bucket))
         ).scalar_one()
     )

backend/app/schemas/report.py

@@ -13,6 +13,7 @@ from app.core.response import PageQuery
 
 # ==== 查询入参 ====
 
+
 class PersonalDetailQuery(PageQuery):
     """个人绩效明细查询条件（FR-029：人员/月份/客户类型/产品类别/岗位）。"""
 
@@ -64,13 +65,21 @@ class PaymentDetailQuery(PageQuery):
 
 # ==== 响应出参 ====
 
+
 class ReportSummary(BaseModel):
-    """全筛选口径合计（与「当前页合计」区分，前端页脚需注明）。"""
+    """全筛选口径合计（与「当前页合计」区分，前端页脚需注明）。
+
+    字段按报表取用：个人明细用绩效/应付/折算毛利；业务统计用发货/回款/
+    销售额；不涉及的留 None。
+    """
 
     row_count: int = Field(default=0, description="筛选结果行数")
     performance_total: Decimal | None = Field(default=None, description="绩效金额合计")
     payable_total: Decimal | None = Field(default=None, description="应付绩效合计")
     converted_profit_total: Decimal | None = Field(default=None, description="折算毛利润合计")
+    ship_total: Decimal | None = Field(default=None, description="发货额合计")
+    received_total: Decimal | None = Field(default=None, description="回款额合计")
+    sale_total: Decimal | None = Field(default=None, description="销售额合计")
 
 
 class ReportPage[T](BaseModel):

backend/app/services/manager_share.py

@@ -0,0 +1,118 @@
+"""经理分成共享口径（月结聚合与报表部门汇总的唯一实现）。
+
+从 settlement_service 迁出（行为零变化）：经理 12.5%/5% 的取参、团队
+分桶、公司基数计算，月结生成与部门汇总报表都要用——抽为单一实现后
+「报表汇总 == 月结聚合行」不靠两处代码各自对齐，而是同源函数的直接
+结果。分成金额仍走 formula.manager_team_share 纯函数。
+"""
+
+from calendar import monthrange
+from dataclasses import dataclass
+from datetime import date
+from decimal import Decimal
+from fractions import Fraction
+
+from sqlalchemy.orm import Session
+
+from app.models import Personnel
+
+
+@dataclass(frozen=True, slots=True)
+class ManagerShareParams:
+    """经理分成取参与参数原文（同一次计算共用一份，随月月末时点化）。"""
+
+    month: str
+    param_date: date
+    team_text: str
+    company_text: str
+    team_pct: Fraction
+    company_pct: Fraction
+    include_manager_personal: bool
+
+
+def month_end(month: str) -> date:
+    """YYYY-MM → 所属月月末（月结与报表取参锚点）。
+
+    月结是对过去某月的金额计算，取参必须锚定业务月份——按 date.today()
+    会让补生成历史月份的结果随操作日期漂移（违背「参数按生效日期时点化」
+    红线）。单笔台账行按回款日取参（裁决 #11）；聚合与月工资口径没有
+    更细的业务日期，取该月最后可确定的生效时点。
+    """
+    year, mon = int(month[:4]), int(month[5:7])
+    return date(year, mon, monthrange(year, mon)[1])
+
+
+def manager_share_params(session: Session, month: str) -> ManagerShareParams:
+    """解析经理分成参数（锚定所属月月末，取参与台账行口径同源）。
+
+    参数按百分数存储（12.5 表示 12.5%）：parse_fraction 得 12.5，
+    manager_team_share 内部再 ÷100 —— 调用方不得再乘 100（否则放大 100 倍）。
+    #12 开关（manager.personal.included）与分成比例同日取参，答复后调整可追溯。
+    """
+    from app.services.param_service import parse_fraction, resolve_snapshot
+
+    param_date = month_end(month)
+    snapshot = resolve_snapshot(session, param_date)
+    team_text = snapshot.get("manager.share.team")
+    company_text = snapshot.get("manager.share.company")
+    return ManagerShareParams(
+        month=month,
+        param_date=param_date,
+        team_text=team_text,
+        company_text=company_text,
+        team_pct=parse_fraction(team_text),
+        company_pct=parse_fraction(company_text),
+        include_manager_personal=snapshot.get("manager.personal.included") == "1",
+    )
+
+
+def add_team_performance(managers: dict[int, dict], manager_id: int, row) -> None:
+    """把一行业绩并入经理的团队桶（首行固化业务锚点，成员名去重收集）。"""
+    bucket = managers.setdefault(
+        manager_id,
+        {
+            "performance": Decimal("0"),
+            "business_id": row.business_id,
+            "business_no": row.business_no,
+            "names": set(),
+        },
+    )
+    bucket["performance"] += row.performance_amount or Decimal("0")
+    bucket["names"].add(row.personnel_name)
+
+
+def collect_team_buckets(
+    session: Session, ledger_rows: list, include_manager_personal: bool
+) -> dict[int, dict]:
+    """按经理采集团队业绩。
+
+    业务员行经 manager_id 上溯到其经理；经理个人户行（#12 开关计入时）
+    归属经理本人——行 personnel_id 即经理，不走上溯（经理没有上级，
+    按上溯会被跳过漏计）。
+    """
+    managers: dict[int, dict] = {}
+    for row in ledger_rows:
+        if row.customer_type == "manager_personal":
+            if include_manager_personal:
+                add_team_performance(managers, row.personnel_id, row)
+            continue  # 默认不计入聚合基数（#12 参数化）
+        person = session.get(Personnel, row.personnel_id)
+        manager_id = person.manager_id if person else None
+        if manager_id is not None:
+            add_team_performance(managers, manager_id, row)
+    return managers
+
+
+def company_base_of(ledger_rows: list, include_manager_personal: bool) -> Decimal:
+    """全公司基数（5% 的底）：默认排除经理个人户行（#12 开关计入时不过滤）。
+
+    口径逐行固定，调用方在循环外算一次。
+    """
+    return sum(
+        (
+            r.performance_amount or Decimal("0")
+            for r in ledger_rows
+            if include_manager_personal or r.customer_type != "manager_personal"
+        ),
+        Decimal("0"),
+    )

backend/app/services/report_export.py

@@ -0,0 +1,309 @@
+"""报表 Excel 模板构建（纯函数，可单测）。
+
+版式（FR-032「导出模板围绕财务复核和领导签字设计」）：
+公司信息页眉 → 报表标题 → 筛选条件 → 导出时间/人/模板版本 → 表头 → 数据行
+→ 合计行 → 财务复核栏 / 领导签字栏 / 日期栏。
+
+精度红线：金额写 ``Decimal``（openpyxl 原生支持，不经 float）+
+``number_format='#,##0.00'``；合计在 Python 侧求和后两位收口，**不写
+Excel 公式**（避免打开时重算漂移与公式注入面）。文本列以 ``= + - @``
+开头时前置单引号，防公式注入。
+"""
+
+from dataclasses import dataclass
+from datetime import datetime
+from decimal import ROUND_HALF_UP, Decimal
+from io import BytesIO
+
+from openpyxl import Workbook
+from openpyxl.styles import Alignment, Border, Font, PatternFill, Side
+from openpyxl.utils import get_column_letter
+
+from app.core.report_meta import CUSTOMER_TYPE_LABELS, PRODUCT_CATEGORY_LABELS
+
+_HEADER_FILL = PatternFill("solid", fgColor="DDE3F0")
+_THIN = Side(style="thin", color="B0B7C3")
+_BORDER = Border(left=_THIN, right=_THIN, top=_THIN, bottom=_THIN)
+_TITLE_FONT = Font(size=14, bold=True)
+_COMPANY_FONT = Font(size=16, bold=True)
+_HEADER_FONT = Font(bold=True)
+_AMOUNT_FORMAT = "#,##0.00"
+_DATE_FORMAT = "YYYY-MM-DD"
+_FORMULA_PREFIXES = ("=", "+", "-", "@", "\t")
+
+BUSINESS_STATUS_LABELS = {1: "进行中", 2: "已完成", 3: "已终止"}
+
+
+@dataclass(frozen=True, slots=True)
+class _Column:
+    """列定义：key 取行 dict 字段，kind 决定单元格类型与格式。"""
+
+    key: str
+    title: str
+    kind: str  # text / amount / int / date / customer / category / status
+    width: float
+
+
+_COLUMNS: dict[str, tuple[_Column, ...]] = {
+    "personal-detail": (
+        _Column("business_no", "合同编号", "text", 20),
+        _Column("ship_date", "发货日", "date", 12),
+        _Column("receipt_date", "回款日", "date", 12),
+        _Column("payment_days", "回款天数", "int", 10),
+        _Column("product_category", "产品类别", "category", 14),
+        _Column("customer_type", "客户类型", "customer", 14),
+        _Column("role_name", "岗位", "text", 12),
+        _Column("rate_text", "计提比例", "text", 14),
+        _Column("weight_r", "支付权重R", "int", 10),
+        _Column("performance_amount", "绩效金额", "amount", 14),
+        _Column("payable_amount", "应付绩效", "amount", 14),
+    ),
+    "department-summary": (
+        _Column("manager_name", "经理团队", "text", 16),
+        _Column("member_name", "成员", "text", 14),
+        _Column("role_code", "岗位编码", "text", 12),
+        _Column("performance_sum", "绩效金额", "amount", 14),
+        _Column("payable_sum", "应付绩效", "amount", 14),
+        _Column("adjust_net", "调整净额", "amount", 14),
+        _Column("net_total", "净合计", "amount", 14),
+    ),
+    "business-stats": (
+        _Column("business_no", "业务编号", "text", 20),
+        _Column("business_date", "业务日期", "date", 12),
+        _Column("customer_name", "客户", "text", 22),
+        _Column("owner_personnel_name", "归属人", "text", 12),
+        _Column("status", "状态", "status", 10),
+        _Column("total_sale_amount", "销售额", "amount", 14),
+        _Column("total_ship_amount", "发货额", "amount", 14),
+        _Column("total_received_amount", "回款额", "amount", 14),
+        _Column("performance_amount", "绩效金额", "amount", 14),
+        _Column("payable_amount", "应付绩效", "amount", 14),
+    ),
+    "payment-stats": (
+        _Column("label", "回款天数区间", "text", 20),
+        _Column("receipt_amount", "回款金额", "amount", 16),
+        _Column("receipt_count", "回款笔数", "int", 10),
+        _Column("performance_amount", "绩效金额", "amount", 14),
+        _Column("payable_amount", "应付绩效", "amount", 14),
+        _Column("ledger_count", "台账行数", "int", 10),
+    ),
+}
+
+
+def build_workbook(
+    *,
+    code: str,
+    title: str,
+    filters: str,
+    rows: list[dict],
+    summary: dict | None,
+    buckets: dict | None,
+    template_version: str,
+    operator_name: str,
+    company_name: str,
+) -> bytes:
+    """构建并返回 xlsx 字节流（业务数据只读，不触碰数据库）。"""
+    workbook = Workbook()
+    sheet = workbook.active
+    sheet.title = _sheet_name(code)
+    columns = _COLUMNS[code]
+    _write_header(sheet, columns, company_name, title, filters, template_version, operator_name)
+    if rows:
+        for row in rows:
+            _write_row(sheet, columns, row)
+    else:
+        sheet.append(["本筛选条件无数据"])
+        _merge_row(sheet, len(columns))
+    _write_total(sheet, columns, rows)
+    _write_signature(sheet, len(columns))
+    _autofit(sheet, columns)
+
+    # 回款统计附维度对比 sheet（客户类型/产品类别），与区间表共用页眉格式
+    if code == "payment-stats" and buckets:
+        for dim_key, dim_title in (
+            ("by_customer_type", "客户类型对比"),
+            ("by_product_category", "产品类别对比"),
+        ):
+            extra = workbook.create_sheet(dim_title)
+            _write_dimension_sheet(
+                extra,
+                dim_title,
+                company_name,
+                buckets.get(dim_key) or [],
+                template_version,
+                operator_name,
+            )
+
+    buffer = BytesIO()
+    workbook.save(buffer)
+    return buffer.getvalue()
+
+
+def _sheet_name(code: str) -> str:
+    return {
+        "personal-detail": "个人绩效明细",
+        "department-summary": "部门绩效汇总",
+        "business-stats": "业务统计",
+        "payment-stats": "回款统计",
+    }[code]
+
+
+def _write_header(
+    sheet,
+    columns: tuple[_Column, ...],
+    company_name: str,
+    title: str,
+    filters: str,
+    template_version: str,
+    operator_name: str,
+) -> None:
+    """页眉四行 + 表头行；返回后 sheet.max_row 即数据行起点。"""
+    sheet.append([company_name])
+    _merge_row(sheet, len(columns))
+    sheet.cell(row=1, column=1).font = _COMPANY_FONT
+    sheet.cell(row=1, column=1).alignment = Alignment(horizontal="left", vertical="center")
+
+    sheet.append([title])
+    _merge_row(sheet, len(columns))
+    sheet.cell(row=2, column=1).font = _TITLE_FONT
+    sheet.cell(row=2, column=1).alignment = Alignment(horizontal="center", vertical="center")
+
+    sheet.append([f"筛选条件：{filters}"])
+    _merge_row(sheet, len(columns))
+    sheet.append(
+        [
+            f"导出时间：{datetime.now():%Y-%m-%d %H:%M:%S} ｜ 导出人：{operator_name} ｜ "
+            f"模板版本：{template_version}"
+        ]
+    )
+    _merge_row(sheet, len(columns))
+    sheet.append([])
+
+    sheet.append([column.title for column in columns])
+    header_row = sheet.max_row
+    for cell in sheet[header_row]:
+        cell.font = _HEADER_FONT
+        cell.fill = _HEADER_FILL
+        cell.border = _BORDER
+        cell.alignment = Alignment(horizontal="center", vertical="center")
+    sheet.freeze_panes = sheet.cell(row=header_row + 1, column=1)
+
+
+def _merge_row(sheet, column_count: int) -> None:
+    """合并当前行（公司名/标题/筛选行跨全表宽）。"""
+    row = sheet.max_row
+    if column_count > 1:
+        sheet.merge_cells(start_row=row, start_column=1, end_row=row, end_column=column_count)
+
+
+def _write_row(sheet, columns: tuple[_Column, ...], row: dict) -> None:
+    """数据行：按列类型写单元格（金额 Decimal、日期 date、文本防注入）。"""
+    values = []
+    for column in columns:
+        raw = row.get(column.key)
+        values.append(_cell_value(column, raw))
+    sheet.append(values)
+    for index, column in enumerate(columns, start=1):
+        cell = sheet.cell(row=sheet.max_row, column=index)
+        cell.border = _BORDER
+        if column.kind == "amount":
+            cell.number_format = _AMOUNT_FORMAT
+        elif column.kind == "date":
+            cell.number_format = _DATE_FORMAT
+
+
+def _cell_value(column: _Column, raw):
+    """单值转换：标签列转中文，文本列防公式注入。"""
+    if raw is None:
+        return None
+    if column.kind == "amount":
+        return raw if isinstance(raw, Decimal) else Decimal(str(raw))
+    if column.kind == "date":
+        return raw
+    if column.kind == "int":
+        return int(raw)
+    if column.kind == "customer":
+        return CUSTOMER_TYPE_LABELS.get(str(raw), str(raw))
+    if column.kind == "category":
+        return PRODUCT_CATEGORY_LABELS.get(str(raw), str(raw))
+    if column.kind == "status":
+        return BUSINESS_STATUS_LABELS.get(int(raw), str(raw))
+    return _safe_text(raw)
+
+
+def _safe_text(value) -> str:
+    """文本防公式注入：以 = + - @ 或制表符开头时前置单引号。"""
+    text = str(value)
+    return "'" + text if text.startswith(_FORMULA_PREFIXES) else text
+
+
+def _write_total(sheet, columns: tuple[_Column, ...], rows: list[dict]) -> None:
+    """合计行：金额列按行求和后两位收口（不写 Excel 公式）。
+
+    「合计」标签只写在第一个非金额列；其余非金额列留空——否则每遇到
+    一个文本列都会重复写一次标签，合计行会满屏「合计」。
+    """
+    if not rows:
+        return
+    totals = []
+    label_written = False
+    for column in columns:
+        if column.kind != "amount":
+            totals.append("合计" if not label_written else None)
+            label_written = True
+            continue
+        total = sum(
+            (
+                row.get(column.key)
+                for row in rows
+                if isinstance(row.get(column.key), (Decimal, int))
+            ),
+            Decimal("0"),
+        )
+        totals.append(total.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP))
+    sheet.append(totals)
+    for index, column in enumerate(columns, start=1):
+        cell = sheet.cell(row=sheet.max_row, column=index)
+        cell.font = _HEADER_FONT
+        cell.border = _BORDER
+        if column.kind == "amount":
+            cell.number_format = _AMOUNT_FORMAT
+
+
+def _write_signature(sheet, column_count: int) -> None:
+    """财务复核栏 / 领导签字栏 / 日期栏（FR-032 模板要素）。"""
+    sheet.append([])
+    sheet.append(["财务复核：________________", "", "日期：______年__月__日"])
+    sheet.append(["领导签字：________________", "", "日期：______年__月__日"])
+    for row_index in (sheet.max_row - 1, sheet.max_row):
+        sheet.cell(row=row_index, column=1).alignment = Alignment(horizontal="left")
+
+
+def _autofit(sheet, columns: tuple[_Column, ...]) -> None:
+    """列宽按定义表设置（CJK 按 2 字符宽，不用默认 8.43）。"""
+    for index, column in enumerate(columns, start=1):
+        sheet.column_dimensions[get_column_letter(index)].width = column.width
+
+
+def _write_dimension_sheet(
+    sheet,
+    title: str,
+    company_name: str,
+    rows: list[dict],
+    template_version: str,
+    operator_name: str,
+) -> None:
+    """维度对比 sheet（客户类型/产品类别）。"""
+    columns = (
+        _Column("label", "维度", "text", 16),
+        _Column("receipt_amount", "回款金额", "amount", 16),
+        _Column("performance_amount", "绩效金额", "amount", 14),
+        _Column("payable_amount", "应付绩效", "amount", 14),
+        _Column("ledger_count", "台账行数", "int", 10),
+    )
+    _write_header(sheet, columns, company_name, title, "全部", template_version, operator_name)
+    for row in rows:
+        _write_row(sheet, columns, row)
+    _write_total(sheet, columns, rows)
+    _write_signature(sheet, len(columns))
+    _autofit(sheet, columns)

backend/app/services/report_service.py

@@ -0,0 +1,698 @@
+"""报表统计服务（M4 ⑤）：四张报表取数编排 + Excel 导出编排。
+
+全局口径（四表共用，改动须同步 backend/README「报表口径」小节）：
+
+1. 数据源 ``performance_ledgers`` 固定 **is_active=1 AND calc_type=1**（单笔
+   生效行）；经理聚合行（calc_type=2）只在部门汇总作冻结对照单读——把
+   聚合额混进明细会让「汇总 = 明细合计」失效；
+2. 金额全程 Decimal（两位收口 ROUND_HALF_UP）；比例原样输出 ``rate_text``
+   （如 88/9000），禁止转 float 或截断展示；
+3. 调整单并入：仅 status=2（生效）；direction=1 补发记 +、2 扣减记 −
+   （金额恒正，方向表达语义）；
+4. 离职人员不按 ``personnel.status`` 过滤——历史业绩留在原团队（FR-030
+   「变更不回溯」原则）；无成员团队仍出 0 值行；
+5. 团队基数直接取 ``manager_share.collect_team_buckets`` 的分桶结果，与
+   月结聚合行同源（不是报表侧另算一遍）。
+"""
+
+from decimal import ROUND_HALF_UP, Decimal
+from fractions import Fraction
+
+from sqlalchemy import select
+
+from app.core import report_meta
+from app.core.constants import ACTION_EXPORT, MODULE_REPORT
+from app.core.exceptions import BadRequestError
+from app.core.report_meta import (
+    REPORT_EXPORT_MAX_ROWS,
+    REPORT_TEMPLATE_VERSION,
+    DayBucket,
+)
+from app.models import BusinessRole, PerformanceLedger, Personnel, User
+from app.models.settlement import ADJUSTMENT_DIRECTION_EXTRA
+from app.repositories import report_repository
+from app.services import audit_service, manager_share
+from app.services.audit_service import AuditRecord, RequestMeta
+from app.services.formula import manager_team_share
+
+# ==== 元数据 ====
+
+
+def get_payment_day_buckets() -> list[dict]:
+    """回款天数区间元数据（前端下拉唯一数据源，禁止前端硬编码档位）。"""
+    return [
+        {"code": b.code, "label": b.label, "min_days": b.min_days, "max_days": b.max_days}
+        for b in report_meta.get_payment_day_buckets()
+    ]
+
+
+# ==== 个人绩效明细（FR-029）====
+
+
+def list_personal_detail(session, query) -> dict:
+    """明细分页 + 全筛选合计（summary 与分页同条件，避免页脚合计≠全部结果）。"""
+    rows = report_repository.list_personal_detail(
+        session, query, offset=query.offset, limit=query.page_size
+    )
+    role_names = _role_names(session)
+    sums = report_repository.sum_personal_detail(session, query)
+    return {
+        "list": [_person_row(r, role_names) for r in rows],
+        "total": report_repository.count_personal_detail(session, query),
+        "pageNum": query.page_num,
+        "pageSize": query.page_size,
+        "summary": {
+            "row_count": sums["row_count"],
+            "performance_total": _q2(_dec(sums["performance_total"])),
+            "payable_total": _q2(_dec(sums["payable_total"])),
+            "converted_profit_total": _q2(_dec(sums["converted_profit_total"])),
+        },
+    }
+
+
+def _person_row(row: PerformanceLedger, role_names: dict[str, str]) -> dict:
+    """台账行 → 明细行出参（金额保持 Decimal，序列化交由信封层）。"""
+    return {
+        "ledger_id": row.id,
+        "business_id": row.business_id,
+        "business_no": row.business_no,
+        "personnel_id": row.personnel_id,
+        "personnel_name": row.personnel_name,
+        "role_code": row.role_code,
+        "role_name": role_names.get(row.role_code, row.role_code),
+        "customer_type": row.customer_type,
+        "product_category": row.product_category,
+        "ship_date": row.ship_date,
+        "receipt_date": row.receipt_date,
+        "payment_days": row.payment_days,
+        "rate_text": row.rate_text,
+        "weight_r": row.weight_r,
+        "payment_weight": row.payment_weight,
+        "performance_amount": row.performance_amount,
+        "payable_amount": row.payable_amount,
+        "converted_profit": row.converted_profit,
+        "is_probation": row.is_probation,
+        "missing_cost": row.missing_cost,
+        "calc_version": row.calc_version,
+        "settlement_id": row.settlement_id,
+    }
+
+
+def _role_names(session) -> dict[str, str]:
+    """岗位编码 → 名称一次查询（明细与部门行共用，避免 N+1）。"""
+    return {
+        r.role_code: r.role_name
+        for r in session.execute(select(BusinessRole)).scalars()
+    }
+
+
+def _dec(value) -> Decimal | None:
+    """SQL 聚合结果转 Decimal（SUM 可能返回 int，None 原样返回）。"""
+    if value is None:
+        return None
+    return value if isinstance(value, Decimal) else Decimal(str(value))
+
+
+def _q2(value: Decimal | None) -> Decimal | None:
+    """两位小数收口（HALF_UP，与核算/月结舍入口径一致）。"""
+    if value is None:
+        return None
+    return value.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
+
+
+# ==== 部门绩效汇总（FR-030）====
+
+
+def department_summary(session, query) -> dict:
+    """按经理团队/月份汇总绩效工资（团队基数与月结聚合行同源）。"""
+    month = query.settlement_month
+    rows = report_repository.list_month_receipt_rows(session, month)
+    share = manager_share.manager_share_params(session, month)
+    buckets = manager_share.collect_team_buckets(session, rows, share.include_manager_personal)
+    company_base = manager_share.company_base_of(rows, share.include_manager_personal)
+
+    aggregates = {
+        row.personnel_id: row
+        for row in report_repository.list_month_manager_aggregates(session, month)
+    }
+    adjustments = _adjustments_by_person(report_repository.list_month_adjustments(session, month))
+    member_map = _member_map(rows)
+    manager_ids = _manager_ids(session, buckets, query.manager_id)
+    # 岗位判定范围必须覆盖「无成员的经理」——只按 member_map 取会让零行团队被跳过
+    manager_of, manager_role = _team_relations(session, member_map, manager_ids)
+
+    teams: list[dict] = []
+    for manager_id in manager_ids:
+        manager = session.get(Personnel, manager_id)
+        if manager is None or manager_role.get(manager_id) != "sales_manager":
+            continue  # 非营销经理的上级不展示团队行（与月结聚合同口径）
+        aggregate = aggregates.get(manager_id)
+        member_rows = _member_rows(
+            manager_id, member_map, manager_of, adjustments, share.include_manager_personal
+        )
+        team_extra = sum((m["adjust_extra"] for m in member_rows), Decimal("0"))
+        team_deduct = sum((m["adjust_deduct"] for m in member_rows), Decimal("0"))
+        team_base = _dec(buckets.get(manager_id, {}).get("performance", Decimal("0")))
+        # 12.5% 与 5% 分列（FR-030「全公司 5% 部分单列」）；两项之和才是月结
+        # 聚合行的金额（frozen_amount 对照），重复计入会双倍展示
+        team_share = _team_share(team_base, share.team_pct)
+        company_share = _company_share(company_base, share.company_pct)
+        teams.append(
+            {
+                "manager_id": manager_id,
+                "manager_name": manager.personnel_name,
+                "manager_status": manager.status,
+                "member_count": len(member_rows),
+                "team_base": _q2(team_base),
+                "team_share_amount": _q2(team_share),
+                "company_share_amount": _q2(company_share),
+                "company_base_ratio": _ratio(team_base, company_base),
+                "adjust_extra": _q2(team_extra),
+                "adjust_deduct": _q2(team_deduct),
+                "adjust_net": _q2(team_extra - team_deduct),
+                "net_total": _q2(team_share + company_share + team_extra - team_deduct),
+                "frozen_amount": aggregate.performance_amount if aggregate else None,
+                "aggregate_ledger_id": aggregate.id if aggregate else None,
+                "members": member_rows,
+            }
+        )
+
+    unassigned = _unassigned_rows(member_map, manager_of, manager_role, adjustments)
+    return {
+        "settlement_month": month,
+        "params": {
+            "team_text": share.team_text,
+            "company_text": share.company_text,
+            "manager_personal_included": 1 if share.include_manager_personal else 0,
+            "params_as_of": str(share.param_date),
+        },
+        "teams": teams,
+        "unassigned": unassigned,
+        "company": {
+            "team_base_total": _q2(sum((t["team_base"] for t in teams), Decimal("0"))),
+            "company_base": _q2(company_base),
+            "company_share_pct": share.company_text,
+            "company_share_amount": _q2(_company_share(company_base, share.company_pct)),
+        },
+        "totals": {
+            "performance_total": _q2(sum((t["team_base"] for t in teams), Decimal("0"))),
+            "payable_total": _q2(sum((t["net_total"] for t in teams), Decimal("0"))),
+        },
+    }
+
+
+def _team_share(team_base: Decimal, team_pct: Fraction) -> Decimal:
+    """12.5% 团队分成：全公司部分传 0，仅取 team_pct 项。"""
+    return manager_team_share(
+        team_base, Decimal("0"), team_pct=team_pct, company_pct=Fraction(0)
+    )
+
+
+def _company_share(company_base: Decimal, company_pct: Fraction) -> Decimal:
+    """5% 全公司分成：团队部分传 0，仅取 company_pct 项。"""
+    return manager_team_share(
+        Decimal("0"), company_base, team_pct=Fraction(0), company_pct=company_pct
+    )
+
+
+def _ratio(part: Decimal, whole: Decimal) -> Decimal:
+    """占比（除零返回 0，不抛 ZeroDivisionError）。"""
+    if not whole:
+        return Decimal("0")
+    return (part / whole).quantize(Decimal("0.0001"), rounding=ROUND_HALF_UP)
+
+
+def _adjustments_by_person(orders) -> dict[int, dict]:
+    """生效调整单按人汇总（补发为正、扣减为负；金额本身恒正）。"""
+    by_person: dict[int, dict] = {}
+    for order in orders:
+        acc = by_person.setdefault(
+            order.personnel_id, {"extra": Decimal("0"), "deduct": Decimal("0")}
+        )
+        if order.direction == ADJUSTMENT_DIRECTION_EXTRA:
+            acc["extra"] += order.amount
+        else:
+            acc["deduct"] += order.amount
+    return by_person
+
+
+def _member_map(rows) -> dict[int, dict]:
+    """台账行按人求和（成员明细的唯一数据源），经理个人户金额单独记账。"""
+    members: dict[int, dict] = {}
+    for row in rows:
+        acc = members.setdefault(
+            row.personnel_id,
+            {
+                "personnel_name": row.personnel_name,
+                "role_code": row.role_code,
+                "is_probation": row.is_probation,
+                "performance": Decimal("0"),
+                "payable": Decimal("0"),
+                "manager_personal_performance": Decimal("0"),
+                "manager_personal_payable": Decimal("0"),
+                "ledger_ids": [],
+                "has_manager_personal": False,
+            },
+        )
+        performance = row.performance_amount or Decimal("0")
+        payable = row.payable_amount or Decimal("0")
+        acc["performance"] += performance
+        acc["payable"] += payable
+        acc["ledger_ids"].append(row.id)
+        if row.customer_type == "manager_personal":
+            acc["has_manager_personal"] = True
+            acc["manager_personal_performance"] += performance
+            acc["manager_personal_payable"] += payable
+    return members
+
+
+def _team_relations(
+    session, member_map: dict[int, dict], manager_ids: list[int]
+) -> tuple[dict[int, int | None], dict[int, str]]:
+    """人员 → 直属经理与人员 → 岗位编码（各一次查询，避免逐人 get 的 N+1）。
+
+    查询范围 = 成员 ∪ 经理枚举：无成员的经理也要有岗位判定，否则「零行团队」
+    会被岗位过滤静默跳过（FR-030 要求无团队成员出 0 行）。
+    经理判定与 ``manager_share.collect_team_buckets`` 的实时上溯同口径：
+    离职人员的 manager_id 保留，历史业绩因此留在原团队。
+    """
+    pids = list(set(member_map) | set(manager_ids))
+    if not pids:
+        return {}, {}
+    people = list(session.execute(select(Personnel).where(Personnel.id.in_(pids))).scalars())
+    manager_of = {p.id: p.manager_id for p in people}
+    role_by_id = {
+        r.id: r.role_code for r in session.execute(select(BusinessRole)).scalars()
+    }
+    manager_role = {p.id: role_by_id.get(p.business_role_id, "") for p in people}
+    return manager_of, manager_role
+
+
+def _manager_ids(session, buckets: dict, only: int | None) -> list[int]:
+    """经理枚举：数据中出现的经理 ∪ 在职营销经理（无成员也要出 0 值行）。"""
+    if only is not None:
+        return [only]
+    found = set(buckets)
+    role = session.execute(
+        select(BusinessRole).where(BusinessRole.role_code == "sales_manager")
+    ).scalar_one_or_none()
+    if role is not None:
+        active = session.execute(
+            select(Personnel.id).where(Personnel.business_role_id == role.id, Personnel.status == 1)
+        ).scalars()
+        found.update(active)
+    return sorted(found)
+
+
+def _member_rows(
+    manager_id: int,
+    member_map: dict[int, dict],
+    manager_of: dict[int, int | None],
+    adjustments: dict[int, dict],
+    include_manager_personal: bool,
+) -> list[dict]:
+    """团队内成员行：业务员按 manager_id 归属；经理个人户按开关归属本人。"""
+    pids = [pid for pid, mid in manager_of.items() if mid == manager_id and pid != manager_id]
+    # 经理本人（经理个人户行）：#12 开关开启时计入本团队基数
+    if include_manager_personal and manager_id in member_map:
+        pids.append(manager_id)
+    return [_member_row(pid, member_map[pid], adjustments) for pid in pids]
+
+
+def _member_row(pid: int, member: dict, adjustments: dict[int, dict]) -> dict:
+    """单条成员行（绩效/应付求和 + 调整净额 + 台账 id 列表）。"""
+    adj = adjustments.get(pid, {"extra": Decimal("0"), "deduct": Decimal("0")})
+    net = member["payable"] + adj["extra"] - adj["deduct"]
+    return {
+        "personnel_id": pid,
+        "personnel_name": member["personnel_name"],
+        "role_code": member["role_code"],
+        "is_probation": member["is_probation"],
+        "performance_sum": _q2(member["performance"]),
+        "payable_sum": _q2(member["payable"]),
+        "adjust_extra": _q2(adj["extra"]),
+        "adjust_deduct": _q2(adj["deduct"]),
+        "adjust_net": _q2(adj["extra"] - adj["deduct"]),
+        "net_total": _q2(net),
+        "ledger_ids": member["ledger_ids"],
+    }
+
+
+def _unassigned_rows(
+    member_map: dict[int, dict],
+    manager_of: dict[int, int | None],
+    manager_role: dict[int, str],
+    adjustments: dict[int, dict],
+) -> list[dict]:
+    """未进入团队基数的成员：无经理业务员 + 非营销经理上级的下属。
+
+    经理个人户业绩默认不计入任何基数（#12 开关关闭时），也不在此展示——
+    其金额仍可在个人明细报表中查询，避免与「团队基数 + 未分配」口径混淆。
+    """
+    rows = []
+    for pid, member in member_map.items():
+        mid = manager_of.get(pid)
+        if mid is not None and manager_role.get(mid) == "sales_manager":
+            continue
+        regular = member["performance"] - member["manager_personal_performance"]
+        if regular == 0 and member["manager_personal_performance"] != 0:
+            continue
+        row = _member_row(pid, member, adjustments)
+        row["performance_sum"] = _q2(regular)
+        row["payable_sum"] = _q2(member["payable"] - member["manager_personal_payable"])
+        rows.append(row)
+    return rows
+
+
+# ==== 业务统计（FR-031 之一）====
+
+
+def business_stats(session, query) -> dict:
+    """业务统计分页 + 全筛选合计；回款/绩效为业务累计值。"""
+    rows = report_repository.list_business_stats(
+        session, query, offset=query.offset, limit=query.page_size
+    )
+    sums = report_repository.sum_business_stats(session, query)
+    return {
+        "list": [_business_row(r) for r in rows],
+        "total": report_repository.count_business_stats(session, query),
+        "pageNum": query.page_num,
+        "pageSize": query.page_size,
+        "summary": {
+            "row_count": sums["row_count"],
+            "ship_total": _q2(_dec(sums["ship_total"])),
+            "received_total": _q2(_dec(sums["received_total"])),
+            "sale_total": _q2(_dec(sums["sale_total"])),
+            "performance_total": _q2(_dec(sums["performance_total"])),
+            "payable_total": _q2(_dec(sums["payable_total"])),
+        },
+    }
+
+
+def _business_row(row) -> dict:
+    """业务统计行（ORM 行与三个联查列）。"""
+    business = row[0]
+    return {
+        "business_id": business.id,
+        "business_no": business.business_no,
+        "business_date": business.business_date,
+        "status": business.status,
+        "customer_name": row.customer_name,
+        "owner_personnel_name": row.owner_personnel_name,
+        "total_sale_amount": business.total_sale_amount,
+        "total_ship_amount": business.total_ship_amount,
+        "total_received_amount": business.total_received_amount,
+        "performance_amount": row.performance_amount,
+        "payable_amount": row.payable_amount,
+    }
+
+
+# ==== 回款统计（FR-031 之二）====
+
+
+def payment_stats(session, query) -> dict:
+    """区间/客户类型/产品类别三维度统计；空桶恒在且为 0。"""
+    perf_rows = report_repository.group_payment_by_days(session, query)
+    receipt_rows = report_repository.group_payment_receipt_by_days(session, query)
+    buckets = _bucket_rows(perf_rows, receipt_rows)
+    by_customer = _group_rows(report_repository.group_payment_by_customer_type(session, query))
+    by_category = _group_rows(report_repository.group_payment_by_product_category(session, query))
+    return {
+        "settlement_month": query.settlement_month,
+        "buckets": buckets,
+        "by_customer_type": by_customer,
+        "by_product_category": by_category,
+        "totals": {
+            "receipt_amount": _q2(
+                sum((b["receipt_amount"] or Decimal("0") for b in buckets), Decimal("0"))
+            ),
+            "receipt_count": sum(b["receipt_count"] for b in buckets),
+            "performance_amount": _q2(
+                sum((b["performance_amount"] or Decimal("0") for b in buckets), Decimal("0"))
+            ),
+            "payable_amount": _q2(
+                sum((b["payable_amount"] or Decimal("0") for b in buckets), Decimal("0"))
+            ),
+        },
+    }
+
+
+def _bucket_rows(perf_rows, receipt_rows) -> list[dict]:
+    """把（天数, 类型, 类别）精确值聚合结果按 8 档区间分桶（Python 侧分桶单点）。"""
+    perf_acc: dict[str, dict] = {}
+    for row in perf_rows:
+        bucket = report_meta.bucket_of(row.payment_days)
+        acc = perf_acc.setdefault(
+            bucket.code, {"performance": Decimal("0"), "payable": Decimal("0"), "count": 0}
+        )
+        acc["performance"] += _dec(row.performance_total)
+        acc["payable"] += _dec(row.payable_total)
+        acc["count"] += int(row.ledger_count)
+    receipt_acc: dict[str, dict] = {}
+    for row in receipt_rows:
+        bucket = report_meta.bucket_of(row.payment_days)
+        acc = receipt_acc.setdefault(bucket.code, {"amount": Decimal("0"), "count": 0})
+        acc["amount"] += _dec(row.receipt_amount)
+        acc["count"] += int(row.receipt_count)
+    zero = Decimal("0")
+    return [
+        {
+            "code": bucket.code,
+            "label": bucket.label,
+            "min_days": bucket.min_days,
+            "max_days": bucket.max_days,
+            # 空桶显示 0 而非 null（FR-031「空区间正常显示」）
+            "receipt_amount": _q2(receipt_acc.get(bucket.code, {}).get("amount", zero)),
+            "receipt_count": receipt_acc.get(bucket.code, {}).get("count", 0),
+            "performance_amount": _q2(perf_acc.get(bucket.code, {}).get("performance", zero)),
+            "payable_amount": _q2(perf_acc.get(bucket.code, {}).get("payable", zero)),
+            "ledger_count": perf_acc.get(bucket.code, {}).get("count", 0),
+        }
+        for bucket in report_meta.get_payment_day_buckets()
+    ]
+
+
+def _group_rows(rows) -> list[dict]:
+    """维度对比行（客户类型/产品类别），标签来自 report_meta。
+
+    仓储返回元组 (dim, 回款额, 绩效, 应付, 台账行数)，按位置解包。
+    """
+    labels = report_meta.CUSTOMER_TYPE_LABELS | report_meta.PRODUCT_CATEGORY_LABELS
+    result = []
+    for dim, receipt_amount, performance_amount, payable_amount, ledger_count in rows:
+        result.append(
+            {
+                "code": dim,
+                "label": labels.get(dim, dim or "未分类"),
+                "receipt_amount": _q2(_dec(receipt_amount)),
+                "performance_amount": _q2(_dec(performance_amount)),
+                "payable_amount": _q2(_dec(payable_amount)),
+                "ledger_count": int(ledger_count),
+            }
+        )
+    return result
+
+
+def payment_detail(session, query) -> dict:
+    """区间下钻分页（bucket 未注册 → 400）。"""
+    bucket = _bucket_by_code(query.bucket)
+    rows = report_repository.list_payment_detail(
+        session, query, bucket, offset=query.offset, limit=query.page_size
+    )
+    return {
+        "list": [_payment_row(r) for r in rows],
+        "total": report_repository.count_payment_detail(session, query, bucket),
+        "pageNum": query.page_num,
+        "pageSize": query.page_size,
+        "summary": {"row_count": len(rows)},
+    }
+
+
+def _payment_row(row) -> dict:
+    """下钻行（台账行 + 回款金额/批次号）。"""
+    ledger = row[0]
+    return {
+        "ledger_id": ledger.id,
+        "receipt_id": ledger.receipt_id,
+        "receipt_batch_no": row.receipt_batch_no,
+        "amount": row.receipt_amount,
+        "business_id": ledger.business_id,
+        "business_no": ledger.business_no,
+        "personnel_id": ledger.personnel_id,
+        "personnel_name": ledger.personnel_name,
+        "customer_type": ledger.customer_type,
+        "product_category": ledger.product_category,
+        "ship_date": ledger.ship_date,
+        "receipt_date": ledger.receipt_date,
+        "payment_days": ledger.payment_days,
+        "performance_amount": ledger.performance_amount,
+        "payable_amount": ledger.payable_amount,
+    }
+
+
+def _bucket_by_code(code: str) -> DayBucket:
+    """按编码取区间桶；未注册编码返回 400（防前端传参与后端档位漂移）。"""
+    for bucket in report_meta.get_payment_day_buckets():
+        if bucket.code == code:
+            return bucket
+    raise BadRequestError(f"未知的回款天数区间：{code}")
+
+
+# ==== 导出编排（FR-032）====
+
+EXPORT_FILENAMES = {
+    "personal-detail": "personal_detail",
+    "department-summary": "department_summary",
+    "business-stats": "business_stats",
+    "payment-stats": "payment_stats",
+}
+
+
+def export_report(session, code: str, query, *, operator: User, meta: RequestMeta | None = None):
+    """导出 Excel：上限校验 → 取数 → 构建 → 审计留痕（独立 commit）。
+
+    返回 (bytes, 文件名)；文件名由 API 层拼日期（ASCII，响应头 latin-1）。
+    """
+    from app.services import report_export
+
+    rows, context = _export_payload(session, code, query)
+    with audit_service.audit(
+        session,
+        operator=operator,
+        record=AuditRecord(
+            module=MODULE_REPORT,
+            action=ACTION_EXPORT,
+            summary=f"导出{context['title']}（{context['row_count']} 行）",
+            request_data=query.model_dump(exclude_none=True),
+        ),
+        meta=meta,
+    ):
+        content = report_export.build_workbook(
+            code=code,
+            title=context["title"],
+            filters=context["filters"],
+            rows=rows,
+            summary=context.get("summary"),
+            buckets=context.get("buckets"),
+            template_version=REPORT_TEMPLATE_VERSION,
+            operator_name=getattr(operator, "username", ""),
+            company_name=context["company_name"],
+        )
+    return content, EXPORT_FILENAMES[code]
+
+
+def _export_payload(session, code: str, query) -> tuple[list[dict], dict]:
+    """按报表编码取全量数据并做上限校验（复用查询函数，不走分页）。"""
+    if code == "personal-detail":
+        total = report_repository.count_personal_detail(session, query)
+        _check_export_limit(total)
+        rows = report_repository.list_personal_detail(
+            session, query, offset=0, limit=REPORT_EXPORT_MAX_ROWS
+        )
+        role_names = _role_names(session)
+        sums = report_repository.sum_personal_detail(session, query)
+        return [_person_row(r, role_names) for r in rows], {
+            "title": "个人绩效工资明细表",
+            "row_count": total,
+            "filters": _filter_text(query),
+            "summary": {
+                "performance_total": _q2(_dec(sums["performance_total"])),
+                "payable_total": _q2(_dec(sums["payable_total"])),
+            },
+            "company_name": _company_name(session),
+        }
+    if code == "department-summary":
+        data = department_summary(session, query)
+        member_total = sum(len(t["members"]) for t in data["teams"])
+        _check_export_limit(member_total)
+        rows = _department_export_rows(data)
+        return rows, {
+            "title": "部门绩效工资汇总表",
+            "row_count": member_total,
+            "filters": f"统计月份 {query.settlement_month}",
+            "summary": data["company"],
+            "company_name": _company_name(session),
+        }
+    if code == "business-stats":
+        total = report_repository.count_business_stats(session, query)
+        _check_export_limit(total)
+        rows = report_repository.list_business_stats(
+            session, query, offset=0, limit=REPORT_EXPORT_MAX_ROWS
+        )
+        sums = report_repository.sum_business_stats(session, query)
+        return [_business_row(r) for r in rows], {
+            "title": "业务统计表",
+            "row_count": total,
+            "filters": _filter_text(query),
+            "summary": {
+                "ship_total": _q2(_dec(sums["ship_total"])),
+                "received_total": _q2(_dec(sums["received_total"])),
+                "performance_total": _q2(_dec(sums["performance_total"])),
+            },
+            "company_name": _company_name(session),
+        }
+    if code == "payment-stats":
+        data = payment_stats(session, query)
+        rows = data["buckets"]
+        return rows, {
+            "title": "回款情况统计表",
+            "row_count": len(rows),
+            "filters": _filter_text(query),
+            "summary": data["totals"],
+            "buckets": data,
+            "company_name": _company_name(session),
+        }
+    raise BadRequestError(f"未知的报表编码：{code}")
+
+
+def _check_export_limit(row_count: int) -> None:
+    """导出上限（需求「超限提示」口径）：超过即 400 引导缩小筛选。"""
+    if row_count > REPORT_EXPORT_MAX_ROWS:
+        raise BadRequestError(
+            f"导出数据量 {row_count} 行超过上限 {REPORT_EXPORT_MAX_ROWS} 行，"
+            "请缩小筛选范围（如按月导出）"
+        )
+
+
+def _department_export_rows(data: dict) -> list[dict]:
+    """部门汇总导出展平：每个成员一行，团队列冗余（便于签字稿核对）。"""
+    rows = []
+    for team in data["teams"]:
+        for member in team["members"]:
+            rows.append(
+                {
+                    "manager_name": team["manager_name"],
+                    "member_name": member["personnel_name"],
+                    "role_code": member["role_code"],
+                    "performance_sum": member["performance_sum"],
+                    "payable_sum": member["payable_sum"],
+                    "adjust_net": member["adjust_net"],
+                    "net_total": member["net_total"],
+                }
+            )
+    return rows
+
+
+def _filter_text(query) -> str:
+    """筛选条件摘要（导出页眉展示；未设置项不展示）。"""
+    parts = []
+    for field, label in (
+        ("settlement_month", "统计月份"),
+        ("customer_type", "客户类型"),
+        ("product_category", "产品类别"),
+        ("role_code", "岗位"),
+        ("keyword", "关键字"),
+        ("date_from", "日期起"),
+        ("date_to", "日期止"),
+    ):
+        value = getattr(query, field, None)
+        if value:
+            parts.append(f"{label} {value}")
+    return " ｜ ".join(parts) or "全部"
+
+
+def _company_name(session) -> str:
+    """导出页眉公司名（未配置时 system_setting_service 返回默认「衍水集团」）。"""
+    from app.services import system_setting_service
+
+    return system_setting_service.get_setting(session).company_name

backend/app/services/settlement_service.py

@@ -9,11 +9,8 @@
 """
 
 import logging
-from calendar import monthrange
-from dataclasses import dataclass
 from datetime import date
 from decimal import ROUND_HALF_UP, Decimal
-from fractions import Fraction
 
 from sqlalchemy import select
 from sqlalchemy.orm import Session
@@ -45,7 +42,7 @@ from app.schemas.settlement import (
     SettlementOut,
     SettlementQuery,
 )
-from app.services import approval_service, audit_service
+from app.services import approval_service, audit_service, manager_share
 from app.services.approval_service import AuditRecord, RequestMeta
 from app.services.formula import manager_team_share, monthly_salary
 
@@ -352,82 +349,28 @@ def _confirm_scope(session: Session, month: str) -> list:
     )
 
 
-@dataclass(frozen=True, slots=True)
-class _ShareParams:
-    """经理分成取参与参数原文（同一次生成共用一份，随月月末时点化）。"""
+def _manager_share_params(session: Session, month: str) -> manager_share.ManagerShareParams:
+    """薄委托 manager_share 共享模块：月结聚合与报表部门汇总同源取参。
 
-    month: str
-    param_date: date
-    team_text: str
-    company_text: str
-    team_pct: Fraction
-    company_pct: Fraction
-    include_manager_personal: bool
-
-
-def _manager_share_params(session: Session, month: str) -> _ShareParams:
-    """解析经理分成参数（锚定所属月月末，取参与台账行口径同源）。
-
-    参数按百分数存储（12.5 表示 12.5%）：parse_fraction 得 12.5，
-    manager_team_share 内部再 ÷100 —— 调用方不得再乘 100（否则放大 100 倍）。
-    #12 开关（manager.personal.included）与分成比例同日取参，答复后调整可追溯。
+    迁移说明（2026-10-19）：实现原样迁至 manager_share，行为零变化；
+    参数按百分数存储（12.5 表示 12.5%），调用方不得再乘 100。
     """
-    from app.services.param_service import parse_fraction, resolve_snapshot
-
-    param_date = _month_end(month)
-    snapshot = resolve_snapshot(session, param_date)
-    team_text = snapshot.get("manager.share.team")
-    company_text = snapshot.get("manager.share.company")
-    return _ShareParams(
-        month=month,
-        param_date=param_date,
-        team_text=team_text,
-        company_text=company_text,
-        team_pct=parse_fraction(team_text),
-        company_pct=parse_fraction(company_text),
-        include_manager_personal=snapshot.get("manager.personal.included") == "1",
-    )
-
-
-def _add_team_performance(managers: dict[int, dict], manager_id: int, row) -> None:
-    """把一行业绩并入经理的团队桶（首行固化业务锚点，成员名去重收集）。"""
-    bucket = managers.setdefault(
-        manager_id,
-        {
-            "performance": Decimal("0"),
-            "business_id": row.business_id,
-            "business_no": row.business_no,
-            "names": set(),
-        },
-    )
-    bucket["performance"] += row.performance_amount or Decimal("0")
-    bucket["names"].add(row.personnel_name)
+    return manager_share.manager_share_params(session, month)
 
 
 def _collect_team_buckets(
     session: Session, ledger_rows: list, include_manager_personal: bool
 ) -> dict[int, dict]:
-    """按经理采集团队业绩。
-
-    业务员行经 manager_id 上溯到其经理；经理个人户行（#12 开关计入时）
-    归属经理本人——行 personnel_id 即经理，不走上溯（经理没有上级，
-    按上溯会被跳过漏计）。
-    """
-    managers: dict[int, dict] = {}
-    for row in ledger_rows:
-        if row.customer_type == "manager_personal":
-            if include_manager_personal:
-                _add_team_performance(managers, row.personnel_id, row)
-            continue  # 默认不计入聚合基数（#12 参数化）
-        person = session.get(Personnel, row.personnel_id)
-        manager_id = person.manager_id if person else None
-        if manager_id is not None:
-            _add_team_performance(managers, manager_id, row)
-    return managers
+    """薄委托 manager_share 共享模块：按经理采集团队业绩（行为零变化）。"""
+    return manager_share.collect_team_buckets(session, ledger_rows, include_manager_personal)
 
 
 def _aggregate_row(
-    session: Session, manager, bucket: dict, company_base: Decimal, share: _ShareParams
+    session: Session,
+    manager,
+    bucket: dict,
+    company_base: Decimal,
+    share: manager_share.ManagerShareParams,
 ):
     """单经理聚合台账行（必须落库：月结 ledger_ids 与报表下钻的锚点）。"""
     from app.models import PerformanceLedger
@@ -487,16 +430,8 @@ def _build_manager_aggregates(session: Session, ledger_rows: list, month: str) -
 
     share = _manager_share_params(session, month)
     managers = _collect_team_buckets(session, ledger_rows, share.include_manager_personal)
-    # 全公司基数：默认排除经理个人户行（#12 开关计入时不过滤）；
-    # 口径逐行固定，循环外算一次
-    company_base = sum(
-        (
-            r.performance_amount or Decimal("0")
-            for r in ledger_rows
-            if share.include_manager_personal or r.customer_type != "manager_personal"
-        ),
-        Decimal("0"),
-    )
+    # 全公司基数：默认排除经理个人户行（#12 开关计入时不过滤）；共享口径单点
+    company_base = manager_share.company_base_of(ledger_rows, share.include_manager_personal)
     aggregates: list = []
     for manager_id, bucket in managers.items():
         manager = session.get(Personnel, manager_id)
@@ -514,15 +449,12 @@ def _build_manager_aggregates(session: Session, ledger_rows: list, month: str) -
 
 
 def _month_end(month: str) -> date:
-    """YYYY-MM → 所属月月末（月结取参锚点）。
+    """薄委托 manager_share 共享模块：YYYY-MM → 所属月月末（取参锚点）。
 
-    月结是对过去某月的金额计算，取参必须锚定业务月份——按 date.today()
-    会让补生成历史月份的结果随操作日期漂移（违背「参数按生效日期时点化」
-    红线）。单笔台账行按回款日取参（裁决 #11）；聚合与月工资口径没有
-    更细的业务日期，取该月最后可确定的生效时点。
+    保留包装：tests/unit/test_settlement_helpers.py 直接 import 本名字；
+    「补生成历史月份结果不随操作日漂移」的原因说明见 manager_share.month_end。
     """
-    year, mon = int(month[:4]), int(month[5:7])
-    return date(year, mon, monthrange(year, mon)[1])
+    return manager_share.month_end(month)
 
 
 def _salary_mode_of(session: Session, role_code: str, month: str) -> str | None: