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 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 # 공개 검색이 한 번에 가져오는 후보 수. 카카오 키워드 검색은 무료 한도를 넘기면 건당 과금이라 # 확정 경로(15건)보다 좁게 잡는다 — 랜딩에서 사람이 훑는 목록은 다섯이면 충분하다. _PUBLIC_SEARCH_SIZE = 5 # IP 당 분당 허용 횟수. 사람이 상호명을 고쳐 가며 치는 속도를 넘지 않게 잡았다. _PUBLIC_SEARCH_PER_MIN = 20 # 도로명주소 → 지역 캐시 키. 외부 장소 DB 는 행정구역 코드를 주지 않으므로 여기서 만든다. from services.external.naver import region_key from services.job_service import enqueue_job # 겨냥 조회를 몇 건까지 할지. 후보 전부(5건)를 부르면 통합검색이 429 를 준다. _TARGETED_LOOKUP_LIMIT = 2 # 공개 검색 캐시 수명. 가게 정보가 이 안에 바뀔 일은 없다. _PUBLIC_SEARCH_CACHE_SEC = 600 class PlaceService: """사업장 등록·조회·동일 업소 검증. ★ 이 서비스의 핵심 규칙: `verified_at` 이 NULL 인 사업장은 수집이 열리지 않는다. 카카오 로컬로 동일 업소임을 확인하지 않으면 남의 가게 정보가 섞인다. """ 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건. 없으면 PLACE_NOT_FOUND(남의 회사 것도 '없음'으로 응답).""" 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( # ★ 주인은 **토큰이 정한다.** 예전엔 요청 body 의 owner_user_id 를 그대로 넣었는데, # 그 값은 아무도 안 보내서 92건 전부 NULL 이었고 스코프는 회사가 대신 하고 있었다. # 회사를 걷어내면서 이 컬럼이 스코프 키가 됐다 — body 로 남의 계정을 적을 수 있으면 # 만들자마자 남의 목록에 들어간다. 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 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: """이 사업장에 **사장님의 것이 쌓였나.** 중복을 접어도 되는지의 판정이다. ★ 무엇을 세나: fact · 객실 · 사진 · 사이트. 채널은 세지 않는다 — 채널은 검증 과정에서 자동으로 붙는 것이라 "사장님이 쌓은 것" 이 아니다. 이걸 세면 방금 만든 빈 행도 비어 있지 않다고 판정돼 중복이 그대로 남는다. ★ 하나라도 있으면 접지 않는다. 지우는 쪽이 틀렸을 때의 비용(사장님이 넣은 값이 사라진다)이 남기는 쪽이 틀렸을 때의 비용(목록에 하나 더 보인다)보다 훨씬 크다. ★ 세지 못하면 **비어 있지 않다고 본다** — 모르면 지우지 않는다. """ 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 를 스칼라로 풀어서 준다(행 튜플이 아니다). # `rows[0][0]` 으로 읽으면 TypeError 로 검증 API 가 통째로 500 이 된다(실측). 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 을 수집 채널로 등록한다. ★ 한 번에 세 가지를 끝낸다: 신원 확정(verified_at) · 채널 등록 · 확정. 쪼개 놓으면 사장님이 같은 판단을 세 번 하게 된다 — 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 # ── 중복 사업장 합치기 ──────────────────────────────────────────────── # ★ 왜 여기인가 # 위저드는 **신원을 알기 전에** 사업장을 먼저 만든다(`ensureServerPlace`) — 이름만 # 아는 빈 행이다. 그리고 이 함수에서 비로소 "이 가게가 누구인지"(네이버 place id)를 # 알게 된다. 그 순간이 "이미 갖고 있는 그 가게인가" 를 물을 수 있는 첫 지점이다. # 여기서 안 묻고 지나가면 위저드를 다시 시작할 때마다 같은 가게가 하나씩 늘어난다 — # 실측(2026-09-10): 로컬 DB 에 '스테이,머뭄' 이 8개였고 그중 7개가 fact 2건짜리 # 빈 행이었다. 사장님은 목록에서 어느 것이 자기 사이트인지 알 수 없다. # # ★ 정본은 **먼저 만든 쪽**이다(`find_by_external` 이 오래된 순으로 준다). # 나중 것을 정본으로 삼으면 앞서 쌓인 fact·사진·발행 이력이 통째로 버려진다. # # ★ 지금 행은 **비어 있을 때만** 지운다. 사장님이 이 행에 뭔가를 쌓았다면(fact·객실· # 사진·발행) 그건 합치기가 아니라 병합이고, 그건 사람이 판단할 일이다 — # 그때는 둘 다 남기고 정본만 돌려준다. 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), ), ) canonical_id = place_id if err_dup == ErrorType.SUCCESS: existing = next((r for r in (dup_rows or []) if str(r.place_id) != str(place_id)), None) if existing is not None: 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, # 네이버 상세의 분류("펜션"·"카페,디저트"). 주변 맛집 경쟁업소 제외의 폴백 근거(실측 2026-09-08: 있음). 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(카카오/네이버) 조회 결과를 박제해 동일 업소를 확정한다. ★ 이걸 통과해야 수집이 열린다(verified_at). 식별 근거가 하나도 없으면 확정하지 않는다 — 외부 고유 id(카카오) 또는 도로명주소(네이버) 중 하나는 있어야 '이 가게가 그 가게'라고 말할 수 있다.""" 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, # ★ 지역 코드는 **서버가 유도한다**. 외부 장소 DB(카카오·네이버)는 행정구역 코드를 # 주지 않으므로 후보에도 없고, 그래서 프론트가 보낼 수가 없다 — 클라이언트가 # 못 채우는 값을 클라이언트에 맡겨 두면 영원히 NULL 로 남는다(실측: 모든 사업장). # # region_code 가 비면 지역 정보 캐시를 찾을 키가 없어서 # 날씨·축제·주변 관광지가 통째로 빈다(local_contents 의 키가 이 값이다). # 발행본에는 날씨 섹션이 아무것도 그리지 않고, 하이드레이션 뒤 실시간 조회도 # 막힌다(use-live-weather 가 regionCode 없이는 fetch 하지 않는다). # # ★ 지어내지 않는다. 도로명주소에서 '시도 + 시군구' 를 뽑는 것뿐이고, # 주소가 없거나 형식이 다르면 None 이다(그때는 지역 정보가 비는 게 맞다). # 요청이 값을 실어 보냈으면 그쪽이 이긴다. "region_code": req.region_code or region_key(road_address), "verified_at": now, "verified_by": uuid.UUID(user_info.user_id), } # ★ 값이 왔을 때만 덮는다 — 네이버 URL 재검증이 분류를 못 읽었다고 카카오가 준 값을 지우면 안 된다. 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 가 준 업체 홈페이지를 공식 홈페이지 채널로 등록해 둔다. # 실측상 Perplexity 는 이 채널을 잘 못 찾는다 — 검증 단계에서 건지는 게 확실하다. # 등록만 하고 확정하지는 않는다(확정은 수집 잡이 어댑터 유무를 보고 판단). 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 등록. Perplexity 가 발견한 것도, 사장님이 직접 붙여넣은 것도 여기로 들어온다. ★ 확정(confirmed_at)은 별도 액션이다 — 등록만으로 크롤링 대상이 되지 않는다.""" 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: """수집 파이프라인을 큐에 넣고 즉시 응답한다. 한 건에 몇 분(Perplexity 10~30s + 크롤링 + Vision 사진 배치)이라 동기로 처리할 수 없다. 클라이언트는 돌려받은 job_id 로 GET /v1/job/{job_id} 를 폴링한다. ★ 진입 게이트는 하나 — 동일 업소 검증(verified_at). 검증 없이 긁으면 남의 가게가 섞인다. 채널 URL 발견(Perplexity)과 확정은 잡 안에서 순서대로 일어난다: Perplexity URL 발견 → 확정 → 확정된 URL 만 크롤링 → fact/사진 후보 적재 """ 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, # 명시적으로 고른 링크가 있을 때만 대상을 제한한다. 기본 요청에서 현재 # 확정 링크를 복사하면, 잡의 discover 단계가 새로 확정한 네이버 링크가 # wanted 필터에서 빠져 사진·정보 수집이 0건으로 끝난다. "link_ids": [str(x) for x in req.link_ids], "force": req.force, # 유료 검색은 요청자가 선택한 한 회차에만 실행한다. 이후 확정 링크 크롤링이나 # 재수집이 이 값을 암묵적으로 물려받으면 같은 URL 을 찾는 데 계속 과금된다. "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 사진 분석을 큐에 넣고 즉시 응답한다. 수집이 사진을 저장하면 자동으로 걸리지만, 사장님이 사진을 직접 올린 뒤 다시 돌리거나 force 로 재분석할 때 이 엔드포인트를 쓴다.""" 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 은 수집 어댑터가 페이지 캡션으로 채운다. # label 로 세면 캡션 있는 사진이 전부 '분석됨'으로 빠져 pending 0 이 된다 # (crud/media_crud.list_media 의 같은 주석 참고). 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 를 찾아 그대로 돌려준다. **인증도, 사업장 행도 없다.** ★ find_candidates 와 무엇이 다른가: 저쪽은 이미 만들어진 사업장의 신원을 확정하는 경로라 place_id 와 로그인이 필요하다. 여기는 아직 아무것도 만들지 않은 사람이 "내 가게가 있나" 를 보는 경로다 — 만들어 보기도 전에 로그인을 요구하지 않기로 한 결정(로그인 관문은 에디터 진입 하나)이 API 까지 내려온 것이다. ★ pick_match 를 돌리지 않는다. 자동 판정은 확정 경로에서만 의미가 있고, 여기서는 사람이 목록에서 고르는 게 전부다. ★ DB 를 읽지도 쓰지도 않는다. 나가는 값은 외부 장소 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 이면 새로고침만으로 요금이 나간다. # 프로세스 메모리 기반이라 완전하지 않다(common/utils/rate_limit.py 주석). 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 # ★ 여기서 네이버 플레이스 주소까지 찾아 실어 보낸다. # 전에는 확정 경로(verify/candidates)에서만 찾았는데, 그건 로그인 뒤라 사장님이 # 후보를 고를 때는 "이 가게의 플레이스가 있는지" 를 알 수 없었다. 화면은 그걸 # "자동으로 못 찾았다" 로 읽고 지도 주소를 물었다 — 찾을 수 있는데도. # 실패는 조용히 넘긴다(None). 유료 API 가 아니라 공개 페이지 조회다. 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: # ★ 빈 결과는 캐시하지 않는다. 네이버가 잠깐 막아서 0건이 나온 것을 굳히면 # 사장님은 TTL 이 끝날 때까지 아무것도 못 한다. 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 에서 이 상호명의 후보를 찾아 그대로 내려준다. ★ 서버가 자동으로 확정하지 않는다. outcome 이 MATCHED 여도 후보 전체를 돌려줘 UI 가 사람에게 보여주고 고르게 한다 — 남의 가게를 붙이는 게 이 서비스에서 제일 비싼 실수다. 자동 판정은 'UI 가 한 번만 물어봐도 되는가'(auto_selectable)를 알려주는 힌트일 뿐이다. """ 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 # ★ 검색어와 판정용 상호명은 다른 값이다. # 화면은 '상호명 + 위치'를 합쳐 query 로 보낸다(후보를 좁히려고). 그런데 판정 # (pick_match)은 후보 상호명과 **정확일치**를 보므로, 지역이 붙은 문자열을 그대로 # 넘기면 정확일치가 영영 성립하지 않는다 — 실측(2026-08-28) 10건 전부 # AMBIGUOUS(name_no_exact) 였고, 후보가 1건뿐인 경우까지 그랬다. # 상호명은 places.name 이 들고 있다(위저드가 검색 직전에 상호만 PATCH 한다). 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를 주지 않는다. 후보를 보여주기 전에 모바일 통합검색에서 # 상호가 정확히 일치하는 ID를 한 번 찾아, 화면이 "지역 후보 발견"과 "플레이스 발견"을 # 구분할 수 있게 한다. 실패는 정상적인 fallback이므로 후보 조회 자체는 실패시키지 않는다. from services.external import naver_place_lookup try: # ★ 여기는 **넓은 검색어**를 쓴다. 통합검색은 한 번만 부르고(5번 부르면 429) 그 # 한 페이지 안에서 후보들의 id 를 찾는 구조라, 지역이 붙어 결과가 그 동네로 # 좁혀질수록 찾을 확률이 올라간다. 판정(pick_match)과는 요구가 정반대다. 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 = {} # ★ 넓은 검색어 한 페이지에서 못 찾은 후보는 **그 후보만 겨냥해** 한 번 더 찾는다. # 넓은 검색은 결과가 다른 지점으로 채워져 이름이 아예 안 실릴 때가 있다 — # 실측(2026-09-03 '스타벅스 판교역점'): 페이지에 id 6개가 있는데 지역검색이 준 # 5개 지점명은 하나도 그 근처에 없었다. 상호+지역으로 겨냥하면 4/4 로 찾는다. # 호출은 상위 후보 몇 건으로 끊는다 — 5건을 다 부르면 429 를 받는다. 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 생성을 큐에 넣는다. ★ 근거로 쓸 확인된 fact 가 없으면 잡을 만들지 않는다 — 근거 없이 문장을 쓰면 그게 환각이고, 유료 호출만 낭비된다.""" from common.database.model.models import place_facts as facts_model 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 if not gemini_text.is_configured(): res.result.SetResult(ErrorType.GENERATOR_NOT_CONFIGURED) 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()) if res.grounded_facts == 0: res.result.SetResult(ErrorType.FAQ_UNGROUNDED) 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