#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 카카오맵 리뷰 분석 API (로깅 강화 버전) ConfigMap과 Secret을 활용한 Kubernetes 배포용 app/main.py """ import os from fastapi import FastAPI, HTTPException, BackgroundTasks from fastapi.responses import HTMLResponse from pydantic import BaseModel, Field from typing import Optional, List, Dict, Any import asyncio import threading import json from datetime import datetime, timedelta import logging import sys import subprocess import platform # 기존 분석기 import from bs4 import BeautifulSoup import re import time # ============================================================================= # .env 파일 로딩 (다른 import보다 먼저) # ============================================================================= from dotenv import load_dotenv # .env 파일에서 환경변수 로드 load_dotenv() # ============================================================================= # 로깅 설정 (가장 먼저) # ============================================================================= logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.StreamHandler(sys.stdout) ] ) logger = logging.getLogger(__name__) # ============================================================================= # 시스템 정보 로깅 # ============================================================================= logger.info("="*60) logger.info("카카오맵 리뷰 분석 API 시작") logger.info("="*60) logger.info(f"Python 버전: {sys.version}") logger.info(f"플랫폼: {platform.platform()}") logger.info(f"아키텍처: {platform.architecture()}") logger.info(f"현재 작업 디렉토리: {os.getcwd()}") # Python 경로 확인 logger.info("Python 경로 정보:") for i, path in enumerate(sys.path): logger.info(f" [{i}] {path}") # ============================================================================= # 패키지 설치 상태 확인 # ============================================================================= def check_package_installation(): """설치된 패키지 확인""" logger.info("패키지 설치 상태 확인 중...") required_packages = [ 'selenium', 'webdriver-manager', 'beautifulsoup4', 'fastapi', 'uvicorn', 'pydantic' ] for package in required_packages: try: result = subprocess.run([sys.executable, '-m', 'pip', 'show', package], capture_output=True, text=True) if result.returncode == 0: version_line = [line for line in result.stdout.split('\n') if line.startswith('Version:')] version = version_line[0].split(':')[1].strip() if version_line else 'Unknown' logger.info(f"✅ {package}: {version}") else: logger.error(f"❌ {package}: 설치되지 않음") except Exception as e: logger.error(f"❌ {package} 확인 실패: {e}") check_package_installation() # ============================================================================= # Selenium 관련 import (세분화된 로깅) # ============================================================================= logger.info("Selenium 관련 모듈 import 시작...") SELENIUM_AVAILABLE = False SELENIUM_IMPORT_ERRORS = [] # 각 모듈을 개별적으로 import하여 어디서 실패하는지 확인 selenium_modules = [ ('selenium', 'selenium'), ('selenium.webdriver', 'webdriver'), ('selenium.webdriver.chrome.options', 'Options'), ('selenium.webdriver.common.by', 'By'), ('selenium.webdriver.support.ui', 'WebDriverWait'), ('selenium.webdriver.support', 'expected_conditions as EC'), ('webdriver_manager.chrome', 'ChromeDriverManager'), ('selenium.webdriver.chrome.service', 'Service') ] imported_modules = {} for module_path, import_name in selenium_modules: try: logger.info(f" 📦 {module_path}.{import_name} import 시도...") if import_name == 'selenium': import selenium imported_modules['selenium'] = selenium logger.info(f" ✅ selenium 버전: {selenium.__version__}") elif import_name == 'webdriver': from selenium import webdriver imported_modules['webdriver'] = webdriver logger.info(f" ✅ webdriver import 성공") elif import_name == 'Options': from selenium.webdriver.chrome.options import Options imported_modules['Options'] = Options logger.info(f" ✅ Options import 성공") elif import_name == 'By': from selenium.webdriver.common.by import By imported_modules['By'] = By logger.info(f" ✅ By import 성공") elif import_name == 'WebDriverWait': from selenium.webdriver.support.ui import WebDriverWait imported_modules['WebDriverWait'] = WebDriverWait logger.info(f" ✅ WebDriverWait import 성공") elif import_name == 'expected_conditions as EC': from selenium.webdriver.support import expected_conditions as EC imported_modules['EC'] = EC logger.info(f" ✅ expected_conditions import 성공") elif import_name == 'ChromeDriverManager': from webdriver_manager.chrome import ChromeDriverManager imported_modules['ChromeDriverManager'] = ChromeDriverManager logger.info(f" ✅ ChromeDriverManager import 성공") elif import_name == 'Service': from selenium.webdriver.chrome.service import Service imported_modules['Service'] = Service logger.info(f" ✅ Service import 성공") except ImportError as e: error_msg = f"{module_path}.{import_name} import 실패: {e}" logger.error(f" ❌ {error_msg}") SELENIUM_IMPORT_ERRORS.append(error_msg) except Exception as e: error_msg = f"{module_path}.{import_name} import 중 예외 발생: {e}" logger.error(f" ❌ {error_msg}") SELENIUM_IMPORT_ERRORS.append(error_msg) # 모든 모듈이 성공적으로 import되었는지 확인 required_modules = ['webdriver', 'Options', 'By', 'WebDriverWait', 'EC', 'ChromeDriverManager', 'Service'] missing_modules = [mod for mod in required_modules if mod not in imported_modules] if missing_modules: logger.error(f"❌ 누락된 모듈들: {missing_modules}") SELENIUM_AVAILABLE = False else: logger.info("✅ 모든 Selenium 모듈 import 성공!") SELENIUM_AVAILABLE = True # 전역 변수로 설정 webdriver = imported_modules['webdriver'] Options = imported_modules['Options'] By = imported_modules['By'] WebDriverWait = imported_modules['WebDriverWait'] EC = imported_modules['EC'] ChromeDriverManager = imported_modules['ChromeDriverManager'] Service = imported_modules['Service'] logger.info(f"Selenium 사용 가능 여부: {SELENIUM_AVAILABLE}") if SELENIUM_IMPORT_ERRORS: logger.error("Import 오류 목록:") for error in SELENIUM_IMPORT_ERRORS: logger.error(f" - {error}") # ============================================================================= # Chrome 설치 상태 확인 # ============================================================================= def check_chrome_installation(): """Chrome 브라우저 설치 상태 확인""" logger.info("Chrome 브라우저 설치 상태 확인 중...") chrome_paths = [ '/usr/bin/google-chrome', '/usr/bin/google-chrome-stable', '/usr/bin/chromium-browser', '/Applications/Google Chrome.app/Contents/MacOS/Google Chrome', 'C:\\Program Files\\Google\\Chrome\\Application\\chrome.exe', 'C:\\Program Files (x86)\\Google\\Chrome\\Application\\chrome.exe' ] chrome_found = False for path in chrome_paths: if os.path.exists(path): logger.info(f" ✅ Chrome 발견: {path}") chrome_found = True # 버전 확인 시도 try: result = subprocess.run([path, '--version'], capture_output=True, text=True, timeout=5) if result.returncode == 0: logger.info(f" 📋 Chrome 버전: {result.stdout.strip()}") except Exception as e: logger.warning(f" ⚠️ Chrome 버전 확인 실패: {e}") break if not chrome_found: logger.error(" ❌ Chrome 브라우저가 설치되지 않음") # 시스템 명령어로 확인 시도 try: result = subprocess.run(['which', 'google-chrome'], capture_output=True, text=True) if result.returncode == 0: logger.info(f" ✅ PATH에서 Chrome 발견: {result.stdout.strip()}") chrome_found = True except: pass try: result = subprocess.run(['google-chrome', '--version'], capture_output=True, text=True, timeout=5) if result.returncode == 0: logger.info(f" ✅ 명령어로 Chrome 실행 가능: {result.stdout.strip()}") chrome_found = True except Exception as e: logger.error(f" ❌ Chrome 명령어 실행 실패: {e}") return chrome_found chrome_available = check_chrome_installation() # ============================================================================= # 환경 변수 설정 # ============================================================================= class Config: """환경 변수 기반 설정 클래스""" # 애플리케이션 메타데이터 APP_TITLE = os.getenv("APP_TITLE", "카카오맵 리뷰 분석 API") APP_VERSION = os.getenv("APP_VERSION", "1.0.0") APP_DESCRIPTION = os.getenv("APP_DESCRIPTION", "교육 목적 전용 - 실제 서비스 사용 금지") # 서버 설정 HOST = os.getenv("HOST", "0.0.0.0") PORT = int(os.getenv("PORT", "8000")) WORKERS = int(os.getenv("WORKERS", "1")) LOG_LEVEL = os.getenv("LOG_LEVEL", "info") # API 기본값 DEFAULT_MAX_TIME = int(os.getenv("DEFAULT_MAX_TIME", "300")) DEFAULT_DAYS_LIMIT = int(os.getenv("DEFAULT_DAYS_LIMIT", "7")) MAX_DAYS_LIMIT = int(os.getenv("MAX_DAYS_LIMIT", "365")) MIN_MAX_TIME = int(os.getenv("MIN_MAX_TIME", "60")) MAX_MAX_TIME = int(os.getenv("MAX_MAX_TIME", "1800")) # Chrome 옵션 (ConfigMap에서 멀티라인 문자열로 받음) CHROME_OPTIONS_RAW = os.getenv("CHROME_OPTIONS", """ --headless --no-sandbox --disable-dev-shm-usage --disable-gpu --window-size=1920,1080 --disable-extensions --disable-plugins --disable-usb-keyboard-detect --no-first-run --no-default-browser-check --disable-logging --log-level=3 """) @property def CHROME_OPTIONS_LIST(self): """Chrome 옵션을 리스트로 파싱""" return [opt.strip() for opt in self.CHROME_OPTIONS_RAW.strip().split('\n') if opt.strip()] # 스크롤링 설정 SCROLL_CHECK_INTERVAL = int(os.getenv("SCROLL_CHECK_INTERVAL", "3")) SCROLL_NO_CHANGE_LIMIT = int(os.getenv("SCROLL_NO_CHANGE_LIMIT", "8")) SCROLL_WAIT_TIME_SHORT = float(os.getenv("SCROLL_WAIT_TIME_SHORT", "1.5")) SCROLL_WAIT_TIME_LONG = float(os.getenv("SCROLL_WAIT_TIME_LONG", "2.0")) # 법적 경고 LEGAL_WARNING_ENABLED = os.getenv("LEGAL_WARNING_ENABLED", "true").lower() == "true" CONTACT_EMAIL = os.getenv("CONTACT_EMAIL", "admin@example.com") # 건강 체크 HEALTH_CHECK_TIMEOUT = int(os.getenv("HEALTH_CHECK_TIMEOUT", "5")) # Secret 값들 (현재는 사용하지 않지만 향후 확장용) EXTERNAL_API_KEY = os.getenv("EXTERNAL_API_KEY", "") DB_USERNAME = os.getenv("DB_USERNAME", "") DB_PASSWORD = os.getenv("DB_PASSWORD", "") JWT_SECRET = os.getenv("JWT_SECRET", "") # 설정 인스턴스 config = Config() logger.info("환경 설정 로드 완료:") logger.info(f" - Chrome 옵션 개수: {len(config.CHROME_OPTIONS_LIST)}") logger.info(f" - 기본 최대 시간: {config.DEFAULT_MAX_TIME}초") logger.info(f" - 로그 레벨: {config.LOG_LEVEL}") # FastAPI 앱 초기화 (환경변수 사용) app = FastAPI( title=config.APP_TITLE, description=f""" **⚠️ 중요 법적 경고사항 ⚠️** {config.APP_DESCRIPTION} **진단 정보:** - Selenium 사용 가능: {SELENIUM_AVAILABLE} - Chrome 설치됨: {chrome_available} - Import 오류 개수: {len(SELENIUM_IMPORT_ERRORS)} **법적 위험:** - 카카오 이용약관 위반 → 계정 정지, 법적 조치 - 개인정보보호법(PIPA) 위반 → 과태료 최대 3억원 - 저작권 및 데이터베이스권 침해 → 손해배상 - 업무방해죄 → 5년 이하 징역 또는 1천500만원 이하 벌금 **합법적 대안:** - 카카오 공식 API 활용 (developers.kakao.com) - 점주 대상 자체 가게 관리 서비스 개발 - 사용자 동의 기반 데이터 수집 앱 - 카카오와 정식 파트너십 체결 **이 API를 실제 서비스에 사용하지 마세요!** **환경:** 로컬 개발 **버전:** {config.APP_VERSION} **연락처:** {config.CONTACT_EMAIL} """, version=config.APP_VERSION, contact={ "name": "관리자", "email": config.CONTACT_EMAIL }, license_info={ "name": "Educational Use Only", "url": "https://developers.kakao.com" } ) # Pydantic 모델 정의 (기존과 동일) class ReviewAnalysisRequest(BaseModel): """리뷰 분석 요청 모델""" store_id: str = Field( ..., description="카카오맵 가게 ID (예: 501745730)", example="501745730", min_length=1, max_length=20 ) days_limit: Optional[int] = Field( None, description=f"며칠 이후의 리뷰만 수집할지 (None이면 모든 날짜, 최대 {config.MAX_DAYS_LIMIT}일)", example=config.DEFAULT_DAYS_LIMIT, ge=1, le=config.MAX_DAYS_LIMIT ) max_time: int = Field( config.DEFAULT_MAX_TIME, description=f"최대 스크롤 시간(초, {config.MIN_MAX_TIME}-{config.MAX_MAX_TIME}초)", example=config.DEFAULT_MAX_TIME, ge=config.MIN_MAX_TIME, le=config.MAX_MAX_TIME ) class ReviewerStats(BaseModel): """리뷰어 통계 정보""" reviews: Optional[int] = None average_rating: Optional[float] = None followers: Optional[int] = None class ReviewData(BaseModel): """개별 리뷰 데이터""" reviewer_name: str reviewer_level: str reviewer_stats: ReviewerStats rating: int date: str content: str badges: List[str] likes: int photo_count: int has_photos: bool class StoreInfo(BaseModel): """가게 정보 (수정됨: id 필드 추가)""" id: str = Field(description="가게 ID") name: str category: str rating: str review_count: str status: str address: str class DateFilter(BaseModel): """날짜 필터 정보""" cutoff_date: Optional[str] = None filtered: bool class ReviewAnalysisResponse(BaseModel): """리뷰 분석 응답 모델""" success: bool = Field(description="분석 성공 여부") message: str = Field(description="응답 메시지") store_info: Optional[StoreInfo] = None reviews: List[ReviewData] = [] analysis_date: str = Field(description="분석 수행 날짜시간") total_reviews: int = Field(description="수집된 총 리뷰 수") analysis_method: str = "selenium" date_filter: DateFilter execution_time: float = Field(description="실행 시간(초)") class ErrorResponse(BaseModel): """에러 응답 모델""" success: bool = False error: str message: str timestamp: str # 분석기 클래스 (로깅 강화) class KakaoReviewAnalyzerAPI: """로깅 강화된 카카오맵 리뷰 분석기""" def __init__(self): self.store_info = {} self.reviews = [] self.cutoff_date = None self.config = config logger.info("KakaoReviewAnalyzerAPI 인스턴스 생성 완료") def analyze_by_store_id(self, store_id: str, days_limit: Optional[int] = None, max_scroll_time: int = None): """Selenium을 사용한 동적 리뷰 분석 (ChromeDriver 경로 수정)""" logger.info("="*60) logger.info("analyze_by_store_id 함수 시작") logger.info("="*60) if max_scroll_time is None: max_scroll_time = self.config.DEFAULT_MAX_TIME logger.info(f"입력 파라미터:") logger.info(f" - store_id: {store_id}") logger.info(f" - days_limit: {days_limit}") logger.info(f" - max_scroll_time: {max_scroll_time}") # 날짜 제한 설정 if days_limit is not None: self.cutoff_date = datetime.now() - timedelta(days=days_limit) logger.info(f"수집 기준일 설정: {self.cutoff_date.strftime('%Y.%m.%d')} 이후 리뷰만 수집") else: self.cutoff_date = None logger.info("날짜 제한 없음: 모든 날짜의 리뷰 수집") target_url = f"https://place.map.kakao.com/{store_id}#comment" logger.info(f"대상 URL 생성: {target_url}") driver = None temp_profile = None step_counter = 0 try: # Step 1: Chrome 옵션 설정 step_counter += 1 logger.info(f"[단계 {step_counter}] Chrome 옵션 설정 중...") if not SELENIUM_AVAILABLE: raise Exception(f"Selenium을 사용할 수 없음. Import 오류: {SELENIUM_IMPORT_ERRORS}") options = Options() logger.info(f" Chrome Options 객체 생성 완료") for i, option in enumerate(self.config.CHROME_OPTIONS_LIST): logger.info(f" 옵션 [{i+1}] 추가: {option}") options.add_argument(option) # User Agent 추가 user_agent = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' logger.info(f" User Agent 설정: {user_agent}") options.add_argument(f'--user-agent={user_agent}') # 추가 최적화 옵션들 experimental_options = ['enable-logging'] logger.info(f" 실험적 옵션 제외: {experimental_options}") options.add_experimental_option('excludeSwitches', experimental_options) options.add_experimental_option('useAutomationExtension', False) logger.info(f" Chrome 옵션 설정 완료 (총 {len(self.config.CHROME_OPTIONS_LIST)+1}개)") # Step 2: WebDriver 초기화 (ChromeDriverManager 사용 중단) step_counter += 1 logger.info(f"[단계 {step_counter}] Chrome WebDriver 초기화 중...") # 🔧 시스템에 설치된 ChromeDriver 사용 chromedriver_paths = [ "/usr/local/bin/chromedriver", # Dockerfile에서 설치한 경로 "/usr/bin/chromedriver", "/opt/chromedriver/chromedriver", "./chromedriver" ] driver_path = None for path in chromedriver_paths: if os.path.exists(path) and os.access(path, os.X_OK): logger.info(f" ChromeDriver 발견: {path}") # 버전 확인 try: import subprocess result = subprocess.run([path, "--version"], capture_output=True, text=True, timeout=5) if result.returncode == 0: logger.info(f" ChromeDriver 버전: {result.stdout.strip()}") driver_path = path break except Exception as version_error: logger.warning(f" {path} 버전 확인 실패: {version_error}") continue if not driver_path: # 마지막 수단: ChromeDriverManager 사용하되 올바른 실행 파일 찾기 logger.info(" 시스템 ChromeDriver 없음 - ChromeDriverManager 사용...") try: # 🔧 ChromeDriverManager 개선 os.environ['WDM_LOCAL'] = '/tmp/.wdm' downloaded_path = ChromeDriverManager(cache_valid_range=7).install() logger.info(f" 다운로드된 경로: {downloaded_path}") # 다운로드된 디렉토리에서 실제 실행 파일 찾기 download_dir = os.path.dirname(downloaded_path) possible_paths = [ os.path.join(download_dir, "chromedriver"), os.path.join(download_dir, "chromedriver-linux64", "chromedriver"), downloaded_path ] for candidate_path in possible_paths: if os.path.exists(candidate_path) and os.access(candidate_path, os.X_OK): # 파일이 실제 실행 파일인지 확인 (텍스트 파일이 아닌지) try: with open(candidate_path, 'rb') as f: header = f.read(4) # ELF 헤더 확인 (리눅스 실행 파일) if header.startswith(b'\x7fELF'): logger.info(f" 실행 가능한 ChromeDriver 발견: {candidate_path}") driver_path = candidate_path break except Exception as file_check_error: logger.warning(f" 파일 확인 실패: {candidate_path} - {file_check_error}") continue if not driver_path: raise Exception(f"다운로드된 디렉토리에서 실행 가능한 chromedriver를 찾을 수 없음: {download_dir}") except Exception as cdm_error: logger.error(f" ChromeDriverManager 실패: {cdm_error}") raise Exception(f"ChromeDriver를 찾을 수 없음. 시스템에 chromedriver를 설치하거나 Docker 이미지를 다시 빌드하세요.") # Service 객체 생성 logger.info(f" ChromeDriver 경로 사용: {driver_path}") service = Service(driver_path) # 🔧 Chrome 실행 환경 설정 logger.info(" Chrome 실행 환경 설정...") # 임시 프로필 디렉토리 생성 import tempfile temp_profile = tempfile.mkdtemp(prefix='chrome_profile_') logger.info(f" 임시 Chrome 프로필 생성: {temp_profile}") # Chrome 옵션에 프로필 경로 및 추가 안정성 옵션 추가 options.add_argument(f'--user-data-dir={temp_profile}') options.add_argument('--disable-web-security') options.add_argument('--allow-running-insecure-content') options.add_argument('--disable-blink-features=AutomationControlled') options.add_experimental_option("excludeSwitches", ["enable-automation"]) options.add_experimental_option('useAutomationExtension', False) logger.info(" 추가 안정성 옵션 적용 완료") # WebDriver 생성 logger.info(" Chrome WebDriver 인스턴스 생성 중...") try: driver = webdriver.Chrome(service=service, options=options) logger.info(" Chrome WebDriver 객체 생성 완료") # 🔧 즉시 연결 상태 확인 logger.info(" Chrome 연결 상태 확인...") driver.set_page_load_timeout(30) driver.set_script_timeout(30) driver.implicitly_wait(10) # 기본 페이지로 이동하여 연결 테스트 logger.info(" 연결 테스트용 빈 페이지 로드...") driver.get("data:text/html,Test

