"""业务管理业务编排（需求 2.2.1/2.2.4）。

事务边界在本层：repository 只 flush，commit/rollback 只写在这里。

- 业务编号自动生成（no_generator.next_no）：MAX+FOR UPDATE 锁当日序号区间，
  残余竞态以唯一键（1062）冲突重试一次兜底
- 五项汇总冗余（total_sale/cost/ship/received/freight）由 ``_recalc_totals``
  单点重算 —— 明细/发货/回款任何写路径收口到此函数，禁止各处手写
- 状态机：1执行中 → 2已完成 / 3已终止；终态不可再变更（一期口径，开放问题 #8）；
  终止必填原因；每次流转写 business_status_logs（创建写首条 old=new=1）
- 归属变更（负责/协助）写 ownership_change_logs（owner_type=2），同事务一次
  commit；只影响变更后核算，不回溯历史（对照表 #19）
- 毛利润视图为只读派生值（复用 formula.profit_rate/converted_profit），
  阈值按 asOf 取 M1 参数快照 —— 落库即漂移，M3 台账才是固化点
"""

import logging
from datetime import date
from decimal import Decimal
from fractions import Fraction

from sqlalchemy import Row, func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from app.core.constants import (
    ACTION_CREATE,
    ACTION_DELETE,
    ACTION_UPDATE,
    MODULE_BUSINESS,
    NO_PREFIX_BUSINESS,
)
from app.core.db_errors import is_duplicate_entry
from app.core.exceptions import BadRequestError, NotFoundError
from app.core.response import PageQuery, PageResult
from app.models import (
    Business,
    BusinessItem,
    BusinessStatusLog,
    Customer,
    OwnershipChangeLog,
    Personnel,
    Receipt,
    Shipment,
    User,
)
from app.models.business import (
    BUSINESS_STATUS_ACTIVE,
    BUSINESS_STATUS_DONE,
)
from app.models.customer import OWNER_TYPE_BUSINESS
from app.repositories import business_repository, customer_repository, personnel_repository
from app.schemas.business import (
    BusinessCreateIn,
    BusinessDetailOut,
    BusinessItemIn,
    BusinessItemOut,
    BusinessOut,
    BusinessOwnershipIn,
    BusinessQuery,
    BusinessStatusIn,
    BusinessStatusLogOut,
    BusinessUpdateIn,
    GrossProfitCategoryOut,
    GrossProfitOut,
    GrossProfitRowOut,
    OwnershipChangeLogOut,
)
from app.services import audit_service, param_service
from app.services.audit_service import AuditRecord, RequestMeta
from app.services.formula import converted_profit, profit_rate
from app.services.no_generator import next_no

logger = logging.getLogger(__name__)

# 编号唯一键冲突后的重试次数：MAX+FOR UPDATE 已覆盖绝大多数竞态，
# 重试是为极端场景（锁等待超时后重入）兜底 —— 无限重试反而掩盖死锁类缺陷
_NO_RETRY = 1

# personnel.status 的在职值（与 customer_service 同口径，注释见彼处）
_PERSONNEL_STATUS_ACTIVE = 1


def _to_out(row: tuple) -> BusinessOut:
    """(business, company_name, owner_name, assistant_name) → 响应模型。

    金额冗余列出参 str；汇总列可能为 NULL（server_default 只对 INSERT 生效，
    存量行可能为 NULL），统一转 "0" 保持契约稳定。
    """
    business, company_name, owner_name, assistant_name = row
    return BusinessOut(
        id=business.id,
        business_no=business.business_no,
        customer_id=business.customer_id,
        customer_name=company_name,
        business_date=business.business_date,
        status=business.status,
        owner_personnel_id=business.owner_personnel_id,
        owner_personnel_name=owner_name,
        assistant_personnel_id=business.assistant_personnel_id,
        assistant_personnel_name=assistant_name,
        is_special=business.is_special,
        special_note=business.special_note,
        total_sale_amount=_str_or_zero(business.total_sale_amount),
        total_cost_amount=_str_or_zero(business.total_cost_amount),
        total_ship_amount=_str_or_zero(business.total_ship_amount),
        total_received_amount=_str_or_zero(business.total_received_amount),
        total_freight_amount=_str_or_zero(business.total_freight_amount),
        remark=business.remark,
    )


