import gradio as gr
import pandas as pd
import os
import time
import threading
import tempfile
import logging
import random
import uuid
import shutil
import glob
from datetime import datetime
from gradio_client import Client
# 로깅 설정
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(),
logging.FileHandler('control_tower_app.log', mode='a')
]
)
logger = logging.getLogger(__name__)
# API 클라이언트 설정
def get_client():
# 환경변수에서 API 엔드포인트 읽기
endpoint = os.environ.get('API_ENDPOINT', '')
# 환경변수에 파일 내용이나 불필요한 텍스트가 들어온 경우 정리
if endpoint:
# 줄바꿈으로 분리해서 첫 번째 유효한 라인 찾기
lines = endpoint.split('\n')
for line in lines:
line = line.strip()
# 주석이나 비어있는 라인 제외
if line and not line.startswith('#') and '/' in line:
# API_ENDPOINT= 같은 키값 제거
if '=' in line:
line = line.split('=', 1)[1].strip()
# 따옴표 제거
line = line.strip('"\'')
if line and '/' in line and len(line) < 50:
return Client(line)
raise ValueError("올바른 API_ENDPOINT를 설정해주세요 (예: username/repo-name)")
# 세션별 임시 파일 관리를 위한 딕셔너리
session_temp_files = {}
session_data = {}
def cleanup_huggingface_temp_folders():
"""허깅페이스 임시 폴더 초기 정리"""
try:
temp_dirs = [
tempfile.gettempdir(),
"/tmp",
"/var/tmp",
os.path.join(os.getcwd(), "temp"),
os.path.join(os.getcwd(), "tmp"),
"/gradio_cached_examples",
"/flagged"
]
cleanup_count = 0
for temp_dir in temp_dirs:
if os.path.exists(temp_dir):
try:
session_files = glob.glob(os.path.join(temp_dir, "session_*.xlsx"))
session_files.extend(glob.glob(os.path.join(temp_dir, "session_*.csv")))
session_files.extend(glob.glob(os.path.join(temp_dir, "*keyword*.xlsx")))
session_files.extend(glob.glob(os.path.join(temp_dir, "*keyword*.csv")))
session_files.extend(glob.glob(os.path.join(temp_dir, "tmp*.xlsx")))
session_files.extend(glob.glob(os.path.join(temp_dir, "tmp*.csv")))
for file_path in session_files:
try:
if os.path.getmtime(file_path) < time.time() - 3600:
os.remove(file_path)
cleanup_count += 1
logger.info(f"초기 정리: 오래된 임시 파일 삭제 - {file_path}")
except Exception as e:
logger.warning(f"파일 삭제 실패 (무시됨): {file_path} - {e}")
except Exception as e:
logger.warning(f"임시 디렉토리 정리 실패 (무시됨): {temp_dir} - {e}")
logger.info(f"✅ 허깅페이스 임시 폴더 초기 정리 완료 - {cleanup_count}개 파일 삭제")
try:
gradio_temp_dir = os.path.join(os.getcwd(), "gradio_cached_examples")
if os.path.exists(gradio_temp_dir):
shutil.rmtree(gradio_temp_dir, ignore_errors=True)
logger.info("Gradio 캐시 폴더 정리 완료")
except Exception as e:
logger.warning(f"Gradio 캐시 폴더 정리 실패 (무시됨): {e}")
except Exception as e:
logger.error(f"초기 임시 폴더 정리 중 오류 (계속 진행): {e}")
def setup_clean_temp_environment():
"""깨끗한 임시 환경 설정"""
try:
cleanup_huggingface_temp_folders()
app_temp_dir = os.path.join(tempfile.gettempdir(), "keyword_app")
if os.path.exists(app_temp_dir):
shutil.rmtree(app_temp_dir, ignore_errors=True)
os.makedirs(app_temp_dir, exist_ok=True)
os.environ['KEYWORD_APP_TEMP'] = app_temp_dir
logger.info(f"✅ 애플리케이션 전용 임시 디렉토리 설정: {app_temp_dir}")
return app_temp_dir
except Exception as e:
logger.error(f"임시 환경 설정 실패: {e}")
return tempfile.gettempdir()
def get_app_temp_dir():
"""애플리케이션 전용 임시 디렉토리 반환"""
return os.environ.get('KEYWORD_APP_TEMP', tempfile.gettempdir())
def get_session_id():
"""세션 ID 생성"""
return str(uuid.uuid4())
def cleanup_session_files(session_id, delay=300):
"""세션별 임시 파일 정리 함수"""
def cleanup():
time.sleep(delay)
if session_id in session_temp_files:
files_to_remove = session_temp_files[session_id].copy()
del session_temp_files[session_id]
for file_path in files_to_remove:
try:
if os.path.exists(file_path):
os.remove(file_path)
logger.info(f"세션 {session_id[:8]}... 임시 파일 삭제: {file_path}")
except Exception as e:
logger.error(f"세션 {session_id[:8]}... 파일 삭제 오류: {e}")
threading.Thread(target=cleanup, daemon=True).start()
def register_session_file(session_id, file_path):
"""세션별 파일 등록"""
if session_id not in session_temp_files:
session_temp_files[session_id] = []
session_temp_files[session_id].append(file_path)
def cleanup_old_sessions():
"""오래된 세션 데이터 정리"""
current_time = time.time()
sessions_to_remove = []
for session_id, data in session_data.items():
if current_time - data.get('last_activity', 0) > 3600:
sessions_to_remove.append(session_id)
for session_id in sessions_to_remove:
if session_id in session_temp_files:
for file_path in session_temp_files[session_id]:
try:
if os.path.exists(file_path):
os.remove(file_path)
logger.info(f"오래된 세션 {session_id[:8]}... 파일 삭제: {file_path}")
except Exception as e:
logger.error(f"오래된 세션 파일 삭제 오류: {e}")
del session_temp_files[session_id]
if session_id in session_data:
del session_data[session_id]
logger.info(f"오래된 세션 데이터 삭제: {session_id[:8]}...")
def update_session_activity(session_id):
"""세션 활동 시간 업데이트"""
if session_id not in session_data:
session_data[session_id] = {}
session_data[session_id]['last_activity'] = time.time()
def create_session_temp_file(session_id, suffix='.xlsx'):
"""세션별 임시 파일 생성 (전용 디렉토리 사용)"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
random_suffix = str(random.randint(1000, 9999))
temp_dir = get_app_temp_dir()
filename = f"session_{session_id[:8]}_{timestamp}_{random_suffix}{suffix}"
temp_file_path = os.path.join(temp_dir, filename)
with open(temp_file_path, 'w') as f:
pass
register_session_file(session_id, temp_file_path)
return temp_file_path
def wrapper_modified(keyword, korean_only, apply_main_keyword_option, exclude_zero_volume, session_id):
"""키워드 검색 및 처리 래퍼 함수 (API 사용)"""
update_session_activity(session_id)
try:
client = get_client()
result = client.predict(
keyword=keyword,
korean_only=korean_only,
apply_main_keyword=apply_main_keyword_option,
exclude_zero_volume=exclude_zero_volume,
api_name="/process_search_results"
)
# API 응답 확인 및 처리
logger.info(f"API 응답 타입: {type(result)}, 길이: {len(result) if isinstance(result, (list, tuple)) else 'N/A'}")
if isinstance(result, (list, tuple)) and len(result) >= 5:
table_html, cat_choices, vol_choices, selected_cat, download_file = result[:5]
else:
# 응답이 예상과 다른 경우 기본값 사용
logger.warning(f"예상과 다른 API 응답: {result}")
table_html = "