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 count_awarded_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 awarded(self, cdb: AsyncSession, company_id, owner, 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 count_awarded_since(self, cdb: AsyncSession, company_id, owner, since) -> Tuple[ErrorType, int]: try: # 이번 달 낙찰 = 마감 + 단독 최저 선정(preferred_sp_yn=True) + 낙찰(마감) 시각이 기준일 이후. where = and_( *_company_scope(company_id, owner), quotations.status == QuotationStatus.CLOSED.value, quotations.preferred_sp_yn.is_(True), quotations.updated_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 awarded(self, cdb: AsyncSession, company_id, owner, limit) -> Tuple[ErrorType, list, int]: try: # 낙찰 = 마감 + 단독 최저가 선정(preferred_sp_yn=True). 동가/결렬과 동형(최신 마감순). where = and_( *_company_scope(company_id, owner), quotations.status == QuotationStatus.CLOSED.value, quotations.preferred_sp_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 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 + 담당자 이메일 보유, 마감 전 견적만. 견적 단위로 묶는다. # total = 미발송 '견적' 수(distinct qt_id) — 리스트와 일치. 견적별 미발송 협력사 수는 행의 unsent_count. 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(func.distinct(quotations.qt_id))))) 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