def _str_or_zero(value: Decimal | None) -> str:
    return str(value) if value is not None else "0"


def _item_to_out(item: BusinessItem) -> BusinessItemOut:
    return BusinessItemOut(
        id=item.id,
        product_id=item.product_id,
        product_name=item.product_name,
        spec_model=item.spec_model,
        unit=item.unit,
        product_category=item.product_category,
        quantity=str(item.quantity),
        sale_price=str(item.sale_price) if item.sale_price is not None else None,
        sale_amount=str(item.sale_amount),
        cost_amount=str(item.cost_amount) if item.cost_amount is not None else None,
        guided_price=str(item.guided_price) if item.guided_price is not None else None,
        remark=item.remark,
    )


def _build_item_rows(business_id: int, items: list[BusinessItemIn]) -> list[BusinessItem]:
    """入参明细行转 ORM 行：sale_amount 服务层按 数量×单价 计算固化。

    不信任前端传值（schema 亦无 sale_amount 入参）——单价可空时销售额记 0，
    由用户在编辑中补价后重算（毛利润视图中销售额 0 的行全部为 0，防除零）。
    """
    rows: list[BusinessItem] = []
    for item in items:
        quantity = Decimal(item.quantity)
        sale_price = Decimal(item.sale_price) if item.sale_price is not None else None
        sale_amount = (quantity * sale_price) if sale_price is not None else Decimal("0")
        rows.append(
            BusinessItem(
                business_id=business_id,
                product_id=item.product_id,
                product_name=item.product_name,
                spec_model=item.spec_model or "",
                unit=item.unit,
                product_category=item.product_category,
                quantity=quantity,
                sale_price=sale_price,
                sale_amount=sale_amount,
                cost_amount=Decimal(item.cost_amount) if item.cost_amount is not None else None,
                guided_price=Decimal(item.guided_price) if item.guided_price is not None else None,
                remark=item.remark,
            )
        )
    return rows


def _recalc_totals(session: Session, business: Business) -> None:
    """重算业务的五项汇总冗余（明细/发货/回款写路径统一收口）。

    五条独立聚合查询（各表单列 SUM，走 business_id 索引）——业务单据量级
    为单业务数行，不值得为省查询做单大 JOIN。
    """
    sale = session.execute(
        _sum(BusinessItem.sale_amount, BusinessItem.business_id, business.id)
    ).scalar_one()
    cost = session.execute(
        _sum(BusinessItem.cost_amount, BusinessItem.business_id, business.id)
    ).scalar_one()
    ship = session.execute(_sum_ship_items(business.id)).scalar_one()
    freight = session.execute(
        _sum(Shipment.freight_amount, Shipment.business_id, business.id)
    ).scalar_one()
    received = session.execute(_sum(Receipt.amount, Receipt.business_id, business.id)).scalar_one()
    business.total_sale_amount = sale or Decimal("0")
    business.total_cost_amount = cost or Decimal("0")
    business.total_ship_amount = ship or Decimal("0")
    business.total_freight_amount = freight or Decimal("0")
    business.total_received_amount = received or Decimal("0")


def _sum(column, match_column, value) -> object:
    return select(func.sum(column)).where(match_column == value)


def _sum_ship_items(business_id: int):
    """发货明细行金额合计（经发货单关联到业务）。"""
    from app.models import ShipmentItem

    return (
        select(func.sum(ShipmentItem.amount))
        .select_from(ShipmentItem)
        .join(Shipment, Shipment.id == ShipmentItem.shipment_id)
        .where(Shipment.business_id == business_id)
    )


def _generate_no(session: Session) -> str:
    """生成业务编号（MAX+FOR UPDATE；1062 兜底重试一次，见模块 docstring）。"""
    return next_no(session, model=Business, column_name="business_no", prefix=NO_PREFIX_BUSINESS)


