o2o-negosium-original/negodata/backend/crud/dashboard_crud.py

from abc import ABC, abstractmethod
from typing import Tuple

from sqlalchemy import select, func, and_
from sqlalchemy.ext.asyncio import AsyncSession

from common.database.db_session_manager import DB_SESSION_MNG
from common.database.model.models import quotations, sessions, suppliers, users
from common.enums import ErrorType, QuotationStatus
from common.logger import LOG


# 대시보드 집계 CRUD. 모든 조회는 회사 스코프(작성자 user_id→users.company_id) 기준이며,
# owner(user_id) 가 주어지면 '내가 만든 견적'으로 더 좁힌다. quotations 엔 company_id 컬럼이 없어 서브쿼리로 건다.
def _company_scope(company_id, owner) -> list:
    conds = [
        quotations.deleted == False,  # noqa: E712
        quotations.user_id.in_(select(users.user_id).where(users.company_id == company_id)),
    ]
    if owner is not None:
        conds.append(quotations.user_id == owner)
    return conds


class IDashboardCRUD(ABC):
    @abstractmethod
    async def count_in_progress(self, cdb: AsyncSession, company_id, owner) -> Tuple[ErrorType, int]:
        pass

    @abstractmethod
    async def count_created_since(self, cdb: AsyncSession, company_id, owner, since) -> Tuple[ErrorType, int]:
        pass

    @abstractmethod
    async def deadline_soon(self, cdb: AsyncSession, company_id, owner, now, horizon, limit) -> Tuple[ErrorType, list, int]:
        pass

    @abstractmethod
    async def equal_bid(self, cdb: AsyncSession, company_id, owner, limit) -> Tuple[ErrorType, list, int]:
        pass

    @abstractmethod
    async def ruptured(self, cdb: AsyncSession, company_id, owner, limit) -> Tuple[ErrorType, list, int]:
        pass

    @abstractmethod
    async def email_unsent(self, cdb: AsyncSession, company_id, owner, limit) -> Tuple[ErrorType, list, int]:
        pass


class DashboardCRUD(IDashboardCRUD):
    async def _count(self, cdb: AsyncSession, where) -> Tuple[ErrorType, int]:
        err, rows = await DB_SESSION_MNG.execute(cdb, select(func.count()).select_from(quotations).where(where))
        if err != ErrorType.SUCCESS:
            return err, 0
        return ErrorType.SUCCESS, (int(rows[0] or 0) if rows else 0)

    async def _list_with_count(self, cdb: AsyncSession, where, cols, order_by, limit) -> Tuple[ErrorType, list, int]:
        c_err, total = await self._count(cdb, where)
        if c_err != ErrorType.SUCCESS:
            return c_err, [], 0
        l_err, rows = await DB_SESSION_MNG.execute(cdb, select(*cols).where(where).order_by(order_by).limit(limit))
        if l_err != ErrorType.SUCCESS:
            return l_err, [], 0
        return ErrorType.SUCCESS, list(rows), total

    async def count_in_progress(self, cdb: AsyncSession, company_id, owner) -> Tuple[ErrorType, int]:
        try:
            where = and_(*_company_scope(company_id, owner), quotations.status != QuotationStatus.CLOSED.value)
            return await self._count(cdb, where)
        except Exception as ex:
            LOG.e_no_callstack(ex)
            return ErrorType.DB_RUN_FAILED, 0

    async def count_created_since(self, cdb: AsyncSession, company_id, owner, since) -> Tuple[ErrorType, int]:
        try:
            where = and_(*_company_scope(company_id, owner), quotations.created_at >= since)
            return await self._count(cdb, where)
        except Exception as ex:
            LOG.e_no_callstack(ex)
            return ErrorType.DB_RUN_FAILED, 0

    async def deadline_soon(self, cdb: AsyncSession, company_id, owner, now, horizon, limit) -> Tuple[ErrorType, list, int]:
        try:
            where = and_(
                *_company_scope(company_id, owner),
                quotations.status != QuotationStatus.CLOSED.value,
                quotations.end_time >= now,
                quotations.end_time <= horizon,
            )
            cols = (quotations.qt_id, quotations.name, quotations.end_time)
            return await self._list_with_count(cdb, where, cols, quotations.end_time.asc(), limit)
        except Exception as ex:
            LOG.e_no_callstack(ex)
            return ErrorType.DB_RUN_FAILED, [], 0

    async def equal_bid(self, cdb: AsyncSession, company_id, owner, limit) -> Tuple[ErrorType, list, int]:
        try:
            where = and_(
                *_company_scope(company_id, owner),
                quotations.status == QuotationStatus.CLOSED.value,
                quotations.equal_bid_yn.is_(True),
            )
            cols = (quotations.qt_id, quotations.name)
            return await self._list_with_count(cdb, where, cols, quotations.updated_at.desc(), limit)
        except Exception as ex:
            LOG.e_no_callstack(ex)
            return ErrorType.DB_RUN_FAILED, [], 0

    async def ruptured(self, cdb: AsyncSession, company_id, owner, limit) -> Tuple[ErrorType, list, int]:
        try:
            # 결렬 = 마감됐는데 단독낙찰(preferred_sp_yn)도 동가(equal_bid_yn)도 아님 → 둘 다 NULL(거부/한도로 그냥 마감).
            where = and_(
                *_company_scope(company_id, owner),
                quotations.status == QuotationStatus.CLOSED.value,
                quotations.preferred_sp_yn.is_(None),
                quotations.equal_bid_yn.is_(None),
            )
            cols = (quotations.qt_id, quotations.name)
            return await self._list_with_count(cdb, where, cols, quotations.updated_at.desc(), limit)
        except Exception as ex:
            LOG.e_no_callstack(ex)
            return ErrorType.DB_RUN_FAILED, [], 0

    async def email_unsent(self, cdb: AsyncSession, company_id, owner, limit) -> Tuple[ErrorType, list, int]:
        try:
            # 미발송 세션 = email_sent_at IS NULL + 담당자 이메일 보유, 마감 전 견적만(보낼 의미 있는 것). 견적 단위로 묶는다.
            conds = [
                sessions.deleted == False,  # noqa: E712
                sessions.email_sent_at.is_(None),
                suppliers.manager_email.isnot(None),
                suppliers.manager_email != "",
                quotations.deleted == False,  # noqa: E712
                quotations.status != QuotationStatus.CLOSED.value,
                quotations.user_id.in_(select(users.user_id).where(users.company_id == company_id)),
            ]
            if owner is not None:
                conds.append(quotations.user_id == owner)
            where = and_(*conds)

            def _joined(stmt):
                return (
                    stmt.select_from(sessions)
                    .join(suppliers, suppliers.supplier_id == sessions.supplier_id)
                    .join(quotations, quotations.qt_id == sessions.quotation_id)
                    .where(where)
                )

            c_err, c_rows = await DB_SESSION_MNG.execute(cdb, _joined(select(func.count())))
            if c_err != ErrorType.SUCCESS:
                return c_err, [], 0
            total = int(c_rows[0] or 0) if c_rows else 0

            cnt = func.count().label("cnt")
            g_err, rows = await DB_SESSION_MNG.execute(
                cdb,
                _joined(select(quotations.qt_id, quotations.name, cnt))
                .group_by(quotations.qt_id, quotations.name)
                .order_by(cnt.desc())
                .limit(limit),
            )
            if g_err != ErrorType.SUCCESS:
                return g_err, [], 0
            return ErrorType.SUCCESS, list(rows), total
        except Exception as ex:
            LOG.e_no_callstack(ex)
            return ErrorType.DB_RUN_FAILED, [], 0