#!/usr/bin/env python3 import concurrent.futures import hashlib import json import re import subprocess import sys import time from pathlib import Path ROOT = Path("/data/video-pipeline/GacMaiAudio/project/gacmai_20260713_050144") STYLE = ( " MANDATORY contemporary 2020s corporate setting with modern business or construction clothing. " "Strict flat hand-drawn 2D Chinese manhua/webtoon, simplified comic anatomy, thick uniform black ink contours, " "hard-edged cel shadows, matte flat colors, graphic two-dimensional background, limited dark teal cream brick-red amber palette. " "No ancient or historical setting, robes, hanfu, palace, fantasy, photorealism, semi-realism, realistic skin, 3D, CGI, painterly style, " "gradients, reflections, glossy materials, depth of field, readable text, letters, numbers, signs, logos, signatures, or watermarks. 16:9 landscape." ) INTRO = "Cảm ơn các bạn đã nghe truyện từ Gác Mái Audio. Chúc các bạn có thời gian nghe truyện vui vẻ." sys.path.insert(0, "/opt/OpenMontage") from tools.tool_registry import registry # noqa: E402 def valid_audio(path: Path) -> bool: if not path.exists() or path.stat().st_size < 10000: return False p = subprocess.run([ "ffprobe", "-v", "error", "-select_streams", "a:0", "-show_entries", "stream=codec_name,sample_rate,channels", "-show_entries", "format=duration", "-of", "json", str(path) ], capture_output=True, text=True) if p.returncode: return False try: d = json.loads(p.stdout) return float(d["format"]["duration"]) > 0 and bool(d["streams"]) except Exception: return False def valid_image(path: Path) -> bool: if not path.exists() or path.stat().st_size < 10000: return False try: from PIL import Image with Image.open(path) as image: image.verify() with Image.open(path) as image: w, h = image.size return w >= 1200 and h >= 675 and 1.6 <= w / h <= 1.9 except Exception: return False def semantic_chunks(text: str, target=1400, hard=1800): paragraphs = [re.sub(r"\s+", " ", p).strip() for p in text.splitlines() if p.strip()] sentences = [] for para in paragraphs: sentences.extend(x.strip() for x in re.split(r"(?<=[.!?…])\s+", para) if x.strip()) chunks, current = [], "" for sentence in sentences: if len(sentence) > hard: clauses = [x.strip() for x in re.split(r"(?<=[,;:])\s+", sentence) if x.strip()] else: clauses = [sentence] for clause in clauses: candidate = f"{current} {clause}".strip() if current and len(candidate) > target: chunks.append(current) current = clause else: current = candidate if current: chunks.append(current) if any(len(c) > hard for c in chunks): raise ValueError("Chunk exceeds hard semantic limit") return chunks def save_status(stage, state, **extra): path = ROOT / "status.json" data = json.loads(path.read_text()) if path.exists() else {} data.update({"stage": stage, "state": state, **extra}) path.write_text(json.dumps(data, ensure_ascii=False, indent=2)) def generate_audio(story, tts): audio_dir = ROOT / "assets/audio/chunks" narration = "\n".join(part["narration"] for part in story["parts"]) chunks = [INTRO] + semantic_chunks(narration) plan = [{"index": i, "chars": len(text), "sha256": hashlib.sha256(text.encode()).hexdigest(), "text": text} for i, text in enumerate(chunks)] (ROOT / "artifacts/audio_chunk_plan.json").write_text(json.dumps(plan, ensure_ascii=False, indent=2)) completed = 0 for i, text in enumerate(chunks): out = audio_dir / f"chunk_{i:04d}.wav" if valid_audio(out): completed += 1 continue for attempt in range(1, 4): res = tts.execute({"text": text, "style": "doc_truyen", "speed": 0.95, "denoise": True, "output_path": str(out), "timeout": 300}) if res.success and valid_audio(out): completed += 1 break if out.exists(): out.unlink() if attempt == 3: raise RuntimeError(f"TTS chunk {i} failed: {res.error}") time.sleep(15 * attempt) save_status("asset_generation", "in_progress", audio_chunks=f"{completed}/{len(chunks)}") return chunks def generate_images(story, image_tool): jobs = [] for part in story["parts"]: for scene_no, prompt in enumerate(part["visual_prompts"], 1): idx = (int(part["id"]) - 1) * 8 + scene_no jobs.append((idx, prompt)) def worker(job): idx, prompt = job out = ROOT / f"assets/images/scene_{idx:03d}.png" if valid_image(out): return idx, "reused", out.stat().st_size for attempt in range(1, 4): res = image_tool.execute({"prompt": prompt + STYLE, "size": "1536x1024", "quality": "high", "output_format": "png", "output_path": str(out)}) if res.success and valid_image(out): return idx, "generated", out.stat().st_size if out.exists(): out.unlink() if attempt == 3: return idx, f"failed:{res.error}", 0 time.sleep(10 * attempt) results = [] with concurrent.futures.ThreadPoolExecutor(max_workers=5) as pool: for result in concurrent.futures.as_completed([pool.submit(worker, j) for j in jobs]): item = result.result() results.append(item) done = sum(1 for _, status, _ in results if not status.startswith("failed")) save_status("asset_generation", "in_progress", images=f"{done}/96") results.sort() failures = [x for x in results if x[1].startswith("failed")] (ROOT / "artifacts/image_manifest.json").write_text(json.dumps([ {"scene": i, "status": status, "bytes": size, "path": str(ROOT / f"assets/images/scene_{i:03d}.png")} for i, status, size in results ], ensure_ascii=False, indent=2)) if failures: raise RuntimeError(f"Image failures: {failures}") def main(): visual = json.loads((ROOT / "artifacts/visual_preflight.json").read_text()) if visual.get("visual_review") != "passed": raise RuntimeError("Visual preflight is not approved") story = json.loads((ROOT / "artifacts/story_package.json").read_text()) registry.discover() tts = registry.get("ngoc_huyen_clone_tts") image_tool = registry.get("cx_image") save_status("asset_generation", "in_progress") chunks = generate_audio(story, tts) generate_images(story, image_tool) save_status("asset_generation", "completed", audio_chunks=len(chunks), images=96) if __name__ == "__main__": main()