88 lines
3.4 KiB
Python
88 lines
3.4 KiB
Python
|
from datetime import datetime
|
||
|
|
|
||
|
|
from sqlalchemy import and_, select
|
||
|
|
from sqlalchemy.orm import Session
|
||
|
|
|
||
|
|
from backend.app.models.business import Customer, CustomerArrears, SalesOrder, SystemReminder
|
||
|
|
|
||
|
|
|
||
|
|
class ReminderRepository:
|
||
|
|
def list_reminders(self, session: Session, filters: dict) -> list[SystemReminder]:
|
||
|
|
stmt = select(SystemReminder)
|
||
|
|
|
||
|
|
if filters.get("reminder_type"):
|
||
|
|
stmt = stmt.where(SystemReminder.reminder_type == filters["reminder_type"])
|
||
|
|
if filters.get("status"):
|
||
|
|
stmt = stmt.where(SystemReminder.status == filters["status"])
|
||
|
|
if filters.get("receiver_user_id") is not None:
|
||
|
|
stmt = stmt.where(SystemReminder.receiver_user_id == filters["receiver_user_id"])
|
||
|
|
|
||
|
|
stmt = stmt.order_by(SystemReminder.id.desc())
|
||
|
|
return list(session.execute(stmt).scalars())
|
||
|
|
|
||
|
|
def get_reminder(self, session: Session, reminder_id: int) -> SystemReminder | None:
|
||
|
|
stmt = select(SystemReminder).where(SystemReminder.id == reminder_id)
|
||
|
|
return session.execute(stmt).scalar_one_or_none()
|
||
|
|
|
||
|
|
def create_reminder(self, session: Session, payload: dict) -> SystemReminder:
|
||
|
|
reminder = SystemReminder(**payload)
|
||
|
|
session.add(reminder)
|
||
|
|
session.flush()
|
||
|
|
return reminder
|
||
|
|
|
||
|
|
def find_active_reminder(
|
||
|
|
self,
|
||
|
|
session: Session,
|
||
|
|
reminder_type: str,
|
||
|
|
biz_type: str,
|
||
|
|
biz_id: int,
|
||
|
|
receiver_user_id: int,
|
||
|
|
) -> SystemReminder | None:
|
||
|
|
stmt = select(SystemReminder).where(
|
||
|
|
SystemReminder.reminder_type == reminder_type,
|
||
|
|
SystemReminder.biz_type == biz_type,
|
||
|
|
SystemReminder.biz_id == biz_id,
|
||
|
|
SystemReminder.receiver_user_id == receiver_user_id,
|
||
|
|
SystemReminder.status.in_(["pending", "sent", "read"]),
|
||
|
|
)
|
||
|
|
return session.execute(stmt).scalar_one_or_none()
|
||
|
|
|
||
|
|
def list_overdue_arrears(self, session: Session) -> list[tuple[CustomerArrears, Customer | None, SalesOrder | None]]:
|
||
|
|
stmt = (
|
||
|
|
select(CustomerArrears, Customer, SalesOrder)
|
||
|
|
.outerjoin(Customer, Customer.id == CustomerArrears.customer_id)
|
||
|
|
.outerjoin(SalesOrder, SalesOrder.id == CustomerArrears.order_id)
|
||
|
|
.where(CustomerArrears.status.in_(["pending", "overdue"]))
|
||
|
|
.order_by(CustomerArrears.id.asc())
|
||
|
|
)
|
||
|
|
return list(session.execute(stmt).all())
|
||
|
|
|
||
|
|
def list_inactive_customers(
|
||
|
|
self,
|
||
|
|
session: Session,
|
||
|
|
inactive_before: datetime,
|
||
|
|
) -> list[tuple[Customer, SalesOrder | None]]:
|
||
|
|
customers = list(
|
||
|
|
session.execute(
|
||
|
|
select(Customer)
|
||
|
|
.where(Customer.deleted == 0)
|
||
|
|
.order_by(Customer.id.asc())
|
||
|
|
).scalars()
|
||
|
|
)
|
||
|
|
|
||
|
|
results: list[tuple[Customer, SalesOrder | None]] = []
|
||
|
|
for customer in customers:
|
||
|
|
latest_order_stmt = (
|
||
|
|
select(SalesOrder)
|
||
|
|
.where(
|
||
|
|
SalesOrder.deleted == 0,
|
||
|
|
SalesOrder.customer_id == customer.id,
|
||
|
|
SalesOrder.order_status != "canceled",
|
||
|
|
)
|
||
|
|
.order_by(SalesOrder.created_at.desc(), SalesOrder.id.desc())
|
||
|
|
)
|
||
|
|
latest_order = session.execute(latest_order_stmt).scalars().first()
|
||
|
|
if latest_order is None or (latest_order.created_at and latest_order.created_at <= inactive_before):
|
||
|
|
results.append((customer, latest_order))
|
||
|
|
return results
|