sqxp/backend/app/crud/ord_settlement.py

"""客户/供应商结算单 CRUD（含 UPDATE 写历史表）"""
from __future__ import annotations

from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple

from sqlalchemy.orm import Session

from app.crud.base import CRUDBase
from app.models.models import (
    BizOrder,
    OrdClientSettlement,
    OrdClientSettlementHis,
    OrdClientSettlementItem,
    OrdClientSettlementItemHis,
    OrdSupplierSettlement,
    OrdSupplierSettlementHis,
    OrdSupplierSettlementItem,
    OrdSupplierSettlementItemHis,
)


class OrdClientSettlementCRUD(CRUDBase[OrdClientSettlement]):
    def _snapshot_main(self, row: OrdClientSettlement) -> Dict[str, Any]:
        return {
            "id": row.id,
            "settlement_no": row.settlement_no,
            "client_id": row.client_id,
            "client_name": row.client_name,
            "order_count": row.order_count,
            "total_amount": row.total_amount,
            "voucher_urls": row.voucher_urls,
            "remark": row.remark,
            "settlement_at": row.settlement_at,
            "status": row.status,
            "create_at": row.create_at,
            "create_by": row.create_by,
            "create_name": row.create_name,
            "update_at": row.update_at,
            "update_by": row.update_by,
        }

    def get_multi_with_filters(
        self,
        db: Session,
        *,
        client_id: Optional[str] = None,
        settlement_no: Optional[str] = None,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None,
        region_id: Optional[str] = None,
        skip: int = 0,
        limit: int = 10,
    ) -> Tuple[List[OrdClientSettlement], int]:
        q = db.query(OrdClientSettlement).filter(OrdClientSettlement.is_del == 0)
        if client_id:
            q = q.filter(OrdClientSettlement.client_id == client_id)
        if settlement_no:
            q = q.filter(OrdClientSettlement.settlement_no.like(f"%{settlement_no}%"))
        if start_date:
            q = q.filter(OrdClientSettlement.settlement_at >= start_date)
        if end_date:
            q = q.filter(OrdClientSettlement.settlement_at <= end_date)
        if region_id:
            sub = (
                db.query(OrdClientSettlementItem.settlement_id)
                .join(BizOrder, BizOrder.id == OrdClientSettlementItem.order_id)
                .filter(
                    OrdClientSettlementItem.is_del == 0,
                    BizOrder.is_del == 0,
                    BizOrder.region_id == region_id,
                )
                .distinct()
            )
            q = q.filter(OrdClientSettlement.id.in_(sub))
        total = q.count()
        rows = q.order_by(OrdClientSettlement.settlement_at.desc()).offset(skip).limit(limit).all()
        return rows, total

    def get_with_items(
        self, db: Session, settlement_id: str
    ) -> Tuple[Optional[OrdClientSettlement], List[OrdClientSettlementItem]]:
        main = self.get(db, settlement_id)
        if not main or main.is_del == 1:
            return None, []
        items = (
            db.query(OrdClientSettlementItem)
            .filter(
                OrdClientSettlementItem.settlement_id == settlement_id,
                OrdClientSettlementItem.is_del == 0,
            )
            .order_by(OrdClientSettlementItem.create_at.asc())
            .all()
        )
        return main, items

    def belongs_to_region(self, db: Session, settlement_id: str, region_id: str) -> bool:
        if not region_id:
            return True
        return (
            db.query(OrdClientSettlementItem.id)
            .join(BizOrder, BizOrder.id == OrdClientSettlementItem.order_id)
            .filter(
                OrdClientSettlementItem.settlement_id == settlement_id,
                OrdClientSettlementItem.is_del == 0,
                BizOrder.is_del == 0,
                BizOrder.region_id == region_id,
            )
            .first()
            is not None
        )

    def update_with_history(
        self,
        db: Session,
        settlement_id: str,
        obj_in: dict,
        updater: Optional[dict] = None,
    ) -> Optional[OrdClientSettlement]:
        original = self.get(db, settlement_id)
        if not original or original.is_del == 1:
            return None
        his = OrdClientSettlementHis(**self._snapshot_main(original))
        db.add(his)
        return self.update(db, settlement_id, obj_in, updater)


class OrdClientSettlementItemCRUD(CRUDBase[OrdClientSettlementItem]):
    def _snapshot_item(self, row: OrdClientSettlementItem) -> Dict[str, Any]:
        return {
            "id": row.id,
            "settlement_id": row.settlement_id,
            "order_id": row.order_id,
            "order_no": row.order_no,
            "client_id": row.client_id,
            "order_amount": row.order_amount,
            "create_at": row.create_at,
            "create_by": row.create_by,
            "create_name": row.create_name,
            "update_at": row.update_at,
            "update_by": row.update_by,
        }

    def update_with_history(
        self,
        db: Session,
        item_id: str,
        obj_in: dict,
        updater: Optional[dict] = None,
    ) -> Optional[OrdClientSettlementItem]:
        original = self.get(db, item_id)
        if not original or original.is_del == 1:
            return None
        his = OrdClientSettlementItemHis(**self._snapshot_item(original))
        db.add(his)
        return self.update(db, item_id, obj_in, updater)

    def soft_delete_with_history(self, db: Session, item_id: str, updater: Optional[dict] = None) -> bool:
        original = self.get(db, item_id)
        if not original or original.is_del == 1:
            return False
        his = OrdClientSettlementItemHis(**self._snapshot_item(original))
        db.add(his)
        return self.update(db, item_id, {"is_del": 1}, updater) is not None


