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

import uuid

from fastapi import Depends

from sqlalchemy import func, select

from common.category_schema import CategorySchemaError, get_schema
from common.database.db_session_manager import DB_SESSION_MNG
from common.database.model.models import place_channels, places, place_units
from common.enums import DBWRType, ErrorType, JobStatus, JobType, LinkChannel, PlaceCategory, PlaceStatus, SourceType
from common.logger import LOG
from common.models.gmodel import PageParams, UserInfo
from common.utils.gtime import GTime
from crud.job_crud import JobQueue
from crud.place_crud import IPlaceCRUD, PlaceCRUD
from services import mail_service
from router.v1.place.protocol import (
    Req_VerifyPlaceByUrl,
    LinkData,
    PlaceData,
    UnitData,
    Req_CreateLink,
    Req_CreatePlace,
    Req_CreateUnit,
    Req_StartCollect,
    Req_StartCopy,
    Req_StartVision,
    Req_UpdatePlace,
    Req_VerifyPlace,
    Res_Link,
    Res_LinkList,
    Res_Place,
    Res_PlaceList,
    Res_PlaceSearch,
    Res_VerifyCandidates,
    PlaceCandidate,
    PlaceSearchItem,
    Res_StartCollect,
    Res_StartCopy,
    Res_StartVision,
    Res_Unit,
    Res_UnitList,
)
from router.v1.job.protocol import JobData, Res_Job

# 공개 검색이 한 번에 가져오는 후보 수.
_PUBLIC_SEARCH_SIZE = 5
# IP 당 분당 허용 횟수.
_PUBLIC_SEARCH_PER_MIN = 20
# 도로명주소 → 지역 캐시 키.
from services.external.naver import region_key
from services.job_service import enqueue_job


# 겨냥 조회를 몇 건까지 할지.
_TARGETED_LOOKUP_LIMIT = 2
# 공개 검색 캐시 수명.
_PUBLIC_SEARCH_CACHE_SEC = 600


