仓储层的 list_*/count_* 此前以 keyword/status/offset/limit 四五个散参数 传参,调用点重复铺开、易漏传且不可扩展;按项目「参数 >4 封装 DTO」的 约定统一收敛为各模块的 Query 对象(如 RoleQuery、BusinessQuery), 服务层与测试调用同步调整,行为不变。 - repositories 层查询方法签名改为 (session, query),分页与筛选条件 由 Query 承载,_apply_filters 同步收口 - schemas 层补齐各模块 Query 模型(approval/business_role/report 等) - services 层调用点同步改为传 Query - tests 随之调整;新增用例:待办角色流转(财务通过后领导出现待办)、 参数预览顺序稳定 验证:全量 pytest 900 passed(11:08);单元 526 passed;ruff check 绿。 Co-Authored-By: Claude Code <noreply@anthropic.com>
248 lines
7.9 KiB
Python
248 lines
7.9 KiB
Python
"""工作台仓储:四区块的只读聚合查询;不 commit、不抛业务异常。
|
||
|
||
总览只做 SQL 聚合(SUM/COUNT DISTINCT),不复用报表的
|
||
``department_summary``——后者会加载整月台账并逐团队查人员,适合报表页,
|
||
不适合每次打开工作台。
|
||
"""
|
||
|
||
from datetime import date, datetime
|
||
|
||
from sqlalchemy import exists, func, or_, select
|
||
from sqlalchemy.orm import Session, defer
|
||
|
||
from app.models import (
|
||
ApprovalOrder,
|
||
Business,
|
||
BusinessRole,
|
||
MonthlySettlement,
|
||
PerformanceLedger,
|
||
Personnel,
|
||
Receipt,
|
||
Shipment,
|
||
)
|
||
from app.models.approval import (
|
||
APPROVAL_STATUS_PENDING_FINANCE,
|
||
APPROVAL_STATUS_PENDING_LEADER,
|
||
)
|
||
|
||
_ACTIVE_SINGLE = (PerformanceLedger.calc_type == 1, PerformanceLedger.is_active == 1)
|
||
_PENDING_STATUSES = (
|
||
APPROVAL_STATUS_PENDING_FINANCE,
|
||
APPROVAL_STATUS_PENDING_LEADER,
|
||
)
|
||
|
||
|
||
# ==== 经营绩效总览 ====
|
||
|
||
|
||
def sum_month_performance(session: Session, month: str) -> dict:
|
||
"""当月单笔生效行合计与人员数(拆行求和不失真,人数需 DISTINCT)。"""
|
||
return (
|
||
session.execute(
|
||
select(
|
||
func.coalesce(func.sum(PerformanceLedger.performance_amount), 0).label(
|
||
"performance_total"
|
||
),
|
||
func.coalesce(func.sum(PerformanceLedger.payable_amount), 0).label("payable_total"),
|
||
func.count(func.distinct(PerformanceLedger.personnel_id)).label("person_count"),
|
||
).where(
|
||
PerformanceLedger.settlement_month == month,
|
||
*_ACTIVE_SINGLE,
|
||
)
|
||
)
|
||
.mappings()
|
||
.one()
|
||
)
|
||
|
||
|
||
def performance_trend(session: Session, months: list[str]) -> dict[str, object]:
|
||
"""近 N 个月绩效合计(按台账归属月)。"""
|
||
rows = session.execute(
|
||
select(
|
||
PerformanceLedger.settlement_month,
|
||
func.coalesce(func.sum(PerformanceLedger.performance_amount), 0),
|
||
)
|
||
.where(
|
||
PerformanceLedger.settlement_month.in_(months),
|
||
*_ACTIVE_SINGLE,
|
||
)
|
||
.group_by(PerformanceLedger.settlement_month)
|
||
).all()
|
||
return dict(rows)
|
||
|
||
|
||
def receipt_trend(session: Session, *, start_date: date, end_date: date) -> dict[str, object]:
|
||
"""近 N 个月回款合计(按到账月,含未核算回款——与台账口径不同,页面注明)。
|
||
|
||
过滤走 ``receipt_date`` 原始列范围(可用索引);``DATE_FORMAT`` 只用于
|
||
分组,不参与 WHERE,避免函数列导致全表扫描。
|
||
"""
|
||
month_expr = func.date_format(Receipt.receipt_date, "%Y-%m")
|
||
rows = session.execute(
|
||
select(month_expr, func.coalesce(func.sum(Receipt.amount), 0))
|
||
.where(Receipt.receipt_date >= start_date, Receipt.receipt_date < end_date)
|
||
.group_by(month_expr)
|
||
).all()
|
||
return dict(rows)
|
||
|
||
|
||
def count_active_managers(session: Session) -> int:
|
||
"""在职营销经理数(团队数口径,与部门汇总的团队枚举一致)。"""
|
||
return int(
|
||
session.execute(
|
||
select(func.count())
|
||
.select_from(Personnel)
|
||
.join(BusinessRole, Personnel.business_role_id == BusinessRole.id)
|
||
.where(
|
||
BusinessRole.role_code == "sales_manager",
|
||
Personnel.status == 1,
|
||
)
|
||
).scalar_one()
|
||
)
|
||
|
||
|
||
def get_month_settlement(session: Session, month: str) -> MonthlySettlement | None:
|
||
"""当月月结头(状态与锁定时间角标)。"""
|
||
return session.execute(
|
||
select(MonthlySettlement).where(MonthlySettlement.settlement_month == month)
|
||
).scalar_one_or_none()
|
||
|
||
|
||
# ==== 待办事项 ====
|
||
|
||
|
||
def list_pending_approvals(
|
||
session: Session, *, statuses: tuple[int, ...], limit: int
|
||
) -> list[ApprovalOrder]:
|
||
"""待处理审批单(按提交时间升序——最久的排前面)。
|
||
|
||
``statuses`` 由调用方按当前用户角色解析(待我审批),空元组返回空列表。
|
||
|
||
排除 snapshot_json 列(快照可达上百 KB,待办只取单号/类型/提交人/时间)。
|
||
"""
|
||
if not statuses:
|
||
return []
|
||
return list(
|
||
session.execute(
|
||
select(ApprovalOrder)
|
||
.options(defer(ApprovalOrder.snapshot_json))
|
||
.where(ApprovalOrder.status.in_(statuses))
|
||
.order_by(ApprovalOrder.create_time.asc(), ApprovalOrder.id.asc())
|
||
.limit(limit)
|
||
).scalars()
|
||
)
|
||
|
||
|
||
def count_pending_approvals(session: Session, *, statuses: tuple[int, ...]) -> int:
|
||
if not statuses:
|
||
return 0
|
||
return int(
|
||
session.execute(
|
||
select(func.count())
|
||
.select_from(ApprovalOrder)
|
||
.where(ApprovalOrder.status.in_(statuses))
|
||
).scalar_one()
|
||
)
|
||
|
||
|
||
def list_unassigned_receipts(session: Session, *, limit: int) -> list:
|
||
"""回款未指定发货批次(shipment_id IS NULL,核算前置缺失)。"""
|
||
return list(
|
||
session.execute(
|
||
select(
|
||
Receipt,
|
||
Business.business_no.label("business_no"),
|
||
)
|
||
.outerjoin(Business, Business.id == Receipt.business_id)
|
||
.where(Receipt.shipment_id.is_(None))
|
||
.order_by(Receipt.receipt_date.asc(), Receipt.id.asc())
|
||
.limit(limit)
|
||
).all()
|
||
)
|
||
|
||
|
||
def count_unassigned_receipts(session: Session) -> int:
|
||
return int(
|
||
session.execute(
|
||
select(func.count()).select_from(Receipt).where(Receipt.shipment_id.is_(None))
|
||
).scalar_one()
|
||
)
|
||
|
||
|
||
# ==== 风险预警 ====
|
||
|
||
|
||
def list_overdue_shipments(session: Session, *, cutoff: date, limit: int) -> list:
|
||
"""发货后超过阈值且该批次无任何回款的发货单(回款逾期规则)。
|
||
|
||
已终止业务(status=3)不再催收,排除在预警之外;业务行缺失时仍保留,
|
||
避免历史脏数据让发货单从预警中静默消失。
|
||
"""
|
||
has_receipt = exists().where(Receipt.shipment_id == Shipment.id)
|
||
return list(
|
||
session.execute(
|
||
select(
|
||
Shipment,
|
||
Business.business_no.label("business_no"),
|
||
)
|
||
.outerjoin(Business, Business.id == Shipment.business_id)
|
||
.where(
|
||
Shipment.ship_date <= cutoff,
|
||
~has_receipt,
|
||
or_(Business.status.is_(None), Business.status != 3),
|
||
)
|
||
.order_by(Shipment.ship_date.asc())
|
||
.limit(limit)
|
||
).all()
|
||
)
|
||
|
||
|
||
def count_overdue_shipments(session: Session, *, cutoff: date) -> int:
|
||
has_receipt = exists().where(Receipt.shipment_id == Shipment.id)
|
||
return int(
|
||
session.execute(
|
||
select(func.count())
|
||
.select_from(Shipment)
|
||
.outerjoin(Business, Business.id == Shipment.business_id)
|
||
.where(
|
||
Shipment.ship_date <= cutoff,
|
||
~has_receipt,
|
||
or_(Business.status.is_(None), Business.status != 3),
|
||
)
|
||
).scalar_one()
|
||
)
|
||
|
||
|
||
def list_overdue_approvals(
|
||
session: Session, *, cutoff: datetime, limit: int
|
||
) -> list[ApprovalOrder]:
|
||
"""提交后超过阈值仍未处理的审批单(审批超时规则)。
|
||
|
||
排除 snapshot_json 列(同待办:预警只用单号/类型/提交人/时间)。
|
||
"""
|
||
return list(
|
||
session.execute(
|
||
select(ApprovalOrder)
|
||
.options(defer(ApprovalOrder.snapshot_json))
|
||
.where(
|
||
ApprovalOrder.status.in_(_PENDING_STATUSES),
|
||
ApprovalOrder.create_time < cutoff,
|
||
)
|
||
.order_by(ApprovalOrder.create_time.asc())
|
||
.limit(limit)
|
||
).scalars()
|
||
)
|
||
|
||
|
||
def count_overdue_approvals(session: Session, *, cutoff: datetime) -> int:
|
||
return int(
|
||
session.execute(
|
||
select(func.count())
|
||
.select_from(ApprovalOrder)
|
||
.where(
|
||
ApprovalOrder.status.in_(_PENDING_STATUSES),
|
||
ApprovalOrder.create_time < cutoff,
|
||
)
|
||
).scalar_one()
|
||
)
|