Connection Test

") # 페이지 제목 확인으로 연결 검증 test_title = driver.title logger.info(f" 연결 테스트 성공 - 페이지 제목: {test_title}") if not test_title: raise Exception("Chrome 연결은 되었으나 페이지 로드 실패") logger.info(" Chrome WebDriver 연결 확인 완료") # Chrome 정보 확인 try: capabilities = driver.capabilities chrome_version = capabilities.get('browserVersion', 'Unknown') driver_version = capabilities.get('chrome', {}).get('chromedriverVersion', 'Unknown') logger.info(f" Chrome 버전: {chrome_version}") logger.info(f" ChromeDriver 버전: {driver_version}") # 창 크기 확인 window_size = driver.get_window_size() logger.info(f" 현재 창 크기: {window_size}") except Exception as info_error: logger.warning(f" Chrome 정보 확인 실패: {info_error}") except Exception as driver_error: logger.error(f" Chrome WebDriver 생성/연결 실패: {driver_error}") # 리소스 정리 if driver: try: driver.quit() except: pass if temp_profile and os.path.exists(temp_profile): try: import shutil shutil.rmtree(temp_profile) logger.info(" 임시 프로필 디렉토리 정리 완료") except Exception as cleanup_error: logger.warning(f" 임시 프로필 정리 실패: {cleanup_error}") error_details = { "error_type": "WEBDRIVER_CONNECTION_FAILED", "chrome_options_count": len(self.config.CHROME_OPTIONS_LIST), "driver_path": driver_path, "temp_profile": temp_profile, "system_info": { "checked_paths": chromedriver_paths, "chrome_exists": os.path.exists("/usr/bin/google-chrome"), "wdm_cache": os.path.exists("/tmp/.wdm") } } raise Exception(f"Chrome WebDriver 연결 실패: {driver_error}\n상세 정보: {error_details}") # Step 3: 페이지 로드 step_counter += 1 logger.info(f"[단계 {step_counter}] 페이지 로드 시작...") logger.info(f" 요청 URL: {target_url}") start_time = time.time() try: driver.get(target_url) load_time = time.time() - start_time logger.info(f" 페이지 로드 완료 (소요시간: {load_time:.2f}초)") # 페이지 정보 확인 current_url = driver.current_url page_title = driver.title logger.info(f" 현재 URL: {current_url}") logger.info(f" 페이지 제목: {page_title}") if "404" in page_title or "오류" in page_title: raise Exception(f"유효하지 않은 페이지: {page_title}") # Body 로드 대기 logger.info(" 페이지 body 로드 대기 중...") WebDriverWait(driver, 15).until( EC.presence_of_element_located((By.TAG_NAME, "body")) ) logger.info(" 페이지 body 로드 완료") except Exception as page_error: logger.error(f" 페이지 로드 실패: {page_error}") current_url = driver.current_url if driver else "N/A" raise Exception(f"페이지 로드 실패 - 요청 URL: {target_url}, 현재 URL: {current_url}, 오류: {page_error}") # Step 4: 후기 탭 클릭 (이하 동일) step_counter += 1 logger.info(f"[단계 {step_counter}] 후기 탭 클릭 시도...") review_tab_selectors = [ "//a[@href='#comment']", "//a[contains(text(), '후기')]", "//button[contains(text(), '후기')]", "//*[@id='comment']", "//a[contains(@class, 'comment')]" ] review_tab_found = False for i, selector in enumerate(review_tab_selectors): try: logger.info(f" 선택자 [{i+1}] 시도: {selector}") review_tab = WebDriverWait(driver, 3).until( EC.element_to_be_clickable((By.XPATH, selector)) ) logger.info(f" 후기 탭 발견: {selector}") driver.execute_script("arguments[0].click();", review_tab) time.sleep(3) logger.info(f" 후기 탭 클릭 성공") review_tab_found = True break except Exception as e: logger.info(f" 선택자 [{i+1}] 실패: {e}") continue if not review_tab_found: logger.warning(" 후기 탭을 찾을 수 없음 - 기본 페이지에서 진행") # Step 5: 리뷰 목록 로드 확인 step_counter += 1 logger.info(f"[단계 {step_counter}] 리뷰 목록 로드 확인...") review_list_selectors = [ ".list_review", "[class*='review']", "[class*='comment']", ".review_list", ".comment_list" ] review_list_found = False for i, selector in enumerate(review_list_selectors): try: logger.info(f" 리뷰 목록 선택자 [{i+1}] 시도: {selector}") elements = driver.find_elements(By.CSS_SELECTOR, selector) if elements: logger.info(f" 리뷰 목록 발견: {selector} (요소 {len(elements)}개)") review_list_found = True break except Exception as e: logger.info(f" 리뷰 목록 선택자 [{i+1}] 실패: {e}") if not review_list_found: logger.warning(" 리뷰 목록을 찾을 수 없음 - HTML에서 직접 추출 시도") # Step 6: 스크롤링 step_counter += 1 logger.info(f"[단계 {step_counter}] 스크롤링 시작...") logger.info(f" 최대 스크롤 시간: {max_scroll_time}초") try: self._smart_infinite_scroll(driver, max_scroll_time) logger.info(" 스크롤링 완료") except Exception as scroll_error: logger.error(f" 스크롤링 중 오류: {scroll_error}") logger.info(" 스크롤링 실패 - 현재까지 로드된 내용으로 분석 진행") # Step 7: HTML 분석 step_counter += 1 logger.info(f"[단계 {step_counter}] HTML 내용 분석...") html_content = driver.page_source html_length = len(html_content) logger.info(f" HTML 내용 추출 완료: {html_length:,}자") if html_length < 1000: raise Exception(f"HTML 내용이 너무 짧음: {html_length}자") # HTML 분석 수행 logger.info(" HTML 파싱 및 데이터 추출 시작...") store_info, reviews = self.analyze_html_content(html_content, store_id) # 결과 검증 if not store_info: logger.warning(" 가게 정보 추출 실패 - 기본값 설정") store_info = { 'id': store_id, 'name': f'Store_{store_id}', 'category': '', 'rating': '', 'review_count': '', 'status': '', 'address': '' } else: logger.info(f" 가게 정보 추출 성공: {store_info.get('name', 'N/A')}") if not reviews: logger.warning(" 리뷰 데이터 추출 실패") reviews = [] else: logger.info(f" 리뷰 데이터 추출 성공: {len(reviews)}개") logger.info("="*60) logger.info("analyze_by_store_id 함수 완료") logger.info(f"최종 결과: 가게 정보={store_info.get('name', 'N/A')}, 리뷰={len(reviews)}개") logger.info("="*60) return store_info, reviews except Exception as e: logger.error("="*60) logger.error(f"[단계 {step_counter}] 오류 발생!") logger.error(f"오류 내용: {str(e)}") logger.error("="*60) # 상세 디버그 정보 수집 debug_info = { "failed_step": step_counter, "store_id": store_id, "target_url": target_url, "max_scroll_time": max_scroll_time, "days_limit": days_limit, "selenium_available": SELENIUM_AVAILABLE, "chrome_available": chrome_available, "chrome_options_count": len(self.config.CHROME_OPTIONS_LIST), "import_errors": SELENIUM_IMPORT_ERRORS, "driver_status": "initialized" if driver else "not_initialized", "temp_profile": temp_profile } if driver: try: debug_info.update({ "current_url": driver.current_url, "page_title": driver.title, "window_size": driver.get_window_size() }) except: debug_info["driver_info"] = "driver_not_accessible" logger.error("디버그 정보:") for key, value in debug_info.items(): logger.error(f" {key}: {value}") raise Exception(f"리뷰 분석 실패 (단계 {step_counter}): {str(e)}") finally: # 🔧 리소스 정리 강화 logger.info("리소스 정리 중...") # WebDriver 정리 if driver: try: logger.info(" Chrome WebDriver 종료 중...") driver.quit() logger.info(" Chrome WebDriver 종료 완료") except Exception as cleanup_error: logger.warning(f" WebDriver 정리 중 오류: {cleanup_error}") else: logger.info(" WebDriver가 초기화되지 않았음 - 정리 작업 불필요") # 임시 프로필 디렉토리 정리 if temp_profile and os.path.exists(temp_profile): try: import shutil shutil.rmtree(temp_profile) logger.info(f" 임시 프로필 디렉토리 정리 완료: {temp_profile}") except Exception as cleanup_error: logger.warning(f" 임시 프로필 정리 실패: {cleanup_error}") logger.info("리소스 정리 완료") def _smart_infinite_scroll(self, driver, max_time): """환경변수 기반 스마트 무한 스크롤링 (로깅 강화)""" logger.info("스크롤링 세부 설정:") logger.info(f" - 체크 간격: {self.config.SCROLL_CHECK_INTERVAL}회마다") logger.info(f" - 변화 없음 제한: {self.config.SCROLL_NO_CHANGE_LIMIT}회") logger.info(f" - 대기 시간 (짧음): {self.config.SCROLL_WAIT_TIME_SHORT}초") logger.info(f" - 대기 시간 (김): {self.config.SCROLL_WAIT_TIME_LONG}초") start_time = time.time() scroll_count = 0 no_change_count = 0 last_review_count = 0 while True: scroll_count += 1 elapsed_time = time.time() - start_time # 시간 제한 확인 if elapsed_time > max_time: logger.info(f"스크롤링 종료: 시간 제한 도달 ({max_time}초)") break try: # 현재 리뷰 개수 확인 review_elements = driver.find_elements(By.CSS_SELECTOR, "ul.list_review li") current_review_count = len(review_elements) # 10회마다 진행 상황 로깅 if scroll_count % 10 == 0: logger.info(f"스크롤링 진행: {scroll_count}회 | 리뷰 {current_review_count}개 | {elapsed_time:.1f}초 경과") # 날짜 기준 확인 if self.cutoff_date and scroll_count % self.config.SCROLL_CHECK_INTERVAL == 0: logger.info(f"날짜 기준 확인 중... (스크롤 {scroll_count}회)") date_cutoff_reached, oldest_date = self._check_date_cutoff_realtime(driver) if date_cutoff_reached: logger.info(f"스크롤링 종료: 날짜 기준 도달 ({oldest_date})") break # 리뷰 개수 변화 확인 if current_review_count == last_review_count: no_change_count += 1 if no_change_count >= self.config.SCROLL_NO_CHANGE_LIMIT: logger.info(f"스크롤링 종료: 더 이상 새로운 리뷰 없음 (총 {current_review_count}개)") break else: no_change_count = 0 last_review_count = current_review_count # 스크롤 실행 driver.execute_script("window.scrollTo(0, document.body.scrollHeight);") # 대기 시간 조정 wait_time = (self.config.SCROLL_WAIT_TIME_SHORT if current_review_count < 30 else self.config.SCROLL_WAIT_TIME_LONG) time.sleep(wait_time) except Exception as e: logger.error(f"스크롤 {scroll_count}회 실행 실패: {e}") break # 최종 결과 final_count = len(driver.find_elements(By.CSS_SELECTOR, "ul.list_review li")) logger.info(f"스크롤링 최종 결과:") logger.info(f" - 총 스크롤 횟수: {scroll_count}회") logger.info(f" - 최종 리뷰 개수: {final_count}개") logger.info(f" - 총 소요 시간: {elapsed_time:.1f}초") def _check_date_cutoff_realtime(self, driver): """실시간 날짜 기준 확인 (로깅 강화)""" if not self.cutoff_date: return False, None try: date_elements = driver.find_elements(By.CSS_SELECTOR, "ul.list_review li .txt_date") recent_dates = date_elements[-10:] if len(date_elements) > 10 else date_elements logger.debug(f"날짜 확인: 최근 {len(recent_dates)}개 날짜 요소 검사") for i, date_elem in enumerate(recent_dates): try: date_text = date_elem.text.strip() if date_text.endswith('.'): date_text = date_text[:-1] if not date_text: continue review_date = datetime.strptime(date_text, '%Y.%m.%d') logger.debug(f" 날짜 [{i+1}]: {date_text} vs 기준일 {self.cutoff_date.strftime('%Y.%m.%d')}") if review_date < self.cutoff_date: logger.info(f"날짜 기준 충족: {date_text} < {self.cutoff_date.strftime('%Y.%m.%d')}") return True, date_text except (ValueError, Exception) as e: logger.debug(f" 날짜 파싱 실패 [{i+1}]: {date_text} - {e}") continue except Exception as e: logger.warning(f"날짜 확인 중 오류: {e}") return False, None def analyze_html_content(self, html_content, store_id): """HTML 내용 분석 (수정됨: store_id 파라미터 추가)""" logger.info("HTML 내용 분석 시작...") try: soup = BeautifulSoup(html_content, 'html.parser') logger.info("BeautifulSoup 파싱 완료") # 가게 정보 추출 logger.info("가게 정보 추출 중...") self.store_info = self._extract_store_info(soup, store_id) logger.info(f"가게 정보 추출 완료: {self.store_info}") # 리뷰 추출 logger.info("리뷰 목록 추출 중...") self.reviews = self._extract_reviews(soup) logger.info(f"리뷰 추출 완료: {len(self.reviews)}개") return self.store_info, self.reviews except Exception as e: logger.error(f"HTML 분석 중 오류: {e}") raise def _extract_store_info(self, soup, store_id): """가게 정보 추출 (수정됨: id 필드 추가)""" store_info = { 'id': store_id, # 추가: 가게 ID 설정 'name': '', 'category': '', 'rating': '', 'review_count': '', 'status': '', 'address': '' } # 가게명 (수정: '장소명' 제거) store_name = soup.select_one('h3.tit_place') if store_name: name_text = store_name.get_text(strip=True) # '장소명' 텍스트 제거 name_text = name_text.replace('장소명', '').strip() store_info['name'] = name_text logger.info(f" 가게ID: {store_info['id']}") logger.info(f" 가게명: {store_info['name']}") else: logger.warning(" 가게명을 찾을 수 없음") # 카테고리 (수정: '장소 카테고리' 제거) category = soup.select_one('span.info_cate') if category: category_text = category.get_text(strip=True) # '장소 카테고리' 텍스트 제거 category_text = category_text.replace('장소 카테고리', '').strip() store_info['category'] = category_text logger.info(f" 카테고리: {store_info['category']}") else: logger.warning(" 카테고리를 찾을 수 없음") # 별점 rating = soup.select_one('span.num_star') if rating: store_info['rating'] = rating.get_text(strip=True) logger.info(f" 별점: {store_info['rating']}") else: logger.warning(" 별점을 찾을 수 없음") # 후기 수 review_count = soup.select_one('span.info_num') if review_count: count_text = review_count.get_text(strip=True) numbers = re.findall(r'\d+', count_text) if numbers: store_info['review_count'] = numbers[0] logger.info(f" 후기 수: {store_info['review_count']}") else: logger.warning(" 후기 수를 찾을 수 없음") # 영업 상태 status = soup.select_one('span.info_state') if status: store_info['status'] = status.get_text(strip=True) logger.info(f" 영업 상태: {store_info['status']}") else: logger.warning(" 영업 상태를 찾을 수 없음") # 주소 address_meta = soup.find('meta', property='og:description') if address_meta: store_info['address'] = address_meta.get('content', '') logger.info(f" 주소: {store_info['address']}") else: logger.warning(" 주소를 찾을 수 없음") return store_info def _extract_reviews(self, soup): """리뷰 목록 추출 (로깅 강화)""" reviews = [] review_list = soup.select_one('ul.list_review') if not review_list: logger.warning("리뷰 목록 컨테이너를 찾을 수 없음") return reviews review_items = review_list.find_all('li', recursive=False) logger.info(f"발견된 리뷰 아이템: {len(review_items)}개") skipped_count = 0 processed_count = 0 for i, item in enumerate(review_items): try: if (i + 1) % 10 == 0: logger.info(f"리뷰 처리 진행: {i+1}/{len(review_items)}") review_data = self._extract_single_review(item) if review_data and review_data.get('reviewer_name'): if self._is_review_within_date_range(review_data): reviews.append(review_data) processed_count += 1 else: skipped_count += 1 if skipped_count > 5: logger.info(f"연속 오래된 리뷰로 추출 중단 (총 {len(reviews)}개)") break except Exception as e: logger.warning(f"리뷰 #{i+1} 추출 실패: {e}") logger.info(f"리뷰 추출 완료: 처리됨={processed_count}개, 제외됨={skipped_count}개, 총={len(reviews)}개") return reviews def _extract_single_review(self, item): """개별 리뷰 데이터 추출 (수정됨: reviewer_name에서 '리뷰어 이름, ' 제거)""" review_data = { 'reviewer_name': '', 'reviewer_level': '', 'reviewer_stats': {}, 'rating': 0, 'date': '', 'content': '', 'badges': [], 'likes': 0, 'photo_count': 0, 'has_photos': False } # 리뷰어 이름 (수정: '리뷰어 이름, ' 제거) name_elem = item.select_one('span.name_user') if name_elem: name_text = name_elem.get_text(strip=True) # '리뷰어 이름, ' 텍스트 제거 (더 확실한 방법) if '리뷰어 이름,' in name_text: review_data['reviewer_name'] = name_text.replace('리뷰어 이름,', '').strip() else: review_data['reviewer_name'] = name_text # 리뷰어 레벨 level_elem = item.select_one('span.txt_badge') if level_elem: review_data['reviewer_level'] = level_elem.get_text(strip=True) # 리뷰어 통계 detail_list = item.select_one('ul.list_detail') if detail_list: stats_json = {} for li in detail_list.find_all('li'): stat = li.get_text(strip=True) if '후기' in stat: numbers = re.findall(r'\d+', stat) if numbers: stats_json['reviews'] = int(numbers[0]) elif '별점평균' in stat: numbers = re.findall(r'\d+\.?\d*', stat) if numbers: stats_json['average_rating'] = float(numbers[0]) elif '팔로워' in stat: numbers = re.findall(r'\d+', stat) if numbers: stats_json['followers'] = int(numbers[0]) review_data['reviewer_stats'] = stats_json # 별점 star_wrapper = item.select_one('span.wrap_grade') if star_wrapper: stars = star_wrapper.select('span.figure_star.on') review_data['rating'] = len(stars) # 날짜 date_elem = item.select_one('span.txt_date') if date_elem: date_text = date_elem.get_text(strip=True) if date_text.endswith('.'): date_text = date_text[:-1] review_data['date'] = date_text # 리뷰 내용 content_elem = item.select_one('p.desc_review') if content_elem: review_data['content'] = content_elem.get_text(strip=True) # 태그/배지 badges = item.select('span.badge_point') for badge in badges: badge_text = badge.get_text(strip=True) if badge_text and badge_text not in review_data['badges']: review_data['badges'].append(badge_text) # 좋아요 수 like_btn = item.select_one('button .txt_btn') if like_btn: like_text = like_btn.get_text(strip=True) try: review_data['likes'] = int(like_text) except ValueError: review_data['likes'] = 0 # 사진 개수 photos = item.select('.review_thumb img, .thumb_img img') review_data['photo_count'] = len(photos) review_data['has_photos'] = len(photos) > 0 return review_data def _is_review_within_date_range(self, review_data): """리뷰가 날짜 범위 내에 있는지 확인""" if not self.cutoff_date: return True date_str = review_data.get('date', '') if not date_str: return True try: if date_str.endswith('.'): date_str = date_str[:-1] review_date = datetime.strptime(date_str, '%Y.%m.%d') return review_date >= self.cutoff_date except ValueError: return True # API 엔드포인트 @app.get("/", response_class=HTMLResponse, include_in_schema=False) async def root(): """메인 페이지 (진단 정보 포함)""" # 진단 정보 생성 diagnosis_info = f"""

