HanJun's picture
Update app.py
37ff9f3 verified
Raw
History Blame Contribute Delete
27.6 kB
import os
import time
import base64
import io
import threading
import atexit
import traceback
import asyncio
import concurrent.futures
from datetime import datetime
from typing import Optional, Dict, Any, Tuple
import gradio as gr
import anthropic
from playwright.sync_api import sync_playwright
from PIL import Image
from dotenv import load_dotenv
# ν™˜κ²½λ³€μˆ˜ λ‘œλ“œ
load_dotenv()
# 뢄석 데이터 μ €μž₯μ†Œ
analyzed_data = {}
last_update_time = None
update_thread = None
is_updating = False
def get_claude_client():
"""Claude API ν΄λΌμ΄μ–ΈνŠΈ μ΄ˆκΈ°ν™”"""
api_key = os.getenv("ANTHROPIC_API_KEY")
if not api_key:
raise ValueError("ANTHROPIC_API_KEY ν™˜κ²½λ³€μˆ˜κ°€ μ„€μ •λ˜μ§€ μ•Šμ•˜μŠ΅λ‹ˆλ‹€.")
return anthropic.Anthropic(
api_key=api_key,
max_retries=3,
timeout=60.0
)
def install_browsers():
"""Playwright λΈŒλΌμš°μ € μ„€μΉ˜"""
try:
print("Playwright λΈŒλΌμš°μ € μ„€μΉ˜ 쀑...")
os.system("playwright install chromium --force")
print("λΈŒλΌμš°μ € μ„€μΉ˜ μ™„λ£Œ!")
return True
except Exception as e:
print(f"λΈŒλΌμš°μ € μ„€μΉ˜ μ‹€νŒ¨: {e}")
return False
def create_browser_for_thread():
"""μŠ€λ ˆλ“œλ³„ 독립적인 λΈŒλΌμš°μ € μΈμŠ€ν„΄μŠ€ 생성"""
try:
print(f"[Thread {threading.current_thread().name}] λΈŒλΌμš°μ € μΈμŠ€ν„΄μŠ€ 생성 쀑...")
playwright = sync_playwright().start()
# λΈŒλΌμš°μ € μ‹€ν–‰ μ˜΅μ…˜
browser_args = [
'--no-sandbox',
'--disable-dev-shm-usage',
'--disable-gpu',
'--disable-extensions',
'--disable-web-security',
'--disable-blink-features=AutomationControlled',
'--disable-background-timer-throttling',
'--disable-backgrounding-occluded-windows',
'--disable-renderer-backgrounding',
'--disable-features=TranslateUI',
'--disable-ipc-flooding-protection'
]
browser = playwright.chromium.launch(
headless=True,
args=browser_args
)
print(f"[Thread {threading.current_thread().name}] λΈŒλΌμš°μ € μΈμŠ€ν„΄μŠ€ 생성 μ™„λ£Œ")
return playwright, browser
except Exception as e:
print(f"[Thread {threading.current_thread().name}] λΈŒλΌμš°μ € μΈμŠ€ν„΄μŠ€ 생성 μ‹€νŒ¨: {str(e)}")
print(f"상세 였λ₯˜: {traceback.format_exc()}")
return None, None
def capture_google_trends() -> Optional[Image.Image]:
"""Google Trends μŠ€ν¬λ¦°μƒ· 캑처 (독립 λΈŒλΌμš°μ € μΈμŠ€ν„΄μŠ€ μ‚¬μš©)"""
playwright = None
browser = None
page = None
try:
print("Google Trends μŠ€ν¬λ¦°μƒ· μ‹œμž‘...")
# μŠ€λ ˆλ“œλ³„ 독립 λΈŒλΌμš°μ € 생성
playwright, browser = create_browser_for_thread()
if not browser:
print("λΈŒλΌμš°μ € μ΄ˆκΈ°ν™” μ‹€νŒ¨")
return None
page = browser.new_page()
page.set_viewport_size({"width": 1400, "height": 900})
page.set_extra_http_headers({
"Accept-Language": "ko-KR,ko;q=0.9,en;q=0.8",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8"
})
url = "https://trends.google.co.kr/trending?geo=KR"
print(f"νŽ˜μ΄μ§€ 접속: {url}")
# νŽ˜μ΄μ§€ λ‘œλ”© (networkidle둜 JavaScript μ™„λ£ŒκΉŒμ§€ λŒ€κΈ°)
try:
page.goto(url, wait_until="networkidle", timeout=45000)
print("νŽ˜μ΄μ§€ λ‘œλ”© μ™„λ£Œ")
except:
# networkidle μ‹€νŒ¨μ‹œ domcontentloaded둜 μž¬μ‹œλ„
page.goto(url, wait_until="domcontentloaded", timeout=30000)
print("κΈ°λ³Έ νŽ˜μ΄μ§€ λ‘œλ”© μ™„λ£Œ")
# 초기 λŒ€κΈ°
time.sleep(5)
# μΏ ν‚€ νŒμ—… 처리
try:
accept_button = page.locator('button:has-text("λͺ¨λ‘ 수락"), button:has-text("Accept all"), button:has-text("수락")')
if accept_button.count() > 0:
accept_button.first.click(timeout=3000)
print("μΏ ν‚€ νŒμ—… 처리됨")
time.sleep(3)
except Exception as e:
print(f"μΏ ν‚€ νŒμ—… 처리 μ‹€νŒ¨ (λ¬΄μ‹œ): {e}")
# νŠΈλ Œλ“œ 데이터 λ‘œλ”© λŒ€κΈ°
try:
# νŠΈλ Œλ“œ 컨텐츠가 λ‘œλ“œλ  λ•ŒκΉŒμ§€ λŒ€κΈ°
page.wait_for_selector("main, .trending-content, .trends-wrapper, [data-topic]", timeout=15000)
print("νŠΈλ Œλ“œ 컨텐츠 λ‘œλ”© 확인됨")
except:
print("νŠΈλ Œλ“œ μš”μ†Œλ₯Ό 찾을 수 μ—†μŒ, 계속 μ§„ν–‰")
# μΆ”κ°€ λŒ€κΈ° (동적 컨텐츠 μ™„μ „ λ‘œλ”©)
time.sleep(5)
# 핡심 μ˜μ—­ μŠ€ν¬λ¦°μƒ·
print("μŠ€ν¬λ¦°μƒ· 캑처 μ‹œμž‘...")
try:
main_content = page.locator('main, .main-content, [data-ve-type="main"], .trends-wrapper')
if main_content.count() > 0 and main_content.first.is_visible():
screenshot_bytes = main_content.first.screenshot(timeout=15000)
print("메인 컨텐츠 μŠ€ν¬λ¦°μƒ· 캑처됨")
else:
screenshot_bytes = page.screenshot(timeout=20000)
print("전체 νŽ˜μ΄μ§€ μŠ€ν¬λ¦°μƒ· 캑처됨")
except Exception as e:
print(f"μŠ€ν¬λ¦°μƒ· 캑처 μž¬μ‹œλ„: {e}")
screenshot_bytes = page.screenshot(timeout=20000)
print("μž¬μ‹œλ„ μŠ€ν¬λ¦°μƒ· 캑처 μ™„λ£Œ")
screenshot = Image.open(io.BytesIO(screenshot_bytes))
print("Google Trends μŠ€ν¬λ¦°μƒ· μ™„λ£Œ")
return screenshot
except Exception as e:
print(f"Google Trends μŠ€ν¬λ¦°μƒ· μ‹€νŒ¨: {str(e)}")
print(f"상세 였λ₯˜: {traceback.format_exc()}")
return None
finally:
# λ¦¬μ†ŒμŠ€ 정리
if page:
try:
page.close()
except:
pass
if browser:
try:
browser.close()
except:
pass
if playwright:
try:
playwright.stop()
except:
pass
def capture_ezme_trends() -> Optional[Image.Image]:
"""rank.ezme.net μŠ€ν¬λ¦°μƒ· 캑처 (독립 λΈŒλΌμš°μ € μΈμŠ€ν„΄μŠ€ μ‚¬μš©)"""
playwright = None
browser = None
page = None
try:
print("rank.ezme.net μŠ€ν¬λ¦°μƒ· μ‹œμž‘...")
# μŠ€λ ˆλ“œλ³„ 독립 λΈŒλΌμš°μ € 생성
playwright, browser = create_browser_for_thread()
if not browser:
print("λΈŒλΌμš°μ € μ΄ˆκΈ°ν™” μ‹€νŒ¨")
return None
page = browser.new_page()
page.set_viewport_size({"width": 1400, "height": 1200})
page.set_extra_http_headers({
"Accept-Language": "ko-KR,ko;q=0.9,en;q=0.8",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8"
})
url = "https://rank.ezme.net/diff/"
print(f"νŽ˜μ΄μ§€ 접속: {url}")
# νŽ˜μ΄μ§€ λ‘œλ”© (networkidle λŒ€κΈ°λ‘œ JavaScript λ‘œλ”© μ™„λ£ŒκΉŒμ§€ λŒ€κΈ°)
try:
page.goto(url, wait_until="networkidle", timeout=45000)
print("νŽ˜μ΄μ§€ λ‘œλ”© μ™„λ£Œ")
except:
# networkidle μ‹€νŒ¨μ‹œ domcontentloaded둜 μž¬μ‹œλ„
page.goto(url, wait_until="domcontentloaded", timeout=30000)
print("κΈ°λ³Έ νŽ˜μ΄μ§€ λ‘œλ”© μ™„λ£Œ")
# JavaScript μ‹€ν–‰ 및 데이터 λ‘œλ”© λŒ€κΈ°
print("동적 컨텐츠 λ‘œλ”© λŒ€κΈ°μ€‘...")
time.sleep(8) # 더 κΈ΄ λŒ€κΈ° μ‹œκ°„
# ν…Œμ΄λΈ” 데이터가 λ‘œλ“œλ˜μ—ˆλŠ”μ§€ 확인
try:
# μ‹œκ°„λ³„ 데이터 ν…Œμ΄λΈ” μš”μ†Œ λŒ€κΈ°
page.wait_for_selector("table, .table, td", timeout=10000)
print("ν…Œμ΄λΈ” 데이터 λ‘œλ”© 확인됨")
except:
print("ν…Œμ΄λΈ” μš”μ†Œλ₯Ό 찾을 수 μ—†μŒ, 계속 μ§„ν–‰")
# νŽ˜μ΄μ§€ μ€‘κ°„μœΌλ‘œ 슀크둀 (μ‹œκ°„λ³„ 검색어 데이터 μ˜μ—­)
try:
page.evaluate("window.scrollTo(0, document.body.scrollHeight * 0.4)")
time.sleep(3)
print("슀크둀 μ™„λ£Œ")
except Exception as e:
print(f"슀크둀 μ‹€νŒ¨: {e}")
# 전체 νŽ˜μ΄μ§€ μŠ€ν¬λ¦°μƒ·
print("μŠ€ν¬λ¦°μƒ· 캑처 μ‹œμž‘...")
screenshot_bytes = page.screenshot(full_page=True, timeout=20000)
screenshot = Image.open(io.BytesIO(screenshot_bytes))
print("rank.ezme.net μŠ€ν¬λ¦°μƒ· μ™„λ£Œ")
return screenshot
except Exception as e:
print(f"rank.ezme.net μŠ€ν¬λ¦°μƒ· μ‹€νŒ¨: {str(e)}")
print(f"상세 였λ₯˜: {traceback.format_exc()}")
return None
finally:
# λ¦¬μ†ŒμŠ€ 정리
if page:
try:
page.close()
except:
pass
if browser:
try:
browser.close()
except:
pass
if playwright:
try:
playwright.stop()
except:
pass
def analyze_google_trends_with_claude(image: Image.Image) -> str:
"""Google Trends 이미지 Claude 뢄석"""
if image is None:
return "Google Trends 뢄석할 이미지가 μ—†μŠ΅λ‹ˆλ‹€."
try:
client = get_claude_client()
buffer = io.BytesIO()
image.save(buffer, format="PNG")
image_base64 = base64.b64encode(buffer.getvalue()).decode()
prompt = """Google Trends μ‹€μ‹œκ°„ 인기 검색어 μŠ€ν¬λ¦°μƒ·μ„ λΆ„μ„ν•΄μ£Όμ„Έμš”.
λ‹€μŒ ν•­λͺ©λ“€μ„ μ€‘μ‹¬μœΌλ‘œ ν•œκ΅­μ–΄λ‘œ λΆ„μ„ν•΄μ£Όμ„Έμš”:
### A. μƒμœ„ 검색어 μˆœμœ„ (κ²€μƒ‰λŸ‰ κΈ°μ€€)
κ²€μƒ‰λŸ‰μ΄ 높은 μˆœμ„œλŒ€λ‘œ μ •λ ¬ν•˜μ—¬ λ‚˜μ—΄:
**1μœ„.** 검색어λͺ… (κ²€μƒ‰λŸ‰)
**2μœ„.** 검색어λͺ… (κ²€μƒ‰λŸ‰)
**3μœ„.** 검색어λͺ… (κ²€μƒ‰λŸ‰)
(이런 μ‹μœΌλ‘œ 10μœ„κΉŒμ§€)
### B. μ£Όμš” νŠΈλ Œλ“œ ν‚€μ›Œλ“œ
- 특히 μ£Όλͺ©ν•  λ§Œν•œ 검색어듀
- κ²€μƒ‰λŸ‰ 증가 νŒ¨ν„΄
### C. μΉ΄ν…Œκ³ λ¦¬λ³„ νŠΉμ§•
- μ—”ν„°ν…ŒμΈλ¨ΌνŠΈ, λ‰΄μŠ€, 슀포츠 λ“± 뢄야별 동ν–₯
### D. νŠΉμ΄μ‚¬ν•­ 및 뢄석
- κΈ‰μƒμŠΉ ν‚€μ›Œλ“œ
- μ‹œκΈ°μ /μ‚¬νšŒμ  λ°°κ²½ μΆ”μΈ‘
- 전체적인 관심사 νŒ¨ν„΄
**μ€‘μš” μ§€μ‹œμ‚¬ν•­:**
- λ§ˆν¬λ‹€μš΄ ν˜•μ‹μ„ μ‚¬μš©ν•˜μ—¬ **ꡡ은 글씨**, *κΈ°μšΈμž„*, `μ½”λ“œ` λ“±μœΌλ‘œ μ€‘μš”ν•œ λ‚΄μš©μ„ κ°•μ‘°ν•΄μ£Όμ„Έμš”
- 검색어λͺ…, 수치, μΉ΄ν…Œκ³ λ¦¬λͺ… λ“± 핡심 μ •λ³΄λŠ” **ꡡ은 글씨**둜 ν‘œμ‹œ
- κΈ‰μƒμŠΉλ₯ μ΄λ‚˜ νŠΉλ³„ν•œ μˆ˜μΉ˜λŠ” `λ°±ν‹±`으둜 κ°μ‹Έμ„œ κ°•μ‘°
- μƒμœ„ κ²€μƒ‰μ–΄λŠ” λ°˜λ“œμ‹œ κ²€μƒ‰λŸ‰ μˆœμ„œλŒ€λ‘œ 1μœ„λΆ€ν„° 10μœ„κΉŒμ§€ μ •λ ¬"""
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1200,
messages=[
{
"role": "user",
"content": [
{
"type": "text",
"text": prompt
},
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": image_base64
}
}
]
}
]
)
return response.content[0].text
except Exception as e:
return f"Google Trends Claude 뢄석 μ‹€νŒ¨: {str(e)}"
def analyze_ezme_trends_with_claude(image: Image.Image) -> str:
"""rank.ezme.net 이미지 Claude 뢄석"""
if image is None:
return "rank.ezme.net 뢄석할 이미지가 μ—†μŠ΅λ‹ˆλ‹€."
try:
client = get_claude_client()
buffer = io.BytesIO()
image.save(buffer, format="PNG")
image_base64 = base64.b64encode(buffer.getvalue()).decode()
prompt = """ν•œκ΅­ ν¬ν„Έμ‚¬μ΄νŠΈλ³„ μ‹œκ°„λ³„ μ‹€μ‹œκ°„ 검색어 데이터λ₯Ό λΆ„μ„ν•΄μ£Όμ„Έμš”.
λ‹€μŒ ν•­λͺ©λ“€μ„ μ€‘μ‹¬μœΌλ‘œ ν•œκ΅­μ–΄λ‘œ λΆ„μ„ν•΄μ£Όμ„Έμš”:
### A. 도메인별 μ£Όμš” 검색어
**Zum (쀌) μ£Όμš” 검색어:**
- μƒμœ„ 검색어와 μ§„μž… 횟수
- μ£Όμš” 관심 λΆ„μ•Ό
**Nate (λ„€μ΄νŠΈ) μ£Όμš” 검색어:**
- μƒμœ„ 검색어와 μ§„μž… 횟수
- μ£Όμš” 관심 λΆ„μ•Ό
**Google ν•œκ΅­ μ£Όμš” 검색어:**
- μƒμœ„ 검색어와 μ§„μž… 횟수
- μ£Όμš” 관심 λΆ„μ•Ό
### B. μ‹œκ°„λŒ€λ³„ νŠΈλ Œλ“œ λ³€ν™”
- 졜근 λͺ‡ μ‹œκ°„ λ™μ•ˆμ˜ λ³€ν™” νŒ¨ν„΄
- μ‹œκ°„λŒ€λ³„ μ£Όμš” 이슈 λ³€ν™”
### C. 검색어 μ§„μž… 횟수 뢄석
- κ°€μž₯ 높은 μ§„μž… 횟수λ₯Ό κΈ°λ‘ν•œ 검색어듀
- μ§€μ†μ μœΌλ‘œ μƒμœ„κΆŒμ„ μœ μ§€ν•˜λŠ” 검색어듀
### D. 포털별 관심사 차이점
- 각 ν¬ν„Έμ‚¬μ΄νŠΈλ³„ νŠΉμ„± 뢄석
- μ‚¬μš©μžμΈ΅λ³„ 관심사 차이
**μ€‘μš” μ§€μ‹œμ‚¬ν•­:**
- λ§ˆν¬λ‹€μš΄ ν˜•μ‹μ„ μ‚¬μš©ν•˜μ—¬ **ꡡ은 글씨**둜 검색어λͺ… κ°•μ‘°
- μ§„μž… νšŸμˆ˜λŠ” `λ°±ν‹±`으둜 κ°μ‹Έμ„œ ν‘œμ‹œ (예: `μ§„μž…νšŸμˆ˜: 15회`)
- μ‹œκ°„λ³„ 데이터가 μžˆλ‹€λ©΄ ꡬ체적으둜 μ–ΈκΈ‰
- 각 ν¬ν„Έμ˜ νŠΉμ§•μ„ λͺ…ν™•νžˆ κ΅¬λΆ„ν•˜μ—¬ 뢄석"""
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1200,
messages=[
{
"role": "user",
"content": [
{
"type": "text",
"text": prompt
},
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": image_base64
}
}
]
}
]
)
return response.content[0].text
except Exception as e:
return f"rank.ezme.net Claude 뢄석 μ‹€νŒ¨: {str(e)}"
def combine_analysis_results(google_analysis: str, ezme_analysis: str) -> str:
"""Google Trends와 ezme 뢄석 κ²°κ³Ό 톡합 (였λ₯˜ 상황 λŒ€μ‘)"""
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
# 성곡/μ‹€νŒ¨ μƒνƒœ 확인
google_failed = "뢄석할 이미지가 μ—†μŠ΅λ‹ˆλ‹€" in google_analysis or "μΊ‘μ²˜μ— μ‹€νŒ¨" in google_analysis
ezme_failed = "뢄석할 이미지가 μ—†μŠ΅λ‹ˆλ‹€" in ezme_analysis or "μΊ‘μ²˜μ— μ‹€νŒ¨" in ezme_analysis
# μƒνƒœ λ©”μ‹œμ§€ 생성
if google_failed and ezme_failed:
status_msg = "**λͺ¨λ“  데이터 μ†ŒμŠ€μ˜ μˆ˜μ§‘μ— μ‹€νŒ¨ν–ˆμŠ΅λ‹ˆλ‹€.** μž μ‹œ ν›„ λ‹€μ‹œ μ‹œλ„ν•΄μ£Όμ„Έμš”."
elif google_failed:
status_msg = "**Google Trends 데이터 μˆ˜μ§‘μ— μ‹€νŒ¨ν–ˆμŠ΅λ‹ˆλ‹€.** ν•œκ΅­ 포털 λ°μ΄ν„°λ§Œ μ œκ³΅λ©λ‹ˆλ‹€."
elif ezme_failed:
status_msg = "**ν•œκ΅­ 포털 데이터 μˆ˜μ§‘μ— μ‹€νŒ¨ν–ˆμŠ΅λ‹ˆλ‹€.** Google Trends λ°μ΄ν„°λ§Œ μ œκ³΅λ©λ‹ˆλ‹€."
else:
status_msg = "**λͺ¨λ“  데이터 μ†ŒμŠ€λ₯Ό μ„±κ³΅μ μœΌλ‘œ μˆ˜μ§‘ν–ˆμŠ΅λ‹ˆλ‹€.**"
combined_result = f"""# μ‹€μ‹œκ°„ 검색어 톡합 뢄석
**μ΅œμ’… μ—…λ°μ΄νŠΈ:** {timestamp}
{status_msg}
---
## Google Trends κΈ€λ‘œλ²Œ 동ν–₯
{google_analysis}
---
## ν•œκ΅­ 포털 μ‹€μ‹œκ°„ 검색어
{ezme_analysis}
"""
return combined_result
def perform_parallel_analysis() -> Tuple[bool, str]:
"""병렬 처리둜 톡합 뢄석 μˆ˜ν–‰ (κ°œμ„ λœ 였λ₯˜ 처리)"""
global analyzed_data, last_update_time, is_updating
is_updating = True
print(f"\n{'='*60}")
print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 톡합 μ‹€μ‹œκ°„ νŠΈλ Œλ“œ 뢄석 μ‹œμž‘")
print(f"{'='*60}")
try:
# λΈŒλΌμš°μ € μ„€μΉ˜ 확인
print("1. λΈŒλΌμš°μ € μ„€μΉ˜ 확인...")
install_browsers()
# 병렬 μŠ€ν¬λ¦°μƒ· 캑처
print("2. 병렬 μŠ€ν¬λ¦°μƒ· 캑처 μ‹œμž‘...")
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
print(" - Google Trends μŠ€ν¬λ¦°μƒ· μž‘μ—… μ‹œμž‘")
google_future = executor.submit(capture_google_trends)
print(" - rank.ezme.net μŠ€ν¬λ¦°μƒ· μž‘μ—… μ‹œμž‘")
ezme_future = executor.submit(capture_ezme_trends)
print(" - 두 μž‘μ—… μ™„λ£Œ λŒ€κΈ° 쀑...")
google_screenshot = google_future.result()
ezme_screenshot = ezme_future.result()
# μŠ€ν¬λ¦°μƒ· κ²°κ³Ό 확인
print("3. μŠ€ν¬λ¦°μƒ· κ²°κ³Ό 확인...")
google_success = google_screenshot is not None
ezme_success = ezme_screenshot is not None
print(f" - Google Trends μŠ€ν¬λ¦°μƒ·: {'성곡' if google_success else 'μ‹€νŒ¨'}")
print(f" - rank.ezme.net μŠ€ν¬λ¦°μƒ·: {'성곡' if ezme_success else 'μ‹€νŒ¨'}")
if not google_success and not ezme_success:
error_msg = "λͺ¨λ“  μŠ€ν¬λ¦°μƒ· μΊ‘μ²˜μ— μ‹€νŒ¨ν–ˆμŠ΅λ‹ˆλ‹€."
print(f" ERROR: {error_msg}")
return False, error_msg
print("4. Claude 뢄석 μ‹œμž‘...")
# Claude 뢄석 (순차 처리 - API μ œν•œ)
if google_success:
print(" - Google Trends 뢄석 쀑...")
google_analysis = analyze_google_trends_with_claude(google_screenshot)
print(" - Google Trends 뢄석 μ™„λ£Œ")
else:
google_analysis = "Google Trends μŠ€ν¬λ¦°μƒ· μΊ‘μ²˜μ— μ‹€νŒ¨ν•˜μ—¬ 뢄석할 수 μ—†μŠ΅λ‹ˆλ‹€."
if ezme_success:
print(" - rank.ezme.net 뢄석 쀑...")
ezme_analysis = analyze_ezme_trends_with_claude(ezme_screenshot)
print(" - rank.ezme.net 뢄석 μ™„λ£Œ")
else:
ezme_analysis = "rank.ezme.net μŠ€ν¬λ¦°μƒ· μΊ‘μ²˜μ— μ‹€νŒ¨ν•˜μ—¬ 뢄석할 수 μ—†μŠ΅λ‹ˆλ‹€."
print("5. 톡합 뢄석 κ²°κ³Ό 생성...")
# 톡합 뢄석 κ²°κ³Ό 생성
combined_analysis = combine_analysis_results(google_analysis, ezme_analysis)
# κ²°κ³Ό μ €μž₯
analyzed_data["ν†΅ν•©μ‹€μ‹œκ°„"] = {
'analysis': combined_analysis,
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'google_analysis': google_analysis,
'ezme_analysis': ezme_analysis,
'google_success': google_success,
'ezme_success': ezme_success
}
last_update_time = datetime.now()
success_count = sum([google_success, ezme_success])
print(f"톡합 뢄석 μ™„λ£Œ - {last_update_time.strftime('%Y-%m-%d %H:%M:%S')}")
print(f"성곡λ₯ : {success_count}/2 μ†ŒμŠ€")
return True, combined_analysis
except Exception as e:
error_msg = f"톡합 뢄석 쀑 μ˜ˆμƒμΉ˜ λͺ»ν•œ 였λ₯˜: {str(e)}"
print(f"ERROR: {error_msg}")
print(f"상세 였λ₯˜: {traceback.format_exc()}")
return False, error_msg
finally:
is_updating = False
print(f"{'='*60}\n")
def auto_update_scheduler():
"""1μ‹œκ°„λ§ˆλ‹€ μžλ™ μ—…λ°μ΄νŠΈ"""
while True:
try:
print(f"[μŠ€μΌ€μ€„λŸ¬] λ‹€μŒ μ—…λ°μ΄νŠΈκΉŒμ§€ 1μ‹œκ°„ λŒ€κΈ° 쀑...")
time.sleep(3600) # 1μ‹œκ°„ λŒ€κΈ°
if not is_updating:
print(f"[μŠ€μΌ€μ€„λŸ¬] μžλ™ μ—…λ°μ΄νŠΈ μ‹œμž‘")
perform_parallel_analysis()
else:
print(f"[μŠ€μΌ€μ€„λŸ¬] 이미 μ—…λ°μ΄νŠΈ μ€‘μ΄λ―€λ‘œ κ±΄λ„ˆλœ€")
except Exception as e:
print(f"[μŠ€μΌ€μ€„λŸ¬] 였λ₯˜: {e}")
time.sleep(300) # 5λΆ„ ν›„ μž¬μ‹œλ„
def get_cached_analysis_result() -> tuple:
"""μ €μž₯된 톡합 뢄석 κ²°κ³Ό λ°˜ν™˜"""
global analyzed_data, last_update_time
if "ν†΅ν•©μ‹€μ‹œκ°„" in analyzed_data:
data = analyzed_data["ν†΅ν•©μ‹€μ‹œκ°„"]
status = f"μ΅œμ‹  데이터 (μ—…λ°μ΄νŠΈ: {data['timestamp']})"
return data['analysis'], status
else:
if last_update_time:
status = f"데이터 μ—†μŒ (λ§ˆμ§€λ§‰ μ—…λ°μ΄νŠΈ: {last_update_time.strftime('%H:%M:%S')})"
else:
status = "아직 뢄석 데이터가 μ€€λΉ„λ˜μ§€ μ•Šμ•˜μŠ΅λ‹ˆλ‹€."
return "**뢄석 데이터가 아직 μ€€λΉ„λ˜μ§€ μ•Šμ•˜μŠ΅λ‹ˆλ‹€.**\n\nμž μ‹œ ν›„ λ‹€μ‹œ μ‹œλ„ν•΄μ£Όμ„Έμš”.", status
def create_interface():
"""Gradio μΈν„°νŽ˜μ΄μŠ€ 생성"""
def show_cached_analysis():
"""μ €μž₯된 톡합 뢄석 κ²°κ³Ό ν‘œμ‹œ"""
if "ν†΅ν•©μ‹€μ‹œκ°„" in analyzed_data:
data = analyzed_data["ν†΅ν•©μ‹€μ‹œκ°„"]
status = f"μΊμ‹œλœ 톡합 뢄석 κ²°κ³Ό (μ—…λ°μ΄νŠΈ: {data['timestamp']})"
return data['analysis'], status
else:
if last_update_time:
status = f"μ €μž₯된 데이터 μ—†μŒ (λ§ˆμ§€λ§‰ μ—…λ°μ΄νŠΈ: {last_update_time.strftime('%H:%M:%S')})"
else:
status = "아직 뢄석 데이터가 μ€€λΉ„λ˜μ§€ μ•Šμ•˜μŠ΅λ‹ˆλ‹€. μž μ‹œ ν›„ λ‹€μ‹œ μ‹œλ„ν•΄μ£Όμ„Έμš”."
return "**μ €μž₯된 톡합 뢄석 데이터가 μ—†μŠ΅λ‹ˆλ‹€.**\n\n'μ‹€μ‹œκ°„ μ—…λ°μ΄νŠΈ' λ²„νŠΌμ„ λˆŒλŸ¬μ£Όμ„Έμš”.", status
def perform_live_update():
"""μ‹€μ‹œκ°„ μ—…λ°μ΄νŠΈ μˆ˜ν–‰"""
if is_updating:
return "**ν˜„μž¬ μ—…λ°μ΄νŠΈκ°€ μ§„ν–‰ μ€‘μž…λ‹ˆλ‹€.**\n\nμž μ‹œ ν›„ λ‹€μ‹œ μ‹œλ„ν•΄μ£Όμ„Έμš”.", "μ—…λ°μ΄νŠΈ μ§„ν–‰ 쀑..."
print("μ‹€μ‹œκ°„ 톡합 μ—…λ°μ΄νŠΈ μš”μ²­λ¨")
success, result = perform_parallel_analysis()
if success and "ν†΅ν•©μ‹€μ‹œκ°„" in analyzed_data:
data = analyzed_data["ν†΅ν•©μ‹€μ‹œκ°„"]
return data['analysis'], f"μ‹€μ‹œκ°„ μ—…λ°μ΄νŠΈ μ™„λ£Œ - {datetime.now().strftime('%H:%M:%S')}"
else:
return "**μ—…λ°μ΄νŠΈμ— μ‹€νŒ¨ν–ˆμŠ΅λ‹ˆλ‹€.**\n\nμž μ‹œ ν›„ λ‹€μ‹œ μ‹œλ„ν•΄μ£Όμ„Έμš”.", f"μ—…λ°μ΄νŠΈ μ‹€νŒ¨ - {datetime.now().strftime('%H:%M:%S')}"
# Gradio μΈν„°νŽ˜μ΄μŠ€ ꡬ성
with gr.Blocks(
title="톡합 μ‹€μ‹œκ°„ 검색어 뢄석",
theme=gr.themes.Soft(),
css="""
@import url('https://fonts.googleapis.com/css2?family=Noto+Sans+KR:wght@300;400;500;700&display=swap');
* {
font-family: 'Noto Sans KR', sans-serif !important;
}
.markdown-content {
font-size: 14px;
line-height: 1.6;
background: #f8f9fa;
padding: 20px;
border-radius: 8px;
border: 1px solid #e9ecef;
}
.markdown-content h1 {
color: #1d4ed8;
margin-bottom: 15px;
}
.markdown-content h2 {
color: #2563eb;
margin-top: 25px;
margin-bottom: 12px;
}
.markdown-content h3 {
color: #3b82f6;
margin-top: 20px;
margin-bottom: 10px;
}
.markdown-content strong {
color: #1d4ed8;
font-weight: 600;
}
"""
) as interface:
gr.Markdown("# 톡합 μ‹€μ‹œκ°„ 검색어 뢄석")
gr.Markdown("Google Trends와 ν•œκ΅­ ν¬ν„Έμ‚¬μ΄νŠΈ 검색어λ₯Ό 톡합 λΆ„μ„ν•©λ‹ˆλ‹€.")
with gr.Row():
with gr.Column(scale=1):
gr.Markdown("### 뢄석 μ˜΅μ…˜")
with gr.Column():
cached_btn = gr.Button("톡합 검색어 뢄석", variant="primary", size="lg")
gr.Markdown("μ €μž₯된 톡합 뢄석 κ²°κ³Όλ₯Ό μ¦‰μ‹œ ν‘œμ‹œν•©λ‹ˆλ‹€.")
gr.Markdown("---")
with gr.Column():
live_btn = gr.Button("μ‹€μ‹œκ°„ μ—…λ°μ΄νŠΈ", variant="secondary", size="lg")
gr.Markdown("Google Trends + ν•œκ΅­ 포털 μ΅œμ‹  λ°μ΄ν„°λ‘œ μ—…λ°μ΄νŠΈν•©λ‹ˆλ‹€. (μ•½ 1λΆ„ μ†Œμš”)")
gr.Markdown("---")
status_output = gr.Textbox(
label="μƒνƒœ",
value="뢄석할 μ˜΅μ…˜μ„ μ„ νƒν•΄μ£Όμ„Έμš”.",
interactive=False
)
# Apeach 이미지 ν‘œμ‹œ
try:
gr.Image(
value="apeach.png",
label=None,
show_label=False,
interactive=False,
height=200,
width=200
)
except:
pass
with gr.Column(scale=2):
analysis_output = gr.Markdown(
value="**뢄석 λ²„νŠΌμ„ λˆŒλŸ¬μ£Όμ„Έμš”.**\n\n- **톡합 검색어 뢄석**: μ €μž₯된 κ²°κ³Ό μ¦‰μ‹œ 확인\n- **μ‹€μ‹œκ°„ μ—…λ°μ΄νŠΈ**: μ΅œμ‹  λ°μ΄ν„°λ‘œ μƒˆλ‘œ 뢄석",
elem_classes=["markdown-content"],
height=700
)
# 이벀트 μ—°κ²°
cached_btn.click(
fn=show_cached_analysis,
outputs=[analysis_output, status_output]
)
live_btn.click(
fn=perform_live_update,
outputs=[analysis_output, status_output]
)
return interface
def main():
"""메인 ν•¨μˆ˜"""
print("=" * 60)
print("톡합 μ‹€μ‹œκ°„ 검색어 뢄석 μ„œλΉ„μŠ€ μ‹œμž‘!")
print("=" * 60)
# API ν‚€ 확인
api_key = os.getenv("ANTHROPIC_API_KEY")
if not api_key:
print("ANTHROPIC_API_KEY ν™˜κ²½λ³€μˆ˜κ°€ μ„€μ •λ˜μ§€ μ•Šμ•˜μŠ΅λ‹ˆλ‹€.")
print(" .env νŒŒμΌμ— ANTHROPIC_API_KEY=your_key_here λ₯Ό μΆ”κ°€ν•˜μ„Έμš”.")
return
else:
print("Claude API ν‚€ 확인됨")
# 초기 톡합 뢄석 μˆ˜ν–‰
print("초기 톡합 뢄석 μ‹œμž‘...")
initial_success, _ = perform_parallel_analysis()
if initial_success:
print("초기 톡합 뢄석 μ™„λ£Œ")
else:
print("초기 뢄석 μ‹€νŒ¨, μ„œλΉ„μŠ€λŠ” 계속 μ§„ν–‰λ©λ‹ˆλ‹€.")
# μžλ™ μ—…λ°μ΄νŠΈ μŠ€μΌ€μ€„λŸ¬ μ‹œμž‘
print("μžλ™ μ—…λ°μ΄νŠΈ μŠ€μΌ€μ€„λŸ¬ μ‹œμž‘ (1μ‹œκ°„ 간격)")
global update_thread
update_thread = threading.Thread(target=auto_update_scheduler, daemon=True)
update_thread.start()
print("λͺ¨λ“  μ€€λΉ„ μ™„λ£Œ!")
print("=" * 60)
# Gradio μΈν„°νŽ˜μ΄μŠ€ μ‹€ν–‰
try:
interface = create_interface()
interface.launch(
server_name="0.0.0.0",
server_port=7860,
share=False,
show_error=True,
quiet=False
)
except Exception as e:
print(f"μΈν„°νŽ˜μ΄μŠ€ μ‹œμž‘ μ‹€νŒ¨: {e}")
print(f"상세 였λ₯˜: {traceback.format_exc()}")
if __name__ == "__main__":
main()