Mina Choi
f1e924931f
- 관측 사실 기준으로 재구성: 내부 점수 제거, 앵커링 % 표기, 탭 URL 분리 - 카드탭을 성과 지표에서 사용 현황으로 전환, 지표 설명 말풍선·카드 상세 링크 - 사이드바 devOnly + 라우트 가드로 개발자 전용 처리
157 lines
7.4 KiB
Python
157 lines
7.4 KiB
Python
from abc import ABC, abstractmethod
|
||
from typing import Tuple
|
||
|
||
from sqlalchemy import and_, asc, column, desc, func, select, table
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from common.database.db_session_manager import DB_SESSION_MNG
|
||
from common.database.model.models import nego_cards, wild_cards
|
||
from common.enums import ErrorType
|
||
|
||
# 학습(learning)·앵커링(anchoring) 스키마는 agent·anchoring 서비스 소유라 negodata ORM 에 없다.
|
||
# 조회 전용이므로 필요한 컬럼만 경량 정의한다 — 이 값들의 주인은 negodata 가 아니다(쓰기 금지).
|
||
_EXPERIENCE_LOGS = table(
|
||
"experience_logs",
|
||
column("company_id"), column("session_id"), column("card_id"), column("reward"),
|
||
column("settled_price"), column("is_invalidated"), column("created_at"),
|
||
column("turn"),
|
||
schema="learning",
|
||
)
|
||
_ANCHORING_CURRENT = table(
|
||
"current_values",
|
||
column("company_id"), column("supplier_type"), column("price_range_index"),
|
||
column("anchoring_value"), column("last_adjusted_at"),
|
||
schema="anchoring",
|
||
)
|
||
_ANCHORING_HISTORY = table(
|
||
"value_history",
|
||
column("company_id"), column("supplier_type"), column("price_range_index"),
|
||
column("anchoring_value_before"), column("anchoring_value_after"),
|
||
column("sample_count"), column("success_count"), column("success_rate"), column("created_at"),
|
||
schema="anchoring",
|
||
)
|
||
|
||
|
||
def _valid(company_id) -> list:
|
||
"""유효 학습 기록 — 무효화 표시된 행은 뺀다(협상 취소·재생성 시 agent 가 표시)."""
|
||
return [
|
||
_EXPERIENCE_LOGS.c.company_id == company_id,
|
||
_EXPERIENCE_LOGS.c.is_invalidated == False, # noqa: E712
|
||
]
|
||
|
||
|
||
class ILearningCRUD(ABC):
|
||
@abstractmethod
|
||
async def learning_summary(self, cdb: AsyncSession, company_id) -> Tuple[ErrorType, tuple]:
|
||
pass
|
||
|
||
@abstractmethod
|
||
async def card_usage(self, cdb: AsyncSession, company_id) -> Tuple[ErrorType, list]:
|
||
pass
|
||
|
||
@abstractmethod
|
||
async def card_names(self, cdb: AsyncSession) -> Tuple[ErrorType, list]:
|
||
pass
|
||
|
||
@abstractmethod
|
||
async def anchoring_current(self, cdb: AsyncSession, company_id) -> Tuple[ErrorType, list]:
|
||
pass
|
||
|
||
@abstractmethod
|
||
async def anchoring_history(self, cdb: AsyncSession, company_id, limit: int) -> Tuple[ErrorType, list]:
|
||
pass
|
||
|
||
|
||
class LearningCRUD(ILearningCRUD):
|
||
async def learning_summary(self, cdb: AsyncSession, company_id) -> Tuple[ErrorType, tuple]:
|
||
"""(학습 협상 수, 기록 수, 타결 협상 수, 마지막 학습 시각)."""
|
||
query = select(
|
||
func.count(func.distinct(_EXPERIENCE_LOGS.c.session_id)),
|
||
func.count(),
|
||
func.count(func.distinct(_EXPERIENCE_LOGS.c.session_id))
|
||
.filter(_EXPERIENCE_LOGS.c.settled_price.isnot(None)),
|
||
func.max(_EXPERIENCE_LOGS.c.created_at),
|
||
).where(and_(*_valid(company_id)))
|
||
err_type, rows = await DB_SESSION_MNG.execute(cdb, query, "learning_summary failed.", raise_error=False)
|
||
if err_type != ErrorType.SUCCESS or not rows:
|
||
return err_type, (0, 0, 0, None)
|
||
return ErrorType.SUCCESS, tuple(rows[0])
|
||
|
||
async def card_usage(self, cdb: AsyncSession, company_id) -> Tuple[ErrorType, list]:
|
||
"""카드별 사용 현황 — (카드번호, 사용 협상 수, 사용 횟수, 평균 라운드, 마지막 사용).
|
||
|
||
협상 결과(타결·가격)는 카드 한 장의 성과로 나눌 수 없어 다루지 않는다 — 한 협상에 여러 장이
|
||
나가 어느 장의 몫인지 가릴 근거가 없고, 카드 배정도 국면에 따라 정해져 무작위가 아니다.
|
||
평균 라운드 = 그 카드가 협상의 몇 번째 라운드에 나갔는지(설정한 국면과 실제가 맞는지 대조).
|
||
"""
|
||
query = (
|
||
select(
|
||
_EXPERIENCE_LOGS.c.card_id,
|
||
func.count(func.distinct(_EXPERIENCE_LOGS.c.session_id)),
|
||
func.count(),
|
||
func.avg(_EXPERIENCE_LOGS.c.turn),
|
||
func.max(_EXPERIENCE_LOGS.c.created_at),
|
||
)
|
||
.where(and_(*_valid(company_id), _EXPERIENCE_LOGS.c.card_id.isnot(None)))
|
||
.group_by(_EXPERIENCE_LOGS.c.card_id)
|
||
.order_by(desc(func.count()))
|
||
)
|
||
err_type, rows = await DB_SESSION_MNG.execute(cdb, query, "card_usage failed.", raise_error=False)
|
||
if err_type != ErrorType.SUCCESS:
|
||
return err_type, []
|
||
return ErrorType.SUCCESS, list(rows or [])
|
||
|
||
async def card_names(self, cdb: AsyncSession) -> Tuple[ErrorType, list]:
|
||
"""카드번호 → (번호, 이름, 와일드 여부, 카드 PK). 학습 로그의 card_id 가 카드번호 문자열이라
|
||
상세 화면(/cards?detail=<PK>) 으로 보내려면 PK 를 같이 들고 와야 한다.
|
||
|
||
UNION 은 실행기가 SELECT 로 인정하지 않아 두 번 나눠 조회한다.
|
||
"""
|
||
out = []
|
||
for model, pk, is_wild in ((nego_cards, nego_cards.nego_card_id, 0), (wild_cards, wild_cards.wild_card_id, 1)):
|
||
query = select(model.number, model.name, pk).where(model.deleted == False) # noqa: E712
|
||
err_type, rows = await DB_SESSION_MNG.execute(cdb, query, "card_names failed.", raise_error=False)
|
||
if err_type != ErrorType.SUCCESS:
|
||
return err_type, []
|
||
out.extend((number, name, is_wild, card_id) for number, name, card_id in (rows or []))
|
||
return ErrorType.SUCCESS, out
|
||
|
||
async def anchoring_current(self, cdb: AsyncSession, company_id) -> Tuple[ErrorType, list]:
|
||
"""현재 앵커링 값 — (협력사유형, 가격대 구간, 값, 마지막 조정 시각). 셀 = 유형 × 가격대."""
|
||
query = (
|
||
select(
|
||
_ANCHORING_CURRENT.c.supplier_type,
|
||
_ANCHORING_CURRENT.c.price_range_index,
|
||
_ANCHORING_CURRENT.c.anchoring_value,
|
||
_ANCHORING_CURRENT.c.last_adjusted_at,
|
||
)
|
||
.where(_ANCHORING_CURRENT.c.company_id == company_id)
|
||
.order_by(asc(_ANCHORING_CURRENT.c.supplier_type), asc(_ANCHORING_CURRENT.c.price_range_index))
|
||
)
|
||
err_type, rows = await DB_SESSION_MNG.execute(cdb, query, "anchoring_current failed.", raise_error=False)
|
||
if err_type != ErrorType.SUCCESS:
|
||
return err_type, []
|
||
return ErrorType.SUCCESS, list(rows or [])
|
||
|
||
async def anchoring_history(self, cdb: AsyncSession, company_id, limit: int) -> Tuple[ErrorType, list]:
|
||
"""앵커링 조정 이력(최근순) — 조정마다 표본 수·성공률·값 변화가 남는다."""
|
||
query = (
|
||
select(
|
||
_ANCHORING_HISTORY.c.supplier_type,
|
||
_ANCHORING_HISTORY.c.price_range_index,
|
||
_ANCHORING_HISTORY.c.anchoring_value_before,
|
||
_ANCHORING_HISTORY.c.anchoring_value_after,
|
||
_ANCHORING_HISTORY.c.sample_count,
|
||
_ANCHORING_HISTORY.c.success_count,
|
||
_ANCHORING_HISTORY.c.success_rate,
|
||
_ANCHORING_HISTORY.c.created_at,
|
||
)
|
||
.where(_ANCHORING_HISTORY.c.company_id == company_id)
|
||
.order_by(desc(_ANCHORING_HISTORY.c.created_at))
|
||
.limit(limit)
|
||
)
|
||
err_type, rows = await DB_SESSION_MNG.execute(cdb, query, "anchoring_history failed.", raise_error=False)
|
||
if err_type != ErrorType.SUCCESS:
|
||
return err_type, []
|
||
return ErrorType.SUCCESS, list(rows or [])
|