241 lines
9.7 KiB
Python
241 lines
9.7 KiB
Python
import sys
|
|
import os
|
|
import logging
|
|
from typing import Any, Dict, Optional, Tuple, List
|
|
from bs4 import BeautifulSoup
|
|
from playwright.sync_api import sync_playwright
|
|
from supabase import create_client, Client
|
|
|
|
from PySide6.QtWidgets import (
|
|
QApplication,
|
|
QWidget,
|
|
QVBoxLayout,
|
|
QFormLayout,
|
|
QLineEdit,
|
|
QPushButton,
|
|
QFileDialog,
|
|
QMessageBox,
|
|
)
|
|
|
|
|
|
# ================= 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 파싱 및 마켓 정보 수집 =================
|
|
def extract_market_urls(html_file_path: str) -> List[str]:
|
|
"""
|
|
지정한 HTML 파일에서 https://smartstore.naver.com 로 시작하는 URL을 추출합니다.
|
|
"""
|
|
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))
|
|
|
|
|
|
def fetch_market_info(url: str, page) -> Tuple[str, str]:
|
|
"""
|
|
Playwright 페이지 인스턴스를 재사용하여 URL에 접속, market_name과 market_grade를 추출합니다.
|
|
market_name은 head > title의 텍스트를,
|
|
market_grade는 지정한 셀렉터의 텍스트를 사용하며, 정보가 없으면 '일반'으로 반환합니다.
|
|
"""
|
|
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
|
|
|
|
|
|
def supabase_insert_markets(supabase_manager: SupabaseManager, market_data: List[Tuple[str, str, str]]) -> None:
|
|
"""
|
|
Supabase에 데이터 삽입 (market_url 중복 검사 포함)
|
|
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()
|
|
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}")
|
|
|
|
|
|
# ================= PySide6 GUI =================
|
|
class MainWindow(QWidget):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.supabase_manager: Optional[SupabaseManager] = None
|
|
self.init_ui()
|
|
|
|
def init_ui(self):
|
|
self.setWindowTitle("Supabase 로그인 및 HTML 처리")
|
|
layout = QVBoxLayout()
|
|
|
|
# 로그인 폼
|
|
form_layout = QFormLayout()
|
|
self.email_input = QLineEdit()
|
|
self.password_input = QLineEdit()
|
|
self.password_input.setEchoMode(QLineEdit.Password)
|
|
form_layout.addRow("Email:", self.email_input)
|
|
form_layout.addRow("Password:", self.password_input)
|
|
layout.addLayout(form_layout)
|
|
|
|
self.login_button = QPushButton("로그인")
|
|
self.login_button.clicked.connect(self.handle_login)
|
|
layout.addWidget(self.login_button)
|
|
|
|
self.setLayout(layout)
|
|
|
|
def handle_login(self):
|
|
email = self.email_input.text().strip()
|
|
password = self.password_input.text().strip()
|
|
if not email or not password:
|
|
QMessageBox.warning(self, "경고", "이메일과 비밀번호를 모두 입력하세요.")
|
|
return
|
|
|
|
# 로그 설정 및 SupabaseManager 초기화
|
|
logging.basicConfig(level=logging.DEBUG)
|
|
logger = logging.getLogger("SupabaseManager")
|
|
self.supabase_manager = SupabaseManager(logger)
|
|
|
|
login_info = self.supabase_manager.login(email, password)
|
|
if not login_info or "error" in login_info:
|
|
QMessageBox.critical(self, "로그인 실패", "로그인에 실패했습니다. 자격증명을 확인하세요.")
|
|
else:
|
|
QMessageBox.information(self, "로그인 성공", "로그인에 성공했습니다.")
|
|
self.select_folder_and_process()
|
|
|
|
def select_folder_and_process(self):
|
|
folder = QFileDialog.getExistingDirectory(self, "HTML 파일이 있는 폴더 선택")
|
|
if not folder:
|
|
QMessageBox.warning(self, "경고", "폴더가 선택되지 않았습니다.")
|
|
return
|
|
|
|
# 폴더 내 모든 .html 파일 경로 리스트
|
|
html_files = [os.path.join(folder, f) for f in os.listdir(folder) if f.lower().endswith(".html")]
|
|
if not html_files:
|
|
QMessageBox.information(self, "알림", "선택한 폴더에 HTML 파일이 없습니다.")
|
|
return
|
|
|
|
# 각 HTML 파일에서 URL 추출 (파일별로 추출한 URL들을 합칩니다)
|
|
all_market_urls = []
|
|
for file_path in html_files:
|
|
urls = extract_market_urls(file_path)
|
|
all_market_urls.extend(urls)
|
|
unique_urls = list(set(all_market_urls))
|
|
print(f"총 {len(unique_urls)}개의 고유 스마트스토어 URL을 찾았습니다.")
|
|
|
|
# Playwright를 한 번 실행하여 각 URL에 대해 마켓 정보 수집
|
|
market_data = []
|
|
with sync_playwright() as p:
|
|
browser = p.chromium.launch(headless=False) # headless 모드 False
|
|
page = browser.new_page()
|
|
for url in unique_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()
|
|
|
|
# Supabase에 데이터 삽입 (중복 검사 포함)
|
|
supabase_insert_markets(self.supabase_manager, market_data)
|
|
QMessageBox.information(self, "완료", "모든 HTML 파일 처리 및 Supabase 입력이 완료되었습니다.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app = QApplication(sys.argv)
|
|
window = MainWindow()
|
|
window.show()
|
|
sys.exit(app.exec())
|