def _ensure_personnel_active(session: Session, personnel_id: int, label: str) -> Personnel:
    """归属人员须存在且在职（人员侧复用校验文案口径）。"""
    person = personnel_repository.get_by_id(session, personnel_id)
    if person is None or person.status != _PERSONNEL_STATUS_ACTIVE:
        raise BadRequestError(f"{label}人员不存在或已离职")
    return person


def _ensure_customer(session: Session, customer_id: int) -> Customer:
    customer = customer_repository.get_by_id(session, customer_id)
    if customer is None:
        raise NotFoundError("客户不存在")
    return customer


def _is_duplicate_entry(exc: IntegrityError) -> bool:
    return is_duplicate_entry(exc)


def list_businesses(session: Session, query: BusinessQuery) -> PageResult[BusinessOut]:
    """分页查询业务（附客户名与归属人名）。"""
    rows = business_repository.list_businesses(
        session,
        offset=query.offset,
        limit=query.page_size,
        keyword=query.keyword,
        status=query.status,
        customer_id=query.customer_id,
        owner_personnel_id=query.owner_personnel_id,
        date_from=query.date_from,
        date_to=query.date_to,
    )
    total = business_repository.count_businesses(
        session,
        keyword=query.keyword,
        status=query.status,
        customer_id=query.customer_id,
        owner_personnel_id=query.owner_personnel_id,
        date_from=query.date_from,
        date_to=query.date_to,
    )
    return PageResult[BusinessOut](
        records=[_to_out(row) for row in rows],
        total=total,
        page_num=query.page_num,
        page_size=query.page_size,
    )


def get_detail(session: Session, business_id: int) -> BusinessDetailOut:
    """业务详情（附明细行，供编辑抽屉回显）。

    :raises NotFoundError: 业务不存在
    """
    row = business_repository.get_with_names(session, business_id)
    if row is None:
        raise NotFoundError("业务不存在")
    out = BusinessDetailOut(**_to_out(row).model_dump())
    out.items = [
        _item_to_out(item) for item in business_repository.list_items(session, business_id)
    ]
    return out


def create_business(
    session: Session,
    payload: BusinessCreateIn,
    *,
    operator: User,
    meta: RequestMeta | None = None,
) -> BusinessOut:
    """创建业务 + 明细行 + 首条状态留痕，一个事务。

    创建时指定的归属不写归属留痕（留痕记「变更」，创建即初始事实）。

    :raises NotFoundError: 客户不存在
    :raises BadRequestError: 归属人员不存在或已离职 / 明细校验不通过
    """
    with audit_service.audit(
        session,
        operator=operator,
        record=AuditRecord(
            module=MODULE_BUSINESS,
            action=ACTION_CREATE,
            summary=f"创建业务（客户ID={payload.customer_id}，明细 {len(payload.items)} 行）",
            request_data=payload.model_dump(exclude_none=True),
        ),
        meta=meta,
    ):
        _ensure_customer(session, payload.customer_id)
        if payload.owner_personnel_id is not None:
            _ensure_personnel_active(session, payload.owner_personnel_id, "负责")
        if payload.assistant_personnel_id is not None:
            _ensure_personnel_active(session, payload.assistant_personnel_id, "协助")

        # 编号唯一键冲突的重试整体收在审计上下文内：rollback 后同一事务重来，
        # 成功/最终失败各只记一条审计（吞异常 return 会被记成假成功，重入
        # 整个 create_* 则违反「不支持嵌套」约定）；上限 _NO_RETRY 防无限重试
        # 掩盖死锁类缺陷。审计摘要预取的是首次编号——重试成功时入库编号见
        # 应用日志（AuditRecord frozen，不回填）。
        for attempt in range(_NO_RETRY + 1):
            try:
                business = Business(
                    business_no=_generate_no(session),
                    customer_id=payload.customer_id,
                    business_date=payload.business_date,
                    status=BUSINESS_STATUS_ACTIVE,
                    owner_personnel_id=payload.owner_personnel_id,
                    assistant_personnel_id=payload.assistant_personnel_id,
                    is_special=payload.is_special,
                    special_note=payload.special_note,
                    remark=payload.remark,
                )
                business_repository.create(session, business)
                business_repository.replace_items(
                    session, business.id, _build_item_rows(business.id, payload.items)
                )
                _recalc_totals(session, business)
                business_repository.add_status_log(
                    session,
                    BusinessStatusLog(
                        business_id=business.id,
                        old_status=BUSINESS_STATUS_ACTIVE,
                        new_status=BUSINESS_STATUS_ACTIVE,
                        operator_id=operator.id,
                        operator_name=operator.username,
                    ),
                )
                customer = customer_repository.get_by_id(session, business.customer_id)
                out = _to_out((business, customer.company_name if customer else None, None, None))
                new_id, new_no = business.id, business.business_no  # 固化：commit 后不碰 ORM
                session.commit()
                break
            except IntegrityError as exc:
                session.rollback()
                if _is_duplicate_entry(exc) and attempt < _NO_RETRY:
                    continue
                raise

        logger.info("创建业务 | business_id=%s | no=%s", new_id, new_no)
        return out


