o2o-negosium-original/lps/services/admin_service.py

"""관리자 FE 도메인 로직 — 조회 중심 + 필수 액션(DEAD 재큐)만.

설정 변경(예산·임계 등)은 FE 에 두지 않는다 — 설정은 toml 단일 소스(재시작 반영) 원칙.
"""

import uuid

from fastapi import Depends
from sqlalchemy.exc import IntegrityError

from common.enums import ErrorType, JobStatus
from crud.bot_detection import BotDetectionLog
from crud.ip_session import IpSessionLog
from crud.job_crud import JobQueue
from crud.price_history import PriceHistory
from router.v1.lps.admin_protocol import (
    BotRow,
    CostBucket,
    HistogramBin,
    HourlyCount,
    IpSessionRow,
    JobListItem,
    ProductItem,
    Res_BotStats,
    Res_CostStats,
    Res_IpSessionStats,
    Res_JobList,
    Res_ProductList,
    Res_Requeue,
)

_STATUS_BY_NAME = {js.name: js.value for js in JobStatus}


class AdminService:
    def __init__(self, queue: JobQueue = Depends(JobQueue), history: PriceHistory = Depends(PriceHistory),
                 ip_log: IpSessionLog = Depends(IpSessionLog), bot_log: BotDetectionLog = Depends(BotDetectionLog)):
        self.queue = queue
        self.history = history
        self.ip_log = ip_log
        self.bot_log = bot_log

    async def list_jobs(self, status: str | None, q: str | None, limit: int, offset: int) -> Res_JobList:
        res = Res_JobList()
        status_code = _STATUS_BY_NAME.get(status.upper()) if status else None
        if status and status_code is None:
            res.result.SetResult(ErrorType.LPS_JOB_NOT_FOUND)   # 알 수 없는 상태 이름
            return res
        rows, total = await self.queue.list_jobs(status_code, q, limit, offset)
        res.total = total
        res.items = [JobListItem(
            job_id=r["job_id"], job_type=r["job_type"], status=JobStatus(r["status"]).name,
            priority=r["priority"], attempts=r["attempts"], max_attempts=r["max_attempts"],
            product_code=r.get("product_code"), product_name=r.get("product_name"),
            outcome=r.get("outcome"), final_lowest=r.get("final_lowest"), cost_usd=r.get("cost_usd"),
            last_error=r.get("last_error"),
            created_at=r["created_at"].isoformat(timespec="seconds"),
            run_started_at=r["run_started_at"].isoformat(timespec="seconds") if r.get("run_started_at") else None,
            updated_at=r["updated_at"].isoformat(timespec="seconds"),
        ) for r in rows]
        return res

    async def requeue(self, job_id: str) -> Res_Requeue:
        res = Res_Requeue(job_id=job_id)
        try:
            uuid.UUID(job_id)
        except (ValueError, TypeError):
            res.result.SetResult(ErrorType.LPS_JOB_NOT_FOUND)
            return res
        try:
            requeued = await self.queue.requeue(job_id)
        except IntegrityError:
            # 같은 상품의 활성 잡(PENDING/RUNNING)이 이미 있음 — 부분 유니크(dedupe) 위반
            res.result.SetResult(ErrorType.DB_ALREADY_SAME_KEY)
            return res
        if requeued is None:   # 없거나 DEAD 가 아님
            res.result.SetResult(ErrorType.LPS_JOB_NOT_FOUND)
            return res
        res.requeued = True
        return res

    async def list_products(self, q: str | None, limit: int) -> Res_ProductList:
        res = Res_ProductList()
        res.items = [ProductItem(
            product_code=r["product_code"], display_name=r.get("display_name"),
            triggered_at=r["triggered_at"].isoformat(timespec="seconds"), outcome=r["outcome"],
            naver_lowest=r.get("naver_lowest"), coupang_lowest=r.get("coupang_lowest"),
            final_lowest=r.get("final_lowest"), final_source=r.get("final_source"),
            searches=int(r.get("searches") or 0),
            sources=r.get("sources"), partial=bool(r.get("partial")),
        ) for r in await self.history.list_products(q, limit)]
        return res

    async def ip_session_stats(self, hours: int) -> Res_IpSessionStats:
        res = Res_IpSessionStats()
        st = await self.ip_log.admin_stats(hours)
        res.by_reason = st["by_reason"]
        res.histogram = [HistogramBin(**b) for b in st["histogram"]]
        res.block_min_requests = st["block_min_requests"]
        res.sessions = [IpSessionRow(
            **{**s, "created_at": s["created_at"].isoformat(timespec="seconds")}
        ) for s in st["sessions"]]
        return res

    async def bot_stats(self, hours: int) -> Res_BotStats:
        res = Res_BotStats()
        st = await self.bot_log.admin_stats(hours)
        res.hourly = [HourlyCount(bucket=h["bucket"], count=h["count"]) for h in st["hourly"]]
        res.items = [BotRow(
            **{**b, "created_at": b["created_at"].isoformat(timespec="seconds")}
        ) for b in st["items"]]
        return res

    async def cost_stats(self, hours: int) -> Res_CostStats:
        res = Res_CostStats()
        res.buckets = [CostBucket(**b) for b in await self.queue.cost_buckets(hours)]
        return res