o2o-site-AEO/solution/backend/scripts/demo_pipeline.py

"""파이프라인 데모 — 실 DB(web4ai_db)에 사업장 하나를 만들어 끝까지 돌린다.

    python scripts/demo_pipeline.py     (backend/ 에서 실행)

★ 실 DB 에 데모 회사·계정·사업장을 만든다. 개발 DB 에서만 쓸 것.
★ 네이버 지역검색과 Perplexity 를 **실제로 호출**한다(요금 발생).
"""
import asyncio, json, os, urllib.parse, urllib.request, uuid
os.environ["APP_ENV"] = "local"

from httpx import ASGITransport, AsyncClient
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine

from common.enums import LinkChannel, PlaceCategory, SourceType, UserRole, UserStatus
from config.server_configs import main_db_config, external_api_config
from services.collector import MockAdapter

NAME = "핑크비치펜션"


def naver_lookup(q):
    """실제 네이버 지역검색 — 동일 업소 검증 재료."""
    cid, sec = external_api_config.naver_client_id, external_api_config.naver_client_secret
    if not (cid and sec):
        return []
    url = f"https://openapi.naver.com/v1/search/local.json?query={urllib.parse.quote(q)}&display=5"
    req = urllib.request.Request(url, headers={"X-Naver-Client-Id": cid, "X-Naver-Client-Secret": sec})
    with urllib.request.urlopen(req, timeout=15) as r:
        return json.load(r).get("items", [])


def strip_tags(s):
    import re
    return re.sub(r"<[^>]+>", "", s or "").replace("&amp;", "&")