def update_business(
    session: Session,
    business_id: int,
    payload: BusinessUpdateIn,
    *,
    operator: User,
    meta: RequestMeta | None = None,
) -> BusinessOut:
    """编辑业务：全量替换明细 + 重算汇总；归属不在此端点（防旁路）。

    客户可改，但有发货或回款记录后禁改 —— 单据 FK 挂在业务上，换客户会让
    发货/回款语义断裂（明细同理随业务走）。

    :raises NotFoundError: 业务不存在
    :raises BadRequestError: 业务已完成或已终止 / 归属人员不合法 / 有发货或回款时换客户
    """
    row = business_repository.get_with_names(session, business_id, for_update=True)
    business = row[0] if row else None
    with audit_service.audit(
        session,
        operator=operator,
        record=AuditRecord(
            module=MODULE_BUSINESS,
            action=ACTION_UPDATE,
            summary=f"修改业务 {business.business_no if business else f'#{business_id}'}",
            request_data=payload.model_dump(exclude_none=True),
        ),
        meta=meta,
    ):
        if row is None:
            raise NotFoundError("业务不存在")
        business, _, owner_name, assistant_name = row

        # 终态门槛与发货/回款写准入同口径：明细与汇总是 M3 核算基线，
        # 已完成/已终止的业务禁改（发货/回款服务以同一理由禁写终态业务）
        if business.status != BUSINESS_STATUS_ACTIVE:
            raise BadRequestError("业务已完成或已终止，禁止编辑业务")

        if payload.customer_id != business.customer_id and business_repository.has_docs(
            session, business_id
        ):
            raise BadRequestError("业务已有发货或回款记录，不能更换客户")
        customer = _ensure_customer(session, payload.customer_id)
        if payload.owner_personnel_id is not None:
            _ensure_personnel_active(session, payload.owner_personnel_id, "负责")
        if payload.assistant_personnel_id is not None:
            _ensure_personnel_active(session, payload.assistant_personnel_id, "协助")

        business.customer_id = payload.customer_id
        business.business_date = payload.business_date
        business.is_special = payload.is_special
        business.special_note = payload.special_note
        business.remark = payload.remark
        # 归属刻意不赋值：留痕只由 /ownership 端点产生，编辑路径改归属构成
        # 「改归属不留痕」的旁路（与 customers 编辑不触归属同一口径）
        business_repository.replace_items(
            session, business_id, _build_item_rows(business_id, payload.items)
        )
        _recalc_totals(session, business)
        # 锁定行上已赋值，同一身份映射实例的 _to_out 即反映新值；名称字段
        # 从首查 row 取（换客户时 _ensure_customer 已给出新公司名）
        out = _to_out((business, customer.company_name, owner_name, assistant_name))
        session.commit()
        logger.info("修改业务 | business_id=%s", business_id)
        return out


