import os import re import random from http import HTTPStatus from typing import Dict, List, Optional, Tuple import base64 import anthropic import openai import asyncio import time from functools import partial import json import gradio as gr import modelscope_studio.components.base as ms import modelscope_studio.components.legacy as legacy import modelscope_studio.components.antd as antd import html import urllib.parse from huggingface_hub import HfApi, create_repo, hf_hub_download import string import requests from selenium import webdriver from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.common.by import By from selenium.common.exceptions import WebDriverException, TimeoutException from PIL import Image from io import BytesIO from datetime import datetime import spaces from safetensors.torch import load_file from diffusers import FluxPipeline import torch from os import path # 이 줄을 추가 from datetime import datetime, timedelta from requests.adapters import HTTPAdapter from requests.packages.urllib3.util.retry import Retry # 캐시 경로 설정 cache_path = path.join(path.dirname(path.abspath(__file__)), "models") os.environ["TRANSFORMERS_CACHE"] = cache_path os.environ["HF_HUB_CACHE"] = cache_path os.environ["HF_HOME"] = cache_path # Hugging Face 토큰 설정 HF_TOKEN = os.getenv("HF_TOKEN") if not HF_TOKEN: print("Warning: HF_TOKEN not found in environment variables") # FLUX 모델 초기화 if not path.exists(cache_path): os.makedirs(cache_path, exist_ok=True) try: pipe = FluxPipeline.from_pretrained( "black-forest-labs/FLUX.1-dev", torch_dtype=torch.bfloat16, use_auth_token=HF_TOKEN # Hugging Face 토큰 추가 ) pipe.load_lora_weights( hf_hub_download( "ByteDance/Hyper-SD", "Hyper-FLUX.1-dev-8steps-lora.safetensors", token=HF_TOKEN # Hugging Face 토큰 추가 ) ) pipe.fuse_lora(lora_scale=0.125) pipe.to(device="cuda", dtype=torch.bfloat16) print("Successfully initialized FLUX model with authentication") except Exception as e: print(f"Error initializing FLUX model: {str(e)}") pipe = None # 이미지 생성 함수 추가 @spaces.GPU def generate_image(prompt, height=512, width=512, steps=8, scales=3.5, seed=3413): with torch.inference_mode(), torch.autocast("cuda", dtype=torch.bfloat16): return pipe( prompt=[prompt], generator=torch.Generator().manual_seed(int(seed)), num_inference_steps=int(steps), guidance_scale=float(scales), height=int(height), width=int(width), max_sequence_length=256 ).images[0] # SystemPrompt 부분을 직접 정의 SystemPrompt = """You are 'MOUSE-I', an advanced AI visualization expert. Your mission is to transform every response into a visually stunning and highly informative presentation. Core Capabilities: - Transform text responses into rich visual experiences - Create interactive data visualizations and charts - Design beautiful and intuitive user interfaces - Utilize engaging animations and transitions - Present information in a clear, structured manner Visual Elements to Include: - Charts & Graphs (using Chart.js, D3.js) - Interactive Data Visualizations - Modern UI Components - Engaging Animations - Informative Icons & Emojis - Color-coded Information Blocks - Progress Indicators - Timeline Visualizations - Statistical Representations - Comparison Tables Technical Requirements: - Modern HTML5/CSS3/JavaScript - Responsive Design - Interactive Elements - Clean Typography - Professional Color Schemes - Smooth Animations - Cross-browser Compatibility Libraries Available: - Chart.js for Data Visualization - D3.js for Complex Graphics - Bootstrap for Layout - jQuery for Interactions - Three.js for 3D Elements Design Principles: - Visual Hierarchy - Clear Information Flow - Consistent Styling - Intuitive Navigation - Engaging User Experience - Accessibility Compliance Remember to: - Present data in the most visually appealing way - Use appropriate charts for different data types - Include interactive elements where relevant - Maintain a professional and modern aesthetic - Ensure responsive design for all devices Return only HTML code wrapped in code blocks, focusing on creating visually stunning and informative presentations. """ from config import DEMO_LIST class Role: SYSTEM = "system" USER = "user" ASSISTANT = "assistant" History = List[Tuple[str, str]] Messages = List[Dict[str, str]] # 이미지 캐시를 메모리에 저장 IMAGE_CACHE = {} # boost_prompt 함수와 handle_boost 함수를 추가합니다 def boost_prompt(prompt: str) -> str: if not prompt: return "" # 증강을 위한 시스템 프롬프트 boost_system_prompt = """ 당신은 웹 개발 프롬프트 전문가입니다. 주어진 프롬프트를 분석하여 더 상세하고 전문적인 요구사항으로 확장하되, 원래 의도와 목적은 그대로 유지하면서 다음 관점들을 고려하여 증강하십시오: 1. 기술적 구현 상세 2. UI/UX 디자인 요소 3. 사용자 경험 최적화 4. 성능과 보안 5. 접근성과 호환성 기존 SystemPrompt의 모든 규칙을 준수하면서 증강된 프롬프트를 생성하십시오. """ try: # Claude API 시도 try: response = claude_client.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=2000, messages=[{ "role": "user", "content": f"다음 프롬프트를 분석하고 증강하시오: {prompt}" }] ) if hasattr(response, 'content') and len(response.content) > 0: return response.content[0].text raise Exception("Claude API 응답 형식 오류") except Exception as claude_error: print(f"Claude API 에러, OpenAI로 전환: {str(claude_error)}") # OpenAI API 시도 completion = openai_client.chat.completions.create( model="gpt-4", messages=[ {"role": "system", "content": boost_system_prompt}, {"role": "user", "content": f"다음 프롬프트를 분석하고 증강하시오: {prompt}"} ], max_tokens=2000, temperature=0.7 ) if completion.choices and len(completion.choices) > 0: return completion.choices[0].message.content raise Exception("OpenAI API 응답 형식 오류") except Exception as e: print(f"프롬프트 증강 중 오류 발생: {str(e)}") return prompt # 오류 발생시 원본 프롬프트 반환 # Boost 버튼 이벤트 핸들러 def handle_boost(prompt: str): try: boosted_prompt = boost_prompt(prompt) return boosted_prompt, gr.update(active_key="empty") except Exception as e: print(f"Boost 처리 중 오류: {str(e)}") return prompt, gr.update(active_key="empty") def get_image_base64(image_path): if image_path in IMAGE_CACHE: return IMAGE_CACHE[image_path] try: with open(image_path, "rb") as image_file: encoded_string = base64.b64encode(image_file.read()).decode() IMAGE_CACHE[image_path] = encoded_string return encoded_string except: return IMAGE_CACHE.get('default.png', '') def history_to_messages(history: History, system: str) -> Messages: messages = [{'role': Role.SYSTEM, 'content': system}] for h in history: messages.append({'role': Role.USER, 'content': h[0]}) messages.append({'role': Role.ASSISTANT, 'content': h[1]}) return messages def messages_to_history(messages: Messages) -> History: assert messages[0]['role'] == Role.SYSTEM history = [] for q, r in zip(messages[1::2], messages[2::2]): history.append([q['content'], r['content']]) return history # API 클라이언트 초기화 YOUR_ANTHROPIC_TOKEN = os.getenv('ANTHROPIC_API_KEY', '') # 기본값 추가 YOUR_OPENAI_TOKEN = os.getenv('OPENAI_API_KEY', '') # 기본값 추가 # API 키 검증 if not YOUR_ANTHROPIC_TOKEN or not YOUR_OPENAI_TOKEN: print("Warning: API keys not found in environment variables") # API 클라이언트 초기화 시 예외 처리 추가 try: claude_client = anthropic.Anthropic(api_key=YOUR_ANTHROPIC_TOKEN) openai_client = openai.OpenAI(api_key=YOUR_OPENAI_TOKEN) except Exception as e: print(f"Error initializing API clients: {str(e)}") claude_client = None openai_client = None # try_claude_api 함수 수정 async def try_claude_api(system_message, claude_messages, timeout=15): try: start_time = time.time() with claude_client.messages.stream( model="claude-3-5-sonnet-20241022", max_tokens=7860, system=system_message, messages=claude_messages ) as stream: collected_content = "" for chunk in stream: current_time = time.time() if current_time - start_time > timeout: print(f"Claude API response time: {current_time - start_time:.2f} seconds") raise TimeoutError("Claude API timeout") if chunk.type == "content_block_delta": collected_content += chunk.delta.text yield collected_content await asyncio.sleep(0) start_time = current_time except Exception as e: print(f"Claude API error: {str(e)}") raise e async def try_openai_api(openai_messages): try: stream = openai_client.chat.completions.create( model="gpt-4o", messages=openai_messages, stream=True, max_tokens=4096, temperature=0.7 ) collected_content = "" for chunk in stream: if chunk.choices[0].delta.content is not None: collected_content += chunk.choices[0].delta.content yield collected_content except Exception as e: print(f"OpenAI API error: {str(e)}") raise e class Demo: def __init__(self): pass async def generation_code(self, query: Optional[str], _setting: Dict[str, str]): if not query or query.strip() == '': query = get_random_placeholder() # 이미지 생성이 필요한지 확인 needs_image = '이미지' in query or '그림' in query or 'image' in query.lower() image_prompt = None # 이미지 프롬프트 추출 if needs_image: for keyword in ['이미지:', '그림:', 'image:']: if keyword in query.lower(): image_prompt = query.split(keyword)[1].strip() break if not image_prompt: image_prompt = query # 명시적 프롬프트가 없으면 전체 쿼리 사용 messages = [{'role': Role.SYSTEM, 'content': _setting['system']}] messages.append({'role': Role.USER, 'content': query}) system_message = messages[0]['content'] claude_messages = [{"role": "user", "content": query}] openai_messages = [ {"role": "system", "content": system_message}, {"role": "user", "content": query} ] try: yield [ "", None, gr.update(active_key="loading"), gr.update(open=True) ] await asyncio.sleep(0) collected_content = None try: async for content in try_claude_api(system_message, claude_messages): yield [ "", None, gr.update(active_key="loading"), gr.update(open=True) ] await asyncio.sleep(0) collected_content = content except Exception as claude_error: print(f"Falling back to OpenAI API due to Claude error: {str(claude_error)}") async for content in try_openai_api(openai_messages): yield [ "", None, gr.update(active_key="loading"), gr.update(open=True) ] await asyncio.sleep(0) collected_content = content if collected_content: # 이미지 생성이 필요한 경우 if needs_image and image_prompt: try: print(f"Generating image for prompt: {image_prompt}") # FLUX 모델을 사용하여 이미지 생성 if pipe is not None: image = generate_image( prompt=image_prompt, height=512, width=512, steps=8, scales=3.5, seed=random.randint(1, 10000) ) # 이미지를 Base64로 인코딩 buffered = BytesIO() image.save(buffered, format="PNG") img_str = base64.b64encode(buffered.getvalue()).decode() # HTML에 이미지 추가 image_html = f'''

