import gradio as gr
import pandas as pd
import os
import logging
from datetime import datetime
import pytz
import time
import tempfile
import zipfile
import re
import json
# 로깅 설정 - 클라이언트 정보 숨김
logging.basicConfig(level=logging.WARNING, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# 외부 라이브러리 로그 비활성화
logging.getLogger('gradio').setLevel(logging.WARNING)
logging.getLogger('gradio_client').setLevel(logging.WARNING)
logging.getLogger('httpx').setLevel(logging.WARNING)
logging.getLogger('urllib3').setLevel(logging.WARNING)
# ===== API 클라이언트 설정 =====
def get_api_client():
"""환경변수에서 API 엔드포인트를 가져와 클라이언트 생성"""
try:
from gradio_client import Client
# 환경변수에서 API 엔드포인트 가져오기
api_endpoint = os.getenv('API_ENDPOINT')
if not api_endpoint:
logger.error("API 엔드포인트가 설정되지 않았습니다.")
raise ValueError("API 엔드포인트가 설정되지 않았습니다.")
client = Client(api_endpoint)
logger.info("원격 API 클라이언트 초기화 성공")
return client
except Exception as e:
logger.error(f"API 클라이언트 초기화 실패: {e}")
return None
# ===== 한국시간 관련 함수 =====
def get_korean_time():
"""한국시간 반환"""
try:
korea_tz = pytz.timezone('Asia/Seoul')
return datetime.now(korea_tz)
except:
return datetime.now()
def format_korean_datetime(dt=None, format_type="filename"):
"""한국시간 포맷팅"""
if dt is None:
dt = get_korean_time()
if format_type == "filename":
return dt.strftime("%y%m%d_%H%M")
elif format_type == "display":
return dt.strftime('%Y년 %m월 %d일 %H시 %M분')
elif format_type == "full":
return dt.strftime('%Y-%m-%d %H:%M:%S')
else:
return dt.strftime("%y%m%d_%H%M")
# ===== 로딩 애니메이션 =====
def create_loading_animation():
"""로딩 애니메이션 HTML"""
return """
분석 중입니다...
원격 서버에서 데이터를 수집하고 AI가 분석하고 있습니다. 잠시만 기다려주세요.
"""
# ===== 에러 처리 함수 =====
def generate_error_response(error_message):
"""에러 응답 생성"""
return f'''
❌ 연결 오류
{error_message}
해결 방법:
네트워크 연결을 확인해주세요
원격 서버 상태를 확인해주세요
잠시 후 다시 시도해주세요
문제가 지속되면 관리자에게 문의하세요
'''
# ===== 파일 출력 함수들 =====
def create_timestamp_filename(analysis_keyword):
"""타임스탬프가 포함된 파일명 생성 - 한국시간 적용"""
timestamp = format_korean_datetime(format_type="filename")
safe_keyword = re.sub(r'[^\w\s-]', '', analysis_keyword).strip()
safe_keyword = re.sub(r'[-\s]+', '_', safe_keyword)
return f"{safe_keyword}_{timestamp}_분석결과"
def export_to_html(analysis_html, filename_base):
"""HTML 파일로 출력 - 한국시간 적용"""
try:
html_filename = f"{filename_base}.html"
html_path = os.path.join(tempfile.gettempdir(), html_filename)
# 한국시간으로 생성 시간 표시
korean_time = format_korean_datetime(format_type="display")
# 완전한 HTML 문서 생성
full_html = f"""
키워드 심층분석 결과
키워드 심층분석 결과
AI 상품 소싱 분석 시스템 v2.10
{analysis_html}
생성 시간: {korean_time} (한국시간)
"""
with open(html_path, 'w', encoding='utf-8') as f:
f.write(full_html)
logger.info(f"HTML 파일 생성 완료: {html_path}")
return html_path
except Exception as e:
logger.error(f"HTML 파일 생성 오류: {e}")
return None
def create_zip_file(html_path, filename_base):
"""압축 파일 생성 (HTML만)"""
try:
zip_filename = f"{filename_base}.zip"
zip_path = os.path.join(tempfile.gettempdir(), zip_filename)
with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
if html_path and os.path.exists(html_path):
zipf.write(html_path, f"{filename_base}.html")
logger.info(f"HTML 파일 압축 추가: {filename_base}.html")
logger.info(f"압축 파일 생성 완료: {zip_path}")
return zip_path
except Exception as e:
logger.error(f"압축 파일 생성 오류: {e}")
return None
def export_analysis_results(export_data):
"""분석 결과 출력 메인 함수 - 세션별 데이터 처리"""
try:
# 출력할 데이터 확인
if not export_data or not isinstance(export_data, dict):
return None, "분석 데이터가 없습니다. 먼저 키워드 심층분석을 실행해주세요."
analysis_keyword = export_data.get("analysis_keyword", "")
analysis_html = export_data.get("analysis_html", "")
if not analysis_keyword:
return None, "분석할 키워드가 설정되지 않았습니다. 먼저 키워드 분석을 실행해주세요."
if not analysis_html:
return None, "분석 결과가 없습니다. 먼저 키워드 심층분석을 실행해주세요."
# 파일명 생성 (한국시간 적용)
filename_base = create_timestamp_filename(analysis_keyword)
logger.info(f"출력 파일명: {filename_base}")
# HTML 파일 생성
html_path = export_to_html(analysis_html, filename_base)
# 압축 파일 생성
if html_path:
zip_path = create_zip_file(html_path, filename_base)
if zip_path:
return zip_path, f"✅ 분석 결과가 성공적으로 출력되었습니다!\n파일명: {filename_base}.zip"
else:
return None, "압축 파일 생성에 실패했습니다."
else:
return None, "출력할 파일이 없습니다."
except Exception as e:
logger.error(f"분석 결과 출력 오류: {e}")
return None, f"출력 중 오류가 발생했습니다: {str(e)}"
# ===== 원격 API 호출 함수들 =====
def call_analyze_keyword_api(analysis_keyword):
"""키워드 심층분석 API 호출"""
try:
client = get_api_client()
if not client:
return generate_error_response("API 클라이언트를 초기화할 수 없습니다."), {}
logger.info("원격 API 호출: 키워드 심층분석")
result = client.predict(
analysis_keyword=analysis_keyword,
api_name="/on_analyze_keyword"
)
logger.info(f"키워드 분석 API 결과 타입: {type(result)}")
# 분석 결과로 export 데이터 생성
if isinstance(result, str) and len(result) > 100:
export_data = {
"analysis_keyword": analysis_keyword,
"analysis_html": result,
"analysis_completed": True,
"created_at": get_korean_time().isoformat()
}
return result, export_data
else:
return str(result), {}
except Exception as e:
logger.error(f"키워드 심층분석 API 호출 오류: {e}")
return generate_error_response(f"원격 서버 연결 실패: {str(e)}"), {}
def call_export_results_api(export_data):
"""분석 결과 출력 API 호출"""
try:
client = get_api_client()
if not client:
return None, "API 클라이언트를 초기화할 수 없습니다."
logger.info("원격 API 호출: 분석 결과 출력")
result = client.predict(
api_name="/on_export_results"
)
logger.info(f"출력 API 결과 타입: {type(result)}")
# 결과가 튜플인 경우 첫 번째 요소는 메시지, 두 번째는 파일
if isinstance(result, tuple) and len(result) == 2:
message, file_path = result
if file_path:
return file_path, message
else:
return None, message
else:
return None, str(result)
except Exception as e:
logger.error(f"분석 결과 출력 API 호출 오류: {e}")
return None, f"원격 서버 연결 실패: {str(e)}"
# ===== 그라디오 인터페이스 =====
def create_interface():
# CSS 파일 로드
try:
with open('style.css', 'r', encoding='utf-8') as f:
custom_css = f.read()
except:
custom_css = """
:root { --primary-color: #FB7F0D; --secondary-color: #ff9a8b; }
.custom-button {
background: linear-gradient(135deg, var(--primary-color), var(--secondary-color)) !important;
color: white !important; border-radius: 30px !important; height: 45px !important;
font-size: 16px !important; font-weight: bold !important; width: 100% !important;
}
.export-button {
background: linear-gradient(135deg, #28a745, #20c997) !important;
color: white !important; border-radius: 25px !important; height: 50px !important;
font-size: 17px !important; font-weight: bold !important; width: 100% !important;
margin-top: 20px !important;
}
.custom-frame {
background-color: white !important;
border: 1px solid #e5e5e5 !important;
border-radius: 18px;
padding: 20px;
margin: 10px 0;
box-shadow: 0 8px 30px rgba(251, 127, 13, 0.08) !important;
}
.section-title {
display: flex;
align-items: center;
font-size: 20px;
font-weight: 700;
color: #334155 !important;
margin-bottom: 10px;
padding-bottom: 5px;
border-bottom: 2px solid var(--primary-color);
font-family: 'Pretendard', 'Noto Sans KR', -apple-system, BlinkMacSystemFont, sans-serif;
}
.section-title img, .section-title i {
margin-right: 10px;
font-size: 20px;
color: var(--primary-color);
}
"""
with gr.Blocks(
css=custom_css,
title="🛒 AI 상품 소싱 분석기 v2.10",
theme=gr.themes.Default(primary_hue="orange", secondary_hue="orange")
) as interface:
# 폰트 및 아이콘 로드
gr.HTML("""
""")
# 세션별 상태 관리 (멀티 사용자 안전)
export_data_state = gr.State({})
# === 키워드 심층분석 입력 ===
with gr.Column(elem_classes="custom-frame fade-in"):
gr.HTML('
')
analysis_result = gr.HTML(label="키워드 심층분석")
# === 결과 출력 섹션 ===
with gr.Column(elem_classes="custom-frame fade-in"):
gr.HTML('
분석 결과 출력
')
export_btn = gr.Button("📊 분석결과 출력하기", elem_classes="export-button", size="lg")
export_result = gr.HTML()
download_file = gr.File(label="다운로드", visible=False)
# ===== 이벤트 핸들러 =====
def on_analyze_keyword(analysis_keyword):
if not analysis_keyword.strip():
return "분석할 키워드를 입력해주세요.", {}
# 로딩 상태 표시
yield create_loading_animation(), {}
# 실제 키워드 분석 실행
keyword_result, session_export_data = call_analyze_keyword_api(analysis_keyword)
# 📈 검색량 트렌드 분석과 🎯 키워드 분석 표시
yield keyword_result, session_export_data
def on_export_results(export_data):
"""분석 결과 출력 핸들러 - 세션별 데이터 처리"""
try:
# 로컬 출력 시도
zip_path, message = export_analysis_results(export_data)
if zip_path:
# 성공 메시지와 함께 다운로드 파일 제공
success_html = f"""
출력 완료!
{message} 포함 파일:
• 🌐 HTML 파일: 키워드 심층분석 결과 (그래프 포함)
아래 다운로드 버튼을 클릭하여 파일을 저장하세요.
⏰ 한국시간 기준으로 파일명이 생성됩니다.
"""
return success_html, gr.update(value=zip_path, visible=True)
else:
# 로컬 출력 실패시 원격 API 시도
try:
remote_file, remote_message = call_export_results_api(export_data)
if remote_file:
success_html = f"""
"""
return error_html, gr.update(visible=False)
# ===== 이벤트 연결 =====
analyze_keyword_btn.click(
fn=on_analyze_keyword,
inputs=[analysis_keyword_input],
outputs=[analysis_result, export_data_state]
)
export_btn.click(
fn=on_export_results,
inputs=[export_data_state],
outputs=[export_result, download_file]
)
return interface
# ===== 메인 실행 =====
if __name__ == "__main__":
# pytz 모듈 설치 확인
try:
import pytz
logger.info("✅ pytz 모듈 로드 성공 - 한국시간 지원")
except ImportError:
logger.warning("⚠️ pytz 모듈이 설치되지 않음 - pip install pytz 실행 필요")
logger.info("시스템 시간을 사용합니다.")
logger.info("===== 상품 소싱 분석 시스템 v2.10 (컨트롤 타워 버전) 시작 =====")
# 필요한 패키지 안내
print("📦 필요한 패키지:")
print(" pip install gradio gradio_client pandas pytz")
print()
# API 엔드포인트 설정 안내
api_endpoint = os.getenv('API_ENDPOINT')
if not api_endpoint:
print("⚠️ API_ENDPOINT 환경변수를 설정하세요.")
print(" export API_ENDPOINT='your-endpoint-url'")
print()
else:
print("✅ API 엔드포인트 설정 완료!")
print()
print("🚀 v2.10 컨트롤 타워 버전 특징:")
print(" • 허깅페이스 그라디오 엔드포인트 활용")
print(" • 완전히 동일한 UI와 기능 구현")
print(" • 📈 검색량 트렌드 분석과 🎯 키워드 분석 표시")
print(" • ✅ 출력 기능: HTML 파일 생성 및 ZIP 다운로드")
print(" • ✅ 한국시간 기준 파일명 생성")
print(" • ✅ 멀티 사용자 안전: gr.State로 세션별 데이터 관리")
print(" • 🔒 클라이언트 정보 환경변수로 완전 숨김")
print(" • 원격 서버와 로컬 처리 하이브리드 방식")
print()
# 앱 실행
app = create_interface()
app.launch(server_name="0.0.0.0", server_port=7860, share=True)