def delete_business(
    session: Session,
    business_id: int,
    *,
    operator: User,
    meta: RequestMeta | None = None,
) -> None:
    """删除业务（物理删除）：有发货/回款 → 400 引导走「已终止」。

    无单据引用的业务（建错的空单）方可删除；删除时连带清理明细、状态留痕
    与业务维度的归属留痕（无外键固化表，须手动清）。

    :raises NotFoundError: 业务不存在
    :raises BadRequestError: 存在发货或回款记录
    """
    business = business_repository.get_by_id(session, business_id, for_update=True)
    with audit_service.audit(
        session,
        operator=operator,
        record=AuditRecord(
            module=MODULE_BUSINESS,
            action=ACTION_DELETE,
            summary=f"删除业务 {business.business_no if business else f'#{business_id}'}",
        ),
        meta=meta,
    ):
        if business is None:
            raise NotFoundError("业务不存在")
        if business_repository.has_docs(session, business_id):
            raise BadRequestError("业务已有发货或回款记录，请改为已终止而非删除")

        business_no = business.business_no
        _hard_delete(session, business_id)
        session.commit()
        logger.info("删除业务 | business_id=%s | no=%s", business_id, business_no)


def _hard_delete(session: Session, business_id: int) -> None:
    """按依赖序删除：明细 → 状态留痕 → 业务维度归属留痕 → 主表。"""
    from sqlalchemy import delete

    session.execute(delete(BusinessItem).where(BusinessItem.business_id == business_id))
    session.execute(delete(BusinessStatusLog).where(BusinessStatusLog.business_id == business_id))
    session.execute(
        delete(OwnershipChangeLog).where(
            OwnershipChangeLog.owner_type == OWNER_TYPE_BUSINESS,
            OwnershipChangeLog.business_id == business_id,
        )
    )
    session.execute(delete(Business).where(Business.id == business_id))


def change_status(
    session: Session,
    business_id: int,
    payload: BusinessStatusIn,
    *,
    operator: User,
    meta: RequestMeta | None = None,
) -> BusinessOut:
    """状态流转：1→2 / 1→3，终态不可再变更；每次流转写留痕。

    :raises NotFoundError: 业务不存在
    :raises BadRequestError: 非执行中状态流转 / 目标与当前相同
    """
    row = business_repository.get_with_names(session, business_id, for_update=True)
    with audit_service.audit(
        session,
        operator=operator,
        record=AuditRecord(
            module=MODULE_BUSINESS,
            action=ACTION_UPDATE,
            summary=(
                f"业务状态变更 {row[0].business_no if row else f'#{business_id}'} → "
                f"{'已完成' if payload.status == BUSINESS_STATUS_DONE else '已终止'}"
            ),
            request_data=payload.model_dump(exclude_none=True),
        ),
        meta=meta,
    ):
        if row is None:
            raise NotFoundError("业务不存在")
        business, _, _, _ = row
        if business.status != BUSINESS_STATUS_ACTIVE:
            raise BadRequestError("已完成或已终止的业务不可再变更状态")
        if business.status == payload.status:
            raise BadRequestError("状态未变化")

        # 变更前状态须在赋值前固化：ORM 属性赋值后 old_status 就取不到旧值
        # （留痕会记成 old==new 失真，时间线无法回溯）
        old_status = business.status
        business.status = payload.status
        business_repository.add_status_log(
            session,
            BusinessStatusLog(
                business_id=business_id,
                old_status=old_status,
                new_status=payload.status,
                change_reason=payload.reason,
                operator_id=operator.id,
                operator_name=operator.username,
            ),
        )
        # 锁定行上已赋值，同一身份映射实例的 _to_out 即反映新状态
        out = _to_out(row)
        session.commit()
        logger.info(
            "业务状态变更 | business_id=%s | %s -> %s",
            business_id,
            old_status,
            payload.status,
        )
        return out


