# @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 )