diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0afbcb8..740435b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -19,6 +19,12 @@ jobs: with: python-version: '3.11' + - name: Install Application Dependencies + run: python -m pip install -r requirements.txt + + - name: Run Offline Verification Suite + run: python -m unittest discover -s tests -v + - name: Validate Python Syntax run: | python -m compileall -q . diff --git a/README.md b/README.md index 0d95c8c..700f15e 100644 --- a/README.md +++ b/README.md @@ -101,6 +101,16 @@ python app.py # Or double-click "Start Video Studio.bat" on Windows ``` +## ๐Ÿงช Verification + +Run the offline verification suite with: + +```bash +python -m unittest discover -s tests -v +``` + +`test_gemini_features.py` and `test_verify.py` are opt-in live smoke checks; they may use network services, API quota, and local media processing, so they are not run during automated test discovery. + --- ## ๐Ÿ“œ License diff --git a/Start Video Studio.bat b/Start Video Studio.bat index 6cb6c0c..08e8f36 100644 --- a/Start Video Studio.bat +++ b/Start Video Studio.bat @@ -1,14 +1,14 @@ @echo off title VideoStudio Pro - 4K AI Video ^& Storyboard Studio -cd /d "C:\Users\chkam\OneDrive\Desktop\BrandFinder\VideoStudio" +cd /d "%~dp0" echo ================================================================ echo VideoStudio Pro - 4K AI Video ^& Storyboard Studio echo Crafted by Kamran Ashraf (Kami) echo ================================================================ -rem --- pick a real Python (no PATH guesswork) --- -set "PY=C:\Users\chkam\AppData\Local\Programs\Python\Python314\python.exe" +rem --- prefer the project's virtual environment, then use system Python --- +set "PY=%~dp0.venv\Scripts\python.exe" if not exist "%PY%" set "PY=python" rem --- already running? just open the browser and exit --- diff --git a/app.py b/app.py index 009826a..e3e9e32 100644 --- a/app.py +++ b/app.py @@ -8,6 +8,7 @@ from __future__ import annotations +import html import os import subprocess import sys @@ -534,10 +535,41 @@ def duration_hint(sec) -> str: return (f"๐ŸŽž๏ธ {int(float(sec))}s 4K ยท " f"{n} AI shots chained & crossfaded") +def job_progress_percent(progress: float) -> int: + try: + return max(0, min(100, int(float(progress) * 100))) + except (TypeError, ValueError, OverflowError): + return 0 + +def render_job_card(job_id: str, prompt: str, stage: Stage, progress: float, message: str, elapsed: Optional[int] = None) -> str: + pct = job_progress_percent(progress) + icon = STAGE_ICONS.get(stage, "โ€ข") + color = "#10B981" if stage == Stage.DONE else "#EF4444" if stage in (Stage.FAILED, Stage.CANCELLED) else "#38BDF8" + elapsed_html = f"โฑ๏ธ {max(0, elapsed)}s" if elapsed is not None else "" + safe_job_id = html.escape(str(job_id)) + safe_prompt = html.escape(str(prompt)[:40]) + safe_stage = html.escape(stage.value) + safe_message = html.escape(str(message)) + return f""" +
+
+ [{safe_job_id}] {safe_prompt} + {icon} {safe_stage} ({pct}%) +
+
+
+ {safe_message} {elapsed_html} +
+
+ """ + def generate_ai_video_live( prompt, negative, preset, quality, aspect, duration, language, gender, custom_script, music, burn_subs, sub_style, tr_lang, interp_60, anti_fp, watermark, seed ): + if not STUDIO.queue.is_worker: + yield "โš ๏ธ Another VideoStudio instance owns the job queue; this instance is view-only.", "
Queue is read-only in this instance.
", None, None, "" + return if not prompt.strip(): yield "โš ๏ธ Please enter a prompt first.", "
Enter a prompt to generate 4K video.
", None, None, "" return @@ -571,23 +603,16 @@ def generate_ai_video_live( time.sleep(1.0) continue - pct = int(status.progress * 100) - icon = STAGE_ICONS.get(status.stage, "โ€ข") - color = "#10B981" if status.stage == Stage.DONE else "#EF4444" if status.stage in (Stage.FAILED, Stage.CANCELLED) else "#38BDF8" + pct = job_progress_percent(status.progress) elapsed = int(time.time() - start_t) - - card_html = f""" -
-
- [{job_id}] {status.settings.prompt[:40] if status.settings else 'Job'} - {icon} {status.stage.value} ({pct}%) -
-
-
- {status.message} โฑ๏ธ {elapsed}s -
-
- """ + card_html = render_job_card( + job_id, + status.settings.prompt if status.settings else "Job", + status.stage, + status.progress, + status.message, + elapsed, + ) if status.stage == Stage.DONE: v_out = status.video_path if (status.video_path and Path(status.video_path).exists()) else None @@ -598,7 +623,7 @@ def generate_ai_video_live( yield f"๐Ÿ›‘ Job Cancelled: {job_id}", card_html, None, None, job_id break elif status.stage == Stage.FAILED: - yield f"โŒ Job Error: {status.error or 'Failed'}", card_html, None, None, job_id + yield f"โŒ Job Error: {html.escape(str(status.error or 'Failed'))}", card_html, None, None, job_id break yield f"โณ Rendering 4K ({pct}%)...", card_html, None, None, job_id @@ -607,6 +632,9 @@ def generate_ai_video_live( def generate_storyboard_video_live( topic, script_text, preset, aspect, language, gender, music, burn_subs, sub_style, watermark, anti_fp ): + if not STUDIO.queue.is_worker: + yield "โš ๏ธ Another VideoStudio instance owns the job queue; this instance is view-only.", "
Queue is read-only in this instance.
", None, "" + return if not script_text.strip() and not topic.strip(): yield "โš ๏ธ Please enter a topic or script.", "
Enter a script to build video.
", None, "" return @@ -634,23 +662,16 @@ def generate_storyboard_video_live( time.sleep(1.0) continue - pct = int(status.progress * 100) - icon = STAGE_ICONS.get(status.stage, "โ€ข") - color = "#10B981" if status.stage == Stage.DONE else "#EF4444" if status.stage in (Stage.FAILED, Stage.CANCELLED) else "#38BDF8" + pct = job_progress_percent(status.progress) elapsed = int(time.time() - start_t) - - card_html = f""" -
-
- [{job_id}] {status.settings.prompt[:40] if status.settings else 'Job'} - {icon} {status.stage.value} ({pct}%) -
-
-
- {status.message} โฑ๏ธ {elapsed}s -
-
- """ + card_html = render_job_card( + job_id, + status.settings.prompt if status.settings else "Job", + status.stage, + status.progress, + status.message, + elapsed, + ) if status.stage == Stage.DONE: v_out = status.video_path if (status.video_path and Path(status.video_path).exists()) else None @@ -660,19 +681,23 @@ def generate_storyboard_video_live( yield f"๐Ÿ›‘ Storyboard Cancelled: {job_id}", card_html, None, job_id break elif status.stage == Stage.FAILED: - yield f"โŒ Storyboard Error: {status.error or 'Failed'}", card_html, None, job_id + yield f"โŒ Storyboard Error: {html.escape(str(status.error or 'Failed'))}", card_html, None, job_id break yield f"โณ Building Storyboard ({pct}%)...", card_html, None, job_id time.sleep(1.0) def cancel_active_job_btn(job_id: str) -> str: + if not STUDIO.queue.is_worker: + return "โš ๏ธ Another VideoStudio instance owns the job queue; this instance is view-only." if not job_id: return "โš ๏ธ No active job selected." success = STUDIO.queue.cancel(job_id) return f"๐Ÿ›‘ Job [{job_id}] stopped." if success else f"Job [{job_id}] not found." def resume_all_interrupted_jobs_btn() -> str: + if not STUDIO.queue.is_worker: + return "โš ๏ธ Another VideoStudio instance owns the job queue; this instance is view-only." STUDIO.queue._auto_resume_interrupted_jobs() return "๐Ÿ”„ Interrupted jobs restored & resumed!" @@ -725,19 +750,15 @@ def render_queue_table() -> str: return "
No jobs in queue. Submit a 4K Video or Storyboard to begin.
" cards = [] for js in reversed(jobs[-8:]): - pct = int(js.progress * 100) - icon = STAGE_ICONS.get(js.stage, "โ€ข") - color = "#10B981" if js.stage == Stage.DONE else "#EF4444" if js.stage == Stage.CANCELLED else "#F59E0B" if js.stage == Stage.FAILED else "#38BDF8" - cards.append(f""" -
-
- [{js.job_id}] {js.settings.prompt[:40] if js.settings else 'Job'} - {icon} {js.stage.value} ({pct}%) -
-
-
{js.message}
-
- """) + cards.append( + render_job_card( + js.job_id, + js.settings.prompt if js.settings else "Job", + js.stage, + js.progress, + js.message, + ) + ) return "".join(cards) # --------------------------------------------------------------------------- diff --git a/assets/test_veo3_free.py b/assets/test_veo3_free.py index f0eba7d..a69f068 100644 --- a/assets/test_veo3_free.py +++ b/assets/test_veo3_free.py @@ -27,9 +27,9 @@ warnings.filterwarnings("ignore", message=".*numpy.*") warnings.filterwarnings("ignore", category=UserWarning) -from video_generator import Veo3FreeVideoEngine, FreeLLMPromptEnhancer - def main(): + from video_generator import Veo3FreeVideoEngine, FreeLLMPromptEnhancer + parser = argparse.ArgumentParser(description="Free VEO 3 AI Video Generator Test") parser.add_argument("prompt", type=str, nargs="?", default="A glowing futuristic starship sailing through a purple nebula, 8k cinematic masterpiece", help="Text prompt for video generation") parser.add_argument("--aspect", type=str, default="16:9", choices=["16:9", "9:16", "1:1"], help="Aspect Ratio") diff --git a/run_videostudio.vbs b/run_videostudio.vbs index 60cb923..69d942c 100644 --- a/run_videostudio.vbs +++ b/run_videostudio.vbs @@ -1,24 +1,16 @@ ' VideoStudio Pro โ€” Resilient Background Launcher Option Explicit -Dim WshShell, fso, q, appDir, py, logPath, cmdLine, i, candidates, cand +Dim WshShell, fso, q, appDir, py, logPath, cmdLine, i q = Chr(34) Set WshShell = CreateObject("WScript.Shell") Set fso = CreateObject("Scripting.FileSystemObject") appDir = fso.GetParentFolderName(WScript.ScriptFullName) If Not fso.FileExists(appDir & "\app.py") Then - candidates = Array( _ - "C:\Users\chkam\OneDrive\Desktop\BrandFinder\VideoStudio", _ - "C:\Users\chkam\OneDrive\Desktop\VideoStudio", _ - "C:\Users\chkam\Desktop\BrandFinder\VideoStudio" _ - ) - For Each cand In candidates - If fso.FileExists(cand & "\app.py") Then - appDir = cand - Exit For - End If - Next + MsgBox "VideoStudio app.py was not found next to this launcher:" & vbCrLf & appDir, _ + vbExclamation, "VideoStudio Pro" + WScript.Quit 1 End If logPath = appDir & "\outputs\logs\launch.log" @@ -41,7 +33,7 @@ If ServerUp() Then WScript.Quit End If -py = "C:\Users\chkam\AppData\Local\Programs\Python\Python314\python.exe" +py = appDir & "\.venv\Scripts\python.exe" If Not fso.FileExists(py) Then py = "python.exe" If Not fso.FolderExists(appDir & "\outputs") Then fso.CreateFolder(appDir & "\outputs") diff --git a/test_gemini_features.py b/test_gemini_features.py index af0f65e..e3220e1 100644 --- a/test_gemini_features.py +++ b/test_gemini_features.py @@ -1,41 +1,57 @@ +"""Opt-in live smoke check for the configured Gemini and Imagen APIs. + +Run explicitly with ``python test_gemini_features.py``; this is intentionally +not executed during test discovery because it consumes network/API quota. +""" + import os -import sys from pathlib import Path + from dotenv import load_dotenv -# Load environment variables safely -load_dotenv(Path(__file__).parent / ".env") -load_dotenv() - -key = os.environ.get("GEMINI_API_KEY") -print(f"GEMINI_API_KEY configured: {bool(key)}") - -from google import genai - -client = genai.Client(api_key=key) - -try: - resp = client.models.generate_content( - model='gemini-2.5-flash', - contents='You are a Veo 3 cinematic prompt director. Expand: "Golden eagle soaring in mountains" into 1 cinematic prompt sentence.' - ) - print("Gemini 2.5 Flash Response:") - print(resp.text) -except Exception as e: - print("Gemini text failed:", e) - -# Test Image generation if supported -try: - img_resp = client.models.generate_images( - model='imagen-3.0-generate-002', - prompt='A majestic golden eagle soaring over snow covered mountain peaks at sunrise, 8k cinematic photography', - config=dict( - number_of_images=1, - aspect_ratio='16:9', + +def main() -> int: + load_dotenv(Path(__file__).parent / ".env") + load_dotenv() + key = os.environ.get("GEMINI_API_KEY") + if not key: + print("GEMINI_API_KEY is not configured; skipping live API checks.") + return 0 + + from google import genai + + client = genai.Client(api_key=key) + try: + resp = client.models.generate_content( + model="gemini-2.5-flash", + contents=( + 'You are a Veo 3 cinematic prompt director. Expand: ' + '"Golden eagle soaring in mountains" into 1 cinematic prompt sentence.' + ), + ) + print("Gemini 2.5 Flash Response:") + print(resp.text) + except Exception as exc: + print("Gemini text check failed:", exc) + return 1 + + try: + img_resp = client.models.generate_images( + model="imagen-3.0-generate-002", + prompt=( + "A majestic golden eagle soaring over snow covered mountain peaks " + "at sunrise, 8k cinematic photography" + ), + config={"number_of_images": 1, "aspect_ratio": "16:9"}, ) - ) - print("Imagen 3 image generation succeeded!") - if img_resp.generated_images: - print("Generated image bytes length:", len(img_resp.generated_images[0].image.image_bytes)) -except Exception as e: - print("Imagen 3 test note:", e) + print("Imagen image generation succeeded.") + if img_resp.generated_images: + print("Generated image bytes length:", len(img_resp.generated_images[0].image.image_bytes)) + except Exception as exc: + print("Imagen check failed:", exc) + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/test_verify.py b/test_verify.py index 548d8d1..922b58a 100644 --- a/test_verify.py +++ b/test_verify.py @@ -1,58 +1,79 @@ -""" -test_verify.py โ€” Complete Verification Suite for VideoStudio Pro. +"""Opt-in end-to-end smoke checks for external services and media generation. + +Run explicitly with ``python test_verify.py``. Automated test discovery only +runs the offline suite in ``tests/``. """ -import sys from pathlib import Path +import sys + + +def main() -> None: + vs_dir = Path(__file__).resolve().parent + if str(vs_dir) not in sys.path: + sys.path.insert(0, str(vs_dir)) + + import app + import video_studio + + print("1. Testing VideoStudio Core Imports...", flush=True) + print(" โœ” Imports succeeded cleanly.", flush=True) + + print("2. Testing Hardware Probe...", flush=True) + probe = video_studio.HardwareProbe() + print(f" โœ” Hardware: {probe.summary()}", flush=True) + + print("3. Testing Prompt Enhancer...", flush=True) + prompt = "Futuristic neon cyber city at night" + enhanced = video_studio.PromptEngine().enhance(prompt, preset="Cyberpunk") + assert enhanced and prompt in enhanced + print(f" โœ” Enhanced prompt: {enhanced[:80]}...", flush=True) + + print("4. Testing Multi-Scene Script Generator...", flush=True) + script = video_studio.FreeLLMPromptEnhancer.generate_script("Deep Sea Wonders", scene_count=2) + assert "Visual:" in script and "VO:" in script + print(" โœ” Generated script preview:") + for line in script.splitlines()[:4]: + print(f" {line}") + + print("5. Testing Neural Audio Voiceover Synthesis...", flush=True) + vo_out, events = video_studio.AudioEngine().generate_voiceover( + "Welcome to VideoStudio Pro.", language="English", gender="Male" + ) + assert vo_out.exists() and vo_out.stat().st_size > 0 + print(f" โœ” Voiceover generated: {vo_out} (Events: {len(events)})", flush=True) + + print("6. Testing Ambient Synth Pad Generator...", flush=True) + bgm_out = video_studio.DEFAULT_OUTPUT_DIR / "audio_temp" / "test_synth_bgm.mp3" + video_studio.AudioEngine().generate_synth_music("Ambient", 3.0, bgm_out) + assert bgm_out.exists() and bgm_out.stat().st_size > 0 + print(f" โœ” Synth BGM generated: {bgm_out}", flush=True) + + print("7. Testing SRT Subtitle Generation...", flush=True) + srt_out = video_studio.DEFAULT_OUTPUT_DIR / "audio_temp" / "test_subs.srt" + video_studio.SubtitleEngine.events_to_srt(events, srt_out) + assert srt_out.exists() and srt_out.stat().st_size > 0 + print(f" โœ” SRT generated: {srt_out}", flush=True) + + print("8. Testing Media Fetcher Fallbacks...", flush=True) + fetcher = video_studio.MediaFetcher() + visual_path, media_type = fetcher.fetch_scene_visual( + "deep ocean glowing jellyfish", prefer_video=False + ) + assert Path(visual_path).exists() + print(f" โœ” Visual asset fetched: {visual_path} (Type: {media_type})", flush=True) + + print("9. Testing Ken Burns 3D Video Motion...", flush=True) + kb_video = video_studio.DEFAULT_OUTPUT_DIR / "test_kb_out.mp4" + video_studio.VideoEngine.image_to_video_ken_burns( + Path(visual_path), 2.5, 640, 360, fps=24, out_path=kb_video + ) + assert kb_video.exists() and kb_video.stat().st_size > 0 + print(f" โœ” Ken Burns video created: {kb_video}", flush=True) + + print("\n๐ŸŽ‰ ALL 9 VERIFICATION CHECKS PASSED!\n", flush=True) + app.STUDIO.queue.shutdown() + -vs_dir = Path(__file__).resolve().parent -if str(vs_dir) not in sys.path: - sys.path.insert(0, str(vs_dir)) - -print("1. Testing VideoStudio Core Imports...", flush=True) -import video_studio -import app -print(" โœ” Imports succeeded cleanly.", flush=True) - -print("2. Testing Hardware Probe...", flush=True) -probe = video_studio.HardwareProbe() -print(f" โœ” Hardware: {probe.summary()}", flush=True) - -print("3. Testing Prompt Enhancer...", flush=True) -prompt = "Futuristic neon cyber city at night" -enhanced = video_studio.PromptEngine().enhance(prompt, preset="Cyberpunk") -print(f" โœ” Enhanced prompt: {enhanced[:80]}...", flush=True) - -print("4. Testing Multi-Scene Script Generator...", flush=True) -script = video_studio.FreeLLMPromptEnhancer.generate_script("Deep Sea Wonders", scene_count=2) -print(" โœ” Generated script preview:") -for line in script.splitlines()[:4]: - print(f" {line}") - -print("5. Testing Neural Audio Voiceover Synthesis...", flush=True) -vo_out, events = video_studio.AudioEngine().generate_voiceover( - "Welcome to VideoStudio Pro.", language="English", gender="Male" -) -print(f" โœ” Voiceover generated: {vo_out} (Exists: {vo_out.exists()}, Events: {len(events)})", flush=True) - -print("6. Testing Ambient Synth Pad Generator...", flush=True) -bgm_out = video_studio.DEFAULT_OUTPUT_DIR / "audio_temp" / "test_synth_bgm.mp3" -video_studio.AudioEngine().generate_synth_music("Ambient", 3.0, bgm_out) -print(f" โœ” Synth BGM generated: {bgm_out} (Exists: {bgm_out.exists()})", flush=True) - -print("7. Testing SRT Subtitle Generation...", flush=True) -srt_out = video_studio.DEFAULT_OUTPUT_DIR / "audio_temp" / "test_subs.srt" -video_studio.SubtitleEngine.events_to_srt(events, srt_out) -print(f" โœ” SRT generated: {srt_out} (Exists: {srt_out.exists()})", flush=True) - -print("8. Testing Media Fetcher Fallbacks...", flush=True) -fetcher = video_studio.MediaFetcher() -visual_path, media_type = fetcher.fetch_scene_visual("deep ocean glowing jellyfish", prefer_video=False) -print(f" โœ” Visual asset fetched: {visual_path} (Type: {media_type})", flush=True) - -print("9. Testing Ken Burns 3D Video Motion...", flush=True) -kb_video = video_studio.DEFAULT_OUTPUT_DIR / "test_kb_out.mp4" -video_studio.VideoEngine.image_to_video_ken_burns(Path(visual_path), 2.5, 640, 360, fps=24, out_path=kb_video) -print(f" โœ” Ken Burns video created: {kb_video} (Exists: {kb_video.exists()})", flush=True) - -print("\n๐ŸŽ‰ ALL 9 VERIFICATION CHECKS PASSED PERFECTLY!\n", flush=True) +if __name__ == "__main__": + main() diff --git a/tests/test_core.py b/tests/test_core.py new file mode 100644 index 0000000..ec9e1f1 --- /dev/null +++ b/tests/test_core.py @@ -0,0 +1,116 @@ +import tempfile +import unittest +from pathlib import Path + +import app +from app import job_progress_percent, render_job_card +from video_studio import ( + JobQueue, + PromptEngine, + Stage, + SubtitleEngine, + VideoStudio, + frames_for, +) + + +class VideoStudioCoreTests(unittest.TestCase): + @classmethod + def tearDownClass(cls): + app.STUDIO.queue.shutdown() + + def test_frames_are_snapped_to_model_requirement(self): + self.assertEqual(frames_for(0), 17) + self.assertEqual(frames_for(5), 81) + self.assertEqual((frames_for(2.5) - 1) % 4, 0) + + def test_prompt_enhancement_uses_requested_style(self): + enhanced = PromptEngine().enhance("A quiet forest", preset="Cyberpunk", seed=7) + self.assertIn("cyberpunk", enhanced.lower()) + self.assertIn("A quiet forest.", enhanced) + self.assertEqual(enhanced, PromptEngine().enhance("A quiet forest", preset="Cyberpunk", seed=7)) + + def test_script_parser_extracts_visual_and_voiceover(self): + studio = object.__new__(VideoStudio) + self.assertEqual( + studio._parse_script_scenes( + "Visual: sunrise over the mountains\nVO: A new day begins.\n\n" + "Visual: a river through the valley\nVO: Water gives life." + ), + [ + ("sunrise over the mountains", "A new day begins."), + ("a river through the valley", "Water gives life."), + ], + ) + + def test_scene_checkpoint_preserves_caption_offsets_for_resume(self): + events = [{"text": "Hello", "offset": 0.25, "duration": 0.4}] + with tempfile.TemporaryDirectory() as tmp: + metadata = Path(tmp) / "scene_00.json" + VideoStudio._write_scene_checkpoint(metadata, events, 1.75) + self.assertEqual(VideoStudio._read_scene_checkpoint(metadata), (events, 1.75)) + + metadata.write_text('{"events": [], "voiceover_duration": NaN}', encoding="utf-8") + self.assertIsNone(VideoStudio._read_scene_checkpoint(metadata)) + + def test_srt_generation_writes_timed_captions(self): + with tempfile.TemporaryDirectory() as tmp: + output = Path(tmp) / "captions.srt" + SubtitleEngine.events_to_srt( + [ + {"text": "Hello", "offset": 0.25, "duration": 0.4}, + {"text": "world.", "offset": 0.7, "duration": 0.5}, + ], + output, + ) + self.assertEqual( + output.read_text(encoding="utf-8"), + "1\n00:00:00,250 --> 00:00:01,200\nHello world.\n", + ) + + def test_job_card_escapes_untrusted_text_and_bounds_progress(self): + card = render_job_card( + "job<1>", + "", + Stage.GENERATING, + 1.5, + "", + ) + self.assertIn("job<1>", card) + self.assertIn("<script>", card) + self.assertIn("<img", card) + self.assertNotIn("