Spaces:
Sleeping
Sleeping
| # @title app.py | |
| import os | |
| import gradio as gr | |
| import asyncio | |
| from pathlib import Path | |
| from typing import Optional, List, Tuple | |
| from core.shooting_board import ShootingBoardGenerator | |
| from core.audio_gen import AudioGenerator | |
| from core.video_gen import VideoGenerator | |
| from core.video_editor import VideoEditor | |
| from utils.logger import SpotMakerLogger | |
| from utils.jwt_handler import JWTHandler | |
| class SpotMakerApp: | |
| def __init__(self): | |
| self.logger = SpotMakerLogger() | |
| self.jwt_handler = JWTHandler() | |
| self.output_dir = Path("output") | |
| self.output_dir.mkdir(exist_ok=True) | |
| def create_interface(self) -> gr.Blocks: | |
| with gr.Blocks(title="SpotMaker - Generator Reklam Video") as interface: | |
| gr.Markdown( | |
| """ | |
| # 🎥 SpotMaker - Generator Reklam Video // by www.Heuristica.pl | |
| Wprowadź swój klucz API i opisz koncepcję reklamy. | |
| """ | |
| ) | |
| with gr.Row(): | |
| with gr.Column(): | |
| # Konfiguracja API | |
| api_key = gr.Textbox( | |
| label="Klucz API Hailuo", | |
| placeholder="Wklej swój klucz API...", | |
| type="password" | |
| ) | |
| # Koncepcja | |
| concept = gr.Textbox( | |
| label="Opis koncepcji", | |
| placeholder="Wprowadź krótki opis koncepcji reklamowej...", | |
| lines=5 | |
| ) | |
| file_input = gr.File( | |
| label="Załącz dodatkowy opis (opcjonalnie)", | |
| file_types=["txt", "docx"] | |
| ) | |
| with gr.Row(): | |
| start_btn = gr.Button("▶️ Generuj spot", variant="primary") | |
| clear_btn = gr.Button("🗑️ Wyczyść") | |
| with gr.Column(): | |
| # Status i logi | |
| progress = gr.Slider( | |
| minimum=0, | |
| maximum=100, | |
| value=0, | |
| label="Postęp", | |
| interactive=False | |
| ) | |
| status = gr.Markdown("Gotowy do rozpoczęcia") | |
| logs = gr.TextArea( | |
| value="", | |
| label="Szczegółowe logi", | |
| interactive=False, | |
| lines=15, | |
| autoscroll=True, | |
| show_copy_button=True, | |
| max_lines=100 | |
| ) | |
| # Funkcja do aktualizacji logów | |
| def add_log(msg: str, current_logs: str) -> str: | |
| """Dodaje nową linię do logów z timestampem.""" | |
| if not current_logs: | |
| return msg | |
| return f"{current_logs}\n{msg}" | |
| # Tworzenie listy do przechowywania callback'ów aktualizacji logów | |
| log_updates = [] | |
| with gr.Row(visible=False) as result_row: | |
| # Wyniki | |
| video_output = gr.Video(label="Wygenerowany spot") | |
| with gr.Column(): | |
| video_info = gr.Markdown("") | |
| download_btn = gr.Button("⬇️ Pobierz spot") | |
| def validate_api_key(api_key: str) -> str: | |
| """Sprawdza poprawność klucza API i wyświetla GroupID.""" | |
| if not api_key: | |
| return "❌ Wprowadź klucz API" | |
| group_id = self.jwt_handler.extract_group_id(api_key) | |
| if not group_id: | |
| return "❌ Nieprawidłowy klucz API" | |
| return f"✅ GroupID: {group_id}" | |
| def update_state(value: float, message: str) -> dict: | |
| """Aktualizuje stan interfejsu.""" | |
| self.logger.update_progress("app", value, message) | |
| return { | |
| progress: value, | |
| status: message, | |
| logs: self.logger.get_module_status("app") | |
| } | |
| def generate(api_key: str, concept: str, file_input: Optional[str]) -> dict: | |
| """Główna funkcja generująca spot.""" | |
| try: | |
| # Walidacja API | |
| group_id = self.jwt_handler.extract_group_id(api_key) | |
| if not group_id: | |
| return { | |
| progress: 0, | |
| status: "❌ Błąd: Nieprawidłowy klucz API", | |
| logs: "Sprawdź czy klucz API jest poprawny", | |
| result_row: gr.Row(visible=False), | |
| video_output: None, | |
| video_info: "", | |
| download_btn: gr.Button(visible=False) | |
| } | |
| # Walidacja koncepcji | |
| if not concept or len(concept.strip()) < 10: | |
| return { | |
| progress: 0, | |
| status: "❌ Błąd: Zbyt krótki opis", | |
| logs: "Wprowadź dłuższy opis koncepcji (min. 10 znaków)", | |
| result_row: gr.Row(visible=False), | |
| video_output: None, | |
| video_info: "", | |
| download_btn: gr.Button(visible=False) | |
| } | |
| # Create event loop | |
| loop = asyncio.new_event_loop() | |
| asyncio.set_event_loop(loop) | |
| # Update progress | |
| update_state(20, "Generuję scenariusz...") | |
| # Create generators | |
| shooting_board = ShootingBoardGenerator(api_key, self.logger) | |
| audio_gen = AudioGenerator(api_key, group_id, "generated_audio", self.logger) | |
| video_gen = VideoGenerator(api_key, "generated_videos", self.logger) | |
| video_editor = VideoEditor("final_spot", self.logger) | |
| # 1. Generate shooting board | |
| scenario = loop.run_until_complete(shooting_board.generate(concept)) | |
| update_state(40, "Generuję ścieżkę audio...") | |
| # 2. Generate audio | |
| audio_result = loop.run_until_complete(audio_gen.generate_voice_over(scenario['voice_over'])) | |
| audio_path, audio_duration = audio_result | |
| update_state(60, "Generuję klipy video...") | |
| # 3. Generate video clips | |
| video_paths = loop.run_until_complete(video_gen.generate_clips(scenario['key_frames'])) | |
| update_state(80, "Montuję finalny spot...") | |
| # 4. Edit final video | |
| final_path = loop.run_until_complete(video_editor.create_spot(video_paths, audio_path, audio_duration)) | |
| # Close loop | |
| loop.close() | |
| # Success | |
| update_state(100, "Zakończono!") | |
| return { | |
| progress: 100, | |
| status: "✅ Generowanie zakończone!", | |
| logs: self.logger.get_module_status("app"), | |
| video_output: final_path, | |
| video_info: f"""### Szczegóły spotu: | |
| - Długość: {audio_duration:.2f}s | |
| - Liczba ujęć: {len(video_paths)}""", | |
| result_row: gr.Row(visible=True), | |
| download_btn: gr.Button(visible=True) | |
| } | |
| # 4. Edit final video | |
| final_path = video_editor.create_spot(video_paths, audio_path, audio_duration) | |
| # Success | |
| update_state(100, "Zakończono!") | |
| return { | |
| progress: 100, | |
| status: "✅ Generowanie zakończone!", | |
| logs: self.logger.get_module_status("app"), | |
| video_output: final_path, | |
| video_info: f"""### Szczegóły spotu: | |
| - Długość: {audio_duration:.2f}s | |
| - Liczba ujęć: {len(video_paths)}""", | |
| result_row: gr.Row(visible=True) | |
| } | |
| except Exception as e: | |
| error_msg = self.logger.format_error("app", e) | |
| return { | |
| progress: 0, | |
| status: "❌ Wystąpił błąd", | |
| logs: f"Szczegóły błędu:\n{error_msg}", | |
| result_row: gr.Row(visible=False), | |
| video_output: None, | |
| video_info: "" | |
| } | |
| def clear() -> dict: | |
| """Czyści interfejs.""" | |
| return { | |
| api_key: "", | |
| concept: "", | |
| file_input: None, | |
| progress: 0, | |
| status: "Gotowy do rozpoczęcia", | |
| logs: "", | |
| video_output: None, | |
| video_info: "", | |
| result_row: gr.Row(visible=False) | |
| } | |
| # Podpięcie zdarzeń | |
| api_key.change(validate_api_key, api_key, status) | |
| start_btn.click( | |
| fn=generate, | |
| inputs=[api_key, concept, file_input], | |
| outputs=[progress, status, logs, video_output, video_info, | |
| result_row, download_btn] | |
| ) | |
| clear_btn.click( | |
| fn=clear, | |
| outputs=[api_key, concept, file_input, progress, status, | |
| logs, video_output, video_info, result_row, download_btn] | |
| ) | |
| return interface | |
| if __name__ == "__main__": | |
| app = SpotMakerApp() | |
| interface = app.create_interface() | |
| interface.queue() | |
| interface.launch( | |
| server_name="0.0.0.0", | |
| server_port=7860, | |
| show_api=False, | |
| share=False | |
| ) |