o2o-site-AEO/solution/backend/services/faq_service.py

"""FAQ 조회 + 승인(검증 상태 전이).

COPY 잡이 faqs 를 UNVERIFIED 로 쌓아두지만, 절대규칙 1 때문에 확인 전에는 사이트에 나가지 못한다
(snapshot 이 PUBLISHABLE 만 싣는다). 즉 **승인 창구가 없으면 FAQPage JSON-LD 는 영영 안 나간다** —
이 서비스가 그 창구다.

fact 와 같은 규칙을 따른다(다른 표를 쓰면 두 도메인의 승인 의미가 갈라진다):
  1. 회사 스코프는 사업장 조회로 강제 — 남의 회사 place 의 FAQ 는 보이지도 만져지지도 않는다
  2. 상태 전이는 FACT_STATUS_TRANSITIONS 에 있는 것만 (FACT_INVALID_TRANSITION)
  3. CORRECTED 는 잠긴 종착 상태(LOCKED_FACT_STATUSES) — 재생성이 사장님 문구를 덮어쓰지 못한다
  4. 사람이 직접 쓴 FAQ 는 바로 노출값이다 — 사람이 곧 출처이자 책임 주체다
"""
import uuid

from fastapi import Depends

from common.database.db_session_manager import DB_SESSION_MNG
from common.database.model.models import place_faqs, places
from common.enums import (
    CANDIDATE_FACT_STATUSES,
    FACT_STATUS_TRANSITIONS,
    PUBLISHABLE_FACT_STATUSES,
    DBWRType,
    ErrorType,
    FactStatus,
    SourceType,
)
from common.logger import LOG
from common.models.gmodel import UserInfo
from common.utils.gtime import GTime
from crud.faq_crud import FaqCRUD, IFaqCRUD
from crud.place_crud import PlaceCRUD
from router.v1.faq.protocol import FaqData, Req_CreateFaq, Req_TransitionFaq, Res_Faq, Res_FaqList