def change_ownership(
    session: Session,
    business_id: int,
    payload: BusinessOwnershipIn,
    *,
    operator: User,
    meta: RequestMeta | None = None,
) -> BusinessOut:
    """归属变更（负责/协助）：同事务「更新主表 + 插留痕」，一次 commit。

    负责与协助之一发生变化即记一条留痕（前后值都含两个字段）；两者均未变化
    → 400「归属未变化」。

    :raises NotFoundError: 业务不存在
    :raises BadRequestError: 归属人员不存在或已离职 / 归属未变化
    """
    row = business_repository.get_with_names(session, business_id, for_update=True)
    with audit_service.audit(
        session,
        operator=operator,
        record=AuditRecord(
            module=MODULE_BUSINESS,
            action=ACTION_UPDATE,
            summary=f"调整业务归属 {row[0].business_no if row else f'#{business_id}'}",
            request_data=payload.model_dump(exclude_none=True),
        ),
        meta=meta,
    ):
        if row is None:
            raise NotFoundError("业务不存在")
        business, _, old_owner_name, old_assistant_name = row
        # 归属变更的出参姓名来自校验函数的返回行（锁定读）：首查姓名是变更前
        # 固化标量，新归属人名必须按新 ID 重取
        new_owner = (
            _ensure_personnel_active(session, payload.new_owner_personnel_id, "负责")
            if payload.new_owner_personnel_id is not None
            else None
        )
        new_assistant = (
            _ensure_personnel_active(session, payload.new_assistant_personnel_id, "协助")
            if payload.new_assistant_personnel_id is not None
            else None
        )

        owner_changed = business.owner_personnel_id != payload.new_owner_personnel_id
        assistant_changed = business.assistant_personnel_id != payload.new_assistant_personnel_id
        if not owner_changed and not assistant_changed:
            raise BadRequestError("归属未变化")

        old_owner_id = business.owner_personnel_id
        old_assistant_id = business.assistant_personnel_id
        business.owner_personnel_id = payload.new_owner_personnel_id
        business.assistant_personnel_id = payload.new_assistant_personnel_id
        business_repository.add_ownership_log(
            session,
            OwnershipChangeLog(
                owner_type=OWNER_TYPE_BUSINESS,
                business_id=business_id,
                old_owner_personnel_id=old_owner_id,
                new_owner_personnel_id=payload.new_owner_personnel_id,
                old_assistant_personnel_id=old_assistant_id,
                new_assistant_personnel_id=payload.new_assistant_personnel_id,
                change_reason=payload.reason,
                operator_id=operator.id,
                operator_name=operator.username,
            ),
        )
        # 出参组装：ID 在锁定行上已赋值（_to_out 读内存值即反映新值），姓名
        # 一侧「变更项用新 ID 的人名、未变项沿用首查名」——首查名是变更前快照
        out = _to_out(
            (
                business,
                row[1],
                new_owner.personnel_name if new_owner is not None else old_owner_name,
                new_assistant.personnel_name if new_assistant is not None else old_assistant_name,
            )
        )
        session.commit()
        logger.info(
            "调整业务归属 | business_id=%s | owner %s->%s | assistant %s->%s",
            business_id,
            old_owner_id,
            payload.new_owner_personnel_id,
            old_assistant_id,
            payload.new_assistant_personnel_id,
        )
        return out


def list_ownership_logs(
    session: Session, business_id: int, query: PageQuery
) -> PageResult[OwnershipChangeLogOut]:
    """某业务的归属变更留痕（分页）。

    :raises NotFoundError: 业务不存在
    """
    if business_repository.get_by_id(session, business_id) is None:
        raise NotFoundError("业务不存在")
    rows = business_repository.list_ownership_logs(
        session, business_id, offset=query.offset, limit=query.page_size
    )
    total = business_repository.count_ownership_logs(session, business_id=business_id)
    return PageResult[OwnershipChangeLogOut](
        records=[_log_to_out(row) for row in rows],
        total=total,
        page_num=query.page_num,
        page_size=query.page_size,
    )


def _log_to_out(row: Row) -> OwnershipChangeLogOut:
    log, old_owner_name, new_owner_name = row
    return OwnershipChangeLogOut(
        id=log.id,
        business_id=log.business_id,
        old_owner_personnel_id=log.old_owner_personnel_id,
        new_owner_personnel_id=log.new_owner_personnel_id,
        old_assistant_personnel_id=log.old_assistant_personnel_id,
        new_assistant_personnel_id=log.new_assistant_personnel_id,
        old_owner_name=old_owner_name,
        new_owner_name=new_owner_name,
        change_reason=log.change_reason,
        operator_id=log.operator_id,
        operator_name=log.operator_name,
        create_time=log.create_time,
    )