🔧 시스템 진단 정보

{f'''

❌ Import 오류 목록:

''' if SELENIUM_IMPORT_ERRORS else ''}
""" warning_section = "" if config.LEGAL_WARNING_ENABLED: warning_section = f"""

⚠️ 중요 법적 경고사항

{config.APP_DESCRIPTION}

합법적 대안을 사용하세요:

""" return f""" {config.APP_TITLE}

🔍 {config.APP_TITLE}

버전: {config.APP_VERSION}

환경: 로컬 개발

연락처: {config.CONTACT_EMAIL}

{diagnosis_info} {warning_section}

🔧 환경 설정 정보

📚 API 문서

Swagger UI 문서 ReDoc 문서 헬스 체크 진단 정보

🛠️ 사용 방법

POST /analyze - 리뷰 분석 수행

{{
  "store_id": "501745730",
  "days_limit": {config.DEFAULT_DAYS_LIMIT},
  "max_time": {config.DEFAULT_MAX_TIME}
}}
""" @app.post( "/analyze", response_model=ReviewAnalysisResponse, summary="카카오맵 리뷰 분석 (로깅 강화)", description=f""" **⚠️ {config.APP_DESCRIPTION}** **진단 정보:** - Selenium 사용 가능: {SELENIUM_AVAILABLE} - Chrome 설치됨: {chrome_available} - Import 오류 개수: {len(SELENIUM_IMPORT_ERRORS)} 지정한 카카오맵 가게의 리뷰를 분석합니다. **주요 기능:** - 실시간 리뷰 수집 및 분석 - 날짜 범위 필터링 (최대 {config.MAX_DAYS_LIMIT}일) - 감정 분석 및 통계 - JSON 형태 응답 - 상세한 로깅 및 디버그 정보 **응답 시간:** 설정에 따라 {config.MIN_MAX_TIME//60}-{config.MAX_MAX_TIME//60}분 소요 **환경:** 로컬 개발 버전 """, responses={ 200: {"description": "분석 성공", "model": ReviewAnalysisResponse}, 400: {"description": "잘못된 요청", "model": ErrorResponse}, 500: {"description": "서버 오류", "model": ErrorResponse} } ) async def analyze_reviews(request: ReviewAnalysisRequest): """리뷰 분석 API (수정됨: environment 섹션 제거)""" logger.info("="*60) logger.info("analyze_reviews API 호출") logger.info("="*60) logger.info(f"요청 파라미터: {request}") # Selenium 사용 가능성 확인 if not SELENIUM_AVAILABLE: error_detail = { "success": False, "error": "SELENIUM_NOT_AVAILABLE", "message": "Selenium이 설치되지 않았습니다. pip install selenium webdriver-manager", "timestamp": datetime.now().isoformat(), "debug_info": { "import_errors": SELENIUM_IMPORT_ERRORS, "python_version": sys.version, "platform": platform.platform(), "chrome_available": chrome_available } } logger.error("Selenium 사용 불가능!") logger.error(f"Import 오류: {SELENIUM_IMPORT_ERRORS}") raise HTTPException(status_code=500, detail=error_detail) # Chrome 사용 가능성 확인 if not chrome_available: error_detail = { "success": False, "error": "CHROME_NOT_AVAILABLE", "message": "Chrome 브라우저가 설치되지 않았습니다.", "timestamp": datetime.now().isoformat() } logger.error("Chrome 브라우저 사용 불가능!") raise HTTPException(status_code=500, detail=error_detail) start_time = time.time() try: logger.info(f"리뷰 분석 시작: store_id={request.store_id}, days_limit={request.days_limit}, max_time={request.max_time}") # 분석기 실행 def run_analysis(): analyzer = KakaoReviewAnalyzerAPI() return analyzer.analyze_by_store_id( store_id=request.store_id, days_limit=request.days_limit, max_scroll_time=request.max_time ) # 스레드에서 실행 loop = asyncio.get_event_loop() store_info, reviews = await loop.run_in_executor(None, run_analysis) execution_time = time.time() - start_time # 응답 데이터 구성 (environment 섹션 제거됨) response_data = ReviewAnalysisResponse( success=True, message="분석이 성공적으로 완료되었습니다.", store_info=StoreInfo(**store_info) if store_info else None, reviews=[ ReviewData( reviewer_name=review['reviewer_name'], reviewer_level=review['reviewer_level'], reviewer_stats=ReviewerStats(**review['reviewer_stats']), rating=review['rating'], date=review['date'], content=review['content'], badges=review['badges'], likes=review['likes'], photo_count=review['photo_count'], has_photos=review['has_photos'] ) for review in reviews ], analysis_date=datetime.now().isoformat(), total_reviews=len(reviews), date_filter=DateFilter( cutoff_date=(datetime.now() - timedelta(days=request.days_limit)).isoformat() if request.days_limit else None, filtered=request.days_limit is not None ), execution_time=execution_time ) logger.info(f"분석 완료: {len(reviews)}개 리뷰, {execution_time:.1f}초 소요") logger.info("="*60) return response_data except Exception as e: execution_time = time.time() - start_time logger.error(f"분석 실패: {str(e)}") logger.error("="*60) raise HTTPException( status_code=500, detail={ "success": False, "error": "ANALYSIS_FAILED", "message": f"리뷰 분석 중 오류가 발생했습니다: {str(e)}", "timestamp": datetime.now().isoformat(), "execution_time": execution_time, "debug_info": { "selenium_available": SELENIUM_AVAILABLE, "chrome_available": chrome_available, "import_errors": SELENIUM_IMPORT_ERRORS, "request_params": { "store_id": request.store_id, "days_limit": request.days_limit, "max_time": request.max_time } } } ) @app.get("/diagnostic", summary="상세 진단 정보", description="시스템 상태를 상세히 진단합니다.") async def diagnostic(): """상세 진단 정보 제공""" # 패키지 버전 확인 package_info = {} required_packages = ['selenium', 'webdriver-manager', 'beautifulsoup4', 'fastapi', 'uvicorn', 'pydantic'] for package in required_packages: try: result = subprocess.run([sys.executable, '-m', 'pip', 'show', package], capture_output=True, text=True) if result.returncode == 0: version_line = [line for line in result.stdout.split('\n') if line.startswith('Version:')] version = version_line[0].split(':')[1].strip() if version_line else 'Unknown' package_info[package] = {"installed": True, "version": version} else: package_info[package] = {"installed": False, "version": None} except Exception as e: package_info[package] = {"installed": False, "error": str(e)} return { "system_info": { "python_version": sys.version, "platform": platform.platform(), "architecture": platform.architecture(), "current_directory": os.getcwd(), "python_path": sys.path[:5] # 처음 5개만 }, "selenium_info": { "available": SELENIUM_AVAILABLE, "import_errors": SELENIUM_IMPORT_ERRORS, "imported_modules": list(imported_modules.keys()) if SELENIUM_AVAILABLE else [] }, "chrome_info": { "available": chrome_available, "common_paths_checked": [ '/usr/bin/google-chrome', '/usr/bin/google-chrome-stable', '/usr/bin/chromium-browser', '/Applications/Google Chrome.app/Contents/MacOS/Google Chrome' ] }, "package_info": package_info, "timestamp": datetime.now().isoformat() } @app.get( "/health", summary="헬스 체크", description="API 서버 상태를 확인합니다." ) async def health_check(): """헬스 체크 엔드포인트 (진단 정보 포함)""" return { "status": "healthy", "timestamp": datetime.now().isoformat(), "selenium_available": SELENIUM_AVAILABLE, "chrome_available": chrome_available, "version": config.APP_VERSION, "message": f"{config.APP_TITLE}이 정상 작동 중입니다." if SELENIUM_AVAILABLE else "Selenium 모듈 문제로 제한된 기능만 사용 가능합니다." } @app.get("/config", summary="환경 설정 확인", description="현재 적용된 환경 변수 설정을 확인합니다.") async def get_config(): """환경 설정 확인 엔드포인트""" return { "app_info": { "title": config.APP_TITLE, "version": config.APP_VERSION, "description": config.APP_DESCRIPTION }, "server_config": { "host": config.HOST, "port": config.PORT, "workers": config.WORKERS, "log_level": config.LOG_LEVEL }, "api_defaults": { "default_max_time": config.DEFAULT_MAX_TIME, "default_days_limit": config.DEFAULT_DAYS_LIMIT, "max_days_limit": config.MAX_DAYS_LIMIT, "min_max_time": config.MIN_MAX_TIME, "max_max_time": config.MAX_MAX_TIME }, "scroll_settings": { "check_interval": config.SCROLL_CHECK_INTERVAL, "no_change_limit": config.SCROLL_NO_CHANGE_LIMIT, "wait_time_short": config.SCROLL_WAIT_TIME_SHORT, "wait_time_long": config.SCROLL_WAIT_TIME_LONG }, "chrome_options": { "count": len(config.CHROME_OPTIONS_LIST), "options": config.CHROME_OPTIONS_LIST }, "features": { "legal_warning_enabled": config.LEGAL_WARNING_ENABLED, "contact_email": config.CONTACT_EMAIL, "health_check_timeout": config.HEALTH_CHECK_TIMEOUT }, "diagnosis": { "selenium_available": SELENIUM_AVAILABLE, "chrome_available": chrome_available, "import_error_count": len(SELENIUM_IMPORT_ERRORS) }, "timestamp": datetime.now().isoformat() } @app.get("/legal-warning", summary="법적 경고사항", description="API 사용 시 주의해야 할 법적 사항을 안내합니다.") async def legal_warning(): """법적 경고사항 (환경변수 활용)""" return { "warning": f"⚠️ {config.APP_DESCRIPTION}", "legal_risks": [ "카카오 이용약관 위반 → 계정 정지, 법적 조치", "개인정보보호법(PIPA) 위반 → 과태료 최대 3억원", "저작권 및 데이터베이스권 침해 → 손해배상", "업무방해죄 → 5년 이하 징역 또는 1천500만원 이하 벌금" ], "legal_alternatives": [ "카카오 공식 API 활용 (developers.kakao.com)", "점주 대상 자체 가게 관리 서비스 개발", "사용자 동의 기반 데이터 수집 앱", "카카오와 정식 파트너십 체결" ], "message": "이 API를 실제 서비스에 사용하지 마세요!", "contact": config.CONTACT_EMAIL, "version": config.APP_VERSION, "timestamp": datetime.now().isoformat() } if __name__ == "__main__": import uvicorn print("🚨 " + "="*60) print(" ⚠️ 중요 법적 경고사항 ⚠️") print("="*64) print(f"❌ {config.APP_DESCRIPTION}") print("❌ 실제 웹사이트 크롤링은 불법입니다.") print("✅ 카카오 공식 API를 사용하세요: developers.kakao.com") print("="*64) print() print(f"🚀 {config.APP_TITLE} 서버 시작") print(f"📊 진단 정보:") print(f" - Selenium 사용 가능: {SELENIUM_AVAILABLE}") print(f" - Chrome 설치됨: {chrome_available}") print(f" - Import 오류: {len(SELENIUM_IMPORT_ERRORS)}개") print() print(f"📚 Swagger 문서: http://{config.HOST}:{config.PORT}/docs") print(f"📖 ReDoc 문서: http://{config.HOST}:{config.PORT}/redoc") print(f"🏠 메인 페이지: http://{config.HOST}:{config.PORT}/") print(f"🔧 진단 정보: http://{config.HOST}:{config.PORT}/diagnostic") print() uvicorn.run( app, host=config.HOST, port=config.PORT, log_level=config.LOG_LEVEL )