class OrdSupplierSettlementCRUD(CRUDBase[OrdSupplierSettlement]):
    def _snapshot_main(self, row: OrdSupplierSettlement) -> Dict[str, Any]:
        return {
            "id": row.id,
            "settlement_no": row.settlement_no,
            "supplier_id": row.supplier_id,
            "supplier_name": row.supplier_name,
            "order_count": row.order_count,
            "total_amount": row.total_amount,
            "voucher_urls": row.voucher_urls,
            "remark": row.remark,
            "settlement_at": row.settlement_at,
            "status": row.status,
            "create_at": row.create_at,
            "create_by": row.create_by,
            "create_name": row.create_name,
            "update_at": row.update_at,
            "update_by": row.update_by,
        }

    def get_multi_with_filters(
        self,
        db: Session,
        *,
        supplier_id: Optional[str] = None,
        settlement_no: Optional[str] = None,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None,
        region_id: Optional[str] = None,
        skip: int = 0,
        limit: int = 10,
    ) -> Tuple[List[OrdSupplierSettlement], int]:
        q = db.query(OrdSupplierSettlement).filter(OrdSupplierSettlement.is_del == 0)
        if supplier_id:
            q = q.filter(OrdSupplierSettlement.supplier_id == supplier_id)
        if settlement_no:
            q = q.filter(OrdSupplierSettlement.settlement_no.like(f"%{settlement_no}%"))
        if start_date:
            q = q.filter(OrdSupplierSettlement.settlement_at >= start_date)
        if end_date:
            q = q.filter(OrdSupplierSettlement.settlement_at <= end_date)
        if region_id:
            sub = (
                db.query(OrdSupplierSettlementItem.settlement_id)
                .join(BizOrder, BizOrder.id == OrdSupplierSettlementItem.order_id)
                .filter(
                    OrdSupplierSettlementItem.is_del == 0,
                    BizOrder.is_del == 0,
                    BizOrder.region_id == region_id,
                )
                .distinct()
            )
            q = q.filter(OrdSupplierSettlement.id.in_(sub))
        total = q.count()
        rows = q.order_by(OrdSupplierSettlement.settlement_at.desc()).offset(skip).limit(limit).all()
        return rows, total

    def get_with_items(
        self, db: Session, settlement_id: str
    ) -> Tuple[Optional[OrdSupplierSettlement], List[OrdSupplierSettlementItem]]:
        main = self.get(db, settlement_id)
        if not main or main.is_del == 1:
            return None, []
        items = (
            db.query(OrdSupplierSettlementItem)
            .filter(
                OrdSupplierSettlementItem.settlement_id == settlement_id,
                OrdSupplierSettlementItem.is_del == 0,
            )
            .order_by(OrdSupplierSettlementItem.create_at.asc())
            .all()
        )
        return main, items

    def belongs_to_region(self, db: Session, settlement_id: str, region_id: str) -> bool:
        if not region_id:
            return True
        return (
            db.query(OrdSupplierSettlementItem.id)
            .join(BizOrder, BizOrder.id == OrdSupplierSettlementItem.order_id)
            .filter(
                OrdSupplierSettlementItem.settlement_id == settlement_id,
                OrdSupplierSettlementItem.is_del == 0,
                BizOrder.is_del == 0,
                BizOrder.region_id == region_id,
            )
            .first()
            is not None
        )

    def update_with_history(
        self,
        db: Session,
        settlement_id: str,
        obj_in: dict,
        updater: Optional[dict] = None,
    ) -> Optional[OrdSupplierSettlement]:
        original = self.get(db, settlement_id)
        if not original or original.is_del == 1:
            return None
        his = OrdSupplierSettlementHis(**self._snapshot_main(original))
        db.add(his)
        return self.update(db, settlement_id, obj_in, updater)


class OrdSupplierSettlementItemCRUD(CRUDBase[OrdSupplierSettlementItem]):
    def _snapshot_item(self, row: OrdSupplierSettlementItem) -> Dict[str, Any]:
        return {
            "id": row.id,
            "settlement_id": row.settlement_id,
            "order_id": row.order_id,
            "order_no": row.order_no,
            "supplier_id": row.supplier_id,
            "order_amount": row.order_amount,
            "create_at": row.create_at,
            "create_by": row.create_by,
            "create_name": row.create_name,
            "update_at": row.update_at,
            "update_by": row.update_by,
        }

    def update_with_history(
        self,
        db: Session,
        item_id: str,
        obj_in: dict,
        updater: Optional[dict] = None,
    ) -> Optional[OrdSupplierSettlementItem]:
        original = self.get(db, item_id)
        if not original or original.is_del == 1:
            return None
        his = OrdSupplierSettlementItemHis(**self._snapshot_item(original))
        db.add(his)
        return self.update(db, item_id, obj_in, updater)

    def soft_delete_with_history(self, db: Session, item_id: str, updater: Optional[dict] = None) -> bool:
        original = self.get(db, item_id)
        if not original or original.is_del == 1:
            return False
        his = OrdSupplierSettlementItemHis(**self._snapshot_item(original))
        db.add(his)
        return self.update(db, item_id, {"is_del": 1}, updater) is not None


ord_client_settlement_crud = OrdClientSettlementCRUD(OrdClientSettlement)
ord_client_settlement_item_crud = OrdClientSettlementItemCRUD(OrdClientSettlementItem)
ord_supplier_settlement_crud = OrdSupplierSettlementCRUD(OrdSupplierSettlement)
ord_supplier_settlement_item_crud = OrdSupplierSettlementItemCRUD(OrdSupplierSettlementItem)