init

src/inventory/services/__init__.py

@@ -0,0 +1,5 @@
+"""进销存业务服务层
+
+将原 api/*_routes.py 中的业务编排（事务、库存增减、流水写入、对账）下沉到此层，
+路由层只做参数校验与响应组装。每个业务域一个 service。
+"""

src/inventory/services/finance_service.py

@@ -0,0 +1,742 @@
+"""财务业务服务层
+
+将原 finance_routes 中的业务编排（收款/付款/核销/作废/对账）下沉到此，
+路由层只做参数校验与响应组装。
+"""
+from fastapi import HTTPException
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select, func
+from sqlalchemy.orm import selectinload
+from typing import Optional, List, Dict, Tuple
+from datetime import datetime
+from decimal import Decimal
+
+from shared.models.database import (
+    User,
+    Customer,
+    Supplier,
+    Product,
+    SalesOrder,
+    SalesOrderItem,
+    PurchaseOrder,
+    PurchaseOrderItem,
+    FinanceTransaction,
+    FinanceAllocation,
+)
+from ..schemas import (
+    ReceiptCreate,
+    PaymentCreate,
+    FinanceTransactionResponse,
+    FinanceSummaryResponse,
+    ReceivableItemResponse,
+    PayableItemResponse,
+    FinancePartnerStatementResponse,
+    PartnerStatementItemResponse,
+    FinancePartnerProductStatementResponse,
+    PartnerProductStatementItemResponse,
+    PaginatedResponse,
+)
+from ..utils import generate_order_no
+from shared.utils.logger import get_logger
+
+logger = get_logger(__name__)
+
+
+class FinanceService:
+    """财务业务服务：收款/付款/核销/作废/对账"""
+
+    @staticmethod
+    def _build_transaction_response(txn: FinanceTransaction, partner_name: Optional[str] = None) -> FinanceTransactionResponse:
+        allocations = [
+            {
+                "id": item.id,
+                "order_type": item.order_type,
+                "order_id": item.order_id,
+                "allocated_amount": item.allocated_amount,
+            }
+            for item in txn.allocations
+        ]
+        return FinanceTransactionResponse(
+            id=txn.id,
+            txn_no=txn.txn_no,
+            txn_type=txn.txn_type,
+            partner_type=txn.partner_type,
+            partner_id=txn.partner_id,
+            partner_name=partner_name,
+            amount=txn.amount,
+            txn_date=txn.txn_date,
+            method=txn.method,
+            account_name=txn.account_name,
+            status=txn.status,
+            remark=txn.remark,
+            created_at=txn.created_at,
+            allocations=allocations,
+        )
+
+    @staticmethod
+    def _validate_allocation_total(transaction_amount: float, allocation_amounts: List[float]):
+        allocated_total = sum(allocation_amounts)
+        if allocated_total - transaction_amount > 1e-6:
+            raise HTTPException(status_code=400, detail="核销总额不能大于单据金额")
+
+    @staticmethod
+    def _resolve_period_scope(year: Optional[int], quarter: Optional[int]) -> Tuple[int, Optional[int], str, datetime, datetime]:
+        now = datetime.now()
+        selected_year = year or now.year
+        if selected_year < 2000 or selected_year > 2100:
+            raise HTTPException(status_code=400, detail="年份超出支持范围")
+
+        if quarter is not None and quarter not in [1, 2, 3, 4]:
+            raise HTTPException(status_code=400, detail="季度必须是1-4")
+
+        if quarter is None:
+            period_start = datetime(selected_year, 1, 1)
+            period_end = datetime(selected_year + 1, 1, 1)
+            period_label = f"{selected_year}年"
+        else:
+            start_month = (quarter - 1) * 3 + 1
+            period_start = datetime(selected_year, start_month, 1)
+            if quarter == 4:
+                period_end = datetime(selected_year + 1, 1, 1)
+            else:
+                period_end = datetime(selected_year, start_month + 3, 1)
+            period_label = f"{selected_year}年Q{quarter}"
+
+        return selected_year, quarter, period_label, period_start, period_end
+
+    @staticmethod
+    async def create_receipt(db_session: AsyncSession, payload: ReceiptCreate, current_user: User) -> FinanceTransactionResponse:
+        customer_result = await db_session.execute(
+            select(Customer).where(Customer.id == payload.customer_id, Customer.is_active == True)
+        )
+        customer = customer_result.scalar_one_or_none()
+        if not customer:
+            raise HTTPException(status_code=404, detail="客户不存在")
+
+        partner_name = customer.name
+        FinanceService._validate_allocation_total(payload.amount, [item.allocated_amount for item in payload.allocations])
+
+        txn = FinanceTransaction(
+            txn_no=generate_order_no("RC"),
+            txn_type="receipt",
+            partner_type="customer",
+            partner_id=payload.customer_id,
+            amount=payload.amount,
+            txn_date=payload.txn_date or datetime.now(),
+            method=payload.method,
+            account_name=payload.account_name,
+            status="confirmed",
+            remark=payload.remark,
+            operator_id=current_user.id,
+        )
+        db_session.add(txn)
+        await db_session.flush()
+
+        for allocation in payload.allocations:
+            if allocation.order_type != "sales":
+                raise HTTPException(status_code=400, detail="收款单只允许核销销售订单")
+
+            order_result = await db_session.execute(
+                select(SalesOrder).where(SalesOrder.id == allocation.order_id, SalesOrder.customer_id == payload.customer_id)
+            )
+            sales_order = order_result.scalar_one_or_none()
+            if not sales_order:
+                raise HTTPException(status_code=404, detail=f"销售订单不存在: {allocation.order_id}")
+
+            remaining = (sales_order.total_amount or 0) - (sales_order.received_amount or 0)
+            if allocation.allocated_amount - remaining > 1e-6:
+                raise HTTPException(status_code=400, detail=f"销售订单核销超额: {sales_order.order_no}")
+
+            db_session.add(
+                FinanceAllocation(
+                    transaction_id=txn.id,
+                    order_type="sales",
+                    order_id=sales_order.id,
+                    allocated_amount=allocation.allocated_amount,
+                )
+            )
+            sales_order.received_amount = (sales_order.received_amount or 0) + allocation.allocated_amount
+
+        await db_session.commit()
+        result = await db_session.execute(
+            select(FinanceTransaction)
+            .options(selectinload(FinanceTransaction.allocations))
+            .where(FinanceTransaction.id == txn.id)
+        )
+        created = result.scalar_one()
+        return FinanceService._build_transaction_response(created, partner_name=partner_name)
+
+    @staticmethod
+    async def create_payment(db_session: AsyncSession, payload: PaymentCreate, current_user: User) -> FinanceTransactionResponse:
+        supplier_result = await db_session.execute(
+            select(Supplier).where(Supplier.id == payload.supplier_id, Supplier.is_active == True)
+        )
+        supplier = supplier_result.scalar_one_or_none()
+        if not supplier:
+            raise HTTPException(status_code=404, detail="供应商不存在")
+
+        partner_name = supplier.name
+        FinanceService._validate_allocation_total(payload.amount, [item.allocated_amount for item in payload.allocations])
+
+        txn = FinanceTransaction(
+            txn_no=generate_order_no("PY"),
+            txn_type="payment",
+            partner_type="supplier",
+            partner_id=payload.supplier_id,
+            amount=payload.amount,
+            txn_date=payload.txn_date or datetime.now(),
+            method=payload.method,
+            account_name=payload.account_name,
+            status="confirmed",
+            remark=payload.remark,
+            operator_id=current_user.id,
+        )
+        db_session.add(txn)
+        await db_session.flush()
+
+        for allocation in payload.allocations:
+            if allocation.order_type != "purchase":
+                raise HTTPException(status_code=400, detail="付款单只允许核销采购订单")
+
+            order_result = await db_session.execute(
+                select(PurchaseOrder).where(PurchaseOrder.id == allocation.order_id, PurchaseOrder.supplier_id == payload.supplier_id)
+            )
+            purchase_order = order_result.scalar_one_or_none()
+            if not purchase_order:
+                raise HTTPException(status_code=404, detail=f"采购订单不存在: {allocation.order_id}")
+
+            remaining = (purchase_order.total_amount or 0) - (purchase_order.paid_amount or 0)
+            if allocation.allocated_amount - remaining > 1e-6:
+                raise HTTPException(status_code=400, detail=f"采购订单核销超额: {purchase_order.order_no}")
+
+            db_session.add(
+                FinanceAllocation(
+                    transaction_id=txn.id,
+                    order_type="purchase",
+                    order_id=purchase_order.id,
+                    allocated_amount=allocation.allocated_amount,
+                )
+            )
+            purchase_order.paid_amount = (purchase_order.paid_amount or 0) + allocation.allocated_amount
+
+        await db_session.commit()
+        result = await db_session.execute(
+            select(FinanceTransaction)
+            .options(selectinload(FinanceTransaction.allocations))
+            .where(FinanceTransaction.id == txn.id)
+        )
+        created = result.scalar_one()
+        return FinanceService._build_transaction_response(created, partner_name=partner_name)
+
+    @staticmethod
+    async def list_transactions(
+        db_session: AsyncSession,
+        txn_type: Optional[str],
+        status: Optional[str],
+        year: Optional[int],
+        quarter: Optional[int],
+        skip: int,
+        limit: int,
+    ) -> PaginatedResponse:
+        base_query = (
+            select(FinanceTransaction)
+            .options(selectinload(FinanceTransaction.allocations))
+            .order_by(FinanceTransaction.created_at.desc())
+        )
+        if txn_type:
+            base_query = base_query.where(FinanceTransaction.txn_type == txn_type)
+        if status:
+            base_query = base_query.where(FinanceTransaction.status == status)
+        if year is not None or quarter is not None:
+            _, _, _, period_start, period_end = FinanceService._resolve_period_scope(year, quarter)
+            base_query = base_query.where(FinanceTransaction.txn_date >= period_start).where(FinanceTransaction.txn_date < period_end)
+
+        count_query = select(func.count()).select_from(base_query.subquery())
+        total = await db_session.scalar(count_query) or 0
+
+        query = base_query.offset(skip).limit(limit)
+        result = await db_session.execute(query)
+        rows = result.scalars().all()
+
+        # 批量查询往来方名称（交易记录只存 partner_id，需关联客户/供应商取名称）
+        partner_names: Dict[int, str] = {}
+        customer_ids = {t.partner_id for t in rows if t.partner_type == "customer"}
+        supplier_ids = {t.partner_id for t in rows if t.partner_type == "supplier"}
+        if customer_ids:
+            cr = await db_session.execute(select(Customer.id, Customer.name).where(Customer.id.in_(customer_ids)))
+            partner_names.update({r[0]: r[1] for r in cr.all()})
+        if supplier_ids:
+            sr = await db_session.execute(select(Supplier.id, Supplier.name).where(Supplier.id.in_(supplier_ids)))
+            partner_names.update({r[0]: r[1] for r in sr.all()})
+
+        return PaginatedResponse(
+            items=[FinanceService._build_transaction_response(item, partner_name=partner_names.get(item.partner_id)) for item in rows],
+            total=total, skip=skip, limit=limit
+        )
+
+    @staticmethod
+    async def void_transaction(db_session: AsyncSession, transaction_id: int, current_user: User) -> dict:
+        result = await db_session.execute(
+            select(FinanceTransaction)
+            .options(selectinload(FinanceTransaction.allocations))
+            .where(FinanceTransaction.id == transaction_id)
+        )
+        txn = result.scalar_one_or_none()
+        if not txn:
+            raise HTTPException(status_code=404, detail="财务单据不存在")
+        if txn.status == "voided":
+            return {"message": "单据已作废"}
+
+        for allocation in txn.allocations:
+            if allocation.order_type == "sales":
+                sales_result = await db_session.execute(select(SalesOrder).where(SalesOrder.id == allocation.order_id))
+                sales_order = sales_result.scalar_one_or_none()
+                if sales_order:
+                    sales_order.received_amount = max((sales_order.received_amount or 0) - allocation.allocated_amount, 0)
+            elif allocation.order_type == "purchase":
+                purchase_result = await db_session.execute(select(PurchaseOrder).where(PurchaseOrder.id == allocation.order_id))
+                purchase_order = purchase_result.scalar_one_or_none()
+                if purchase_order:
+                    purchase_order.paid_amount = max((purchase_order.paid_amount or 0) - allocation.allocated_amount, 0)
+
+        txn.status = "voided"
+        await db_session.commit()
+        logger.warning(
+            "财务单据已作废: txn_no=%s txn_type=%s amount=%s operator_id=%s",
+            txn.txn_no, txn.txn_type, txn.amount, current_user.id
+        )
+        return {"message": "单据已作废"}
+
+    @staticmethod
+    async def list_receivables(
+        db_session: AsyncSession,
+        year: Optional[int],
+        quarter: Optional[int],
+        skip: int,
+        limit: int,
+    ) -> List[ReceivableItemResponse]:
+        query = (
+            select(SalesOrder, Customer)
+            .join(Customer, SalesOrder.customer_id == Customer.id)
+            .where((SalesOrder.total_amount - SalesOrder.received_amount) > 0)
+            .order_by(SalesOrder.created_at.desc())
+        )
+        if year is not None or quarter is not None:
+            _, _, _, period_start, period_end = FinanceService._resolve_period_scope(year, quarter)
+            query = query.where(SalesOrder.order_date >= period_start).where(SalesOrder.order_date < period_end)
+        result = await db_session.execute(query.offset(skip).limit(limit))
+        rows = []
+        for order, customer in result.all():
+            receivable_amount = (order.total_amount or 0) - (order.received_amount or 0)
+            rows.append(
+                ReceivableItemResponse(
+                    order_id=order.id,
+                    order_no=order.order_no,
+                    customer_id=customer.id,
+                    customer_name=customer.name,
+                    order_date=order.order_date,
+                    total_amount=order.total_amount or 0,
+                    received_amount=order.received_amount or 0,
+                    receivable_amount=receivable_amount,
+                    status=order.status,
+                )
+            )
+        return rows
+
+    @staticmethod
+    async def list_payables(
+        db_session: AsyncSession,
+        year: Optional[int],
+        quarter: Optional[int],
+        skip: int,
+        limit: int,
+    ) -> List[PayableItemResponse]:
+        query = (
+            select(PurchaseOrder, Supplier)
+            .join(Supplier, PurchaseOrder.supplier_id == Supplier.id)
+            .where((PurchaseOrder.total_amount - PurchaseOrder.paid_amount) > 0)
+            .order_by(PurchaseOrder.created_at.desc())
+        )
+        if year is not None or quarter is not None:
+            _, _, _, period_start, period_end = FinanceService._resolve_period_scope(year, quarter)
+            query = query.where(PurchaseOrder.order_date >= period_start).where(PurchaseOrder.order_date < period_end)
+        result = await db_session.execute(query.offset(skip).limit(limit))
+        rows = []
+        for order, supplier in result.all():
+            payable_amount = (order.total_amount or 0) - (order.paid_amount or 0)
+            rows.append(
+                PayableItemResponse(
+                    order_id=order.id,
+                    order_no=order.order_no,
+                    supplier_id=supplier.id,
+                    supplier_name=supplier.name,
+                    order_date=order.order_date,
+                    total_amount=order.total_amount or 0,
+                    paid_amount=order.paid_amount or 0,
+                    payable_amount=payable_amount,
+                    status=order.status,
+                )
+            )
+        return rows
+
+    @staticmethod
+    async def get_summary(db_session: AsyncSession, year: Optional[int], quarter: Optional[int]) -> FinanceSummaryResponse:
+        receivable_total = await db_session.scalar(
+            select(func.coalesce(func.sum(SalesOrder.total_amount - SalesOrder.received_amount), 0))
+            .where((SalesOrder.total_amount - SalesOrder.received_amount) > 0)
+        ) or 0
+        payable_total = await db_session.scalar(
+            select(func.coalesce(func.sum(PurchaseOrder.total_amount - PurchaseOrder.paid_amount), 0))
+            .where((PurchaseOrder.total_amount - PurchaseOrder.paid_amount) > 0)
+        ) or 0
+
+        now = datetime.now()
+        month_start = datetime(now.year, now.month, 1)
+
+        monthly_receipt_total = await db_session.scalar(
+            select(func.coalesce(func.sum(FinanceTransaction.amount), 0))
+            .where(FinanceTransaction.txn_type == "receipt")
+            .where(FinanceTransaction.status == "confirmed")
+            .where(FinanceTransaction.txn_date >= month_start)
+        ) or 0
+        monthly_payment_total = await db_session.scalar(
+            select(func.coalesce(func.sum(FinanceTransaction.amount), 0))
+            .where(FinanceTransaction.txn_type == "payment")
+            .where(FinanceTransaction.status == "confirmed")
+            .where(FinanceTransaction.txn_date >= month_start)
+        ) or 0
+
+        selected_year, selected_quarter, period_label, period_start, period_end = FinanceService._resolve_period_scope(year, quarter)
+        period_receipt_total = await db_session.scalar(
+            select(func.coalesce(func.sum(FinanceTransaction.amount), 0))
+            .where(FinanceTransaction.txn_type == "receipt")
+            .where(FinanceTransaction.status == "confirmed")
+            .where(FinanceTransaction.txn_date >= period_start)
+            .where(FinanceTransaction.txn_date < period_end)
+        ) or 0
+        period_payment_total = await db_session.scalar(
+            select(func.coalesce(func.sum(FinanceTransaction.amount), 0))
+            .where(FinanceTransaction.txn_type == "payment")
+            .where(FinanceTransaction.status == "confirmed")
+            .where(FinanceTransaction.txn_date >= period_start)
+            .where(FinanceTransaction.txn_date < period_end)
+        ) or 0
+
+        return FinanceSummaryResponse(
+            receivable_total=Decimal(str(receivable_total)),
+            payable_total=Decimal(str(payable_total)),
+            monthly_receipt_total=Decimal(str(monthly_receipt_total)),
+            monthly_payment_total=Decimal(str(monthly_payment_total)),
+            selected_year=selected_year,
+            selected_quarter=selected_quarter,
+            period_label=period_label,
+            period_receipt_total=Decimal(str(period_receipt_total)),
+            period_payment_total=Decimal(str(period_payment_total)),
+            overdue_receivable_count=0,
+            overdue_payable_count=0,
+        )
+
+    @staticmethod
+    async def get_partner_statement(
+        db_session: AsyncSession,
+        partner_type: str,
+        year: Optional[int],
+        quarter: Optional[int],
+    ) -> FinancePartnerStatementResponse:
+        if partner_type not in ["customer", "supplier"]:
+            raise HTTPException(status_code=400, detail="partner_type 必须是 customer 或 supplier")
+
+        selected_year, selected_quarter, period_label, period_start, period_end = FinanceService._resolve_period_scope(year, quarter)
+        stats_map: Dict[int, Dict] = {}
+
+        if partner_type == "customer":
+            order_rows = await db_session.execute(
+                select(SalesOrder, Customer)
+                .join(Customer, SalesOrder.customer_id == Customer.id)
+                .where(SalesOrder.order_date >= period_start)
+                .where(SalesOrder.order_date < period_end)
+                .where(Customer.is_active == True)
+            )
+            for order, customer in order_rows.all():
+                partner_stat = stats_map.setdefault(
+                    customer.id,
+                    {
+                        "partner_id": customer.id,
+                        "partner_name": customer.name,
+                        "order_count": 0,
+                        "transaction_count": 0,
+                        "order_total": 0.0,
+                        "settled_total": 0.0,
+                        "transaction_total": 0.0,
+                        "outstanding_total": 0.0,
+                    },
+                )
+                total_amount = Decimal(str(order.total_amount or 0))
+                settled_amount = Decimal(str(order.received_amount or 0))
+                outstanding = max(total_amount - settled_amount, Decimal("0"))
+                partner_stat["order_count"] += 1
+                partner_stat["order_total"] += total_amount
+                partner_stat["settled_total"] += settled_amount
+                partner_stat["outstanding_total"] += outstanding
+
+            transaction_rows = await db_session.execute(
+                select(FinanceTransaction)
+                .where(FinanceTransaction.partner_type == "customer")
+                .where(FinanceTransaction.txn_type == "receipt")
+                .where(FinanceTransaction.status == "confirmed")
+                .where(FinanceTransaction.txn_date >= period_start)
+                .where(FinanceTransaction.txn_date < period_end)
+            )
+            for txn in transaction_rows.scalars().all():
+                partner_stat = stats_map.setdefault(
+                    txn.partner_id,
+                    {
+                        "partner_id": txn.partner_id,
+                        "partner_name": f"客户#{txn.partner_id}",
+                        "order_count": 0,
+                        "transaction_count": 0,
+                        "order_total": 0.0,
+                        "settled_total": 0.0,
+                        "transaction_total": 0.0,
+                        "outstanding_total": 0.0,
+                    },
+                )
+                partner_stat["transaction_count"] += 1
+                partner_stat["transaction_total"] += Decimal(str(txn.amount or 0))
+        else:
+            order_rows = await db_session.execute(
+                select(PurchaseOrder, Supplier)
+                .join(Supplier, PurchaseOrder.supplier_id == Supplier.id)
+                .where(PurchaseOrder.order_date >= period_start)
+                .where(PurchaseOrder.order_date < period_end)
+                .where(Supplier.is_active == True)
+            )
+            for order, supplier in order_rows.all():
+                partner_stat = stats_map.setdefault(
+                    supplier.id,
+                    {
+                        "partner_id": supplier.id,
+                        "partner_name": supplier.name,
+                        "order_count": 0,
+                        "transaction_count": 0,
+                        "order_total": 0.0,
+                        "settled_total": 0.0,
+                        "transaction_total": 0.0,
+                        "outstanding_total": 0.0,
+                    },
+                )
+                total_amount = Decimal(str(order.total_amount or 0))
+                settled_amount = Decimal(str(order.paid_amount or 0))
+                outstanding = max(total_amount - settled_amount, Decimal("0"))
+                partner_stat["order_count"] += 1
+                partner_stat["order_total"] += total_amount
+                partner_stat["settled_total"] += settled_amount
+                partner_stat["outstanding_total"] += outstanding
+
+            transaction_rows = await db_session.execute(
+                select(FinanceTransaction)
+                .where(FinanceTransaction.partner_type == "supplier")
+                .where(FinanceTransaction.txn_type == "payment")
+                .where(FinanceTransaction.status == "confirmed")
+                .where(FinanceTransaction.txn_date >= period_start)
+                .where(FinanceTransaction.txn_date < period_end)
+            )
+            for txn in transaction_rows.scalars().all():
+                partner_stat = stats_map.setdefault(
+                    txn.partner_id,
+                    {
+                        "partner_id": txn.partner_id,
+                        "partner_name": f"供应商#{txn.partner_id}",
+                        "order_count": 0,
+                        "transaction_count": 0,
+                        "order_total": 0.0,
+                        "settled_total": 0.0,
+                        "transaction_total": 0.0,
+                        "outstanding_total": 0.0,
+                    },
+                )
+                partner_stat["transaction_count"] += 1
+                partner_stat["transaction_total"] += Decimal(str(txn.amount or 0))
+
+        missing_partner_ids = [pid for pid, item in stats_map.items() if "#" in item["partner_name"]]
+        if missing_partner_ids:
+            if partner_type == "customer":
+                name_rows = await db_session.execute(
+                    select(Customer.id, Customer.name).where(Customer.id.in_(missing_partner_ids))
+                )
+            else:
+                name_rows = await db_session.execute(
+                    select(Supplier.id, Supplier.name).where(Supplier.id.in_(missing_partner_ids))
+                )
+            name_map = {row[0]: row[1] for row in name_rows.all()}
+            for pid in missing_partner_ids:
+                if pid in name_map:
+                    stats_map[pid]["partner_name"] = name_map[pid]
+
+        items = [
+            PartnerStatementItemResponse(
+                partner_id=item["partner_id"],
+                partner_name=item["partner_name"],
+                order_count=item["order_count"],
+                transaction_count=item["transaction_count"],
+                order_total=Decimal(str(item["order_total"])),
+                settled_total=Decimal(str(item["settled_total"])),
+                transaction_total=Decimal(str(item["transaction_total"])),
+                outstanding_total=Decimal(str(item["outstanding_total"])),
+                period_year=selected_year,
+                period_quarter=selected_quarter,
+            )
+            for item in sorted(stats_map.values(), key=lambda x: (x["outstanding_total"], x["order_total"]), reverse=True)
+        ]
+
+        return FinancePartnerStatementResponse(
+            partner_type=partner_type,
+            year=selected_year,
+            quarter=selected_quarter,
+            period_label=period_label,
+            order_total=Decimal(str(sum(item.order_total for item in items))),
+            settled_total=Decimal(str(sum(item.settled_total for item in items))),
+            transaction_total=Decimal(str(sum(item.transaction_total for item in items))),
+            outstanding_total=Decimal(str(sum(item.outstanding_total for item in items))),
+            items=items,
+        )
+
+    @staticmethod
+    async def get_partner_product_statement(
+        db_session: AsyncSession,
+        partner_type: str,
+        partner_id: Optional[int],
+        year: Optional[int],
+        quarter: Optional[int],
+    ) -> FinancePartnerProductStatementResponse:
+        if partner_type not in ["customer", "supplier"]:
+            raise HTTPException(status_code=400, detail="partner_type 必须是 customer 或 supplier")
+
+        selected_year, selected_quarter, period_label, period_start, period_end = FinanceService._resolve_period_scope(year, quarter)
+        stats_map: Dict[Tuple[int, int], Dict] = {}
+
+        if partner_type == "customer":
+            query = (
+                select(SalesOrderItem, SalesOrder, Product, Customer)
+                .join(SalesOrder, SalesOrderItem.order_id == SalesOrder.id)
+                .join(Product, SalesOrderItem.product_id == Product.id)
+                .join(Customer, SalesOrder.customer_id == Customer.id)
+                .where(SalesOrder.order_date >= period_start)
+                .where(SalesOrder.order_date < period_end)
+                .where(Customer.is_active == True)
+            )
+            if partner_id:
+                query = query.where(Customer.id == partner_id)
+
+            result = await db_session.execute(query)
+            for item, order, product, customer in result.all():
+                map_key = (customer.id, product.id)
+                stat = stats_map.setdefault(
+                    map_key,
+                    {
+                        "partner_id": customer.id,
+                        "partner_name": customer.name,
+                        "product_id": product.id,
+                        "product_sku": product.sku,
+                        "product_name": product.name,
+                        "order_ids": set(),
+                        "order_quantity": 0.0,
+                        "order_amount": 0.0,
+                        "settled_amount": 0.0,
+                        "outstanding_amount": 0.0,
+                    },
+                )
+
+                item_amount = Decimal(str(item.amount or 0))
+                order_total = Decimal(str(order.total_amount or 0))
+                order_settled = max(Decimal(str(order.received_amount or 0)), Decimal("0"))
+                ratio = (item_amount / order_total) if order_total > Decimal("1e-9") else Decimal("0")
+                item_settled = min(item_amount, order_settled * ratio)
+                item_outstanding = max(item_amount - item_settled, Decimal("0"))
+
+                stat["order_ids"].add(order.id)
+                stat["order_quantity"] += Decimal(str(item.quantity or 0))
+                stat["order_amount"] += item_amount
+                stat["settled_amount"] += item_settled
+                stat["outstanding_amount"] += item_outstanding
+        else:
+            query = (
+                select(PurchaseOrderItem, PurchaseOrder, Product, Supplier)
+                .join(PurchaseOrder, PurchaseOrderItem.order_id == PurchaseOrder.id)
+                .join(Product, PurchaseOrderItem.product_id == Product.id)
+                .join(Supplier, PurchaseOrder.supplier_id == Supplier.id)
+                .where(PurchaseOrder.order_date >= period_start)
+                .where(PurchaseOrder.order_date < period_end)
+                .where(Supplier.is_active == True)
+            )
+            if partner_id:
+                query = query.where(Supplier.id == partner_id)
+
+            result = await db_session.execute(query)
+            for item, order, product, supplier in result.all():
+                map_key = (supplier.id, product.id)
+                stat = stats_map.setdefault(
+                    map_key,
+                    {
+                        "partner_id": supplier.id,
+                        "partner_name": supplier.name,
+                        "product_id": product.id,
+                        "product_sku": product.sku,
+                        "product_name": product.name,
+                        "order_ids": set(),
+                        "order_quantity": 0.0,
+                        "order_amount": 0.0,
+                        "settled_amount": 0.0,
+                        "outstanding_amount": 0.0,
+                    },
+                )
+
+                item_amount = Decimal(str(item.amount or 0))
+                order_total = Decimal(str(order.total_amount or 0))
+                order_settled = max(Decimal(str(order.paid_amount or 0)), Decimal("0"))
+                ratio = (item_amount / order_total) if order_total > Decimal("1e-9") else Decimal("0")
+                item_settled = min(item_amount, order_settled * ratio)
+                item_outstanding = max(item_amount - item_settled, Decimal("0"))
+
+                stat["order_ids"].add(order.id)
+                stat["order_quantity"] += Decimal(str(item.quantity or 0))
+                stat["order_amount"] += item_amount
+                stat["settled_amount"] += item_settled
+                stat["outstanding_amount"] += item_outstanding
+
+        items = [
+            PartnerProductStatementItemResponse(
+                partner_id=item["partner_id"],
+                partner_name=item["partner_name"],
+                product_id=item["product_id"],
+                product_sku=item["product_sku"],
+                product_name=item["product_name"],
+                order_count=len(item["order_ids"]),
+                order_quantity=Decimal(str(item["order_quantity"])),
+                order_amount=Decimal(str(item["order_amount"])),
+                settled_amount=Decimal(str(item["settled_amount"])),
+                outstanding_amount=Decimal(str(item["outstanding_amount"])),
+                period_year=selected_year,
+                period_quarter=selected_quarter,
+            )
+            for item in sorted(
+                stats_map.values(),
+                key=lambda x: (x["outstanding_amount"], x["order_amount"]),
+                reverse=True
+            )
+        ]
+
+        return FinancePartnerProductStatementResponse(
+            partner_type=partner_type,
+            year=selected_year,
+            quarter=selected_quarter,
+            period_label=period_label,
+            partner_id=partner_id,
+            order_amount_total=Decimal(str(sum(item.order_amount for item in items))),
+            settled_amount_total=Decimal(str(sum(item.settled_amount for item in items))),
+            outstanding_amount_total=Decimal(str(sum(item.outstanding_amount for item in items))),
+            items=items,
+        )
+
+
+finance_service = FinanceService()