"""군산 공통 맛집 한일옥 등록.""" import argparse import asyncio import json import sys import uuid from datetime import datetime, timezone from pathlib import Path from types import SimpleNamespace sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from sqlalchemy import select from sqlalchemy.dialects.postgresql import insert from sqlalchemy.engine import URL from sqlalchemy.ext.asyncio import create_async_engine from common.database.db_session_manager import DB_SESSION_MNG from common.database.model.models import area_contents, place_area_refs, places from common.enums import LocalContentStatus, LocalContentType, LocalSource from common.utils.geo import haversine_m from config.server_configs import main_db_config as cfg from services.collector.naver_place_adapter import NaverPlaceAdapter from services.external import naver_place_lookup from services.local_content_service import LocalContentService from services.site_payload import _local from services.snapshot import _local_contents, _site_places REGION = '52군산시' NAVER_ID = '11861452' CONTENT_ID = uuid.uuid5(uuid.NAMESPACE_URL, 'web4ai:52군산시:restaurant:11861452') def _as_float(value) -> float | None: try: return float(value) if value is not None else None except (TypeError, ValueError): return None async def _fetch_coordinates() -> tuple[float, float] | None: """한일옥 좌표.""" summary = await NaverPlaceAdapter().fetch_summary(naver_place_lookup.place_url(NAVER_ID)) if not summary: return None lat, lng = _as_float(summary.get('latitude')), _as_float(summary.get('longitude')) if lat is None or lng is None: return None return lat, lng async def main(apply: bool): lat_lng = await _fetch_coordinates() if lat_lng is None: print(json.dumps({'error': '네이버에서 한일옥 좌표를 가져오지 못했습니다.'}, ensure_ascii=False)) return lat, lng = lat_lng engine = create_async_engine(URL.create( 'postgresql+asyncpg', username=cfg.write_id, password=cfg.write_pw, host=cfg.write_host, port=cfg.write_port, database=cfg.name, )) try: async with engine.begin() as conn: rows = (await conn.execute(select( area_contents.local_content_id, area_contents.title, area_contents.body, ).where( area_contents.region_code == REGION, area_contents.kind == 'restaurant', area_contents.source == LocalSource.OFFICIAL_WEB.value, area_contents.deleted.is_(False), ))).mappings().all() if rows and any(r['body'].get('naverPlaceId') != NAVER_ID for r in rows): raise RuntimeError('다른 군산 공통 맛집이 이미 있어 덮어쓰지 않습니다.') content_id = rows[0]['local_content_id'] if rows else CONTENT_ID now = datetime.now(timezone.utc) values = dict( local_content_id=content_id, region_code=REGION, content_type=LocalContentType.RESTAURANT.value, kind='restaurant', # 사용자 확인을 거친 수기 웹 등록. source=LocalSource.OFFICIAL_WEB.value, external_id=NAVER_ID, title='한일옥', body={**(rows[0]['body'] if rows else {}), 'name': '한일옥', 'searchQuery': '군산 한일옥', 'naverPlaceId': NAVER_ID, 'sourceUrl': f'https://m.place.naver.com/restaurant/{NAVER_ID}/home', 'registration': 'owner_confirmed_region_default'}, latitude=lat, longitude=lng, status=LocalContentStatus.PUBLISHED.value, published_at=now, collected_at=now, display_start_at=None, display_end_at=None, expires_at=None, deleted=False, updated_at=now, ) if apply: await conn.execute(insert(area_contents).values(**values).on_conflict_do_update( index_elements=[area_contents.local_content_id], set_={k: v for k, v in values.items() if k != 'local_content_id'}, )) targets = (await conn.execute(select( places.place_id, places.name, places.latitude, places.longitude, ).where( places.deleted.is_(False), places.region_code == REGION, ))).mappings().all() distances: dict[str, int | None] = {} for row in targets: plat, plng = _as_float(row['latitude']), _as_float(row['longitude']) distances[str(row['place_id'])] = ( round(haversine_m(plat, plng, lat, lng)) if plat is not None and plng is not None else None ) if apply: for row in targets: stmt = insert(place_area_refs).values( place_id=row['place_id'], local_content_id=content_id, distance_m=distances[str(row['place_id'])], deleted=False, ).on_conflict_do_update( index_elements=[place_area_refs.place_id, place_area_refs.local_content_id], set_={'distance_m': distances[str(row['place_id'])], 'deleted': False, 'updated_at': now}, ) await conn.execute(stmt) print(json.dumps({ 'applied': apply, 'database': cfg.name, 'region': REGION, 'contentId': str(content_id), 'naverPlaceId': NAVER_ID, 'coordinates': {'latitude': lat, 'longitude': lng}, 'existingPlaces': [ {**{k: v for k, v in r.items() if k not in ('latitude', 'longitude')}, 'distanceMeters': distances[str(r['place_id'])]} for r in targets ], }, default=str, ensure_ascii=False)) if apply: # 업장마다 다른 거리라 사이트 개인화 맵(site_sections)에도 얹어야 캔버스·발행본이 읽는다 (services/local_content_service.py::_write_site_places 규약과 동일). service = LocalContentService() for row in targets: place_id = row['place_id'] places_map = dict(await _site_places(place_id)) prev = places_map.get(str(content_id)) or {} places_map[str(content_id)] = { 'kind': 'restaurant', 'distanceMeters': distances[str(place_id)], 'hidden': bool(prev.get('hidden', False)), } await service._write_site_places(place_id, places_map) # 업장마다 거리를 포함해 정확히 한 번 실렸는지 확인한다. mismatches = [] for row in targets: place = SimpleNamespace( place_id=row['place_id'], region_code=REGION, latitude=row['latitude'], longitude=row['longitude'], ) snapshot = await _local_contents(place) local, _ = _local(snapshot, _as_float(row['latitude']), _as_float(row['longitude'])) matches = [r for r in local['restaurants'] if r['name'] == '한일옥'] expected = distances[str(row['place_id'])] ok = len(matches) == 1 and (expected is None or matches[0].get('distanceMeters') == expected) if not ok: mismatches.append({'place': row['name'], 'matches': matches, 'expectedDistanceMeters': expected}) print(json.dumps({'verified': not mismatches, 'mismatches': mismatches}, ensure_ascii=False)) finally: await engine.dispose() await DB_SESSION_MNG.dispose_all() if __name__ == '__main__': parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--apply', action='store_true') asyncio.run(main(parser.parse_args().apply))