#!/usr/bin/env python3 import json import time import urllib.error import urllib.request from pathlib import Path PROJECT = Path('/data/video-pipeline/Channel-Loi-Co-Nhan/project/13062026-086-Dung-De-Viec-Cu-Lam-Duc-Ngay-Moi') RENDER_ENDPOINT = 'http://192.168.1.104:8019/render-story-video' PROBE_ENDPOINT = 'http://192.168.1.104:8002/probe' PACKAGE_ENDPOINT = 'http://192.168.1.104:8018/package-audio-safe' PACKAGE_PROBE_ENDPOINT = 'http://192.168.1.104:8018/probe' BLACKDETECT_ENDPOINT = 'http://192.168.1.104:8018/blackdetect' BGM = Path('/data/video-pipeline/Channel-Loi-Co-Nhan/data/bgm/paulyudin-sentimental-story-182512.mp3') THUMB = PROJECT / 'output/thumbnails/thumb_9x16_final.png' INTRO_AUDIO = PROJECT / 'output/intro/intro_audio_with_2s_tail.wav' INTRO_AUDIO_RAW = PROJECT / 'output/intro/intro_audio.wav' INTRO = PROJECT / 'output/intro/intro.mp4' MAIN = PROJECT / 'output/main_story.mp4' FINAL = PROJECT / 'output/final/final_video.mp4' SUB = PROJECT / 'subtitles/master_subtitles_asr_timing_script_text.srt' MASTER = PROJECT / 'audio/master_narration.wav' TIMELINE = PROJECT / 'audio/master_narration_timeline.json' LOG = PROJECT / 'logs/strict_worker_rebuild.log' BGM_VOLUME = 0.30 INTRO_TAIL_SILENCE_SEC = 1.0 SCENE_VISUAL_TAIL_SEC = 0.35 def log(msg): line = time.strftime('%F %T ') + msg print(line, flush=True) LOG.parent.mkdir(parents=True, exist_ok=True) with LOG.open('a', encoding='utf-8') as f: f.write(line + '\n') def post_json(url, payload, timeout=7200): req = urllib.request.Request( url, data=json.dumps(payload, ensure_ascii=False).encode('utf-8'), headers={'Content-Type': 'application/json'}, method='POST', ) try: with urllib.request.urlopen(req, timeout=timeout) as resp: return json.loads(resp.read().decode('utf-8')) except urllib.error.HTTPError as e: body = e.read().decode('utf-8', 'ignore') raise RuntimeError(f'{url} HTTP {e.code}: {body}') from e def duration_from_probe(probe): return float(probe.get('format', {}).get('duration') or probe.get('duration') or 0) def has_stream(probe, typ): return any(s.get('codec_type') == typ for s in probe.get('streams', [])) def require_inputs(): missing = [str(p) for p in [THUMB, MASTER, TIMELINE, SUB, BGM] if not p.exists()] if missing: raise SystemExit('missing_inputs: ' + ', '.join(missing)) def ensure_intro_audio_tail(): import wave raw = INTRO_AUDIO_RAW out = INTRO_AUDIO with wave.open(str(raw), 'rb') as src: params = src.getparams() frames = src.readframes(src.getnframes()) nch, sw, fr, nframes, comptype, compname = params tail_frames = int(fr * INTRO_TAIL_SILENCE_SEC) silence = b'\x00' * tail_frames * nch * sw out.parent.mkdir(parents=True, exist_ok=True) with wave.open(str(out), 'wb') as dst: dst.setparams(params) dst.writeframes(frames + silence) return out def render_intro_worker(): # Render intro with tail silence baked into the audio; 8019 cuts video to audio duration. ensure_intro_audio_tail() intro_audio_probe = post_json(PROBE_ENDPOINT, {'input_path': str(INTRO_AUDIO)}, timeout=120) intro_dur = duration_from_probe(intro_audio_probe) payload = { 'job_id': f'{PROJECT.name}_step17_worker_intro', 'scenes': [{'image_path': str(THUMB), 'duration': round(intro_dur, 3)}], 'audio_path': str(INTRO_AUDIO), 'subtitle_path': None, 'output_path': str(INTRO), 'width': 1440, 'height': 2560, 'fps': 30, 'crf': 20, 'preset': 'veryfast', 'burn_subtitles': False, 'ken_burns': False, 'motion_preset': 'none', 'transition': 'crossfade', 'transition_duration': 0.5, 'fade_in': 0.0, 'fade_out': 0.0, 'timeline_start_sec': 0.0, 'timeline_end_sec': None, 'audio_start_sec': None, 'audio_end_sec': None, 'slice_audio': False, 'slice_subtitles': False, } log('STEP17 worker render intro.mp4') response = post_json(RENDER_ENDPOINT, payload, timeout=7200) probe = post_json(PROBE_ENDPOINT, {'input_path': str(INTRO)}, timeout=120) report = {'endpoint': RENDER_ENDPOINT, 'payload': payload, 'response': response, 'probe': probe} (PROJECT / 'logs/step17_worker_intro_manifest.json').write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding='utf-8') (PROJECT / 'output/intro/intro_probe.json').write_text(json.dumps(probe, ensure_ascii=False, indent=2), encoding='utf-8') return probe def render_main_worker(): timeline = json.loads(TIMELINE.read_text(encoding='utf-8'))['items'] master_probe = post_json(PROBE_ENDPOINT, {'input_path': str(MASTER)}, timeout=120) master_duration = duration_from_probe(master_probe) scenes = [] for i, item in enumerate(timeline): image_path = item.get('image_path') or str(PROJECT / 'images/scenes' / f"{item['scene_id']}.png") start = float(item.get('start', 0.0)) next_start = float(timeline[i + 1].get('start', master_duration)) if i + 1 < len(timeline) else master_duration hold = max(float(item.get('duration', 0.5)), next_start - start) + SCENE_VISUAL_TAIL_SEC scenes.append({'image_path': image_path, 'duration': round(hold, 3)}) payload = { 'job_id': f'{PROJECT.name}_step14_worker_onepass_strict', 'scenes': scenes, 'audio_path': str(MASTER), 'subtitle_path': str(SUB), 'output_path': str(MAIN), 'width': 1440, 'height': 2560, 'fps': 30, 'crf': 20, 'preset': 'veryfast', 'burn_subtitles': True, 'ken_burns': True, 'motion_preset': 'storybook_smooth', 'transition': 'crossfade', 'transition_duration': 0.5, 'fade_in': 0.0, 'fade_out': 0.0, 'timeline_start_sec': 0.0, 'timeline_end_sec': None, 'audio_start_sec': None, 'audio_end_sec': None, 'subtitle_time_offset_sec': 0.0, 'subtitle_advance_sec': 0.0, 'slice_audio': False, 'slice_subtitles': False, 'watermark_logo_path': '/data/video-pipeline/Channel-Loi-Co-Nhan/data/logo/logoloiconhan.png', 'watermark_opacity': 1.0, } log('STEP14 worker render main_story.mp4') response = post_json(RENDER_ENDPOINT, payload, timeout=7200) probe = post_json(PROBE_ENDPOINT, {'input_path': str(MAIN)}, timeout=120) report = {'endpoint': RENDER_ENDPOINT, 'payload': payload, 'response': response, 'probe': probe} (PROJECT / 'logs/step14_worker_onepass_manifest.json').write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding='utf-8') (PROJECT / 'output/main_story_probe.json').write_text(json.dumps(probe, ensure_ascii=False, indent=2), encoding='utf-8') return probe def package_final_worker(intro_probe, main_probe): FINAL.parent.mkdir(parents=True, exist_ok=True) payload = { 'intro_video_path': str(INTRO), 'main_video_path': str(MAIN), 'bgm_audio_path': str(BGM), 'output_video_path': str(FINAL), 'sample_rate': 44100, 'channels': 2, 'width': 1080, 'height': 1920, 'bgm_volume': BGM_VOLUME, 'crf': 20, 'preset': 'veryfast', } log('STEP18 worker package intro + main + BGM') response = post_json(PACKAGE_ENDPOINT, payload, timeout=7200) probe = response.get('probe') or post_json(PACKAGE_PROBE_ENDPOINT, {'input_path': str(FINAL)}, timeout=120) black = post_json(BLACKDETECT_ENDPOINT, {'input_path': str(FINAL), 'log_path': str(FINAL.parent / 'final_blackdetect.log')}, timeout=7200) expected = duration_from_probe(intro_probe) + duration_from_probe(main_probe) report = { 'project_id': PROJECT.name, 'status': 'completed_worker_only', 'render_endpoint': RENDER_ENDPOINT, 'package_endpoint': PACKAGE_ENDPOINT, 'output_video_path': str(FINAL), 'expected_duration_sec': expected, 'final_duration_sec': duration_from_probe(probe), 'duration_diff_sec': duration_from_probe(probe) - expected, 'bgm_path': str(BGM), 'bgm_volume': BGM_VOLUME, 'probe': probe, 'blackdetect': black, 'package_response': response, } (FINAL.parent / 'final_video_probe.json').write_text(json.dumps(probe, ensure_ascii=False, indent=2), encoding='utf-8') (FINAL.parent / 'final_package_report.json').write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding='utf-8') (PROJECT / 'logs/step18_final_package_manifest.json').write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding='utf-8') return report def slug_words(text): import re s = text.lower() repl = {'đ':'d','á':'a','à':'a','ả':'a','ã':'a','ạ':'a','ă':'a','ắ':'a','ằ':'a','ẳ':'a','ẵ':'a','ặ':'a','â':'a','ấ':'a','ầ':'a','ẩ':'a','ẫ':'a','ậ':'a','é':'e','è':'e','ẻ':'e','ẽ':'e','ẹ':'e','ê':'e','ế':'e','ề':'e','ể':'e','ễ':'e','ệ':'e','í':'i','ì':'i','ỉ':'i','ĩ':'i','ị':'i','ó':'o','ò':'o','ỏ':'o','õ':'o','ọ':'o','ô':'o','ố':'o','ồ':'o','ổ':'o','ỗ':'o','ộ':'o','ơ':'o','ớ':'o','ờ':'o','ở':'o','ỡ':'o','ợ':'o','ú':'u','ù':'u','ủ':'u','ũ':'u','ụ':'u','ư':'u','ứ':'u','ừ':'u','ử':'u','ữ':'u','ự':'u','ý':'y','ỳ':'y','ỷ':'y','ỹ':'y','ỵ':'y'} for k, v in repl.items(): s = s.replace(k, v) return [w for w in re.findall(r'[a-z0-9]+', s) if len(w) > 1][:5] def build_meaningful_hashtags(topic_clean): topic_lower = topic_clean.lower() hashtags = ['#loiconhan', '#xuhuong'] keyword_map = [ (['lời lành', 'miệng nói', 'lời nói', 'giữ miệng'], ['#loilanh', '#giumieng', '#songtute']), (['nhận lỗi', 'sửa lỗi', 'xin lỗi'], ['#nhanloi', '#sualoi', '#songtute']), (['chữ tín', 'bất tín', 'giữ lời', 'lời hứa'], ['#chutin', '#giuloi', '#giuchutin']), (['vợ chồng', 'hôn nhân'], ['#vochong', '#honnhan', '#giadinhhanhphuc']), (['cha mẹ', 'hiếu'], ['#cha_me', '#dao_hieu', '#giadinh']), (['gia đình', 'nhà'], ['#giadinh', '#nepnhaxua', '#phuckhi']), (['nhẫn'], ['#chunhan', '#nhannhin', '#songbinhan']), (['lòng tham', 'tham'], ['#longtham', '#bietdu', '#giudao']), (['thiện', 'lòng tốt'], ['#songthien', '#longtot', '#phucduc']), ] for needles, tags in keyword_map: if any(n in topic_lower for n in needles): hashtags.extend(tags) break hashtags.extend(['#daolycuocsong', '#trietlycuocsong', '#songtute', '#phucduc', '#shorts']) return list(dict.fromkeys(hashtags)) def build_social_meta(): import re scenes_path = PROJECT / 'script/scenes.json' data = json.loads(scenes_path.read_text(encoding='utf-8')) if scenes_path.exists() else {} metadata = data.get('metadata', {}) title = data.get('title') or metadata.get('title') or PROJECT.name topic = (metadata.get('topic_sentence') or title.split('|')[0]).strip().rstrip('.') topic_clean = topic.split(':', 1)[1].strip() if topic.lower().startswith('lời cổ nhân dạy:') else topic first_narration = '' for scene in data.get('scenes', []): first_narration = (scene.get('narration') or '').strip() if first_narration: break description = first_narration or f'Cổ nhân dạy: {topic_clean}. Biết giữ lòng bình thản, giữ lời nói cho hiền, đời người tự khắc bớt sóng gió và thêm phúc khí.' description = re.sub(r'\s+', ' ', description).strip() if len(description) > 280: description = description[:277].rsplit(' ', 1)[0] + '...' hashtags = build_meaningful_hashtags(topic_clean) tags = ['lời cổ nhân', 'cổ nhân dạy', topic_clean.lower(), 'đạo lý làm người', 'triết lý cuộc sống', 'sống tử tế', 'phúc đức'] caption = f'{topic_clean}. ' + ' '.join(hashtags[:10]) return { 'title': title, 'upload_title': title, 'description': description, 'short_description': f'{topic_clean}. Lời cổ nhân dạy để sống chậm, sống tỉnh và giữ phúc cho mình.', 'caption': caption, 'hashtags': hashtags, 'tags': tags, 'category': 'Education', 'language': 'vi', 'privacy_status': 'private', 'made_for_kids': False, 'social_meta': { 'youtube': {'title': title, 'description': description + '\n\n' + ' '.join(hashtags[:10]), 'tags': tags[:7], 'category': 'Education'}, 'tiktok': {'caption': caption}, 'facebook_reels': {'caption': f'Cổ nhân dạy: {topic_clean}.\n\n' + ' '.join(hashtags[:10])}, 'instagram_reels': {'caption': f'{topic_clean}.\n\n' + ' '.join(hashtags[:10])}, }, } def main(): LOG.write_text('', encoding='utf-8') require_inputs() intro_probe = render_intro_worker() if not has_stream(intro_probe, 'audio') or not has_stream(intro_probe, 'video'): raise SystemExit('worker_intro_missing_audio_or_video') main_probe = render_main_worker() if not has_stream(main_probe, 'audio') or not has_stream(main_probe, 'video'): raise SystemExit('worker_main_missing_audio_or_video') report = package_final_worker(intro_probe, main_probe) info_path = PROJECT / 'output/info.json' info = json.loads(info_path.read_text(encoding='utf-8')) if info_path.exists() else {} info.update(build_social_meta()) info.update({ 'status': 'completed_worker_only_needs_final_human_qa', 'main_story': str(MAIN), 'final_video': str(FINAL), 'duration_sec': report['final_duration_sec'], 'render_endpoint': RENDER_ENDPOINT, 'package_endpoint': PACKAGE_ENDPOINT, 'host_ffmpeg_render_used': False, 'worker_only_rebuild': True, }) info_path.write_text(json.dumps(info, ensure_ascii=False, indent=2), encoding='utf-8') log(f'COMPLETE worker-only rebuild final_duration={report["final_duration_sec"]:.3f}') if __name__ == '__main__': main()