AutoPercenty3/test/in_market.py

import sys
import getpass
import logging
from typing import Any, Dict, Optional
from bs4 import BeautifulSoup
from playwright.sync_api import sync_playwright
from supabase import create_client, Client

# SupabaseManager 클래스 (제공해주신 코드 기반)
class SupabaseManager:
    """
    SupabaseManager는 Supabase 클라이언트를 래핑하여,
    회원가입, 로그인, 사용자 정보 조회, 공지사항, 라이센스, 기타 API 요청을 수행합니다.
    """
    def __init__(self, logger: logging.Logger) -> None:
        """
        초기화합니다.
        :param logger: 로깅을 위한 Logger 객체.
        """
        self.logger = logger
        self.url: str = "http://146.56.101.199:8000"
        self.key: str = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyAgCiAgICAicm9sZSI6ICJhbm9uIiwKICAgICJpc3MiOiAic3VwYWJhc2UtZGVtbyIsCiAgICAiaWF0IjogMTY0MTc2OTIwMCwKICAgICJleHAiOiAxNzk5NTM1NjAwCn0.dc_X5iR_VP_qT0zsiyj_I_OZ2T9FtRU2BBNWN8Bu4GE"
        self.client: Client = create_client(self.url, self.key)
        self.access_token: Optional[str] = None
        self.refresh_token: Optional[str] = None

    def update_client_with_token(self, access_token: str) -> None:
        """
        클라이언트의 인증 토큰 업데이트
        """
        self.client.auth.session = {"access_token": access_token}

    def login(self, email: str, password: str) -> Optional[Dict[str, Any]]:
        """
        Supabase Auth를 사용하여 로그인합니다.
        이메일 인증이 완료되지 않은 경우 에러 메시지를 반환합니다.
        :param email: 사용자 이메일.
        :param password: 사용자 비밀번호.
        :return: 로그인 성공 시 사용자 정보를 담은 딕셔너리, 실패 시 None 또는 error dict.
        """
        try:
            response = self.client.auth.sign_in_with_password({"email": email, "password": password})
            if response.session:
                if response.user.email_confirmed_at is None:
                    self.logger.warning("이메일 인증이 완료되지 않았습니다. 이메일을 확인하세요.")
                    return {"error": "이메일 인증을 먼저 완료해 주세요."}
                self.access_token = response.session.access_token
                self.refresh_token = response.session.refresh_token
                self.update_client_with_token(self.access_token)
                user_info: Dict[str, Any] = {
                    "id": response.user.id,
                    "email": response.user.email,
                    "nickname": response.user.user_metadata.get("nickname", "Unknown"),
                }
                self.logger.debug("로그인 성공")
                self.logger.debug(f"response : {response}")
                self.logger.debug(f"user_info : {user_info}")
                return user_info
            else:
                self.logger.warning("로그인 실패: 세션 없음")
                return None
        except Exception as e:
            self.logger.error(f"Login error: {e}", exc_info=True)
            return None

# HTML 파일에서 스마트스토어 URL 추출 함수
def extract_market_urls(html_file_path: str):
    with open(html_file_path, "r", encoding="utf-8") as f:
        html_content = f.read()
    soup = BeautifulSoup(html_content, "html.parser")
    links = soup.find_all("a", href=True)
    market_urls = []
    for link in links:
        href = link["href"]
        if href.startswith("https://smartstore.naver.com"):
            market_urls.append(href)
    # 중복 제거
    return list(set(market_urls))

# Playwright를 사용하여 마켓 정보 수집 함수 (페이지 인스턴스를 외부에서 전달)
def fetch_market_info(url: str, page) -> (str, str):
    market_name = ""
    market_grade = ""
    
    try:
        page.goto(url, timeout=60000)  # 60초 timeout
        page.wait_for_load_state("domcontentloaded", timeout=60000)
        
        # market_name 추출 (head > title)
        name_elem = page.query_selector("head > title")
        if name_elem:
            market_name = name_elem.inner_text().strip()
        
        # market_grade 추출 (셀렉터 수정)
        grade_elem = page.query_selector("div#pc-sellerInfoWidget div:nth-child(1) > span._3CfLtIh1fI")
        if grade_elem:
            market_grade = grade_elem.inner_text().strip()
        else:
            # 마켓 등급 정보가 없으면 "일반"으로 설정
            market_grade = "일반"
    except Exception as e:
        print(f"Error processing {url}: {e}")
    
    return market_name, market_grade

# Supabase 데이터 삽입 함수 (market_url 중복 검사 포함)
def supabase_insert_markets(supabase_manager: SupabaseManager, market_data: list):
    """
    market_data: 리스트로 [(market_url, market_name, market_grade), ...]
    """
    for url, name, grade in market_data:
        # 중복 검사: market_url이 이미 존재하는지 확인
        existing = supabase_manager.client.table("markets").select("*").eq("market_url", url).execute()
        if existing.data:
            print(f"{url} 은(는) 이미 존재합니다. 건너뜁니다.")
            continue
        
        data = {
            "market_name": name,
            "market_url": url,
            "market_grade": grade,
            "market_memo": ""
        }
        try:
            response = supabase_manager.client.table("markets").insert(data).execute()
            print(f"response : {response}")
            if response.error:
                print(f"Failed to insert {url}: {response['error']['message']}")
            else:
                print(f"Inserted {url} successfully.")
        except Exception as e:
            print(f"Exception inserting {url}: {e}")

def main():
    if len(sys.argv) < 2:
        print("Usage: python module.py <html_file_path>")
        sys.exit(1)
    
    html_file_path = sys.argv[1]
    
    # 1. HTML 파일에서 마켓 URL 추출
    market_urls = extract_market_urls(html_file_path)
    print(f"총 {len(market_urls)}개의 스마트스토어 URL을 찾았습니다.")
    
    market_data = []
    
    # Playwright 브라우저와 페이지 한 번만 생성하여 재사용
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=False)  # headless=False로 설정
        page = browser.new_page()
        
        # 2. 각 URL에 대해 페이지 재사용하여 정보 수집
        for url in market_urls:
            print(f"Processing {url} ...")
            name, grade = fetch_market_info(url, page)
            print(f"  market_name: {name}")
            print(f"  market_grade: {grade}")
            market_data.append((url, name, grade))
        
        browser.close()
    
    # 3. SupabaseManager 초기화 및 로그인 (로그 출력 위해 logging 설정)
    logging.basicConfig(level=logging.DEBUG)
    logger = logging.getLogger("SupabaseManager")
    supabase_manager = SupabaseManager(logger)
    
    print("Supabase 로그인 정보를 입력하세요.")
    supabase_email = input("Supabase Email (ID): ").strip()
    supabase_pw = getpass.getpass("Supabase Password: ").strip()
    
    login_info = supabase_manager.login(supabase_email, supabase_pw)
    if not login_info or "error" in login_info:
        print("Supabase 로그인 실패. 프로그램을 종료합니다.")
        sys.exit(1)
    else:
        print("Supabase 로그인 성공!")
    
    # 4. 수집한 데이터 Supabase에 삽입 (중복 검사 포함)
    supabase_insert_markets(supabase_manager, market_data)

if __name__ == "__main__":
    main()