Generated Image:

Prompt: {html.escape(image_prompt)}

''' # HTML 응답에 이미지 삽입 if '```html' in collected_content: # HTML 코드 블록 내부에 이미지 추가 collected_content = collected_content.replace('```html\n', f'```html\n{image_html}') else: # HTML 코드 블록으로 감싸서 이미지 추가 collected_content = f'```html\n{image_html}\n```\n{collected_content}' print("Image generation successful") else: raise Exception("FLUX model not initialized") except Exception as e: print(f"Image generation error: {str(e)}") error_message = f'''

Failed to generate image: {str(e)}

''' if '```html' in collected_content: collected_content = collected_content.replace('```html\n', f'```html\n{error_message}') else: collected_content = f'```html\n{error_message}\n```\n{collected_content}' # 최종 결과 표시 yield [ collected_content, send_to_sandbox(remove_code_block(collected_content)), gr.update(active_key="render"), gr.update(open=False) ] else: raise ValueError("No content was generated from either API") except Exception as e: print(f"Error details: {str(e)}") raise ValueError(f'Error calling APIs: {str(e)}') def clear_history(self): return [] def remove_code_block(text): pattern = r'```html\n(.+?)\n```' match = re.search(pattern, text, re.DOTALL) if match: return match.group(1).strip() else: return text.strip() def history_render(history: History): return gr.update(open=True), history def send_to_sandbox(code): encoded_html = base64.b64encode(code.encode('utf-8')).decode('utf-8') data_uri = f"data:text/html;charset=utf-8;base64,{encoded_html}" return f""" """ # 배포 관련 함수 추가 def generate_space_name(): """6자리 랜덤 영문 이름 생성""" letters = string.ascii_lowercase return ''.join(random.choice(letters) for i in range(6)) def deploy_to_vercel(code: str): try: token = "A8IFZmgW2cqA4yUNlLPnci0N" if not token: return "Vercel 토큰이 설정되지 않았습니다." # 6자리 영문 프로젝트 이름 생성 project_name = ''.join(random.choice(string.ascii_lowercase) for i in range(6)) # Vercel API 엔드포인트 deploy_url = "https://api.vercel.com/v13/deployments" # 헤더 설정 headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json" } # package.json 파일 생성 package_json = { "name": project_name, "version": "1.0.0", "private": True, # true -> True로 수정 "dependencies": { "vite": "^5.0.0" }, "scripts": { "dev": "vite", "build": "echo 'No build needed' && mkdir -p dist && cp index.html dist/", "preview": "vite preview" } } # 배포할 파일 데이터 구조 files = [ { "file": "index.html", "data": code }, { "file": "package.json", "data": json.dumps(package_json, indent=2) # indent 추가로 가독성 향상 } ] # 프로젝트 설정 project_settings = { "buildCommand": "npm run build", "outputDirectory": "dist", "installCommand": "npm install", "framework": None } # 배포 요청 데이터 deploy_data = { "name": project_name, "files": files, "target": "production", "projectSettings": project_settings } deploy_response = requests.post(deploy_url, headers=headers, json=deploy_data) if deploy_response.status_code != 200: return f"배포 실패: {deploy_response.text}" # URL 형식 수정 - 6자리.vercel.app 형태로 반환 deployment_url = f"{project_name}.vercel.app" time.sleep(5) return f"""배포 완료! https://{deployment_url}""" except Exception as e: return f"배포 중 오류 발생: {str(e)}" theme = gr.themes.Soft() def get_random_placeholder(): return random.choice(DEMO_LIST)['description'] def update_placeholder(): return gr.update(placeholder=get_random_placeholder()) def create_main_interface(): """메인 인터페이스 생성 함수""" #NEW - 검색 결과를 통합한 응답 생성 함수 async def execute_search_and_generate(query, setting): try: print(f"Executing search for query: {query}") # 검색 실행 url = "https://api.serphouse.com/serp/live" payload = { "data": { "q": query, "domain": "google.com", "lang": "en", "device": "desktop", "serp_type": "news", "loc": "United States", "page": "1", "num": "10" } } headers = { "Authorization": "Bearer V38CNn4HXpLtynJQyOeoUensTEYoFy8PBUxKpDqAW1pawT1vfJ2BWtPQ98h6", "Content-Type": "application/json" } response = requests.post(url, headers=headers, json=payload) results = response.json() print(f"Search results: {results}") # 디버깅용 # 검색 결과를 HTML로 변환 search_content = "```html\n
\n" search_content += "

최신 뉴스 검색 결과

\n" # API 응답 구조에 맞게 수정 if 'results' in results: news_items = results['results'].get('news', []) for item in news_items[:5]: search_content += f"""

{item['title']}

{item['snippet']}

{item['channel']} {item['time']}
""" search_content += "
\n```" # 검색 결과를 포함한 프롬프트 생성 enhanced_prompt = f"""Based on these news search results, create a comprehensive visual summary: {search_content} Please create a visually appealing HTML response that: 1. Summarizes the key points from the news 2. Organizes information in a clear structure 3. Uses appropriate HTML formatting and styling 4. Includes relevant quotes and statistics 5. Provides proper source attribution The response should be in HTML format with appropriate styling.""" print("Generating response with search results...") # 디버깅용 # async generator를 처리하기 위한 수정 async for result in demo_instance.generation_code(enhanced_prompt, setting): final_result = result print(f"Generated result: {final_result}") # 디버깅용 print("Response generation completed") # 디버깅용 return final_result except Exception as e: print(f"Search error: {str(e)}") print(f"Full error details: {str(e.__class__.__name__)}: {str(e)}") return [ "", None, gr.update(active_key="error"), gr.update(open=False) ] def execute_code(query: str): if not query or query.strip() == '': return None, gr.update(active_key="empty") try: if '```html' in query and '```' in query: code = remove_code_block(query) else: code = query.strip() return send_to_sandbox(code), gr.update(active_key="render") except Exception as e: print(f"Error executing code: {str(e)}") return None, gr.update(active_key="empty") async def handle_generation(query, setting, is_search): try: print(f"Mode: {'Web Search' if is_search else 'Generate'}") # 디버깅용 if is_search: print("Executing search and generate...") # 디버깅용 return await execute_search_and_generate(query, setting) else: print("Executing normal generation...") # 디버깅용 async for result in demo_instance.generation_code(query, setting): final_result = result return final_result except Exception as e: print(f"Generation error: {str(e)}") return ["", None, gr.update(active_key="error"), gr.update(open=False)] # CSS 파일 내용을 직접 적용 with open('app.css', 'r', encoding='utf-8') as f: custom_css = f.read() custom_css = """ /* 전체 컨테이너 */ .container { max-width: 1400px; margin: 0 auto; padding: 20px; } /* 메인 레이아웃 */ .main-tabs { display: flex; gap: 30px; min-height: 90vh; background: #f5f7fa; border-radius: 20px; padding: 30px; box-shadow: 0 10px 40px rgba(0,0,0,0.1); } /* 좌측 패널 */ .left-panel { flex: 1; background: white; border-radius: 15px; padding: 25px; box-shadow: 0 4px 15px rgba(0,0,0,0.05); display: flex; flex-direction: column; gap: 20px; } /* 우측 패널 */ .right-panel { flex: 2; background: white; border-radius: 15px; padding: 25px; box-shadow: 0 4px 15px rgba(0,0,0,0.05); } /* 모드 선택기 */ .mode-selector { background: #f8f9fa; padding: 20px; border-radius: 12px; border: 1px solid #e0e5ec; } /* 입력 영역 */ .input-area { display: flex; flex-direction: column; gap: 15px; } .custom-textarea { min-height: 200px !important; padding: 20px !important; border: 2px solid #e0e5ec !important; border-radius: 12px !important; font-size: 16px !important; line-height: 1.6 !important; resize: vertical !important; transition: all 0.3s ease !important; } .custom-textarea:focus { border-color: #007aff !important; box-shadow: 0 0 0 3px rgba(0,122,255,0.1) !important; } /* 버튼 그룹 */ .button-group { display: flex; gap: 12px; margin-top: 20px; } .generate-btn { background: linear-gradient(45deg, #007aff, #00a2ff) !important; color: white !important; padding: 12px 24px !important; border-radius: 10px !important; font-weight: 600 !important; border: none !important; box-shadow: 0 4px 15px rgba(0,122,255,0.3) !important; transition: all 0.3s ease !important; } .enhance-btn { background: white !important; color: #007aff !important; padding: 12px 24px !important; border-radius: 10px !important; font-weight: 600 !important; border: 2px solid #007aff !important; transition: all 0.3s ease !important; } .share-btn { background: white !important; color: #28c840 !important; padding: 12px 24px !important; border-radius: 10px !important; font-weight: 600 !important; border: 2px solid #28c840 !important; transition: all 0.3s ease !important; } /* 버튼 호버 효과 */ .generate-btn:hover { transform: translateY(-2px); box-shadow: 0 6px 20px rgba(0,122,255,0.4) !important; } .enhance-btn:hover { background: rgba(0,122,255,0.1) !important; transform: translateY(-2px); } .share-btn:hover { background: rgba(40,200,64,0.1) !important; transform: translateY(-2px); } /* 프리뷰 영역 */ .preview-container { background: white; border-radius: 15px; overflow: hidden; } .preview-header { background: #f8f9fa; padding: 15px; border-bottom: 1px solid #e0e5ec; display: flex; align-items: center; gap: 15px; } .window-controls { display: flex; gap: 8px; } .control { width: 12px; height: 12px; border-radius: 50%; transition: all 0.3s ease; } .close { background: #ff5f57; } .minimize { background: #febc2e; } .maximize { background: #28c840; } .preview-content { padding: 25px; min-height: 500px; } /* 결과 영역 */ .result-area { background: #f8f9fa; border-radius: 12px; padding: 20px; margin-top: 20px; } /* 로딩 상태 */ .loading-container { display: flex; flex-direction: column; align-items: center; justify-content: center; padding: 40px; gap: 20px; } /* 반응형 디자인 */ @media (max-width: 1200px) { .main-tabs { flex-direction: column; } .left-panel, .right-panel { width: 100%; } } """ demo = gr.Blocks(css=custom_css, theme=theme) with demo: with gr.Row(elem_classes="container"): with gr.Column(elem_classes="main-tabs"): # 좌측 패널 with gr.Column(scale=1, elem_classes="left-panel"): mode = gr.Radio( choices=["Generate", "Generate + Web Search"], label="Mode", value="Generate", info="Select mode for content generation", elem_classes="mode-selector" ) with gr.Column(elem_classes="input-area"): input = gr.Textbox( label="Enter your prompt", placeholder="Type your request here...", lines=8, elem_classes="custom-textarea" ) with gr.Row(elem_classes="button-group"): btn = gr.Button("Generate", elem_classes="generate-btn") boost_btn = gr.Button("Enhance", elem_classes="enhance-btn") deploy_btn = gr.Button("Share", elem_classes="share-btn") deploy_result = gr.HTML( label="Share Result", elem_classes="result-area" ) # 우측 패널 with gr.Column(scale=2, elem_classes="right-panel"): with gr.Column(elem_classes="preview-container"): gr.HTML("""
Preview
""") with gr.Tabs() as state_tab: with gr.Tab("empty"): gr.Markdown("Enter your prompt to begin", elem_classes="preview-content") with gr.Tab("loading"): with gr.Column(elem_classes="loading-container"): gr.Markdown("Creating visual presentation...") with gr.Tab("render"): sandbox = gr.HTML(elem_classes="preview-content") with gr.Tab("error"): gr.Markdown("An error occurred. Please try again.", elem_classes="preview-content") # 상태 변수들 setting = gr.State({"system": SystemPrompt}) search_mode = gr.State(False) code_output = gr.State("") # 이벤트 핸들러 mode.change( fn=lambda x: x == "Generate + Web Search", inputs=[mode], outputs=[search_mode] ) btn.click( fn=handle_generation, inputs=[input, setting, search_mode], outputs=[code_output, sandbox, state_tab, code_drawer] ).then( fn=update_placeholder, inputs=[], outputs=[input] ) boost_btn.click( fn=handle_boost, inputs=[input], outputs=[input, state_tab] ) deploy_btn.click( fn=lambda code: deploy_to_vercel(remove_code_block(code)) if code else "No code to share.", inputs=[code_output], outputs=[deploy_result] ) return demo if __name__ == "__main__": try: demo_instance = Demo() demo = create_main_interface() demo.queue( default_concurrency_limit=20, status_update_rate=10, api_open=False ).launch( server_name="0.0.0.0", server_port=7860, share=False, debug=False ) except Exception as e: print(f"Initialization error: {e}") raise