class PlaceService:
    """사업장 등록·조회·동일 업소 검증."""

    def __init__(self, crud: IPlaceCRUD = Depends(PlaceCRUD), queue: JobQueue = Depends(JobQueue)):
        self.crud = crud
        self.queue = queue

    # 조회
    async def list_places(self, user_info: UserInfo, pg: PageParams, search=None, category=None, status=None) -> Res_PlaceList:
        res = Res_PlaceList(page=pg.page, size=pg.size)
        uid = uuid.UUID(user_info.user_id)
        err_type, rows, total = await DB_SESSION_MNG.execute_lambda(
            places.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.crud.list_places(
                s, uid, search,
                category.value if isinstance(category, PlaceCategory) else category,
                status.value if isinstance(status, PlaceStatus) else status,
                pg.skip, pg.size,
            ),
        )
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res
        res.places = [PlaceData.model_validate(r) for r in rows]
        res.total = total
        return res

    async def get_place(self, user_info: UserInfo, place_id: str) -> Res_Place:
        res = Res_Place()
        err_type, place = await self._load(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res
        res.place = PlaceData.model_validate(place)
        return res

    async def _load(self, user_info: UserInfo, place_id: str):
        """회사 스코프로 사업장 1건."""
        err_type, place = await DB_SESSION_MNG.execute_lambda(
            places.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.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 create_place(self, user_info: UserInfo, req: Req_CreatePlace) -> Res_Place:
        res = Res_Place()
        if not req.name.strip():
            res.result.SetResult(ErrorType.INVALID_REQUEST_DATA)
            return res
        # 업종 스키마가 없는 업종은 받지 않는다 — fact 를 하나도 쓸 수 없다.
        try:
            get_schema(req.category)
        except CategorySchemaError:
            res.result.SetResult(ErrorType.PLACE_INVALID_CATEGORY)
            return res

        place = places(
            owner_user_id=uuid.UUID(user_info.user_id),
            name=req.name.strip(),
            category=req.category.value,
            status=PlaceStatus.DRAFT.value,
        )
        err_type = await DB_SESSION_MNG.execute_lambda_run(
            [places.DBType()],
            [lambda s: self.crud.add_place(s, place)],
        )
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res
        res.place = PlaceData.model_validate(place)
        return res

    async def update_place(self, user_info: UserInfo, place_id: str, req: Req_UpdatePlace) -> Res_Place:
        res = Res_Place()
        data = req.model_dump(exclude_unset=True, exclude_none=True)
        if "status" in data:
            data["status"] = req.status.value
        if "name" in data:
            data["name"] = str(data["name"]).strip()
        if "notify_email" in data:
            data["notify_email"] = str(data["notify_email"]).strip()
            if data["notify_email"] and not mail_service.is_valid_address(data["notify_email"]):
                res.result.SetResult(ErrorType.INVALID_REQUEST_DATA)
                return res
            data["notify_email"] = data["notify_email"] or None

        if data:
            err_type, rowcount = await DB_SESSION_MNG.execute_lambda_claim(
                places.DBType(),
                lambda s: self.crud.update_place(s, uuid.UUID(user_info.user_id), uuid.UUID(place_id), data),
            )
            if err_type != ErrorType.SUCCESS:
                res.result.SetResult(err_type)
                return res
            if rowcount == 0:
                res.result.SetResult(ErrorType.PLACE_NOT_FOUND)
                return res
        return await self.get_place(user_info, place_id)

    async def delete_place(self, user_info: UserInfo, place_id: str) -> Res_Place:
        res = Res_Place()
        err_type, rowcount = await DB_SESSION_MNG.execute_lambda_claim(
            places.DBType(),
            lambda s: self.crud.delete_place(
                s, uuid.UUID(user_info.user_id), uuid.UUID(place_id)
            ),
        )
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
        elif rowcount == 0:
            res.result.SetResult(ErrorType.PLACE_NOT_FOUND)
        return res

    async def get_active_collect(self, user_info: UserInfo, place_id: str) -> Res_Job:
        """재접속한 화면이 진행 중인 수집 잡에 다시 연결할 수 있게 한다."""
        res = Res_Job()
        err_type, _place = await self._load(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res
        row = await self.queue.find_active(f"collect:{place_id}")
        if row:
            res.job = JobData(**row)
        return res

    # 동일 업소 검증
    async def _is_empty(self, place_id: str) -> bool:
        """이 사업장에 **사장님의 것이 쌓였나.** 중복을 접어도 되는지의 판정이다."""
        from common.database.model.models import place_facts, place_photos, sites

        pid = uuid.UUID(place_id)

        def _count(model):
            return (
                select(func.count())
                .select_from(model)
                .where(model.place_id == pid, model.deleted == False)  # noqa: E712
                .scalar_subquery()
            )

        query = select(
            _count(place_facts) + _count(place_units) + _count(place_photos) + _count(sites)
        )
        err, rows = await DB_SESSION_MNG.execute_lambda(
            places.DBType(), DBWRType.DB_READ.value,
            lambda s: DB_SESSION_MNG.execute(s, query),
        )
        if err != ErrorType.SUCCESS or not rows:
            LOG.w(f"[verify_by_url] 중복 판정용 계수 실패 place={place_id} — 접지 않는다")
            return False
        # 이 세션 헬퍼는 한 칸짜리 select 를 스칼라로 풀어서 준다(행 튜플이 아니다).
        row = rows[0]
        total = row if isinstance(row, int) else row[0]
        return int(total) == 0

    async def verify_place_by_url(self, user_info: UserInfo, place_id: str, req: Req_VerifyPlaceByUrl) -> Res_Place:
        """네이버 플레이스 URL → 상호·주소·좌표를 읽어 동일 업소를 확정하고, 그 URL 을 수집 채널로 등록한다."""
        from decimal import Decimal

        from common.enums import ExternalPlaceSource, LinkChannel, SourceType
        from services.collector.naver_place_adapter import NaverPlaceAdapter

        res = Res_Place()
        err_type, place = await self._load(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        url = (req.url or "").strip()
        adapter = NaverPlaceAdapter()
        if not url or not adapter.can_handle(url):
            res.result.SetResult(ErrorType.INVALID_REQUEST_DATA)
            res.msg = "네이버 플레이스 주소가 아닙니다."
            return res

        try:
            naver_id = await adapter._resolve_place_id(url)
            state = await adapter._load_state(naver_id)
        except Exception as ex:
            LOG.w(f"[verify_by_url] 상세를 읽지 못했다: {ex}")
            res.result.SetResult(ErrorType.PLACE_VERIFY_NO_CANDIDATE)
            res.msg = "이 주소에서 가게 정보를 읽지 못했습니다. 주소를 다시 확인해 주세요."
            return res

        base = state.get(f"PlaceDetailBase:{naver_id}") or next(
            (v for k, v in state.items() if k.startswith("PlaceDetailBase")), None
        )
        if not base or not str(base.get("name") or "").strip():
            res.result.SetResult(ErrorType.PLACE_VERIFY_NO_CANDIDATE)
            res.msg = "이 주소에서 상호를 찾지 못했습니다."
            return res

        # ── 중복 사업장 합치기 ────────────────────────────────────────────────
        err_dup, dup_rows = await DB_SESSION_MNG.execute_lambda(
            places.DBType(), DBWRType.DB_READ.value,
            lambda sess: self.crud.find_by_external(
                sess, uuid.UUID(user_info.user_id), ExternalPlaceSource.NAVER.value, str(naver_id),
            ),
        )
        # 끄더라도 **비어 있는 중복 행은 치운다.** 원래 막으려던 누적이 그것이고, 빈 행은 잃을 것이 없다.
        others = []
        if err_dup == ErrorType.SUCCESS:
            others = [r for r in (dup_rows or []) if str(r.place_id) != str(place_id)]

        if others and not req.reuse_existing:
            for row in others:
                if await self._is_empty(str(row.place_id)):
                    await DB_SESSION_MNG.execute_lambda_claim(
                        places.DBType(),
                        lambda sess, rid=row.place_id: self.crud.delete_place(
                            sess, uuid.UUID(user_info.user_id), rid,
                        ),
                    )
                    LOG.i(f"[verify_by_url] 새로 만들기 — 빈 중복 행 {row.place_id} 를 치운다 "
                          f"(naver {naver_id})")
            LOG.i(f"[verify_by_url] 새로 만들기 — 같은 업소가 {len(others)}개 있지만 "
                  f"{place_id} 로 새로 간다 (naver {naver_id})")
            others = []

        canonical_id = place_id
        if others:
            existing = others[0]
            canonical_id = str(existing.place_id)
            if await self._is_empty(place_id):
                await DB_SESSION_MNG.execute_lambda_claim(
                    places.DBType(),
                    lambda sess: self.crud.delete_place(
                        sess, uuid.UUID(user_info.user_id), uuid.UUID(place_id),
                    ),
                )
                LOG.i(f"[verify_by_url] 같은 업소가 이미 있다 — 빈 행 {place_id} 를 접고 "
                      f"{canonical_id} 로 잇는다 (naver {naver_id})")
            else:
                LOG.w(f"[verify_by_url] 같은 업소가 둘이다 — {place_id} 에 쌓인 것이 있어 "
                      f"지우지 않는다. 정본 {canonical_id} 를 돌려준다 (naver {naver_id})")
            place_id = canonical_id
            err_type, place = await self._load(user_info, place_id)
            if err_type != ErrorType.SUCCESS:
                res.result.SetResult(err_type)
                return res

        coord = base.get("coordinate") or {}
        verify_req = Req_VerifyPlace(
            source=ExternalPlaceSource.NAVER,
            external_place_id=str(naver_id),
            road_address=base.get("roadAddress") or None,
            address=base.get("address") or None,
            phone=base.get("phone") or base.get("virtualPhone") or None,
            latitude=Decimal(str(coord.get("y"))) if coord.get("y") else None,
            longitude=Decimal(str(coord.get("x"))) if coord.get("x") else None,
            # 네이버 상세의 분류("펜션"·"카페,디저트").
            category_name=str(base.get("category") or "").strip() or None,
        )
        verified = await self.verify_place(user_info, place_id, verify_req)
        if not verified.result.success:
            return verified

        # 상호도 네이버 표기로 맞춘다 — 사장님이 검색창에 친 이름과 실제 등록 상호가 다를 수 있다.
        official = str(base.get("name")).strip()
        if official and official != place.name:
            await DB_SESSION_MNG.execute_lambda_claim(
                places.DBType(),
                lambda s: self.crud.update_place(
                    s, uuid.UUID(user_info.user_id), uuid.UUID(place_id), {"name": official}
                ),
            )
            if verified.place:
                verified.place.name = official

        # 붙여넣은 URL 을 채널로 등록·확정한다.
        canonical = f"https://m.place.naver.com/place/{naver_id}/home"
        await self.create_link(
            user_info, place_id,
            Req_CreateLink(channel=LinkChannel.NAVER_PLACE, url=canonical,
                           title=f"{official} 네이버 플레이스", discovered_by=SourceType.OWNER),
        )
        await DB_SESSION_MNG.execute_lambda_claim(
            place_channels.DBType(),
            lambda s: self.crud.confirm_link_by_url(
                s, uuid.UUID(place_id), canonical, uuid.UUID(user_info.user_id), GTime.UTC()
            ),
        )
        LOG.i(f"[verify_by_url] '{official}' 확정 + 채널 등록 — naver place {naver_id}")
        return verified

    async def verify_place(self, user_info: UserInfo, place_id: str, req: Req_VerifyPlace) -> Res_Place:
        """외부 장소 DB(카카오/네이버) 조회 결과를 박제해 동일 업소를 확정한다."""
        res = Res_Place()
        external_id = req.external_place_id.strip()
        road_address = (req.road_address or "").strip()
        if not external_id and not road_address:
            res.result.SetResult(ErrorType.PLACE_VERIFY_NO_CANDIDATE)
            return res

        err_type, place = await self._load(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        uid = uuid.UUID(user_info.user_id)
        now = GTime.UTC()
        data = {
            "external_source": req.source.value,
            "external_place_id": external_id or None,
            "road_address": road_address or None,
            "address": req.address,
            "phone": req.phone,
            "latitude": req.latitude,
            "longitude": req.longitude,
            # 지역 코드는 **서버가 유도한다**.
            "region_code": req.region_code or region_key(road_address),
            "verified_at": now,
            "verified_by": uuid.UUID(user_info.user_id),
        }
        if (req.category_name or "").strip():
            data["external_category"] = req.category_name.strip()[:200]
        err_type, rowcount = await DB_SESSION_MNG.execute_lambda_claim(
            places.DBType(),
            lambda s: self.crud.update_place(s, uid, uuid.UUID(place_id), data),
        )
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res
        if rowcount == 0:
            res.result.SetResult(ErrorType.PLACE_NOT_FOUND)
            return res

        # 외부 장소 DB 가 준 업체 홈페이지를 공식 홈페이지 채널로 등록해 둔다.
        if (req.place_url or "").strip():
            link = place_channels(
                place_id=uuid.UUID(place_id),
                channel=LinkChannel.OFFICIAL_SITE.value,
                url=req.place_url.strip(),
                title=place.name,
                discovered_by=SourceType.API.value,
                discovered_at=now,
            )
            await DB_SESSION_MNG.execute_lambda_run(
                [place_channels.DBType()],
                [lambda s: self.crud.add_link(s, link)],
            )
        return await self.get_place(user_info, place_id)

    # 하위 단위(객실·메뉴·프로그램)
    async def list_units(self, user_info: UserInfo, place_id: str) -> Res_UnitList:
        res = Res_UnitList()
        err_type, _place = await self._load(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res
        list_err, rows = await DB_SESSION_MNG.execute_lambda(
            place_units.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.crud.list_units(s, uuid.UUID(place_id)),
        )
        if list_err != ErrorType.SUCCESS:
            res.result.SetResult(list_err)
            return res
        res.units = [UnitData.model_validate(r) for r in rows]
        return res

    async def create_unit(self, user_info: UserInfo, place_id: str, req: Req_CreateUnit) -> Res_Unit:
        res = Res_Unit()
        if not req.name.strip():
            res.result.SetResult(ErrorType.INVALID_REQUEST_DATA)
            return res
        err_type, _place = await self._load(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        unit = place_units(place_id=uuid.UUID(place_id), name=req.name.strip(), sort_order=req.sort_order)
        run_err = await DB_SESSION_MNG.execute_lambda_run(
            [place_units.DBType()],
            [lambda s: self.crud.add_unit(s, unit)],
        )
        if run_err != ErrorType.SUCCESS:
            res.result.SetResult(run_err)
            return res
        res.unit = UnitData.model_validate(unit)
        return res

    # 채널 링크
    async def list_links(self, user_info: UserInfo, place_id: str, confirmed_only: bool = False) -> Res_LinkList:
        res = Res_LinkList()
        err_type, _place = await self._load(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res
        list_err, rows = await DB_SESSION_MNG.execute_lambda(
            place_channels.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.crud.list_links(s, uuid.UUID(place_id), confirmed_only),
        )
        if list_err != ErrorType.SUCCESS:
            res.result.SetResult(list_err)
            return res
        res.links = [LinkData.model_validate(r) for r in rows]
        res.confirmed = sum(1 for r in rows if r.confirmed_at is not None)
        return res

    async def create_link(self, user_info: UserInfo, place_id: str, req: Req_CreateLink) -> Res_Link:
        """채널 URL 등록."""
        res = Res_Link()
        if not req.url.strip():
            res.result.SetResult(ErrorType.INVALID_REQUEST_DATA)
            return res
        err_type, _place = await self._load(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        link = place_channels(
            place_id=uuid.UUID(place_id),
            channel=req.channel.value,
            url=req.url.strip(),
            title=req.title,
            discovered_by=req.discovered_by.value,
            discovered_at=GTime.UTC(),
        )
        run_err = await DB_SESSION_MNG.execute_lambda_run(
            [place_channels.DBType()],
            [lambda s: self.crud.add_link(s, link)],
        )
        if run_err != ErrorType.SUCCESS:
            res.result.SetResult(run_err)
            return res
        res.link = LinkData.model_validate(link)
        return res

    async def confirm_link(self, user_info: UserInfo, place_id: str, link_id: str) -> Res_Link:
        """동일 업소로 확인된 URL 만 크롤링 대상이 된다."""
        res = Res_Link()
        err_type, place = await self._load(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res
        if place.verified_at is None:
            res.result.SetResult(ErrorType.PLACE_NOT_VERIFIED)
            return res

        run_err, rowcount = await DB_SESSION_MNG.execute_lambda_claim(
            place_channels.DBType(),
            lambda s: self.crud.confirm_link(
                s, uuid.UUID(place_id), uuid.UUID(link_id), uuid.UUID(user_info.user_id), GTime.UTC()
            ),
        )
        if run_err != ErrorType.SUCCESS:
            res.result.SetResult(run_err)
            return res
        if rowcount == 0:
            # 없거나 이미 확정됨 — 어느 쪽이든 이 호출로 바뀐 건 없다.
            res.result.SetResult(ErrorType.LINK_NOT_FOUND)
            return res

        links = await self.list_links(user_info, place_id)
        res.link = next((x for x in links.links if str(x.link_id) == str(link_id)), None)
        return res

    # 수집 시작
    async def start_collect(self, user_info: UserInfo, place_id: str, req: Req_StartCollect) -> Res_StartCollect:
        """수집 파이프라인을 큐에 넣고 즉시 응답한다."""
        res = Res_StartCollect()
        err_type, place = await self._load(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        if place.verified_at is None:
            res.result.SetResult(ErrorType.PLACE_NOT_VERIFIED)
            return res

        # 채널 URL 발견(Perplexity)은 **잡의 첫 단계**다 — 링크가 하나도 없어도 수집을 시작할 수 있다.
        link_err, links = await DB_SESSION_MNG.execute_lambda(
            place_channels.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.crud.list_links(s, uuid.UUID(place_id), True),
        )
        if link_err != ErrorType.SUCCESS:
            res.result.SetResult(link_err)
            return res

        wanted = {str(x) for x in req.link_ids}
        targets = [x for x in links if not wanted or str(x.link_id) in wanted]
        res.confirmed_links = len(targets)

        # link_ids 를 콕 집었는데 그중 확정된 게 없으면 시작할 이유가 없다(재크롤 요청 경로).
        if wanted and not targets:
            res.result.SetResult(ErrorType.LINK_NOT_CONFIRMED)
            return res

        payload = {
            "place_id": place_id,
            "owner_user_id": user_info.user_id,
            "category": place.category,
            # 명시적으로 고른 링크가 있을 때만 대상을 제한한다.
            "link_ids": [str(x) for x in req.link_ids],
            "force": req.force,
            # 유료 검색은 요청자가 선택한 한 회차에만 실행한다.
            "discover_channels": req.discover_channels,
            "requested_by": user_info.user_id,
        }
        # 사업장당 활성 수집 잡 1건 — 버튼을 두 번 눌러도 두 번 돌지 않는다.
        job_id, created = await enqueue_job(
            self.queue, JobType.COLLECT, payload, dedupe_key=f"collect:{place_id}"
        )
        if job_id is None:
            res.result.SetResult(ErrorType.COLLECT_ALREADY_RUNNING)
            return res

        res.job_id = uuid.UUID(job_id)
        res.status = JobStatus.PENDING
        res.created = created

        # 수집 진행 중임을 사업장 상태에 반영(관리 화면 배지).
        if created and place.status == PlaceStatus.DRAFT.value:
            await DB_SESSION_MNG.execute_lambda_claim(
                places.DBType(),
                lambda s: self.crud.update_place(
                    s, uuid.UUID(user_info.user_id), uuid.UUID(place_id),
                    {"status": PlaceStatus.COLLECTING.value},
                ),
            )
        return res

    # 사진 분석 시작
    async def start_vision(self, user_info: UserInfo, place_id: str, req: Req_StartVision) -> Res_StartVision:
        """Gemini Vision 사진 분석을 큐에 넣고 즉시 응답한다."""
        from sqlalchemy import func, select

        from common.database.model.models import place_photos
        from services.external import gemini

        res = Res_StartVision()
        err_type, _place = await self._load(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        if not gemini.is_configured():
            res.result.SetResult(ErrorType.GENERATOR_NOT_CONFIGURED)
            return res

        conds = [place_photos.place_id == uuid.UUID(place_id), place_photos.deleted == False]  # noqa: E712
        if not req.force:
            # 미분석 기준은 alt_text 다 — label 은 수집 어댑터가 페이지 캡션으로 채운다.
            from sqlalchemy import func as sa_func
            from sqlalchemy import or_ as sa_or
            conds.append(sa_or(place_photos.alt_text.is_(None), sa_func.btrim(place_photos.alt_text) == ""))
        cnt_err, rows = await DB_SESSION_MNG.execute_lambda(
            place_photos.DBType(),
            DBWRType.DB_READ.value,
            lambda s: DB_SESSION_MNG.execute(s, select(func.count()).select_from(place_photos).where(*conds)),
        )
        res.pending_media = int(rows[0] or 0) if cnt_err == ErrorType.SUCCESS and rows else 0
        if res.pending_media == 0:
            res.result.SetResult(ErrorType.MEDIA_NOT_FOUND)
            return res

        job_id, created = await enqueue_job(
            self.queue, JobType.VISION,
            {"place_id": place_id, "owner_user_id": user_info.user_id, "force": req.force},
            dedupe_key=f"vision:{place_id}",
        )
        if job_id is None:
            res.result.SetResult(ErrorType.COLLECT_ALREADY_RUNNING)
            return res
        res.job_id = uuid.UUID(job_id)
        res.status = JobStatus.PENDING
        res.created = created
        return res

    # 동일 업소 후보 조회 (UI 가 사람에게 고르게 한다)
    async def search_places_public(self, query: str, client_ip: str) -> Res_PlaceSearch:
        """상호명으로 외부 장소 DB 를 찾아 그대로 돌려준다."""
        from common.enums import ExternalPlaceSource
        from common.utils import rate_limit, ttl_cache
        from services.external import kakao as kakao_client
        from services.external import naver as naver_client
        from services.external import naver_place_lookup
        from services.place_category import guess_category

        res = Res_PlaceSearch()
        q = (query or "").strip()
        if len(q) < 2:
            res.result.SetResult(ErrorType.INVALID_REQUEST_DATA)
            return res

        # 같은 검색어는 캐시로 받는다.
        cache_key = f"place-search:{q}"
        cached = ttl_cache.get(cache_key)
        if cached is not None:
            res.source, res.items = cached
            return res

        # 인증이 없는데 유료 외부 API 를 부른다 — 방어가 0 이면 새로고침만으로 요금이 나간다.
        if not rate_limit.allow(f"place-search:{client_ip}", _PUBLIC_SEARCH_PER_MIN, 60.0):
            res.result.SetResult(ErrorType.HTTP_TO_MANY_REQUEST)
            return res

        kakao = kakao_client.KakaoLocalClient()
        try:
            if kakao.enabled:
                res.source = ExternalPlaceSource.KAKAO
                rows = await kakao.search_keyword(q, size=_PUBLIC_SEARCH_SIZE)
                await kakao.aclose()
            else:
                client = naver_client.NaverLocalClient()
                if not client.enabled:
                    res.result.SetResult(ErrorType.LOCAL_NOT_CONFIGURED)
                    return res
                res.source = ExternalPlaceSource.NAVER
                rows = await client.search_local(q, display=_PUBLIC_SEARCH_SIZE)
                await client.aclose()
        except (kakao_client.KakaoNotConfigured, naver_client.NaverNotConfigured):
            res.result.SetResult(ErrorType.LOCAL_NOT_CONFIGURED)
            return res
        except (kakao_client.KakaoRequestFailed, naver_client.NaverRequestFailed) as ex:
            LOG.w(f"[search] 공개 검색 실패 q={q!r}: {type(ex).__name__}: {ex}")
            res.result.SetResult(ErrorType.LOCAL_FETCH_FAILED)
            return res

        # 여기서 네이버 플레이스 주소까지 찾아 실어 보낸다.
        naver_ids: dict[str, str] = {}
        try:
            naver_ids = await naver_place_lookup.find_place_ids(q, [r.name for r in rows])
            for row in rows[:_TARGETED_LOOKUP_LIMIT]:
                if row.name in naver_ids:
                    continue
                found = await naver_place_lookup.find_place_id(row.name, row.road_address or row.address)
                if found:
                    naver_ids[row.name] = found
        except Exception as ex:  # noqa: BLE001 — 지도 주소 붙여넣기로 이어진다
            LOG.w(f"[search] 플레이스 자동 발견 실패(후보는 유지): {type(ex).__name__}: {ex}")

        res.items = [
            PlaceSearchItem(
                name=row.name,
                road_address=row.road_address or row.address,
                category_name=row.category_name,
                # 네이버는 그룹코드를 주지 않는다 — 그때는 분류 문자열만으로 추정한다.
                category=guess_category(row.category_name, getattr(row, "category_group_code", None)),
                naver_place_url=(
                    naver_place_lookup.place_url(naver_ids[row.name]) if row.name in naver_ids else None
                ),
            )
            for row in rows
        ]
        if not res.items:
            # 빈 결과는 캐시하지 않는다.
            res.result.SetResult(ErrorType.PLACE_VERIFY_NO_CANDIDATE)
            return res

        ttl_cache.put(cache_key, (res.source, res.items), _PUBLIC_SEARCH_CACHE_SEC)
        return res

    async def find_candidates(self, user_info: UserInfo, place_id: str, query: str | None = None) -> Res_VerifyCandidates:
        """외부 장소 DB 에서 이 상호명의 후보를 찾아 그대로 내려준다."""
        from common.enums import ExternalPlaceSource
        from services.external import kakao as kakao_client
        from services.external import naver as naver_client

        res = Res_VerifyCandidates()
        err_type, place = await self._load(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        # 검색어와 판정용 상호명은 다른 값이다.
        search_query = (query or place.name or "").strip()
        name = (place.name or "").strip() or search_query
        if not search_query:
            res.result.SetResult(ErrorType.INVALID_REQUEST_DATA)
            return res

        # 카카오 키가 있으면 카카오(전화번호·고유 id 가 있어 판정이 강하다), 없으면 네이버.
        kakao = kakao_client.KakaoLocalClient()
        try:
            if kakao.enabled:
                res.source = ExternalPlaceSource.KAKAO
                match = await kakao.verify_place(name, search_query=search_query)
                await kakao.aclose()
            else:
                client = naver_client.NaverLocalClient()
                if not client.enabled:
                    res.result.SetResult(ErrorType.LOCAL_NOT_CONFIGURED)
                    return res
                res.source = ExternalPlaceSource.NAVER
                match = await client.verify_place(
                    name, address_hint=place.road_address, search_query=search_query
                )
                await client.aclose()
        except (kakao_client.KakaoNotConfigured, naver_client.NaverNotConfigured):
            res.result.SetResult(ErrorType.LOCAL_NOT_CONFIGURED)
            return res
        except (kakao_client.KakaoRequestFailed, naver_client.NaverRequestFailed) as ex:
            LOG.w(f"[verify] 후보 조회 실패 place={place_id}: {type(ex).__name__}: {ex}")
            res.result.SetResult(ErrorType.LOCAL_FETCH_FAILED)
            return res

        res.outcome = match.outcome.value if hasattr(match.outcome, "value") else str(match.outcome)
        res.reason = match.reason
        res.auto_selectable = bool(match.is_matched)

        # MATCHED 여도 후보를 전부 내려보낸다 — 사람이 다른 걸 고를 수 있어야 한다.
        rows = match.candidates or ([match.place] if match.place else [])
        # 지역검색 API는 네이버 Place ID를 주지 않는다.
        from services.external import naver_place_lookup

        try:
            # 여기는 **넓은 검색어**를 쓴다.
            naver_ids = await naver_place_lookup.find_place_ids(search_query, [c.name for c in rows])
        except Exception as ex:  # noqa: BLE001 — 지도 URL 직접 입력으로 이어진다
            LOG.w(f"[verify] 네이버 플레이스 자동 발견 실패(후보는 유지): {type(ex).__name__}: {ex}")
            naver_ids = {}
        # 넓은 검색어 한 페이지에서 못 찾은 후보는 **그 후보만 겨냥해** 한 번 더 찾는다.
        for c in rows[:_TARGETED_LOOKUP_LIMIT]:
            if c.name in naver_ids:
                continue
            try:
                found = await naver_place_lookup.find_place_id(c.name, c.road_address or c.address)
            except Exception as ex:  # noqa: BLE001 — 지도 URL 직접 입력으로 이어진다
                LOG.w(f"[verify] 겨냥 조회 실패({c.name}): {type(ex).__name__}: {ex}")
                break
            if found:
                naver_ids[c.name] = found

        res.candidates = [
            PlaceCandidate(
                external_place_id=getattr(c, "kakao_place_id", None) or getattr(c, "naver_place_id", None),
                name=c.name,
                road_address=c.road_address,
                address=c.address,
                phone=c.phone,
                latitude=c.latitude,
                longitude=c.longitude,
                category_name=c.category_name,
                place_url=c.place_url,
                naver_place_url=(
                    naver_place_lookup.place_url(naver_ids[c.name]) if c.name in naver_ids else None
                ),
            )
            for c in rows
        ]
        if not res.candidates:
            res.result.SetResult(ErrorType.PLACE_VERIFY_NO_CANDIDATE)
        return res

    # 소개문·FAQ 생성 시작
    async def start_copy(self, user_info: UserInfo, place_id: str, req: Req_StartCopy) -> Res_StartCopy:
        """소개문·FAQ 생성을 큐에 넣는다."""
        from common.database.model.models import place_facts as facts_model
        from common.faq_catalog import find_catalog
        from crud.fact_crud import FactCRUD
        from services.external import gemini_text

        res = Res_StartCopy()
        err_type, place = await self._load(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        # URL 에 잡 ID 를 쓰기 전에 응답이 끊겨도 완료된 작업을 다시 만들지 않는다.
        if req.resume:
            existing = await self.queue.find_latest(f"copy:{place_id}")
            if existing:
                res.job_id = uuid.UUID(existing["job_id"])
                res.status = JobStatus(existing["status"])
                res.created = False
                return res

        crud = FactCRUD()
        f_err, rows = await DB_SESSION_MNG.execute_lambda(
            facts_model.DBType(),
            DBWRType.DB_READ.value,
            lambda s: crud.list_facts(s, uuid.UUID(place_id), None, None, True, True),
        )
        if f_err != ErrorType.SUCCESS:
            res.result.SetResult(f_err)
            return res
        res.grounded_facts = sum(1 for r in rows if r.unit_id is None and (r.value or "").strip())
        # 객실 fact 만 있어도 LLM 이 쓴다(copy_service 의 unit_summaries) — 키 검사는 그 기준으로 한다.
        needs_llm = any((r.value or "").strip() for r in rows)
        if not needs_llm and find_catalog(place.category, place.external_category) is None:
            # 쓸 근거도, 채울 공통 질문도 없다(카페·음식점·호텔).
            res.result.SetResult(ErrorType.FAQ_UNGROUNDED)
            return res
        if needs_llm and not gemini_text.is_configured():
            res.result.SetResult(ErrorType.GENERATOR_NOT_CONFIGURED)
            return res

        job_id, created = await enqueue_job(
            self.queue, JobType.COPY,
            {"place_id": place_id, "owner_user_id": user_info.user_id, "requested_by": user_info.user_id},
            dedupe_key=f"copy:{place_id}",
        )
        if job_id is None:
            res.result.SetResult(ErrorType.COLLECT_ALREADY_RUNNING)
            return res
        res.job_id = uuid.UUID(job_id)
        res.status = JobStatus.PENDING
        res.created = created
        return res