class FaqService:
    def __init__(self, crud: IFaqCRUD = Depends(FaqCRUD), place_crud: PlaceCRUD = Depends(PlaceCRUD)):
        self.crud = crud
        self.place_crud = place_crud

    # ---- 사업장 로드(사장님 스코프) ----
    async def _load_place(self, user_info: UserInfo, place_id: str):
        """owner_user_id 를 WHERE 에 걸어 조회한다 — 남의 place_id 를 넣으면 PLACE_NOT_FOUND.
        '없다'와 '권한 없다'를 구분해 주지 않는 것도 의도다(존재 여부를 흘리지 않는다)."""
        err_type, place = await DB_SESSION_MNG.execute_lambda(
            places.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.place_crud.get_place(s, uuid.UUID(user_info.user_id), uuid.UUID(place_id)),
        )
        if err_type != ErrorType.SUCCESS:
            return ErrorType.PLACE_NOT_FOUND, None
        return ErrorType.SUCCESS, place

    async def _mark_content_updated(self, place_id: str, ts):
        """★ 사이트에 나가는 내용이 바뀌었다 — 이 사업장만 다시 빌드하면 된다는 표시.
        FAQ 도 페이지에 렌더되고 FAQPage JSON-LD 로 나가므로 fact 와 같은 취급이다.
        부가 효과라 실패해도 본 흐름을 막지 않는다(다음 변경 때 다시 찍힌다)."""
        err = await DB_SESSION_MNG.execute_lambda_run([places.DBType()], [lambda s: self._touch(s, place_id, ts)])
        if err != ErrorType.SUCCESS:
            LOG.e_no_callstack(f"[faq] content_updated_at 갱신 실패 place={place_id}")

    async def _touch(self, s, place_id: str, ts):
        from sqlalchemy import update

        query = update(places).where(places.place_id == uuid.UUID(place_id)).values(content_updated_at=ts, updated_at=ts)
        return await DB_SESSION_MNG.add(s, query)

    async def _reload(self, res: Res_Faq, pid, faq_id) -> Res_Faq:
        _e, row = await DB_SESSION_MNG.execute_lambda(
            place_faqs.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.crud.get_faq(s, pid, faq_id),
        )
        res.faq = FaqData.model_validate(row) if row is not None else None
        return res

    # ---- 조회 ----
    async def list_faqs(self, user_info: UserInfo, place_id: str, publishable_only: bool = False) -> Res_FaqList:
        """publishable_only=True → ★ 사이트에 나갈 수 있는 것만(빌드가 보는 것과 같은 집합).
        기본(False)은 승인 대기까지 함께 준다 — 관리 화면이 승인할 대상을 봐야 하기 때문이다."""
        res = Res_FaqList()
        err_type, _place = await self._load_place(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        pid = uuid.UUID(place_id)
        list_err, rows = await DB_SESSION_MNG.execute_lambda(
            place_faqs.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.crud.list_faqs(s, pid, publishable_only),
        )
        if list_err != ErrorType.SUCCESS:
            res.result.SetResult(list_err)
            return res

        res.faqs = [FaqData.model_validate(r) for r in rows]
        res.publishable = sum(1 for r in rows if FactStatus(r.status) in PUBLISHABLE_FACT_STATUSES)
        res.pending_review = sum(1 for r in rows if FactStatus(r.status) in CANDIDATE_FACT_STATUSES)
        return res

    # ---- 직접 추가 ----
    async def create_faq(self, user_info: UserInfo, place_id: str, req: Req_CreateFaq) -> Res_Faq:
        """사장님이 직접 쓴 FAQ 를 넣는다 — fact 의 owner 입력과 같은 철학으로 바로 노출값(VERIFIED)이 된다.

        근거 fact(source_fact_ids)를 요구하지 않는 것도 그래서다. 근거를 요구하는 이유는
        LLM 이 지어냈는지 확인하기 위해서인데, 여기서는 사람 본인이 출처다."""
        res = Res_Faq()
        err_type, _place = await self._load_place(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        question = (req.question or "").strip()
        answer = (req.answer or "").strip()
        if not question or not answer:
            # 빈 문답은 사이트에 렌더도 안 되고 고유 콘텐츠로도 안 세진다 — 애초에 받지 않는다.
            res.result.SetResult(ErrorType.INVALID_REQUEST_DATA)
            return res

        pid = uuid.UUID(place_id)
        sort_order = req.sort_order
        if sort_order is None:
            # 순서를 안 주면 맨 뒤에 붙인다 — 기존 FAQ 사이에 끼어들어 순서를 흔들지 않게.
            _e, rows = await DB_SESSION_MNG.execute_lambda(
                place_faqs.DBType(),
                DBWRType.DB_READ.value,
                lambda s: self.crud.list_faqs(s, pid, False),
            )
            sort_order = max((r.sort_order for r in rows), default=-1) + 1

        row = place_faqs(
            place_id=pid,
            question=question,
            answer=answer,
            generated_by=SourceType.OWNER.value,
            status=FactStatus.VERIFIED.value,
            sort_order=sort_order,
        )
        run_err = await DB_SESSION_MNG.execute_lambda_run([place_faqs.DBType()], [lambda s: self.crud.add_faq(s, row)])
        if run_err != ErrorType.SUCCESS:
            res.result.SetResult(run_err)
            return res

        # 바로 노출 대상이 늘었다 → 재빌드 대상 표시.
        await self._mark_content_updated(place_id, GTime.UTC())
        return await self._reload(res, pid, row.faq_id)

    # ---- 검증 상태 전이(승인 · 정정 · 반려) ----
    async def transition(self, user_info: UserInfo, place_id: str, faq_id: str, req: Req_TransitionFaq) -> Res_Faq:
        """생성된 FAQ 를 승인하거나, 문구를 고쳐 승인하거나, 반려한다.

        고쳐서 승인하면 CORRECTED — ★ 잠긴 상태다. 이후 재생성(expire_generated)은 미확인 FAQ 만
        건드리므로 사장님이 고친 문구는 살아남는다. generated_by 도 OWNER 로 바꿔 둔다:
        문장의 책임 주체가 LLM 에서 사람으로 넘어왔다는 기록이다."""
        res = Res_Faq()
        err_type, _place = await self._load_place(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        pid = uuid.UUID(place_id)
        fid = uuid.UUID(faq_id)
        get_err, faq = await DB_SESSION_MNG.execute_lambda(
            place_faqs.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.crud.get_faq(s, pid, fid),
        )
        if get_err != ErrorType.SUCCESS:
            res.result.SetResult(ErrorType.FAQ_NOT_FOUND)
            return res

        current = FactStatus(faq.status)
        target = req.status
        if target not in FACT_STATUS_TRANSITIONS.get(current, set()):
            res.result.SetResult(ErrorType.FACT_INVALID_TRANSITION)
            return res

        data: dict = {}
        if target == FactStatus.CORRECTED:
            question = (req.question or "").strip()
            answer = (req.answer or "").strip()
            if not question and not answer:
                # '정정'인데 고친 문구가 없으면 그냥 승인(VERIFIED)이어야 한다 — 잠금만 얻는 길을 막는다.
                res.result.SetResult(ErrorType.INVALID_REQUEST_DATA)
                return res
            if question:
                data["question"] = question
            if answer:
                data["answer"] = answer
            data["generated_by"] = SourceType.OWNER.value

        now = GTime.UTC()
        run_err = await DB_SESSION_MNG.execute_lambda_run(
            [place_faqs.DBType()],
            [lambda s: self._transition_ok(s, fid, current, target, data)],
        )
        if run_err != ErrorType.SUCCESS:
            # 적용행수 0 = 그 사이 다른 사람이 상태를 바꿨다 — 요청한 전이는 더 이상 유효하지 않다.
            res.result.SetResult(ErrorType.FACT_INVALID_TRANSITION if run_err == ErrorType.DB_EMPTY_DATA else run_err)
            return res

        # 노출 집합이 바뀌는 전이(승격 / 정정 / 노출값 내림)면 재빌드 대상으로 표시한다.
        if target in PUBLISHABLE_FACT_STATUSES or current in PUBLISHABLE_FACT_STATUSES:
            await self._mark_content_updated(place_id, now)
        return await self._reload(res, pid, fid)

    async def _transition_ok(self, s, fid, current: FactStatus, target: FactStatus, data: dict):
        """출발 상태를 WHERE 에 걸어 조건부로만 바꾼다(동시 처리 가드).
        execute_lambda_run 은 ErrorType 만 받으므로 적용행수 0 을 DB_EMPTY_DATA 로 바꿔 트랜잭션을 중단시킨다."""
        err_type, rowcount = await self.crud.transition(s, fid, (current.value,), target.value, data)
        if err_type != ErrorType.SUCCESS:
            return err_type
        return ErrorType.SUCCESS if rowcount else ErrorType.DB_EMPTY_DATA