async def main():
    pw = f":{main_db_config.write_pw}" if main_db_config.write_pw else ""
    dsn = f"postgresql+asyncpg://{main_db_config.write_id}{pw}@{main_db_config.write_host}:{main_db_config.write_port}/{main_db_config.name}"
    engine = create_async_engine(dsn)

    from router.v1.validator.dependencies import GetHashedPW
    uid, login = uuid.uuid4(), f"demo{uuid.uuid4().hex[:6]}"
    async with engine.begin() as c:
        await c.execute(text("INSERT INTO company.users (user_id,id,password,name,status,role,last_accessed_at) "
                             "VALUES (:u,:i,:p,:n,:s,:r,now())"),
                        {"u": uid, "i": login, "p": await GetHashedPW("pw1234"),
                         "n": "데모", "s": UserStatus.ACTIVE.value, "r": UserRole.OWNER.value})
    await engine.dispose()

    from router.router import app
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://demo") as cl:
        tok = (await cl.post("/v1/auth/login", json={"id": login, "password": "pw1234"})).json()["access_token"]
        h = {"Authorization": f"Bearer {tok}"}

        print("\n" + "="*74)
        print(f"  1. 사업장 등록 — 상호명 하나만 넣는다")
        print("="*74)
        pl = (await cl.post("/v1/place", headers=h, json={"name": NAME, "category": PlaceCategory.LODGING.value})).json()["place"]
        pid = pl["place_id"]
        print(f"  place_id   : {pid}")
        print(f"  status     : {pl['status']} (DRAFT) · verified_at: {pl.get('verified_at') or '없음 → 수집 잠김'}")

        print("\n" + "="*74)
        print(f"  2. 네이버 지역검색 — 동일 업소 검증 (실제 API 호출)")
        print("="*74)
        items = naver_lookup(NAME)
        exact = [i for i in items if strip_tags(i["title"]).replace(" ", "") == NAME.replace(" ", "")]
        for i in items:
            mark = "★" if i in exact else " "
            print(f"  {mark} {strip_tags(i['title'])[:28]:30s} {i.get('category',''):14s} {strip_tags(i.get('roadAddress',''))[:38]}")
        print(f"\n  정확 일치 {len(exact)}건 → {'MATCHED (자동 확정)' if len(exact)==1 else 'AMBIGUOUS (사람이 골라야 함)'}")

        pick = exact[0] if len(exact) == 1 else (items[0] if items else None)
        if len(exact) != 1 and pick:
            print(f"  ⚠ 자동 확정 불가 → 사람이 고른 셈 치고 진행: {strip_tags(pick['title'])}")
        if pick:
            it = pick
            road = strip_tags(it["roadAddress"])
            v = (await cl.post(f"/v1/place/{pid}/verify", headers=h, json={
                "external_place_id": f"naver:{abs(hash(it['link']))%10**8}",
                "road_address": road, "address": strip_tags(it.get("address","")),
                "latitude": str(int(it["mapy"])/1e7), "longitude": str(int(it["mapx"])/1e7),
                "region_code": "4283000",
            })).json()["place"]
            print(f"  → 확정: {v['road_address']}")
            print(f"          좌표 {v['latitude']}, {v['longitude']} · verified_at {v['verified_at']}")

        print("\n" + "="*74)
        print(f"  3. 크롤링 대상 링크 (MockAdapter — 실크롤러는 법무 대기)")
        print("="*74)
        murl = MockAdapter.url_for(PlaceCategory.LODGING, pid, channel="naver_place")
        lk = (await cl.post(f"/v1/place/{pid}/link", headers=h, json={
            "channel": LinkChannel.NAVER_PLACE.value, "url": murl, "discovered_by": SourceType.API.value})).json()["link"]
        await cl.post(f"/v1/place/{pid}/link/{lk['link_id']}/confirm", headers=h)
        print(f"  {murl}  → 확정됨")

        print("\n" + "="*74)
        print(f"  4. 수집 시작 → 잡 적재 (즉시 응답)")
        print("="*74)
        st = (await cl.post(f"/v1/place/{pid}/collect", headers=h, json={})).json()
        print(f"  job_id: {st['job_id']} · status: PENDING · 확정링크 {st['confirmed_links']}건")

        print("\n  워커 실행 중 (실제 Perplexity 호출 포함, 10~30초)...")
        from crud.job_crud import JobQueue
        from worker.handlers import build_handler
        from worker.runner import Worker
        await Worker("demo", JobQueue(), build_handler(), job_deadline_sec=180).process_one()

        job = (await cl.get(f"/v1/job/{st['job_id']}", headers=h)).json()["job"]
        r = job.get("result") or {}
        print(f"\n  잡 상태: {job['status']} (3=DONE)")
        d = r.get("discover", {})
        print(f"  · Perplexity : URL {d.get('discovered',0)}건 발견 (검색 {d.get('searches',0)}회)"
              + (f" — {d['error'][:50]}" if d.get('error') else ""))
        print(f"  · 확정       : {r.get('confirm',{})}")
        print(f"  · 크롤링     : {r.get('fetch',{})}")
        print(f"  · 객실       : {r.get('units',0)}개")
        print(f"  · fact       : {r.get('facts',{})}")
        print(f"  · 사진       : {r.get('media',{})}")

        print("\n" + "="*74)
        print(f"  5. 발견된 채널 URL")
        print("="*74)
        for l in (await cl.get(f"/v1/place/{pid}/link/list", headers=h)).json()["links"]:
            ch = LinkChannel(l["channel"]).name
            print(f"  [{'확정' if l.get('confirmed_at') else '미확정'}] {ch:14s} {l['url'][:56]}")

        print("\n" + "="*74)
        print(f"  6. 수집된 fact — ★ 전부 후보다. 사이트에 안 나간다")
        print("="*74)
        fl = (await cl.get(f"/v1/place/{pid}/fact/list", headers=h)).json()
        units = {u["unit_id"]: u["name"] for u in (await cl.get(f"/v1/place/{pid}/unit/list", headers=h)).json().get("units", [])}
        for f in fl["facts"][:14]:
            scope = units.get(f.get("unit_id"), "사업장")
            print(f"  {scope:12s} {f['key']:20s} = {str(f.get('value'))[:30]:32s} [{'UNVERIFIED'}]")
        print(f"  ... 총 {len(fl['facts'])}건")
        print(f"\n  ★ 사이트에 나갈 수 있는 fact: {fl['publishable']}건  ← 사람이 승인해야 올라간다")

        print("\n" + "="*74)
        print(f"  7. 사람이 체크인 시간을 승인해본다")
        print("="*74)
        ci = next(f for f in fl["facts"] if f["key"] == "check_in_time")
        t = (await cl.post(f"/v1/place/{pid}/fact/{ci['fact_id']}/transition", headers=h, json={"status": 3})).json()
        print(f"  check_in_time = {t['fact']['value']} → VERIFIED (검증자 {t['fact']['verified_by'][:8]}…)")
        fl2 = (await cl.get(f"/v1/place/{pid}/fact/list", headers=h)).json()
        print(f"  ★ 사이트에 나갈 수 있는 fact: {fl2['publishable']}건")
        pl2 = (await cl.get(f"/v1/place/{pid}", headers=h)).json()["place"]
        print(f"  ★ content_updated_at: {pl2.get('content_updated_at')}  ← 이 사업장만 재빌드하면 된다")
        print("="*74 + "\n")

asyncio.run(main())