def list_status_logs(
    session: Session, business_id: int, query: PageQuery
) -> PageResult[BusinessStatusLogOut]:
    """某业务的状态流转时间线（分页，时间正序）。

    :raises NotFoundError: 业务不存在
    """
    if business_repository.get_by_id(session, business_id) is None:
        raise NotFoundError("业务不存在")
    rows = business_repository.list_status_logs(
        session, business_id, offset=query.offset, limit=query.page_size
    )
    total = business_repository.count_status_logs(session, business_id=business_id)
    return PageResult[BusinessStatusLogOut](
        records=[BusinessStatusLogOut.model_validate(row) for row in rows],
        total=total,
        page_num=query.page_num,
        page_size=query.page_size,
    )


def gross_profit(session: Session, business_id: int, as_of: date | None) -> GrossProfitOut:
    """毛利润数据视图（需求 2.2.4）：行级派生值 + 类别汇总，实时计算不落库。

    阈值按 asOf（默认今日）取 M1 参数快照（product.threshold，按行类别维度，
    存储口径为 % 数如 30，直接进 converted_profit）；毛利润率按 M1 预览同一
    出参口径输出 Fraction 分数字符串（如 "97/10000"）；成本缺失的行派生列为
    None 并标注原因；汇总只聚合可计算行。

    :raises NotFoundError: 业务不存在
    """
    if business_repository.get_by_id(session, business_id) is None:
        raise NotFoundError("业务不存在")
    as_of = as_of or date.today()
    snapshot = param_service.resolve_snapshot(session, as_of)

    def threshold_of(category: str) -> Fraction:
        return snapshot.fraction("product.threshold", product_category=category)

    rows_out: list[GrossProfitRowOut] = []
    missing_count = 0
    # 类别聚合容器：{类别: [sale, cost]}，成本缺失行不计入该类别聚合
    agg: dict[str, list[Decimal]] = {}
    for item in business_repository.list_items(session, business_id):
        threshold = threshold_of(item.product_category)
        row_kwargs: dict = {
            "item_id": item.id,
            "product_category": item.product_category,
            "product_name": item.product_name,
            "spec_model": item.spec_model,
            "quantity": str(item.quantity),
            "sale_amount": str(item.sale_amount),
            "threshold": str(threshold),
        }
        if item.cost_amount is None:
            missing_count += 1
            row_kwargs.update(
                cost_amount=None,
                profit_amount=None,
                profit_rate=None,
                converted_profit=None,
                missing_reason="成本缺失，无法计算毛利润",
            )
        else:
            profit = item.sale_amount - item.cost_amount
            rate = profit_rate(profit, item.sale_amount)
            conv = converted_profit(item.sale_amount, rate, threshold)
            row_kwargs.update(
                cost_amount=str(item.cost_amount),
                profit_amount=str(profit),
                profit_rate=str(rate),
                converted_profit=str(conv),
                missing_reason=None,
            )
            bucket = agg.setdefault(item.product_category, [Decimal("0"), Decimal("0")])
            bucket[0] += item.sale_amount
            bucket[1] += item.cost_amount
        rows_out.append(GrossProfitRowOut(**row_kwargs))

    categories = [
        GrossProfitCategoryOut(
            product_category=category,
            sale_amount=str(sale),
            cost_amount=str(cost),
            profit_amount=str(sale - cost),
            profit_rate=str(profit_rate(sale - cost, sale)),
            threshold=str(threshold_of(category)),
            converted_profit=str(
                converted_profit(sale, profit_rate(sale - cost, sale), threshold_of(category))
            ),
        )
        for category, [sale, cost] in agg.items()
    ]
    return GrossProfitOut(
        business_id=business_id,
        as_of=as_of,
        rows=rows_out,
        categories=categories,
        missing_count=missing_count,
    )