diff --git a/README.md b/README.md index 3d106eb2..fe2aea8b 100644 --- a/README.md +++ b/README.md @@ -120,6 +120,16 @@ python setup_env.py Or double-click `OneKeyStart_uv.bat` on Windows. +### Task Workspaces + +When you create a task from the Streamlit sidebar, VideoLingo stores its files under `workspace/jobs//`. Each task keeps its own config snapshot and `output/` folder so separate videos do not overwrite each other's intermediate files. + +### Human Review Workbench + +When a task workspace is active, VideoLingo shows a Review Workbench for terminology, translation, subtitle rows, and TTS text. Each review file belongs to the active workspace, and saving changes creates a backup under `artifacts/reviews/backups/`. + +The app pauses at review checkpoints so you can confirm terminology before translation, translation before subtitle splitting, subtitles before SRT generation, and TTS text before dubbing. + ### Option B: Using Conda > ⚠️ **Not recommended.** This method will not be maintained going forward. Please use uv (Option A) above. diff --git a/batch/utils/batch_processor.py b/batch/utils/batch_processor.py index 3f2917cd..03f03eda 100644 --- a/batch/utils/batch_processor.py +++ b/batch/utils/batch_processor.py @@ -2,64 +2,59 @@ import gc from batch.utils.settings_check import check_settings from batch.utils.video_processor import process_video -from core.utils.config_utils import load_key, update_key +from core.utils.config_utils import update_key +from core import workspace import pandas as pd from rich.console import Console from rich.panel import Panel import time -import shutil console = Console() +def ensure_workspace_columns(df): + if 'Workspace' not in df.columns: + df['Workspace'] = '' + return df + + +def get_or_create_batch_workspace(row, index): + current = row.get('Workspace', '') + if isinstance(current, str) and current and os.path.exists(current): + return current + + video_file = str(row['Video File']) + source_is_url = video_file.startswith('http') + job = workspace.create_job( + name=os.path.splitext(os.path.basename(video_file))[0] or f"Batch Task {index + 1}", + source_type='batch', + source_path='' if source_is_url else video_file, + source_url=video_file if source_is_url else '', + ) + return job['path'] + + def record_and_update_config(source_language, target_language): - original_source_lang = load_key('whisper.language') - original_target_lang = load_key('target_language') - if source_language and not pd.isna(source_language): update_key('whisper.language', source_language) if target_language and not pd.isna(target_language): update_key('target_language', target_language) - - return original_source_lang, original_target_lang def process_batch(): if not check_settings(): raise Exception("Settings check failed") - df = pd.read_excel('batch/tasks_setting.xlsx') + df = ensure_workspace_columns(pd.read_excel('batch/tasks_setting.xlsx')) for index, row in df.iterrows(): if pd.isna(row['Status']) or 'Error' in str(row['Status']): total_tasks = len(df) video_file = row['Video File'] + workspace_path = get_or_create_batch_workspace(row, index) + df.at[index, 'Workspace'] = workspace_path + workspace.set_active_workspace(workspace_path) if not pd.isna(row['Status']) and 'Error' in str(row['Status']): console.print(Panel(f"Retrying failed task: {video_file}\nTask {index + 1}/{total_tasks}", title="[bold yellow]Retry Task", expand=False)) - - # Restore files from batch/output/ERROR to output - error_folder = os.path.join('batch', 'output', 'ERROR', os.path.splitext(video_file)[0]) - - if os.path.exists(error_folder): - # Ensure the output folder exists - os.makedirs('output', exist_ok=True) - - # Copy all contents from ERROR folder for the specific video to output - for item in os.listdir(error_folder): - src_path = os.path.join(error_folder, item) - dst_path = os.path.join('output', item) - - if os.path.isdir(src_path): - if os.path.exists(dst_path): - shutil.rmtree(dst_path) - shutil.copytree(src_path, dst_path) - else: - if os.path.exists(dst_path): - os.remove(dst_path) - shutil.copy2(src_path, dst_path) - - console.print(f"[green]Restored files from ERROR folder for {video_file}") - else: - console.print(f"[yellow]Warning: Error folder not found: {error_folder}") else: console.print(Panel(f"Now processing task: {video_file}\nTask {index + 1}/{total_tasks}", title="[bold blue]Current Task", expand=False)) @@ -67,7 +62,7 @@ def process_batch(): source_language = row['Source Language'] target_language = row['Target Language'] - original_source_lang, original_target_lang = record_and_update_config(source_language, target_language) + record_and_update_config(source_language, target_language) try: dubbing = 0 if pd.isna(row['Dubbing']) else int(row['Dubbing']) @@ -78,8 +73,7 @@ def process_batch(): status_msg = f"Error: Unhandled exception - {str(e)}" console.print(f"[bold red]Error processing {video_file}: {status_msg}") finally: - update_key('whisper.language', original_source_lang) - update_key('target_language', original_target_lang) + workspace.clear_active_workspace() df.at[index, 'Status'] = status_msg df.to_excel('batch/tasks_setting.xlsx', index=False) @@ -90,8 +84,8 @@ def process_batch(): else: print(f"Skipping task: {row['Video File']} - Status: {row['Status']}") - console.print(Panel("All tasks processed!\nCheck out in `batch/output`!", + console.print(Panel("All tasks processed!\nCheck each row's `Workspace` folder for outputs.", title="[bold green]Batch Processing Complete", expand=False)) if __name__ == "__main__": - process_batch() \ No newline at end of file + process_batch() diff --git a/batch/utils/video_processor.py b/batch/utils/video_processor.py index 48ab9760..0e4abd2f 100644 --- a/batch/utils/video_processor.py +++ b/batch/utils/video_processor.py @@ -1,7 +1,7 @@ import os from core.st_utils.imports_and_utils import * -from core.utils.onekeycleanup import cleanup from core.utils import load_key +from core.workspace import output_path import shutil from functools import partial from rich.panel import Panel @@ -11,14 +11,11 @@ console = Console() INPUT_DIR = 'batch/input' -OUTPUT_DIR = 'output' -SAVE_DIR = 'batch/output' -ERROR_OUTPUT_DIR = 'batch/output/ERROR' YTB_RESOLUTION_KEY = "ytb_resolution" def process_video(file, dubbing=False, is_retry=False): if not is_retry: - prepare_output_folder(OUTPUT_DIR) + prepare_output_folder(str(output_path())) text_steps = [ ("🎥 Processing input file", partial(process_input_file, file)), @@ -60,7 +57,6 @@ def process_video(file, dubbing=False, is_retry=False): border_style="red" ) console.print(error_panel) - cleanup(ERROR_OUTPUT_DIR) return False, current_step, str(e) console.print(Panel( f"[yellow]Attempt {attempt + 1} failed. Retrying...[/]", @@ -68,7 +64,6 @@ def process_video(file, dubbing=False, is_retry=False): )) console.print(Panel("[bold green]All steps completed successfully! 🎉[/]", border_style="green")) - cleanup(SAVE_DIR) return True, "", "" def prepare_output_folder(output_folder): @@ -82,7 +77,7 @@ def process_input_file(file): video_file = _1_ytdlp.find_video_files() else: input_file = os.path.join('batch', 'input', file) - output_file = os.path.join(OUTPUT_DIR, file) + output_file = os.path.join(str(output_path()), file) shutil.copy(input_file, output_file) video_file = output_file return {'video_file': video_file} diff --git a/core/_10_gen_audio.py b/core/_10_gen_audio.py index 94e52454..9022c51c 100644 --- a/core/_10_gen_audio.py +++ b/core/_10_gen_audio.py @@ -17,10 +17,17 @@ console = Console() -TEMP_FILE_TEMPLATE = f"{_AUDIO_TMP_DIR}/{{}}_temp.wav" -OUTPUT_FILE_TEMPLATE = f"{_AUDIO_SEGS_DIR}/{{}}.wav" WARMUP_SIZE = 5 + +def temp_file_path(number, line_index): + return os.path.join(str(_AUDIO_TMP_DIR), f"{number}_{line_index}_temp.wav") + + +def output_file_path(number, line_index): + return os.path.join(str(_AUDIO_SEGS_DIR), f"{number}_{line_index}.wav") + + def parse_df_srt_time(time_str: str) -> float: """Convert SRT time format to seconds""" hours, minutes, seconds = time_str.strip().split(':') @@ -68,7 +75,7 @@ def process_row(row: pd.Series, tasks_df: pd.DataFrame) -> Tuple[int, float]: lines = eval(row['lines']) if isinstance(row['lines'], str) else row['lines'] real_dur = 0 for line_index, line in enumerate(lines): - temp_file = TEMP_FILE_TEMPLATE.format(f"{number}_{line_index}") + temp_file = temp_file_path(number, line_index) tts_main(line, temp_file, number, tasks_df) real_dur += get_audio_duration(temp_file) return number, real_dur @@ -165,8 +172,8 @@ def merge_chunks(tasks_df: pd.DataFrame) -> pd.DataFrame: lines = eval(row['lines']) if isinstance(row['lines'], str) else row['lines'] for line_index, line in enumerate(lines): # 🔄 Step2: Start speed change and save as OUTPUT_FILE_TEMPLATE - temp_file = TEMP_FILE_TEMPLATE.format(f"{number}_{line_index}") - output_file = OUTPUT_FILE_TEMPLATE.format(f"{number}_{line_index}") + temp_file = temp_file_path(number, line_index) + output_file = output_file_path(number, line_index) adjust_audio_speed(temp_file, output_file, speed_factor) ad_dur = get_audio_duration(output_file) new_sub_times.append([cur_time, cur_time+ad_dur]) @@ -186,7 +193,7 @@ def merge_chunks(tasks_df: pd.DataFrame) -> pd.DataFrame: last_number = tasks_df.iloc[index]['number'] last_lines = eval(tasks_df.iloc[index]['lines']) if isinstance(tasks_df.iloc[index]['lines'], str) else tasks_df.iloc[index]['lines'] last_line_index = len(last_lines) - 1 - last_file = OUTPUT_FILE_TEMPLATE.format(f"{last_number}_{last_line_index}") + last_file = output_file_path(last_number, last_line_index) # Calculate the duration to keep audio = AudioSegment.from_wav(last_file) diff --git a/core/_11_merge_audio.py b/core/_11_merge_audio.py index 41c8ac16..2f0cd20b 100644 --- a/core/_11_merge_audio.py +++ b/core/_11_merge_audio.py @@ -6,12 +6,16 @@ from rich.console import Console from core.utils import * from core.utils.models import * +from core.workspace import output_path console = Console() -DUB_VOCAL_FILE = 'output/dub.mp3' +DUB_VOCAL_FILE = output_path("dub.mp3") -DUB_SUB_FILE = 'output/dub.srt' -OUTPUT_FILE_TEMPLATE = f"{_AUDIO_SEGS_DIR}/{{}}.wav" +DUB_SUB_FILE = output_path("dub.srt") + + +def output_file_path(number, line_index): + return os.path.join(str(_AUDIO_SEGS_DIR), f"{number}_{line_index}.wav") def load_and_flatten_data(excel_file): """Load and flatten Excel data""" @@ -31,7 +35,7 @@ def get_audio_files(df): number = row['number'] line_count = len(eval(row['lines']) if isinstance(row['lines'], str) else row['lines']) for line_index in range(line_count): - temp_file = OUTPUT_FILE_TEMPLATE.format(f"{number}_{line_index}") + temp_file = output_file_path(number, line_index) audios.append(temp_file) return audios @@ -123,9 +127,9 @@ def merge_full_audio(): with console.status("[bold cyan]💾 Exporting final audio file...[/bold cyan]"): merged_audio = merged_audio.set_frame_rate(16000).set_channels(1) - merged_audio.export(DUB_VOCAL_FILE, format="mp3", parameters=["-b:a", "64k"]) + merged_audio.export(str(DUB_VOCAL_FILE), format="mp3", parameters=["-b:a", "64k"]) console.print(f"[bold green]✅ Audio file successfully merged![/bold green]") console.print(f"[bold green]📁 Output file: {DUB_VOCAL_FILE}[/bold green]") if __name__ == "__main__": - merge_full_audio() \ No newline at end of file + merge_full_audio() diff --git a/core/_12_dub_to_vid.py b/core/_12_dub_to_vid.py index da7b2895..769a603a 100644 --- a/core/_12_dub_to_vid.py +++ b/core/_12_dub_to_vid.py @@ -9,12 +9,13 @@ from core.asr_backend.audio_preprocess import normalize_audio_volume from core.utils import * from core.utils.models import * +from core.workspace import output_path console = Console() -DUB_VIDEO = "output/output_dub.mp4" -DUB_SUB_FILE = 'output/dub.srt' -DUB_AUDIO = 'output/dub.mp3' +DUB_VIDEO = output_path("output_dub.mp4") +DUB_SUB_FILE = output_path("dub.srt") +DUB_AUDIO = output_path("dub.mp3") TRANS_FONT_SIZE = 17 TRANS_FONT_NAME = 'Arial' @@ -31,7 +32,10 @@ def merge_video_audio(): """Merge video and audio, and reduce video volume""" VIDEO_FILE = find_video_files() - background_file = _BACKGROUND_AUDIO_FILE + background_file = str(_BACKGROUND_AUDIO_FILE) + dub_video = str(DUB_VIDEO) + dub_audio = str(DUB_AUDIO) + dub_sub_file = str(DUB_SUB_FILE) if not load_key("burn_subtitles"): rprint("[bold yellow]Warning: A 0-second black video will be generated as a placeholder as subtitles are not burned in.[/bold yellow]") @@ -39,7 +43,7 @@ def merge_video_audio(): # Create a black frame frame = np.zeros((1080, 1920, 3), dtype=np.uint8) fourcc = cv2.VideoWriter_fourcc(*'mp4v') - out = cv2.VideoWriter(DUB_VIDEO, fourcc, 1, (1920, 1080)) + out = cv2.VideoWriter(dub_video, fourcc, 1, (1920, 1080)) out.write(frame) out.release() @@ -47,8 +51,8 @@ def merge_video_audio(): return # Normalize dub audio - normalized_dub_audio = 'output/normalized_dub.wav' - normalize_audio_volume(DUB_AUDIO, normalized_dub_audio) + normalized_dub_audio = str(output_path("normalized_dub.wav")) + normalize_audio_volume(dub_audio, normalized_dub_audio) # Merge video and audio with translated subtitles video = cv2.VideoCapture(VIDEO_FILE) @@ -58,7 +62,7 @@ def merge_video_audio(): rprint(f"[bold green]Video resolution: {TARGET_WIDTH}x{TARGET_HEIGHT}[/bold green]") subtitle_filter = ( - f"subtitles={DUB_SUB_FILE}:force_style='FontSize={TRANS_FONT_SIZE}," + f"subtitles={dub_sub_file}:force_style='FontSize={TRANS_FONT_SIZE}," f"FontName={TRANS_FONT_NAME},PrimaryColour={TRANS_FONT_COLOR}," f"OutlineColour={TRANS_OUTLINE_COLOR},OutlineWidth={TRANS_OUTLINE_WIDTH}," f"BackColour={TRANS_BACK_COLOR},Alignment=2,MarginV=27,BorderStyle=4'" @@ -79,10 +83,10 @@ def merge_video_audio(): else: cmd.extend(['-map', '[v]', '-map', '[a]']) - cmd.extend(['-c:a', 'aac', '-b:a', '96k', DUB_VIDEO]) + cmd.extend(['-c:a', 'aac', '-b:a', '96k', dub_video]) subprocess.run(cmd) - rprint(f"[bold green]Video and audio successfully merged into {DUB_VIDEO}[/bold green]") + rprint(f"[bold green]Video and audio successfully merged into {dub_video}[/bold green]") if __name__ == '__main__': merge_video_audio() diff --git a/core/_1_ytdlp.py b/core/_1_ytdlp.py index efe47a8a..15439139 100644 --- a/core/_1_ytdlp.py +++ b/core/_1_ytdlp.py @@ -3,6 +3,7 @@ import re import subprocess from core.utils import * +from core.workspace import output_path def sanitize_filename(filename): # Remove or replace illegal characters @@ -23,7 +24,8 @@ def update_ytdlp(): from yt_dlp import YoutubeDL return YoutubeDL -def download_video_ytdlp(url, save_path='output', resolution='1080'): +def download_video_ytdlp(url, save_path=None, resolution='1080'): + save_path = str(save_path or output_path()) os.makedirs(save_path, exist_ok=True) ydl_opts = { 'format': 'bestvideo+bestaudio/best' if resolution == 'best' else f'bestvideo[height<={resolution}]+bestaudio/best[height<={resolution}]', @@ -51,12 +53,13 @@ def download_video_ytdlp(url, save_path='output', resolution='1080'): if new_filename != filename: os.rename(os.path.join(save_path, file), os.path.join(save_path, new_filename + ext)) -def find_video_files(save_path='output'): +def find_video_files(save_path=None): + save_path = str(save_path or output_path()) video_files = [file for file in glob.glob(save_path + "/*") if os.path.splitext(file)[1][1:].lower() in load_key("allowed_video_formats")] # change \\ to /, this happen on windows if sys.platform.startswith('win'): video_files = [file.replace("\\", "/") for file in video_files] - video_files = [file for file in video_files if not file.startswith("output/output")] + video_files = [file for file in video_files if not os.path.basename(file).startswith("output")] if len(video_files) != 1: raise ValueError(f"Number of videos found {len(video_files)} is not unique. Please check.") return video_files[0] diff --git a/core/_7_sub_into_vid.py b/core/_7_sub_into_vid.py index 7a2e253f..536b6369 100644 --- a/core/_7_sub_into_vid.py +++ b/core/_7_sub_into_vid.py @@ -4,6 +4,7 @@ import numpy as np import platform from core.utils import * +from core.workspace import output_path SRC_FONT_SIZE = 15 TRANS_FONT_SIZE = 17 @@ -28,10 +29,9 @@ TRANS_OUTLINE_WIDTH = 1 TRANS_BACK_COLOR = '&H33000000' -OUTPUT_DIR = "output" -OUTPUT_VIDEO = f"{OUTPUT_DIR}/output_sub.mp4" -SRC_SRT = f"{OUTPUT_DIR}/src.srt" -TRANS_SRT = f"{OUTPUT_DIR}/trans.srt" +OUTPUT_VIDEO = output_path("output_sub.mp4") +SRC_SRT = output_path("src.srt") +TRANS_SRT = output_path("trans.srt") def check_gpu_available(): try: @@ -42,7 +42,10 @@ def check_gpu_available(): def merge_subtitles_to_video(): video_file = find_video_files() - os.makedirs(os.path.dirname(OUTPUT_VIDEO), exist_ok=True) + output_video = str(OUTPUT_VIDEO) + src_srt = str(SRC_SRT) + trans_srt = str(TRANS_SRT) + os.makedirs(os.path.dirname(output_video), exist_ok=True) # Check resolution if not load_key("burn_subtitles"): @@ -51,14 +54,14 @@ def merge_subtitles_to_video(): # Create a black frame frame = np.zeros((1080, 1920, 3), dtype=np.uint8) fourcc = cv2.VideoWriter_fourcc(*'mp4v') - out = cv2.VideoWriter(OUTPUT_VIDEO, fourcc, 1, (1920, 1080)) + out = cv2.VideoWriter(output_video, fourcc, 1, (1920, 1080)) out.write(frame) out.release() rprint("[bold green]Placeholder video has been generated.[/bold green]") return - if not os.path.exists(SRC_SRT) or not os.path.exists(TRANS_SRT): + if not os.path.exists(src_srt) or not os.path.exists(trans_srt): rprint("Subtitle files not found in the 'output' directory.") exit(1) @@ -72,10 +75,10 @@ def merge_subtitles_to_video(): '-vf', ( f"scale={TARGET_WIDTH}:{TARGET_HEIGHT}:force_original_aspect_ratio=decrease," f"pad={TARGET_WIDTH}:{TARGET_HEIGHT}:(ow-iw)/2:(oh-ih)/2," - f"subtitles={SRC_SRT}:force_style='FontSize={SRC_FONT_SIZE},FontName={FONT_NAME}," + f"subtitles={src_srt}:force_style='FontSize={SRC_FONT_SIZE},FontName={FONT_NAME}," f"PrimaryColour={SRC_FONT_COLOR},OutlineColour={SRC_OUTLINE_COLOR},OutlineWidth={SRC_OUTLINE_WIDTH}," f"ShadowColour={SRC_SHADOW_COLOR},BorderStyle=1'," - f"subtitles={TRANS_SRT}:force_style='FontSize={TRANS_FONT_SIZE},FontName={TRANS_FONT_NAME}," + f"subtitles={trans_srt}:force_style='FontSize={TRANS_FONT_SIZE},FontName={TRANS_FONT_NAME}," f"PrimaryColour={TRANS_FONT_COLOR},OutlineColour={TRANS_OUTLINE_COLOR},OutlineWidth={TRANS_OUTLINE_WIDTH}," f"BackColour={TRANS_BACK_COLOR},Alignment=2,MarginV=27,BorderStyle=4'" ).encode('utf-8'), @@ -85,7 +88,7 @@ def merge_subtitles_to_video(): if ffmpeg_gpu: rprint("[bold green]will use GPU acceleration.[/bold green]") ffmpeg_cmd.extend(['-c:v', 'h264_nvenc']) - ffmpeg_cmd.extend(['-y', OUTPUT_VIDEO]) + ffmpeg_cmd.extend(['-y', output_video]) rprint("🎬 Start merging subtitles to video...") start_time = time.time() @@ -103,4 +106,4 @@ def merge_subtitles_to_video(): process.kill() if __name__ == "__main__": - merge_subtitles_to_video() \ No newline at end of file + merge_subtitles_to_video() diff --git a/core/_8_1_audio_task.py b/core/_8_1_audio_task.py index b570eb05..7a2a912b 100644 --- a/core/_8_1_audio_task.py +++ b/core/_8_1_audio_task.py @@ -7,12 +7,13 @@ from core.tts_backend.estimate_duration import init_estimator, estimate_duration from core.utils import * from core.utils.models import * +from core.workspace import output_path console = Console() speed_factor = load_key("speed_factor") -TRANS_SUBS_FOR_AUDIO_FILE = 'output/audio/trans_subs_for_audio.srt' -SRC_SUBS_FOR_AUDIO_FILE = 'output/audio/src_subs_for_audio.srt' +TRANS_SUBS_FOR_AUDIO_FILE = output_path("audio", "trans_subs_for_audio.srt") +SRC_SUBS_FOR_AUDIO_FILE = output_path("audio", "src_subs_for_audio.srt") ESTIMATOR = None def check_len_then_trim(text, duration): @@ -140,4 +141,4 @@ def gen_audio_task_main(): rprint(Panel(f"Successfully generated {_8_1_AUDIO_TASK}", title="Success", border_style="green")) if __name__ == '__main__': - gen_audio_task_main() \ No newline at end of file + gen_audio_task_main() diff --git a/core/_8_2_dub_chunks.py b/core/_8_2_dub_chunks.py index aba23fc1..c6292833 100644 --- a/core/_8_2_dub_chunks.py +++ b/core/_8_2_dub_chunks.py @@ -6,9 +6,10 @@ from core.tts_backend.estimate_duration import init_estimator, estimate_duration from core.utils import * from core.utils.models import * +from core.workspace import output_path -SRC_SRT = "output/src.srt" -TRANS_SRT = "output/trans.srt" +SRC_SRT = output_path("src.srt") +TRANS_SRT = output_path("trans.srt") MAX_MERGE_COUNT = 5 ESTIMATOR = None @@ -203,4 +204,4 @@ def clean_text(text): rprint("[✅ Complete] Matching completed successfully!") if __name__ == "__main__": - gen_dub_chunks() \ No newline at end of file + gen_dub_chunks() diff --git a/core/__init__.py b/core/__init__.py index d42b9c33..140b9492 100644 --- a/core/__init__.py +++ b/core/__init__.py @@ -1,5 +1,7 @@ # use try-except to avoid error when installing try: + from . import workspace + from . import review from . import ( _1_ytdlp, _2_asr, @@ -43,5 +45,7 @@ '_9_refer_audio', '_10_gen_audio', '_11_merge_audio', - '_12_dub_to_vid' + '_12_dub_to_vid', + 'workspace', + 'review', ] diff --git a/core/asr_backend/audio_preprocess.py b/core/asr_backend/audio_preprocess.py index 0d0db2ff..f80ec39d 100644 --- a/core/asr_backend/audio_preprocess.py +++ b/core/asr_backend/audio_preprocess.py @@ -158,7 +158,7 @@ def process_transcription(result: Dict) -> pd.DataFrame: return pd.DataFrame(all_words) def save_results(df: pd.DataFrame): - os.makedirs('output/log', exist_ok=True) + os.makedirs(os.path.dirname(str(_2_CLEANED_CHUNKS)), exist_ok=True) # Remove rows where 'text' is empty initial_rows = len(df) @@ -178,4 +178,4 @@ def save_results(df: pd.DataFrame): rprint(f"[green]📊 Excel file saved to {_2_CLEANED_CHUNKS}[/green]") def save_language(language: str): - update_key("whisper.detected_language", language) \ No newline at end of file + update_key("whisper.detected_language", language) diff --git a/core/asr_backend/elevenlabs_asr.py b/core/asr_backend/elevenlabs_asr.py index 5a4c5dda..285194cc 100644 --- a/core/asr_backend/elevenlabs_asr.py +++ b/core/asr_backend/elevenlabs_asr.py @@ -7,6 +7,7 @@ import soundfile as sf from rich import print as rprint from core.utils import * +from core.workspace import output_path # ---------------------------------------- # ISO 639-2 to 1 @@ -66,7 +67,7 @@ def elev2whisper(elev_json, word_level_timestamp = False): def transcribe_audio_elevenlabs(raw_audio_path, vocal_audio_path, start = None, end = None): rprint(f"[cyan]🎤 Processing audio transcription, file path: {vocal_audio_path}[/cyan]") - LOG_FILE = f"output/log/elevenlabs_transcribe_{start}_{end}.json" + LOG_FILE = str(output_path("log", f"elevenlabs_transcribe_{start}_{end}.json")) if os.path.exists(LOG_FILE): with open(LOG_FILE, "r", encoding="utf-8") as f: return json.load(f) @@ -141,5 +142,5 @@ def transcribe_audio_elevenlabs(raw_audio_path, vocal_audio_path, start = None, print(result) # Save result to file - with open("output/transcript.json", "w", encoding="utf-8") as f: + with open(str(output_path("transcript.json")), "w", encoding="utf-8") as f: json.dump(result, f, ensure_ascii=False, indent=4) diff --git a/core/asr_backend/whisperX_302.py b/core/asr_backend/whisperX_302.py index 3501470c..4200a1f8 100644 --- a/core/asr_backend/whisperX_302.py +++ b/core/asr_backend/whisperX_302.py @@ -8,11 +8,13 @@ from rich import print as rprint from core.utils import * from core.utils.models import * +from core.workspace import output_path -OUTPUT_LOG_DIR = "output/log" +OUTPUT_LOG_DIR = output_path("log") def transcribe_audio_302(raw_audio_path: str, vocal_audio_path: str, start: float = None, end: float = None): - os.makedirs(OUTPUT_LOG_DIR, exist_ok=True) - LOG_FILE = f"{OUTPUT_LOG_DIR}/whisperx302_{start}_{end}.json" + output_log_dir = str(OUTPUT_LOG_DIR) + os.makedirs(output_log_dir, exist_ok=True) + LOG_FILE = os.path.join(output_log_dir, f"whisperx302_{start}_{end}.json") if os.path.exists(LOG_FILE): with open(LOG_FILE, "r", encoding="utf-8") as f: return json.load(f) diff --git a/core/review.py b/core/review.py new file mode 100644 index 00000000..ef71df3c --- /dev/null +++ b/core/review.py @@ -0,0 +1,265 @@ +from __future__ import annotations + +import json +import shutil +from datetime import datetime +from pathlib import Path +from typing import Any + +import pandas as pd +from ruamel.yaml import YAML + +from core import workspace + +REVIEW_ROOT = Path("artifacts") / "reviews" +BACKUP_DIR = REVIEW_ROOT / "backups" +STATUS_FILE = REVIEW_ROOT / "review_status.yaml" + +STAGES = { + "terminology": "output/log/terminology.json", + "translation": "output/log/translation_results.xlsx", + "subtitles": "output/log/translation_results_for_subtitles.xlsx", + "tts_text": "output/audio/tts_tasks.xlsx", +} + +DOWNSTREAM = { + "terminology": ("translation", "subtitles", "tts_text"), + "translation": ("subtitles", "tts_text"), + "subtitles": ("tts_text",), + "tts_text": (), +} + +yaml = YAML() +yaml.preserve_quotes = True + + +def now_iso() -> str: + return datetime.now().astimezone().isoformat(timespec="seconds") + + +def active_workspace_dir() -> Path: + active = workspace.get_active_workspace() + if not active: + raise RuntimeError("Select or create a task workspace before reviewing artifacts.") + return Path(active) + + +def review_root() -> Path: + return active_workspace_dir() / REVIEW_ROOT + + +def review_status_path() -> Path: + return active_workspace_dir() / STATUS_FILE + + +def default_status() -> dict[str, Any]: + return { + "stages": { + stage: { + "status": "pending", + "artifact": artifact, + "confirmed_at": "", + "updated_at": "", + } + for stage, artifact in STAGES.items() + } + } + + +def save_status(status: dict[str, Any]) -> None: + path = review_status_path() + path.parent.mkdir(parents=True, exist_ok=True) + with open(path, "w", encoding="utf-8") as file: + yaml.dump(status, file) + + +def normalize_status(status: dict[str, Any] | None) -> dict[str, Any]: + merged = default_status() + for stage, values in (status or {}).get("stages", {}).items(): + if stage in merged["stages"] and isinstance(values, dict): + merged["stages"][stage].update(values) + return merged + + +def load_status() -> dict[str, Any]: + path = review_status_path() + if not path.exists(): + status = default_status() + save_status(status) + return status + with open(path, "r", encoding="utf-8") as file: + status = normalize_status(yaml.load(file) or {}) + save_status(status) + return status + + +def validate_stage(stage: str) -> None: + if stage not in STAGES: + raise ValueError(f"Unknown review stage: {stage}") + + +def mark_confirmed(stage: str) -> dict[str, Any]: + validate_stage(stage) + status = load_status() + stamp = now_iso() + status["stages"][stage]["status"] = "confirmed" + status["stages"][stage]["confirmed_at"] = stamp + status["stages"][stage]["updated_at"] = stamp + save_status(status) + return status + + +def mark_needs_review(stage: str) -> dict[str, Any]: + validate_stage(stage) + status = load_status() + status["stages"][stage]["status"] = "pending" + status["stages"][stage]["confirmed_at"] = "" + status["stages"][stage]["updated_at"] = now_iso() + save_status(status) + return status + + +def invalidate_downstream( + stage: str, status: dict[str, Any] | None = None +) -> dict[str, Any]: + validate_stage(stage) + status = status or load_status() + stamp = now_iso() + for downstream in DOWNSTREAM[stage]: + status["stages"][downstream]["status"] = "stale" + status["stages"][downstream]["confirmed_at"] = "" + status["stages"][downstream]["updated_at"] = stamp + save_status(status) + return status + + +def artifact_path(stage: str) -> Path: + validate_stage(stage) + return active_workspace_dir() / STAGES[stage] + + +def backup_artifact(stage: str, source: Path | None = None) -> Path | None: + validate_stage(stage) + source = source or artifact_path(stage) + if not source.exists(): + return None + backup_dir = active_workspace_dir() / BACKUP_DIR + backup_dir.mkdir(parents=True, exist_ok=True) + backup = ( + backup_dir + / f"{stage}-{datetime.now().astimezone().strftime('%Y%m%d-%H%M%S')}{source.suffix}" + ) + shutil.copy2(source, backup) + return backup + + +def mark_saved(stage: str, status_value: str = "pending") -> dict[str, Any]: + validate_stage(stage) + status = load_status() + status["stages"][stage]["status"] = status_value + status["stages"][stage]["confirmed_at"] = "" + status["stages"][stage]["updated_at"] = now_iso() + save_status(status) + return invalidate_downstream(stage, status) + + +def load_terminology() -> dict[str, Any]: + path = artifact_path("terminology") + if not path.exists(): + raise FileNotFoundError(path) + with open(path, "r", encoding="utf-8") as file: + data = json.load(file) + data.setdefault("terms", []) + return data + + +def save_terminology( + terms: list[dict[str, Any]], theme: str | None = None +) -> dict[str, Any]: + path = artifact_path("terminology") + path.parent.mkdir(parents=True, exist_ok=True) + existing = {} + if path.exists(): + with open(path, "r", encoding="utf-8") as file: + existing = json.load(file) or {} + backup_artifact("terminology", path) + + cleaned_terms = [] + for term in terms: + cleaned_terms.append( + { + "src": str(term.get("src", "")).strip(), + "tgt": str(term.get("tgt", "")).strip(), + "note": str(term.get("note", "")).strip(), + } + ) + + existing["terms"] = cleaned_terms + if theme is not None: + existing["theme"] = theme + + with open(path, "w", encoding="utf-8") as file: + json.dump(existing, file, ensure_ascii=False, indent=4) + mark_saved("terminology") + return existing + + +def load_table(stage: str) -> pd.DataFrame: + path = artifact_path(stage) + if not path.exists(): + raise FileNotFoundError(path) + return pd.read_excel(path) + + +def validate_required_columns( + df: pd.DataFrame, required_columns: tuple[str, ...] +) -> None: + missing = [column for column in required_columns if column not in df.columns] + if missing: + raise ValueError(f"Missing required columns: {', '.join(missing)}") + + +def save_table( + stage: str, + df: pd.DataFrame, + required_columns: tuple[str, ...], + invalidate: bool = True, +) -> None: + validate_stage(stage) + validate_required_columns(df, required_columns) + path = artifact_path(stage) + path.parent.mkdir(parents=True, exist_ok=True) + if path.exists(): + backup_artifact(stage, path) + df.to_excel(path, index=False) + if invalidate: + mark_saved(stage) + + +def save_subtitles(df: pd.DataFrame) -> None: + validate_required_columns(df, ("Source", "Translation")) + save_table( + "subtitles", + df, + required_columns=("Source", "Translation"), + invalidate=False, + ) + + remerged_path = active_workspace_dir() / "output/log/translation_results_remerged.xlsx" + if remerged_path.exists(): + remerged = pd.read_excel(remerged_path) + if len(remerged) == len(df) and {"Source", "Translation"}.issubset( + remerged.columns + ): + backup_dir = active_workspace_dir() / BACKUP_DIR + backup_dir.mkdir(parents=True, exist_ok=True) + backup = ( + backup_dir + / f"subtitles-remerged-{datetime.now().astimezone().strftime('%Y%m%d-%H%M%S')}.xlsx" + ) + shutil.copy2(remerged_path, backup) + remerged["Source"] = df["Source"] + remerged["Translation"] = df["Translation"] + remerged.to_excel(remerged_path, index=False) + + mark_saved("subtitles") diff --git a/core/spacy_utils/load_nlp_model.py b/core/spacy_utils/load_nlp_model.py index c296a886..b2f6d51d 100644 --- a/core/spacy_utils/load_nlp_model.py +++ b/core/spacy_utils/load_nlp_model.py @@ -1,6 +1,7 @@ import spacy from spacy.cli import download from core.utils import rprint, load_key, except_handler +from core.workspace import output_path SPACY_MODEL_MAP = load_key("spacy_model_map") @@ -28,6 +29,6 @@ def init_nlp(): # -------------------- # define the intermediate files # -------------------- -SPLIT_BY_COMMA_FILE = "output/log/split_by_comma.txt" -SPLIT_BY_CONNECTOR_FILE = "output/log/split_by_connector.txt" -SPLIT_BY_MARK_FILE = "output/log/split_by_mark.txt" +SPLIT_BY_COMMA_FILE = output_path("log", "split_by_comma.txt") +SPLIT_BY_CONNECTOR_FILE = output_path("log", "split_by_connector.txt") +SPLIT_BY_MARK_FILE = output_path("log", "split_by_mark.txt") diff --git a/core/spacy_utils/split_by_mark.py b/core/spacy_utils/split_by_mark.py index 3cd31275..311d0875 100644 --- a/core/spacy_utils/split_by_mark.py +++ b/core/spacy_utils/split_by_mark.py @@ -3,6 +3,7 @@ import warnings from core.spacy_utils.load_nlp_model import init_nlp, SPLIT_BY_MARK_FILE from core.utils.config_utils import load_key, get_joiner +from core.utils.models import _2_CLEANED_CHUNKS from rich import print as rprint warnings.filterwarnings("ignore", category=FutureWarning) @@ -12,7 +13,7 @@ def split_by_mark(nlp): language = load_key("whisper.detected_language") if whisper_language == 'auto' else whisper_language # consider force english case joiner = get_joiner(language) rprint(f"[blue]🔍 Using {language} language joiner: '{joiner}'[/blue]") - chunks = pd.read_excel("output/log/cleaned_chunks.xlsx") + chunks = pd.read_excel(_2_CLEANED_CHUNKS) chunks.text = chunks.text.apply(lambda x: x.strip('"').strip("")) # join with joiner diff --git a/core/st_utils/download_video_section.py b/core/st_utils/download_video_section.py index 5f3023e5..fba4c85d 100644 --- a/core/st_utils/download_video_section.py +++ b/core/st_utils/download_video_section.py @@ -7,11 +7,11 @@ import streamlit as st from core._1_ytdlp import download_video_ytdlp, find_video_files from core.utils import * +from core.workspace import output_path from translations.translations import translate as t -OUTPUT_DIR = "output" - def download_video_section(): + output_dir = str(output_path()) st.header(t("a. Download or Upload Video")) with st.container(border=True): try: @@ -19,8 +19,8 @@ def download_video_section(): st.video(video_file) if st.button(t("Delete and Reselect"), key="delete_video_button"): os.remove(video_file) - if os.path.exists(OUTPUT_DIR): - shutil.rmtree(OUTPUT_DIR) + if os.path.exists(output_dir): + shutil.rmtree(output_dir) sleep(1) st.rerun() return True @@ -47,25 +47,25 @@ def download_video_section(): uploaded_file = st.file_uploader(t("Or upload video"), type=load_key("allowed_video_formats") + load_key("allowed_audio_formats")) if uploaded_file: - if os.path.exists(OUTPUT_DIR): - shutil.rmtree(OUTPUT_DIR) - os.makedirs(OUTPUT_DIR, exist_ok=True) + if os.path.exists(output_dir): + shutil.rmtree(output_dir) + os.makedirs(output_dir, exist_ok=True) raw_name = uploaded_file.name.replace(' ', '_') name, ext = os.path.splitext(raw_name) clean_name = re.sub(r'[^\w\-_\.]', '', name) + ext.lower() - with open(os.path.join(OUTPUT_DIR, clean_name), "wb") as f: + with open(os.path.join(output_dir, clean_name), "wb") as f: f.write(uploaded_file.getbuffer()) if ext.lower() in load_key("allowed_audio_formats"): - convert_audio_to_video(os.path.join(OUTPUT_DIR, clean_name)) + convert_audio_to_video(os.path.join(output_dir, clean_name)) st.rerun() else: return False def convert_audio_to_video(audio_file: str) -> str: - output_video = os.path.join(OUTPUT_DIR, 'black_screen.mp4') + output_video = os.path.join(str(output_path()), 'black_screen.mp4') if not os.path.exists(output_video): print(f"🎵➡️🎬 Converting audio to video with FFmpeg ......") ffmpeg_cmd = ['ffmpeg', '-y', '-f', 'lavfi', '-i', 'color=c=black:s=640x360', '-i', audio_file, '-shortest', '-c:v', 'libx264', '-c:a', 'aac', '-pix_fmt', 'yuv420p', output_video] diff --git a/core/st_utils/imports_and_utils.py b/core/st_utils/imports_and_utils.py index c6929094..73fe20a7 100644 --- a/core/st_utils/imports_and_utils.py +++ b/core/st_utils/imports_and_utils.py @@ -3,18 +3,20 @@ import io, zipfile from core.st_utils.download_video_section import download_video_section from core.st_utils.sidebar_setting import page_setting +from core.workspace import output_path from translations.translations import translate as t def download_subtitle_zip_button(text: str): zip_buffer = io.BytesIO() - output_dir = "output" + output_dir = str(output_path()) with zipfile.ZipFile(zip_buffer, "w") as zip_file: - for file_name in os.listdir(output_dir): - if file_name.endswith(".srt"): - file_path = os.path.join(output_dir, file_name) - with open(file_path, "rb") as file: - zip_file.writestr(file_name, file.read()) + if os.path.isdir(output_dir): + for file_name in os.listdir(output_dir): + if file_name.endswith(".srt"): + file_path = os.path.join(output_dir, file_name) + with open(file_path, "rb") as file: + zip_file.writestr(file_name, file.read()) zip_buffer.seek(0) @@ -116,4 +118,4 @@ def download_subtitle_zip_button(text: str): box-shadow: none !important; } -""" \ No newline at end of file +""" diff --git a/core/st_utils/review_section.py b/core/st_utils/review_section.py new file mode 100644 index 00000000..33056800 --- /dev/null +++ b/core/st_utils/review_section.py @@ -0,0 +1,155 @@ +from __future__ import annotations + +import pandas as pd +import streamlit as st + +from core import review, workspace +from translations.translations import translate as t + + +STAGE_LABELS = { + "terminology": "Terminology", + "translation": "Translation", + "subtitles": "Subtitles", + "tts_text": "TTS Text", +} + +STATUS_ICON = { + "pending": "○", + "confirmed": "✓", + "stale": "!", +} + + +def render_status_row() -> None: + status = review.load_status() + cols = st.columns(4) + for column, stage in zip(cols, review.STAGES): + item = status["stages"][stage] + icon = STATUS_ICON.get(item["status"], "○") + column.metric(t(STAGE_LABELS[stage]), f"{icon} {t(item['status'])}") + + +def _missing_artifact(stage: str) -> bool: + path = review.artifact_path(stage) + if path.exists(): + return False + st.info(t("Waiting for the upstream step to generate this review file.")) + st.caption(str(path)) + return True + + +def _confirm_controls(stage: str) -> None: + col1, col2 = st.columns(2) + with col1: + if st.button(t("Confirm"), key=f"review_confirm_{stage}", width="stretch"): + review.mark_confirmed(stage) + st.rerun() + with col2: + if st.button( + t("Mark as needs review"), + key=f"review_pending_{stage}", + width="stretch", + ): + review.mark_needs_review(stage) + st.rerun() + + +def terminology_tab() -> None: + stage = "terminology" + if _missing_artifact(stage): + return + try: + data = review.load_terminology() + except Exception as exc: + st.error(f"{t('Unable to load review file')}: {exc}") + return + + theme = st.text_area(t("Theme"), value=str(data.get("theme", "")), key="review_theme") + terms = pd.DataFrame(data.get("terms", []), columns=["src", "tgt", "note"]) + edited = st.data_editor( + terms, + num_rows="dynamic", + width="stretch", + key="review_terms_editor", + ) + if st.button(t("Save changes"), key="review_save_terminology", width="stretch"): + review.save_terminology(edited.fillna("").to_dict("records"), theme=theme) + st.success(t("Saved")) + st.rerun() + _confirm_controls(stage) + + +def table_tab( + stage: str, + required_columns: tuple[str, ...], + editable_columns: tuple[str, ...], +) -> None: + if _missing_artifact(stage): + return + try: + df = review.load_table(stage) + except Exception as exc: + st.error(f"{t('Unable to load review file')}: {exc}") + return + + disabled = [column for column in df.columns if column not in editable_columns] + edited = st.data_editor( + df, + disabled=disabled, + width="stretch", + hide_index=False, + key=f"review_editor_{stage}", + ) + if st.button(t("Save changes"), key=f"review_save_{stage}", width="stretch"): + if stage == "subtitles": + review.save_subtitles(edited) + else: + review.save_table(stage, edited, required_columns=required_columns) + st.success(t("Saved")) + st.rerun() + _confirm_controls(stage) + + +def review_section() -> None: + st.header(t("Review Workbench")) + if not workspace.get_active_workspace(): + st.info(t("Create or select a task workspace before reviewing.")) + return + + render_status_row() + tab_terms, tab_translation, tab_subtitles, tab_tts = st.tabs( + [ + t("Terminology"), + t("Translation"), + t("Subtitles"), + t("TTS Text"), + ] + ) + with tab_terms: + terminology_tab() + with tab_translation: + table_tab( + "translation", + required_columns=("Source", "Translation"), + editable_columns=("Translation",), + ) + with tab_subtitles: + table_tab( + "subtitles", + required_columns=("Source", "Translation"), + editable_columns=("Source", "Translation"), + ) + with tab_tts: + table_tab( + "tts_text", + required_columns=( + "number", + "start_time", + "end_time", + "duration", + "text", + "origin", + ), + editable_columns=("text", "origin"), + ) diff --git a/core/tts_backend/gpt_sovits_tts.py b/core/tts_backend/gpt_sovits_tts.py index 0d08474d..3fa5d211 100644 --- a/core/tts_backend/gpt_sovits_tts.py +++ b/core/tts_backend/gpt_sovits_tts.py @@ -5,6 +5,7 @@ import socket import time from core.utils import * +from core.workspace import output_path def check_lang(text_lang, prompt_lang): # only support zh and en @@ -61,7 +62,6 @@ def gpt_sovits_tts_for_videolingo(text, save_as, number, task_df): DUBBING_CHARACTER = sovits_set["character"] REFER_MODE = sovits_set["refer_mode"] - current_dir = Path.cwd() prompt_lang = load_key("whisper.detected_language") if WHISPER_LANGUAGE == 'auto' else WHISPER_LANGUAGE prompt_text = task_df.loc[task_df['number'] == number, 'origin'].values[0] @@ -86,7 +86,7 @@ def gpt_sovits_tts_for_videolingo(text, save_as, number, task_df): prompt_text = content elif REFER_MODE in [2, 3]: # Check if the reference audio file exists - ref_audio_path = current_dir / ("output/audio/refers/1.wav" if REFER_MODE == 2 else f"output/audio/refers/{number}.wav") + ref_audio_path = Path(str(output_path("audio", "refers", "1.wav" if REFER_MODE == 2 else f"{number}.wav"))) if not ref_audio_path.exists(): # If the file does not exist, try to extract the reference audio try: @@ -102,7 +102,7 @@ def gpt_sovits_tts_for_videolingo(text, save_as, number, task_df): success = gpt_sovits_tts(text, TARGET_LANGUAGE, save_as, ref_audio_path, prompt_lang, prompt_text) if not success and REFER_MODE == 3: rprint(f"[bold red]TTS request failed, switching back to mode 2 and retrying[/bold red]") - ref_audio_path = current_dir / "output/audio/refers/1.wav" + ref_audio_path = Path(str(output_path("audio", "refers", "1.wav"))) gpt_sovits_tts(text, TARGET_LANGUAGE, save_as, ref_audio_path, prompt_lang, prompt_text) diff --git a/core/tts_backend/sf_cosyvoice2.py b/core/tts_backend/sf_cosyvoice2.py index 75af1f7c..52715e82 100644 --- a/core/tts_backend/sf_cosyvoice2.py +++ b/core/tts_backend/sf_cosyvoice2.py @@ -2,6 +2,7 @@ from pathlib import Path import base64 from core.utils import * +from core.workspace import output_path def wav_to_base64(wav_file_path): with open(wav_file_path, 'rb') as audio_file: @@ -14,12 +15,11 @@ def cosyvoice_tts_for_videolingo(text, save_as, number, task_df): prompt_text = task_df.loc[task_df['number'] == number, 'origin'].values[0] API_KEY = load_key("sf_cosyvoice2.api_key") # 设置参考音频路径 - current_dir = Path.cwd() - ref_audio_path = current_dir / f"output/audio/refers/{number}.wav" + ref_audio_path = Path(str(output_path("audio", "refers", f"{number}.wav"))) # 如果参考音频不存在，使用第一个音频作为备选 if not ref_audio_path.exists(): - ref_audio_path = current_dir / "output/audio/refers/1.wav" + ref_audio_path = Path(str(output_path("audio", "refers", "1.wav"))) if not ref_audio_path.exists(): try: from core._9_refer_audio import extract_refer_audio_main @@ -45,4 +45,4 @@ def cosyvoice_tts_for_videolingo(text, save_as, number, task_df): response.stream_to_file(save_path) print(f"音频已成功保存至: {save_path}") - return True \ No newline at end of file + return True diff --git a/core/utils/ask_gpt.py b/core/utils/ask_gpt.py index 972bd20c..6b4c0db4 100644 --- a/core/utils/ask_gpt.py +++ b/core/utils/ask_gpt.py @@ -6,18 +6,19 @@ from core.utils.config_utils import load_key from rich import print as rprint from core.utils.decorator import except_handler +from core.workspace import output_path # ------------ # cache gpt response # ------------ LOCK = Lock() -GPT_LOG_FOLDER = 'output/gpt_log' +GPT_LOG_FOLDER = output_path("gpt_log") def _save_cache(model, prompt, resp_content, resp_type, resp, message=None, log_title="default"): with LOCK: logs = [] - file = os.path.join(GPT_LOG_FOLDER, f"{log_title}.json") + file = os.path.join(str(GPT_LOG_FOLDER), f"{log_title}.json") os.makedirs(os.path.dirname(file), exist_ok=True) if os.path.exists(file): with open(file, 'r', encoding='utf-8') as f: @@ -28,7 +29,7 @@ def _save_cache(model, prompt, resp_content, resp_type, resp, message=None, log_ def _load_cache(prompt, resp_type, log_title): with LOCK: - file = os.path.join(GPT_LOG_FOLDER, f"{log_title}.json") + file = os.path.join(str(GPT_LOG_FOLDER), f"{log_title}.json") if os.path.exists(file): with open(file, 'r', encoding='utf-8') as f: for item in json.load(f): diff --git a/core/utils/config_utils.py b/core/utils/config_utils.py index 2b6135f9..87a146de 100644 --- a/core/utils/config_utils.py +++ b/core/utils/config_utils.py @@ -1,5 +1,6 @@ from ruamel.yaml import YAML import threading +from pathlib import Path CONFIG_PATH = 'config.yaml' lock = threading.Lock() @@ -11,9 +12,23 @@ # load & update config # ----------------------- +def _active_config_path(): + try: + from core.workspace import CONFIG_SNAPSHOT_FILE, get_active_workspace + + active_workspace = get_active_workspace() + if active_workspace: + snapshot = Path(active_workspace) / CONFIG_SNAPSHOT_FILE + if snapshot.exists(): + return str(snapshot) + except Exception: + pass + return CONFIG_PATH + + def load_key(key): with lock: - with open(CONFIG_PATH, 'r', encoding='utf-8') as file: + with open(_active_config_path(), 'r', encoding='utf-8') as file: data = yaml.load(file) keys = key.split('.') @@ -27,7 +42,8 @@ def load_key(key): def update_key(key, new_value): with lock: - with open(CONFIG_PATH, 'r', encoding='utf-8') as file: + config_path = _active_config_path() + with open(config_path, 'r', encoding='utf-8') as file: data = yaml.load(file) keys = key.split('.') @@ -40,7 +56,7 @@ def update_key(key, new_value): if isinstance(current, dict) and keys[-1] in current: current[keys[-1]] = new_value - with open(CONFIG_PATH, 'w', encoding='utf-8') as file: + with open(config_path, 'w', encoding='utf-8') as file: yaml.dump(data, file) return True else: diff --git a/core/utils/delete_retry_dubbing.py b/core/utils/delete_retry_dubbing.py index 67fa3875..c6e421f6 100644 --- a/core/utils/delete_retry_dubbing.py +++ b/core/utils/delete_retry_dubbing.py @@ -1,10 +1,11 @@ import os import shutil +from core.workspace import output_path def delete_dubbing_files(): files_to_delete = [ - os.path.join("output", "dub.wav"), - os.path.join("output", "output_dub.mp4") + str(output_path("dub.wav")), + str(output_path("output_dub.mp4")), ] for file_path in files_to_delete: @@ -17,7 +18,7 @@ def delete_dubbing_files(): else: print(f"File not found: {file_path}") - segs_folder = os.path.join("output", "audio", "segs") + segs_folder = str(output_path("audio", "segs")) if os.path.exists(segs_folder): try: shutil.rmtree(segs_folder) @@ -28,4 +29,4 @@ def delete_dubbing_files(): print(f"Folder not found: {segs_folder}") if __name__ == "__main__": - delete_dubbing_files() \ No newline at end of file + delete_dubbing_files() diff --git a/core/utils/models.py b/core/utils/models.py index 8e15c829..8adcda1d 100644 --- a/core/utils/models.py +++ b/core/utils/models.py @@ -1,29 +1,32 @@ +from core.workspace import workspace_path + + # ------------------------------------------ # 定义中间产出文件 # ------------------------------------------ -_2_CLEANED_CHUNKS = "output/log/cleaned_chunks.xlsx" -_3_1_SPLIT_BY_NLP = "output/log/split_by_nlp.txt" -_3_2_SPLIT_BY_MEANING = "output/log/split_by_meaning.txt" -_4_1_TERMINOLOGY = "output/log/terminology.json" -_4_2_TRANSLATION = "output/log/translation_results.xlsx" -_5_SPLIT_SUB = "output/log/translation_results_for_subtitles.xlsx" -_5_REMERGED = "output/log/translation_results_remerged.xlsx" +_2_CLEANED_CHUNKS = workspace_path("output/log/cleaned_chunks.xlsx") +_3_1_SPLIT_BY_NLP = workspace_path("output/log/split_by_nlp.txt") +_3_2_SPLIT_BY_MEANING = workspace_path("output/log/split_by_meaning.txt") +_4_1_TERMINOLOGY = workspace_path("output/log/terminology.json") +_4_2_TRANSLATION = workspace_path("output/log/translation_results.xlsx") +_5_SPLIT_SUB = workspace_path("output/log/translation_results_for_subtitles.xlsx") +_5_REMERGED = workspace_path("output/log/translation_results_remerged.xlsx") -_8_1_AUDIO_TASK = "output/audio/tts_tasks.xlsx" +_8_1_AUDIO_TASK = workspace_path("output/audio/tts_tasks.xlsx") # ------------------------------------------ # 定义音频文件 # ------------------------------------------ -_OUTPUT_DIR = "output" -_AUDIO_DIR = "output/audio" -_RAW_AUDIO_FILE = "output/audio/raw.mp3" -_VOCAL_AUDIO_FILE = "output/audio/vocal.mp3" -_BACKGROUND_AUDIO_FILE = "output/audio/background.mp3" -_AUDIO_REFERS_DIR = "output/audio/refers" -_AUDIO_SEGS_DIR = "output/audio/segs" -_AUDIO_TMP_DIR = "output/audio/tmp" +_OUTPUT_DIR = workspace_path("output") +_AUDIO_DIR = workspace_path("output/audio") +_RAW_AUDIO_FILE = workspace_path("output/audio/raw.mp3") +_VOCAL_AUDIO_FILE = workspace_path("output/audio/vocal.mp3") +_BACKGROUND_AUDIO_FILE = workspace_path("output/audio/background.mp3") +_AUDIO_REFERS_DIR = workspace_path("output/audio/refers") +_AUDIO_SEGS_DIR = workspace_path("output/audio/segs") +_AUDIO_TMP_DIR = workspace_path("output/audio/tmp") # ------------------------------------------ # 导出 diff --git a/core/utils/onekeycleanup.py b/core/utils/onekeycleanup.py index 286a8e76..5c0d778d 100644 --- a/core/utils/onekeycleanup.py +++ b/core/utils/onekeycleanup.py @@ -2,13 +2,19 @@ import glob from core._1_ytdlp import find_video_files import shutil +from core.workspace import get_active_workspace, output_path def cleanup(history_dir="history"): # Get video file name video_file = find_video_files() - video_name = video_file.split("/")[1] + video_name = os.path.basename(video_file) video_name = os.path.splitext(video_name)[0] video_name = sanitize_filename(video_name) + + output_dir = str(output_path()) + active_workspace = get_active_workspace() + if active_workspace: + history_dir = os.path.join(active_workspace, "artifacts", "history") # Create required folders os.makedirs(history_dir, exist_ok=True) @@ -19,23 +25,23 @@ def cleanup(history_dir="history"): os.makedirs(gpt_log_dir, exist_ok=True) # Move non-log files - for file in glob.glob("output/*"): + for file in glob.glob(os.path.join(output_dir, "*")): if not file.endswith(('log', 'gpt_log')): move_file(file, video_history_dir) # Move log files - for file in glob.glob("output/log/*"): + for file in glob.glob(os.path.join(output_dir, "log", "*")): move_file(file, log_dir) # Move gpt_log files - for file in glob.glob("output/gpt_log/*"): + for file in glob.glob(os.path.join(output_dir, "gpt_log", "*")): move_file(file, gpt_log_dir) # Delete empty output directories try: - os.rmdir("output/log") - os.rmdir("output/gpt_log") - os.rmdir("output") + os.rmdir(os.path.join(output_dir, "log")) + os.rmdir(os.path.join(output_dir, "gpt_log")) + os.rmdir(output_dir) except OSError: pass # Ignore errors when deleting directories @@ -77,4 +83,4 @@ def sanitize_filename(filename): return filename if __name__ == "__main__": - cleanup() \ No newline at end of file + cleanup() diff --git a/core/workspace.py b/core/workspace.py new file mode 100644 index 00000000..1f3c8cbb --- /dev/null +++ b/core/workspace.py @@ -0,0 +1,172 @@ +from __future__ import annotations + +import os +import re +import shutil +from datetime import datetime +from pathlib import Path +from typing import Any + +from ruamel.yaml import YAML + +ACTIVE_WORKSPACE_ENV = "VIDEOLINGO_ACTIVE_WORKSPACE" +DEFAULT_WORKSPACE_ROOT = Path("workspace") / "jobs" +JOB_FILE = "job.yaml" +CONFIG_SNAPSHOT_FILE = "config.snapshot.yaml" + +yaml = YAML() +yaml.preserve_quotes = True + + +class WorkspacePath: + def __init__(self, relative_path: str): + self.relative_path = relative_path.replace("\\", "/").strip("/") + + def resolve(self) -> str: + return resolve_path(self.relative_path) + + def __fspath__(self) -> str: + return self.resolve() + + def __str__(self) -> str: + return self.resolve() + + def __repr__(self) -> str: + return repr(self.resolve()) + + +def _now() -> str: + return datetime.now().astimezone().isoformat(timespec="seconds") + + +def _slugify(value: str) -> str: + slug = re.sub(r"[^A-Za-z0-9._-]+", "-", value.strip()).strip("-._") + return slug[:60] or "task" + + +def get_active_workspace() -> str | None: + value = os.environ.get(ACTIVE_WORKSPACE_ENV, "").strip() + return value or None + + +def set_active_workspace(path: str | os.PathLike[str]) -> str: + job_dir = Path(path) + if not (job_dir / JOB_FILE).is_file(): + raise FileNotFoundError(f"Workspace metadata not found: {job_dir / JOB_FILE}") + os.environ[ACTIVE_WORKSPACE_ENV] = str(job_dir) + return str(job_dir) + + +def clear_active_workspace() -> None: + os.environ.pop(ACTIVE_WORKSPACE_ENV, None) + + +def workspace_path(relative_path: str) -> WorkspacePath: + return WorkspacePath(relative_path) + + +def output_path(*parts: str) -> WorkspacePath: + clean = [part.strip("/\\") for part in parts if part] + return workspace_path("/".join(["output", *clean])) + + +def resolve_path(relative_path: str) -> str: + active = get_active_workspace() + if active: + return str(Path(active) / relative_path) + return relative_path + + +def create_job( + name: str | None = None, + source_type: str = "", + source_path: str = "", + source_url: str = "", + workspace_root: str | os.PathLike[str] = DEFAULT_WORKSPACE_ROOT, + config_path: str | os.PathLike[str] = "config.yaml", +) -> dict[str, Any]: + title = name or "Untitled Task" + timestamp = datetime.now().astimezone().strftime("%Y%m%d-%H%M%S") + job_id = f"{timestamp}-{_slugify(title)}" + job_dir = Path(workspace_root) / job_id + suffix = 2 + while job_dir.exists(): + job_dir = Path(workspace_root) / f"{job_id}-{suffix}" + suffix += 1 + + (job_dir / "output").mkdir(parents=True, exist_ok=False) + (job_dir / "logs").mkdir(parents=True, exist_ok=True) + (job_dir / "artifacts").mkdir(parents=True, exist_ok=True) + + created_at = _now() + job = { + "id": job_dir.name, + "name": title, + "path": str(job_dir), + "source": { + "type": source_type, + "path": source_path, + "url": source_url, + }, + "status": "created", + "created_at": created_at, + "updated_at": created_at, + "current_stage": "", + "error": "", + } + save_job(job_dir, job) + save_config_snapshot(job_dir, config_path) + return job + + +def save_config_snapshot( + job_dir: str | os.PathLike[str], + config_path: str | os.PathLike[str] = "config.yaml", +) -> None: + src = Path(config_path) + dst = Path(job_dir) / CONFIG_SNAPSHOT_FILE + if src.exists(): + shutil.copy2(src, dst) + else: + dst.write_text("{}\n", encoding="utf-8") + + +def load_job(job_dir: str | os.PathLike[str]) -> dict[str, Any]: + with open(Path(job_dir) / JOB_FILE, "r", encoding="utf-8") as file: + data = yaml.load(file) or {} + data["path"] = str(Path(job_dir)) + return data + + +def save_job(job_dir: str | os.PathLike[str], data: dict[str, Any]) -> None: + data = dict(data) + data["path"] = str(Path(job_dir)) + data["updated_at"] = data.get("updated_at") or _now() + with open(Path(job_dir) / JOB_FILE, "w", encoding="utf-8") as file: + yaml.dump(data, file) + + +def update_job(job_dir: str | os.PathLike[str], **updates: Any) -> dict[str, Any]: + job = load_job(job_dir) + job.update(updates) + job["updated_at"] = _now() + save_job(job_dir, job) + return job + + +def archive_job(job_dir: str | os.PathLike[str]) -> dict[str, Any]: + return update_job(job_dir, status="archived") + + +def list_jobs( + workspace_root: str | os.PathLike[str] = DEFAULT_WORKSPACE_ROOT, +) -> list[dict[str, Any]]: + root = Path(workspace_root) + if not root.exists(): + return [] + + jobs = [] + for child in sorted(root.iterdir(), reverse=True): + if child.is_dir() and (child / JOB_FILE).is_file(): + jobs.append(load_job(child)) + return jobs diff --git a/docs/superpowers/plans/2026-05-03-human-review-workbench.md b/docs/superpowers/plans/2026-05-03-human-review-workbench.md new file mode 100644 index 00000000..5f534e8a --- /dev/null +++ b/docs/superpowers/plans/2026-05-03-human-review-workbench.md @@ -0,0 +1,1043 @@ +# Human Review Workbench Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Add a workspace-local human review workbench for terminology, translation, subtitle rows, and TTS text. + +**Architecture:** Add a focused `core/review.py` module that owns review status, artifact loading/saving, backups, and invalidation. Add `core/st_utils/review_section.py` for Streamlit UI, then wire `st.py` so text and audio workflows pause at review checkpoints and resume from the confirmed stage. + +**Tech Stack:** Python, Streamlit, pandas Excel files, ruamel.yaml, JSON artifacts, existing VideoLingo workspace path helpers. + +--- + +## File Structure + +- Create `core/review.py` + - Review stage metadata. + - `review_status.yaml` creation/loading/saving. + - Workspace-aware artifact paths. + - JSON and Excel artifact read/write helpers. + - Backups under `artifacts/reviews/backups/`. + - Downstream invalidation rules. +- Create `core/st_utils/review_section.py` + - Streamlit status row and four tabs. + - Editable data tables for terminology, translation, subtitles, and TTS text. + - Save, confirm, and mark-needs-review buttons. +- Modify `st.py` + - Import review helpers. + - Split text processing into review-aware step groups. + - Split audio processing into review-aware step groups. + - Render review workbench after download section. + - Show action buttons for "generate next artifact", "review", and "continue". +- Modify `core/__init__.py` + - Export `review`. +- Create `tests/test_review.py` + - Unit tests for status, backups, artifact saves, and invalidation. +- Modify `tests/test_workspace.py` only if workspace setup needs a small helper; otherwise leave it unchanged. + +--- + +### Task 1: Add Review Status Core + +**Files:** +- Create: `core/review.py` +- Create: `tests/test_review.py` +- Modify: `core/__init__.py` + +- [ ] **Step 1: Write failing tests for default review status and confirmations** + +Add this to `tests/test_review.py`: + +```python +import tempfile +import unittest +from pathlib import Path + +from ruamel.yaml import YAML + +from core import review, workspace + + +class ReviewTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name) + self.config_path = self.root / "config.yaml" + self.config_path.write_text("display_language: en\n", encoding="utf-8") + self.job = workspace.create_job( + name="Review Video", + workspace_root=self.root / "jobs", + config_path=self.config_path, + ) + workspace.set_active_workspace(self.job["path"]) + + def tearDown(self): + workspace.clear_active_workspace() + + def test_default_review_status_is_created_in_active_workspace(self): + status = review.load_status() + + status_path = Path(self.job["path"]) / "artifacts" / "reviews" / "review_status.yaml" + self.assertTrue(status_path.is_file()) + self.assertEqual(status["stages"]["terminology"]["status"], "pending") + self.assertEqual( + status["stages"]["translation"]["artifact"], + "output/log/translation_results.xlsx", + ) + + def test_mark_confirmed_records_timestamp(self): + review.mark_confirmed("terminology") + status = review.load_status() + + self.assertEqual(status["stages"]["terminology"]["status"], "confirmed") + self.assertTrue(status["stages"]["terminology"]["confirmed_at"]) +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: + +```bash +/Users/oliverchow/VideoLingo/.venv/bin/python -m unittest tests.test_review -v +``` + +Expected: fail because `core.review` does not exist. + +- [ ] **Step 3: Implement minimal review status core** + +Create `core/review.py`: + +```python +from __future__ import annotations + +import shutil +from datetime import datetime +from pathlib import Path +from typing import Any + +from ruamel.yaml import YAML + +from core import workspace + +REVIEW_ROOT = Path("artifacts") / "reviews" +BACKUP_DIR = REVIEW_ROOT / "backups" +STATUS_FILE = REVIEW_ROOT / "review_status.yaml" + +STAGES = { + "terminology": "output/log/terminology.json", + "translation": "output/log/translation_results.xlsx", + "subtitles": "output/log/translation_results_for_subtitles.xlsx", + "tts_text": "output/audio/tts_tasks.xlsx", +} + +DOWNSTREAM = { + "terminology": ("translation", "subtitles", "tts_text"), + "translation": ("subtitles", "tts_text"), + "subtitles": ("tts_text",), + "tts_text": (), +} + +yaml = YAML() +yaml.preserve_quotes = True + + +def now_iso() -> str: + return datetime.now().astimezone().isoformat(timespec="seconds") + + +def active_workspace_dir() -> Path: + active = workspace.get_active_workspace() + if not active: + raise RuntimeError("Select or create a task workspace before reviewing artifacts.") + return Path(active) + + +def review_root() -> Path: + return active_workspace_dir() / REVIEW_ROOT + + +def review_status_path() -> Path: + return active_workspace_dir() / STATUS_FILE + + +def default_status() -> dict[str, Any]: + return { + "stages": { + stage: { + "status": "pending", + "artifact": artifact, + "confirmed_at": "", + "updated_at": "", + } + for stage, artifact in STAGES.items() + } + } + + +def save_status(status: dict[str, Any]) -> None: + path = review_status_path() + path.parent.mkdir(parents=True, exist_ok=True) + with open(path, "w", encoding="utf-8") as file: + yaml.dump(status, file) + + +def normalize_status(status: dict[str, Any] | None) -> dict[str, Any]: + merged = default_status() + for stage, values in (status or {}).get("stages", {}).items(): + if stage in merged["stages"] and isinstance(values, dict): + merged["stages"][stage].update(values) + return merged + + +def load_status() -> dict[str, Any]: + path = review_status_path() + if not path.exists(): + status = default_status() + save_status(status) + return status + with open(path, "r", encoding="utf-8") as file: + status = normalize_status(yaml.load(file) or {}) + save_status(status) + return status + + +def validate_stage(stage: str) -> None: + if stage not in STAGES: + raise ValueError(f"Unknown review stage: {stage}") + + +def mark_confirmed(stage: str) -> dict[str, Any]: + validate_stage(stage) + status = load_status() + stamp = now_iso() + status["stages"][stage]["status"] = "confirmed" + status["stages"][stage]["confirmed_at"] = stamp + status["stages"][stage]["updated_at"] = stamp + save_status(status) + return status + + +def mark_needs_review(stage: str) -> dict[str, Any]: + validate_stage(stage) + status = load_status() + status["stages"][stage]["status"] = "pending" + status["stages"][stage]["confirmed_at"] = "" + status["stages"][stage]["updated_at"] = now_iso() + save_status(status) + return status + + +def invalidate_downstream(stage: str, status: dict[str, Any] | None = None) -> dict[str, Any]: + validate_stage(stage) + status = status or load_status() + stamp = now_iso() + for downstream in DOWNSTREAM[stage]: + status["stages"][downstream]["status"] = "stale" + status["stages"][downstream]["confirmed_at"] = "" + status["stages"][downstream]["updated_at"] = stamp + save_status(status) + return status + + +def artifact_path(stage: str) -> Path: + validate_stage(stage) + return active_workspace_dir() / STAGES[stage] + + +def backup_artifact(stage: str, source: Path | None = None) -> Path | None: + validate_stage(stage) + source = source or artifact_path(stage) + if not source.exists(): + return None + backup_dir = active_workspace_dir() / BACKUP_DIR + backup_dir.mkdir(parents=True, exist_ok=True) + backup = backup_dir / f"{stage}-{datetime.now().astimezone().strftime('%Y%m%d-%H%M%S')}{source.suffix}" + shutil.copy2(source, backup) + return backup +``` + +Modify `core/__init__.py` so `review` can be imported from `core`: + +```python +try: + from . import review +except Exception: + review = None +``` + +Also add `"review"` to `__all__`. + +- [ ] **Step 4: Run tests** + +Run: + +```bash +/Users/oliverchow/VideoLingo/.venv/bin/python -m unittest tests.test_review -v +``` + +Expected: both tests pass. + +- [ ] **Step 5: Commit** + +```bash +git add core/review.py core/__init__.py tests/test_review.py +git commit -m "feat: add review status core" +``` + +--- + +### Task 2: Add Artifact Save, Backup, and Invalidation Helpers + +**Files:** +- Modify: `core/review.py` +- Modify: `tests/test_review.py` + +- [ ] **Step 1: Write failing artifact tests** + +Append these tests to `tests/test_review.py`: + +```python + def test_save_terminology_preserves_theme_and_invalidates_downstream(self): + import json + + path = Path(self.job["path"]) / "output" / "log" + path.mkdir(parents=True, exist_ok=True) + terminology = path / "terminology.json" + terminology.write_text( + json.dumps( + { + "theme": "Physics lecture", + "terms": [{"src": "force", "tgt": "力", "note": "physics"}], + }, + ensure_ascii=False, + ), + encoding="utf-8", + ) + review.mark_confirmed("translation") + + review.save_terminology( + [{"src": "energy", "tgt": "能量", "note": "physics"}], + theme="Physics lecture", + ) + + saved = json.loads(terminology.read_text(encoding="utf-8")) + self.assertEqual(saved["theme"], "Physics lecture") + self.assertEqual(saved["terms"][0]["src"], "energy") + status = review.load_status() + self.assertEqual(status["stages"]["translation"]["status"], "stale") + backups = list((Path(self.job["path"]) / "artifacts" / "reviews" / "backups").glob("terminology-*.json")) + self.assertEqual(len(backups), 1) + + def test_save_table_creates_backup_and_validates_required_columns(self): + import pandas as pd + + path = Path(self.job["path"]) / "output" / "log" + path.mkdir(parents=True, exist_ok=True) + artifact = path / "translation_results.xlsx" + pd.DataFrame({"Source": ["hello"], "Translation": ["你好"]}).to_excel(artifact, index=False) + + review.save_table( + "translation", + pd.DataFrame({"Source": ["hello"], "Translation": ["您好"]}), + required_columns=("Source", "Translation"), + ) + + saved = pd.read_excel(artifact) + self.assertEqual(saved.at[0, "Translation"], "您好") + backups = list((Path(self.job["path"]) / "artifacts" / "reviews" / "backups").glob("translation-*.xlsx")) + self.assertEqual(len(backups), 1) + + def test_save_table_rejects_missing_required_columns(self): + import pandas as pd + + with self.assertRaises(ValueError): + review.save_table( + "translation", + pd.DataFrame({"Source": ["hello"]}), + required_columns=("Source", "Translation"), + ) +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: + +```bash +/Users/oliverchow/VideoLingo/.venv/bin/python -m unittest tests.test_review -v +``` + +Expected: fail because `save_terminology` and `save_table` do not exist. + +- [ ] **Step 3: Implement artifact helpers** + +Add imports and functions to `core/review.py`: + +```python +import json + +import pandas as pd + + +def mark_saved(stage: str, status_value: str = "pending") -> dict[str, Any]: + validate_stage(stage) + status = load_status() + status["stages"][stage]["status"] = status_value + status["stages"][stage]["confirmed_at"] = "" + status["stages"][stage]["updated_at"] = now_iso() + save_status(status) + return invalidate_downstream(stage, status) + + +def load_terminology() -> dict[str, Any]: + path = artifact_path("terminology") + if not path.exists(): + raise FileNotFoundError(path) + with open(path, "r", encoding="utf-8") as file: + data = json.load(file) + data.setdefault("terms", []) + return data + + +def save_terminology(terms: list[dict[str, Any]], theme: str | None = None) -> dict[str, Any]: + path = artifact_path("terminology") + path.parent.mkdir(parents=True, exist_ok=True) + existing = {} + if path.exists(): + with open(path, "r", encoding="utf-8") as file: + existing = json.load(file) or {} + backup_artifact("terminology", path) + cleaned_terms = [] + for term in terms: + cleaned_terms.append( + { + "src": str(term.get("src", "")).strip(), + "tgt": str(term.get("tgt", "")).strip(), + "note": str(term.get("note", "")).strip(), + } + ) + existing["terms"] = cleaned_terms + if theme is not None: + existing["theme"] = theme + with open(path, "w", encoding="utf-8") as file: + json.dump(existing, file, ensure_ascii=False, indent=4) + mark_saved("terminology") + return existing + + +def load_table(stage: str) -> pd.DataFrame: + path = artifact_path(stage) + if not path.exists(): + raise FileNotFoundError(path) + return pd.read_excel(path) + + +def validate_required_columns(df: pd.DataFrame, required_columns: tuple[str, ...]) -> None: + missing = [column for column in required_columns if column not in df.columns] + if missing: + raise ValueError(f"Missing required columns: {', '.join(missing)}") + + +def save_table( + stage: str, + df: pd.DataFrame, + required_columns: tuple[str, ...], + invalidate: bool = True, +) -> None: + validate_stage(stage) + validate_required_columns(df, required_columns) + path = artifact_path(stage) + path.parent.mkdir(parents=True, exist_ok=True) + if path.exists(): + backup_artifact(stage, path) + df.to_excel(path, index=False) + if invalidate: + mark_saved(stage) +``` + +- [ ] **Step 4: Run tests** + +Run: + +```bash +/Users/oliverchow/VideoLingo/.venv/bin/python -m unittest tests.test_review -v +``` + +Expected: all review tests pass. + +- [ ] **Step 5: Commit** + +```bash +git add core/review.py tests/test_review.py +git commit -m "feat: add review artifact persistence" +``` + +--- + +### Task 3: Add Subtitle and TTS Review Helpers + +**Files:** +- Modify: `core/review.py` +- Modify: `tests/test_review.py` + +- [ ] **Step 1: Write failing tests for subtitle sync and TTS save** + +Append these tests to `tests/test_review.py`: + +```python + def test_save_subtitles_updates_remerged_when_row_counts_match(self): + import pandas as pd + + log_dir = Path(self.job["path"]) / "output" / "log" + log_dir.mkdir(parents=True, exist_ok=True) + pd.DataFrame({"Source": ["A"], "Translation": ["甲"]}).to_excel( + log_dir / "translation_results_for_subtitles.xlsx", + index=False, + ) + pd.DataFrame({"Source": ["A"], "Translation": ["甲"]}).to_excel( + log_dir / "translation_results_remerged.xlsx", + index=False, + ) + + review.save_subtitles(pd.DataFrame({"Source": ["A"], "Translation": ["乙"]})) + + display = pd.read_excel(log_dir / "translation_results_for_subtitles.xlsx") + remerged = pd.read_excel(log_dir / "translation_results_remerged.xlsx") + self.assertEqual(display.at[0, "Translation"], "乙") + self.assertEqual(remerged.at[0, "Translation"], "乙") + self.assertEqual(review.load_status()["stages"]["tts_text"]["status"], "stale") + + def test_save_subtitles_leaves_remerged_when_row_counts_differ(self): + import pandas as pd + + log_dir = Path(self.job["path"]) / "output" / "log" + log_dir.mkdir(parents=True, exist_ok=True) + pd.DataFrame({"Source": ["A"], "Translation": ["甲"]}).to_excel( + log_dir / "translation_results_for_subtitles.xlsx", + index=False, + ) + pd.DataFrame({"Source": ["A", "B"], "Translation": ["甲", "乙"]}).to_excel( + log_dir / "translation_results_remerged.xlsx", + index=False, + ) + + review.save_subtitles(pd.DataFrame({"Source": ["A"], "Translation": ["丙"]})) + + remerged = pd.read_excel(log_dir / "translation_results_remerged.xlsx") + self.assertEqual(len(remerged), 2) + self.assertEqual(remerged.at[0, "Translation"], "甲") + + def test_save_tts_text_preserves_timing_columns(self): + import pandas as pd + + audio_dir = Path(self.job["path"]) / "output" / "audio" + audio_dir.mkdir(parents=True, exist_ok=True) + pd.DataFrame( + { + "number": [1], + "start_time": ["00:00:00.000"], + "end_time": ["00:00:02.000"], + "duration": [2.0], + "text": ["old"], + "origin": ["source"], + } + ).to_excel(audio_dir / "tts_tasks.xlsx", index=False) + + review.save_table( + "tts_text", + pd.DataFrame( + { + "number": [1], + "start_time": ["00:00:00.000"], + "end_time": ["00:00:02.000"], + "duration": [2.0], + "text": ["new"], + "origin": ["source"], + } + ), + required_columns=("number", "start_time", "end_time", "duration", "text", "origin"), + ) + + saved = pd.read_excel(audio_dir / "tts_tasks.xlsx") + self.assertEqual(saved.at[0, "text"], "new") + self.assertEqual(saved.at[0, "duration"], 2.0) +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: + +```bash +/Users/oliverchow/VideoLingo/.venv/bin/python -m unittest tests.test_review -v +``` + +Expected: fail because `save_subtitles` does not exist. + +- [ ] **Step 3: Implement subtitle save helper** + +Add this to `core/review.py`: + +```python +def save_subtitles(df: pd.DataFrame) -> None: + validate_required_columns(df, ("Source", "Translation")) + save_table("subtitles", df, required_columns=("Source", "Translation"), invalidate=False) + + remerged_path = active_workspace_dir() / "output/log/translation_results_remerged.xlsx" + if remerged_path.exists(): + remerged = pd.read_excel(remerged_path) + if len(remerged) == len(df) and {"Source", "Translation"}.issubset(remerged.columns): + backup_dir = active_workspace_dir() / BACKUP_DIR + backup_dir.mkdir(parents=True, exist_ok=True) + backup = backup_dir / f"subtitles-remerged-{datetime.now().astimezone().strftime('%Y%m%d-%H%M%S')}.xlsx" + shutil.copy2(remerged_path, backup) + remerged["Source"] = df["Source"] + remerged["Translation"] = df["Translation"] + remerged.to_excel(remerged_path, index=False) + + mark_saved("subtitles") +``` + +- [ ] **Step 4: Run tests** + +Run: + +```bash +/Users/oliverchow/VideoLingo/.venv/bin/python -m unittest tests.test_review -v +``` + +Expected: all review tests pass. + +- [ ] **Step 5: Commit** + +```bash +git add core/review.py tests/test_review.py +git commit -m "feat: add subtitle and tts review persistence" +``` + +--- + +### Task 4: Add Streamlit Review Workbench UI + +**Files:** +- Create: `core/st_utils/review_section.py` +- Modify: `st.py` + +- [ ] **Step 1: Create the review UI module** + +Create `core/st_utils/review_section.py`: + +```python +from __future__ import annotations + +import pandas as pd +import streamlit as st + +from core import review, workspace +from translations.translations import translate as t + + +STAGE_LABELS = { + "terminology": "Terminology", + "translation": "Translation", + "subtitles": "Subtitles", + "tts_text": "TTS Text", +} + +STATUS_ICON = { + "pending": "○", + "confirmed": "✓", + "stale": "!", +} + + +def render_status_row() -> None: + status = review.load_status() + cols = st.columns(4) + for column, stage in zip(cols, review.STAGES): + item = status["stages"][stage] + icon = STATUS_ICON.get(item["status"], "○") + column.metric(t(STAGE_LABELS[stage]), f"{icon} {t(item['status'])}") + + +def _missing_artifact(stage: str) -> bool: + path = review.artifact_path(stage) + if path.exists(): + return False + st.info(t("Waiting for the upstream step to generate this review file.")) + st.caption(str(path)) + return True + + +def _confirm_controls(stage: str) -> None: + col1, col2 = st.columns(2) + with col1: + if st.button(t("Confirm"), key=f"review_confirm_{stage}", width="stretch"): + review.mark_confirmed(stage) + st.rerun() + with col2: + if st.button(t("Mark as needs review"), key=f"review_pending_{stage}", width="stretch"): + review.mark_needs_review(stage) + st.rerun() + + +def terminology_tab() -> None: + stage = "terminology" + if _missing_artifact(stage): + return + try: + data = review.load_terminology() + except Exception as exc: + st.error(f"{t('Unable to load review file')}: {exc}") + return + + theme = st.text_area(t("Theme"), value=str(data.get("theme", "")), key="review_theme") + terms = pd.DataFrame(data.get("terms", []), columns=["src", "tgt", "note"]) + edited = st.data_editor( + terms, + num_rows="dynamic", + width="stretch", + key="review_terms_editor", + ) + if st.button(t("Save changes"), key="review_save_terminology", width="stretch"): + review.save_terminology(edited.fillna("").to_dict("records"), theme=theme) + st.success(t("Saved")) + st.rerun() + _confirm_controls(stage) + + +def table_tab(stage: str, required_columns: tuple[str, ...], editable_columns: tuple[str, ...]) -> None: + if _missing_artifact(stage): + return + try: + df = review.load_table(stage) + except Exception as exc: + st.error(f"{t('Unable to load review file')}: {exc}") + return + + disabled = [column for column in df.columns if column not in editable_columns] + edited = st.data_editor( + df, + disabled=disabled, + width="stretch", + hide_index=False, + key=f"review_editor_{stage}", + ) + if st.button(t("Save changes"), key=f"review_save_{stage}", width="stretch"): + if stage == "subtitles": + review.save_subtitles(edited) + else: + review.save_table(stage, edited, required_columns=required_columns) + st.success(t("Saved")) + st.rerun() + _confirm_controls(stage) + + +def review_section() -> None: + st.header(t("Review Workbench")) + if not workspace.get_active_workspace(): + st.info(t("Create or select a task workspace before reviewing.")) + return + + render_status_row() + tab_terms, tab_translation, tab_subtitles, tab_tts = st.tabs( + [ + t("Terminology"), + t("Translation"), + t("Subtitles"), + t("TTS Text"), + ] + ) + with tab_terms: + terminology_tab() + with tab_translation: + table_tab( + "translation", + required_columns=("Source", "Translation"), + editable_columns=("Translation",), + ) + with tab_subtitles: + table_tab( + "subtitles", + required_columns=("Source", "Translation"), + editable_columns=("Source", "Translation"), + ) + with tab_tts: + table_tab( + "tts_text", + required_columns=("number", "start_time", "end_time", "duration", "text", "origin"), + editable_columns=("text", "origin"), + ) +``` + +- [ ] **Step 2: Wire the section into `st.py`** + +Add: + +```python +from core.st_utils.review_section import review_section +``` + +Then call `review_section()` in `main()` after `download_video_section()` and before `text_processing_section()`. + +- [ ] **Step 3: Compile to catch import errors** + +Run: + +```bash +/Users/oliverchow/VideoLingo/.venv/bin/python -m compileall -q core/st_utils/review_section.py st.py +``` + +Expected: exit 0. + +- [ ] **Step 4: Commit** + +```bash +git add core/st_utils/review_section.py st.py +git commit -m "feat: add Streamlit review workbench" +``` + +--- + +### Task 5: Add Review-Aware Pipeline Buttons + +**Files:** +- Modify: `st.py` + +- [ ] **Step 1: Add helper functions near `_get_text_steps()`** + +Add this code to `st.py`: + +```python +def _stage_confirmed(stage: str) -> bool: + try: + status = review.load_status() + return status["stages"][stage]["status"] == "confirmed" + except Exception: + return False + + +def _artifact_exists(stage: str) -> bool: + try: + return review.artifact_path(stage).exists() + except Exception: + return False +``` + +- [ ] **Step 2: Replace `_get_text_steps()` with review-aware groups** + +Use this shape: + +```python +def _get_text_generation_steps(): + return [ + (t("WhisperX word-level transcription"), _2_asr.transcribe), + ( + t("Sentence segmentation using NLP and LLM"), + lambda: ( + _3_1_split_nlp.split_by_spacy(), + _3_2_split_meaning.split_sentences_by_meaning(), + ), + ), + (t("Summarization and terminology generation"), _4_1_summarize.get_summary), + ] + + +def _get_translation_steps(): + return [(t("Multi-step translation"), _4_2_translate.translate_all)] + + +def _get_subtitle_split_steps(): + return [(t("Cutting and aligning long subtitles"), _5_split_sub.split_for_sub_main)] + + +def _get_subtitle_finalize_steps(): + return [ + (t("Generating timeline and subtitles"), _6_gen_sub.align_timestamp_main), + (t("Merging subtitles into the video"), _7_sub_into_vid.merge_subtitles_to_video), + ] +``` + +- [ ] **Step 3: Replace text processing button logic** + +Inside `text_processing_section()`, keep the existing completed-output branch. In the not-completed branch, show one action at a time: + +```python +if runner.is_active or runner.is_done: + _task_control_panel("_text_runner") +elif not _artifact_exists("terminology"): + if st.button(t("Generate terminology for review"), key="generate_terms_button"): + runner.start(_get_text_generation_steps()) + st.rerun() +elif not _stage_confirmed("terminology"): + st.info(t("Review and confirm terminology before translation.")) +elif not _artifact_exists("translation"): + if st.button(t("Continue to translation"), key="continue_translation_button"): + runner.start(_get_translation_steps()) + st.rerun() +elif not _stage_confirmed("translation"): + st.info(t("Review and confirm translation before subtitle splitting.")) +elif not _artifact_exists("subtitles"): + if st.button(t("Generate subtitle split for review"), key="continue_subtitle_split_button"): + runner.start(_get_subtitle_split_steps()) + st.rerun() +elif not _stage_confirmed("subtitles"): + st.info(t("Review and confirm subtitles before final subtitle generation.")) +else: + if st.button(t("Generate final subtitles and video"), key="continue_subtitle_finalize_button"): + runner.start(_get_subtitle_finalize_steps()) + st.rerun() +``` + +- [ ] **Step 4: Replace `_get_audio_steps()` with review-aware groups** + +Use this shape: + +```python +def _get_tts_task_steps(): + return [(t("Generate audio tasks"), _8_1_audio_task.gen_audio_task_main)] + + +def _get_audio_finalize_steps(): + return [ + (t("Generate audio chunks"), _8_2_dub_chunks.gen_dub_chunks), + (t("Extract reference audio"), _9_refer_audio.extract_refer_audio_main), + (t("Generate and merge audio files"), _10_gen_audio.gen_audio), + (t("Merge full audio"), _11_merge_audio.merge_full_audio), + (t("Merge final audio into video"), _12_dub_to_vid.merge_video_audio), + ] +``` + +- [ ] **Step 5: Replace audio processing button logic** + +Inside `audio_processing_section()`, keep the completed-output branch. In the not-completed branch: + +```python +if runner.is_active or runner.is_done: + _task_control_panel("_audio_runner") +elif not _artifact_exists("tts_text"): + if st.button(t("Generate TTS text for review"), key="generate_tts_text_button"): + runner.start(_get_tts_task_steps()) + st.rerun() +elif not _stage_confirmed("tts_text"): + st.info(t("Review and confirm TTS text before dubbing.")) +else: + if st.button(t("Generate dubbing"), key="continue_audio_finalize_button"): + runner.start(_get_audio_finalize_steps()) + st.rerun() +``` + +- [ ] **Step 6: Compile** + +Run: + +```bash +/Users/oliverchow/VideoLingo/.venv/bin/python -m compileall -q st.py +``` + +Expected: exit 0. + +- [ ] **Step 7: Commit** + +```bash +git add st.py +git commit -m "feat: add review-aware pipeline checkpoints" +``` + +--- + +### Task 6: Full Verification and Documentation + +**Files:** +- Modify: `README.md` + +- [ ] **Step 1: Add README documentation** + +Add a short section after Task Workspaces: + +```markdown +### Human Review Workbench + +When a task workspace is active, VideoLingo shows a Review Workbench for terminology, translation, subtitle rows, and TTS text. Each review file belongs to the active workspace, and saving changes creates a backup under `artifacts/reviews/backups/`. + +The app pauses at review checkpoints so you can confirm terminology before translation, translation before subtitle splitting, subtitles before SRT generation, and TTS text before dubbing. +``` + +- [ ] **Step 2: Run unit tests** + +Run: + +```bash +/Users/oliverchow/VideoLingo/.venv/bin/python -m unittest discover -s tests -v +``` + +Expected: all tests pass. + +- [ ] **Step 3: Compile changed modules** + +Run: + +```bash +/Users/oliverchow/VideoLingo/.venv/bin/python -m compileall -q core batch st.py +``` + +Expected: exit 0. + +- [ ] **Step 4: Streamlit smoke test** + +Run: + +```bash +/Users/oliverchow/VideoLingo/.venv/bin/streamlit run st.py --server.headless=true --server.port=8765 --logger.level=error +``` + +Then request: + +```bash +/Users/oliverchow/VideoLingo/.venv/bin/python - <<'PY' +import urllib.request +print(urllib.request.urlopen("http://localhost:8765", timeout=10).status) +PY +``` + +Expected: `200`. Stop the Streamlit process after the check. + +- [ ] **Step 5: Commit documentation** + +```bash +git add README.md +git commit -m "docs: document human review workbench" +``` + +- [ ] **Step 6: Final status check** + +Run: + +```bash +git status --short --branch +git log --oneline main..HEAD +``` + +Expected: working tree clean and review workbench commits present. + +--- + +## Self-Review + +Spec coverage: + +- Terminology, translation, subtitle, and TTS text review are covered by Tasks 2, 3, and 4. +- Workspace-local status and backups are covered by Tasks 1 and 2. +- Downstream invalidation is covered by Tasks 2 and 3. +- Review-aware pipeline checkpoints are covered by Task 5. +- Documentation and verification are covered by Task 6. + +Placeholder scan: + +- No marker-only steps, fill-in instructions, or unspecified "add tests" steps. +- Every code-changing task includes concrete snippets and commands. + +Type consistency: + +- Stage names are consistently `terminology`, `translation`, `subtitles`, and `tts_text`. +- Required artifact columns match current pipeline files. +- Status values are consistently `pending`, `confirmed`, and `stale`. diff --git a/docs/superpowers/plans/2026-05-03-task-workspace.md b/docs/superpowers/plans/2026-05-03-task-workspace.md new file mode 100644 index 00000000..8ad00c97 --- /dev/null +++ b/docs/superpowers/plans/2026-05-03-task-workspace.md @@ -0,0 +1,732 @@ +# Task Workspace Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Add isolated task workspaces so Streamlit and batch runs can write artifacts under `workspace/jobs//` while legacy `output/` runs keep working. + +**Architecture:** Add `core/workspace.py` as the single workspace boundary. Existing pipeline code gets workspace-aware paths through `core/utils/models.py` and small call-site updates for hardcoded `output/` paths. Streamlit owns active job selection through session state and the workspace module mirrors it into an environment variable so pipeline functions can resolve paths without threading a job object through every step. + +**Tech Stack:** Python 3.10, ruamel.yaml, Streamlit session state, unittest, existing VideoLingo pipeline modules. + +--- + +## File Structure + +- Create `core/workspace.py`: workspace metadata, active workspace state, path resolution, config snapshots. +- Modify `core/utils/models.py`: replace static output strings with dynamic workspace-aware path objects. +- Modify `core/_1_ytdlp.py`: use workspace output paths for downloads and video discovery. +- Modify `core/st_utils/download_video_section.py`: upload/delete files in the active workspace output directory. +- Modify `core/st_utils/imports_and_utils.py`: zip SRT files from active workspace output. +- Modify `core/utils/onekeycleanup.py`: archive active workspace output into workspace artifacts/history. +- Modify `core/utils/delete_retry_dubbing.py`: delete dubbing files from active workspace output. +- Modify `st.py`: add task selector UI and activate selected workspace before pipeline actions. +- Modify `batch/utils/batch_processor.py`: create or reuse one workspace per batch row and snapshot row config. +- Modify `batch/utils/video_processor.py`: process each video inside the active workspace output directory. +- Create `tests/test_workspace.py`: focused workspace unit tests. + +--- + +### Task 1: Workspace Core + +**Files:** +- Create: `core/workspace.py` +- Test: `tests/test_workspace.py` + +- [ ] **Step 1: Write failing workspace creation tests** + +Create `tests/test_workspace.py` with tests that use temporary roots and config files: + +```python +import os +import tempfile +import unittest +from pathlib import Path + +from ruamel.yaml import YAML + +from core import workspace + + +class WorkspaceTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name) + self.config_path = self.root / "config.yaml" + self.config_path.write_text("display_language: en\napi:\n key: ''\n", encoding="utf-8") + workspace.clear_active_workspace() + + def tearDown(self): + workspace.clear_active_workspace() + + def test_create_job_writes_metadata_and_config_snapshot(self): + job = workspace.create_job( + name="My Video!", + source_type="upload", + source_path="input.mp4", + workspace_root=self.root / "jobs", + config_path=self.config_path, + ) + + self.assertTrue(Path(job["path"]).is_dir()) + self.assertTrue((Path(job["path"]) / "output").is_dir()) + self.assertTrue((Path(job["path"]) / "job.yaml").is_file()) + self.assertTrue((Path(job["path"]) / "config.snapshot.yaml").is_file()) + self.assertEqual(job["name"], "My Video!") + self.assertEqual(job["source"]["type"], "upload") + + def test_active_workspace_resolves_output_path(self): + job = workspace.create_job( + name="Video", + workspace_root=self.root / "jobs", + config_path=self.config_path, + ) + workspace.set_active_workspace(job["path"]) + + resolved = workspace.output_path("log", "file.txt") + + self.assertEqual( + os.fspath(resolved), + str(Path(job["path"]) / "output" / "log" / "file.txt"), + ) + + def test_legacy_output_path_without_active_workspace(self): + workspace.clear_active_workspace() + + self.assertEqual(os.fspath(workspace.output_path("log", "file.txt")), "output/log/file.txt") + + def test_archive_job_marks_status_without_deleting_files(self): + job = workspace.create_job( + name="Video", + workspace_root=self.root / "jobs", + config_path=self.config_path, + ) + + workspace.archive_job(job["path"]) + loaded = workspace.load_job(job["path"]) + + self.assertEqual(loaded["status"], "archived") + self.assertTrue(Path(job["path"]).exists()) + + def test_config_snapshot_is_independent_from_global_config(self): + job = workspace.create_job( + name="Video", + workspace_root=self.root / "jobs", + config_path=self.config_path, + ) + self.config_path.write_text("display_language: zh-CN\napi:\n key: changed\n", encoding="utf-8") + + yaml = YAML() + snapshot = yaml.load((Path(job["path"]) / "config.snapshot.yaml").read_text(encoding="utf-8")) + + self.assertEqual(snapshot["display_language"], "en") + self.assertEqual(snapshot["api"]["key"], "") +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: + +```bash +/Users/oliverchow/VideoLingo/.venv/bin/python -m unittest tests.test_workspace -v +``` + +Expected: FAIL because `core.workspace` does not exist. + +- [ ] **Step 3: Implement `core/workspace.py`** + +Add: + +```python +from __future__ import annotations + +import os +import re +import shutil +from datetime import datetime +from pathlib import Path +from typing import Any + +from ruamel.yaml import YAML + +ACTIVE_WORKSPACE_ENV = "VIDEOLINGO_ACTIVE_WORKSPACE" +DEFAULT_WORKSPACE_ROOT = Path("workspace") / "jobs" +JOB_FILE = "job.yaml" +CONFIG_SNAPSHOT_FILE = "config.snapshot.yaml" + +yaml = YAML() +yaml.preserve_quotes = True + + +class WorkspacePath: + def __init__(self, relative_path: str): + self.relative_path = relative_path.replace("\\", "/").strip("/") + + def resolve(self) -> str: + return resolve_path(self.relative_path) + + def __fspath__(self) -> str: + return self.resolve() + + def __str__(self) -> str: + return self.resolve() + + def __repr__(self) -> str: + return repr(self.resolve()) + + +def _now() -> str: + return datetime.now().astimezone().isoformat(timespec="seconds") + + +def _slugify(value: str) -> str: + slug = re.sub(r"[^A-Za-z0-9._-]+", "-", value.strip()).strip("-._") + return slug[:60] or "task" + + +def get_active_workspace() -> str | None: + value = os.environ.get(ACTIVE_WORKSPACE_ENV, "").strip() + return value or None + + +def set_active_workspace(path: str | os.PathLike[str]) -> str: + job_dir = Path(path) + if not (job_dir / JOB_FILE).is_file(): + raise FileNotFoundError(f"Workspace metadata not found: {job_dir / JOB_FILE}") + os.environ[ACTIVE_WORKSPACE_ENV] = str(job_dir) + return str(job_dir) + + +def clear_active_workspace() -> None: + os.environ.pop(ACTIVE_WORKSPACE_ENV, None) + + +def workspace_path(relative_path: str) -> WorkspacePath: + return WorkspacePath(relative_path) + + +def output_path(*parts: str) -> WorkspacePath: + clean = [part.strip("/\\") for part in parts if part] + return workspace_path("/".join(["output", *clean])) + + +def resolve_path(relative_path: str) -> str: + active = get_active_workspace() + if active: + return str(Path(active) / relative_path) + return relative_path + + +def create_job( + name: str | None = None, + source_type: str = "", + source_path: str = "", + source_url: str = "", + workspace_root: str | os.PathLike[str] = DEFAULT_WORKSPACE_ROOT, + config_path: str | os.PathLike[str] = "config.yaml", +) -> dict[str, Any]: + title = name or "Untitled Task" + timestamp = datetime.now().astimezone().strftime("%Y%m%d-%H%M%S") + job_id = f"{timestamp}-{_slugify(title)}" + job_dir = Path(workspace_root) / job_id + suffix = 2 + while job_dir.exists(): + job_dir = Path(workspace_root) / f"{job_id}-{suffix}" + suffix += 1 + + (job_dir / "output").mkdir(parents=True, exist_ok=False) + (job_dir / "logs").mkdir(parents=True, exist_ok=True) + (job_dir / "artifacts").mkdir(parents=True, exist_ok=True) + + created_at = _now() + job = { + "id": job_dir.name, + "name": title, + "path": str(job_dir), + "source": {"type": source_type, "path": source_path, "url": source_url}, + "status": "created", + "created_at": created_at, + "updated_at": created_at, + "current_stage": "", + "error": "", + } + save_job(job_dir, job) + save_config_snapshot(job_dir, config_path) + return job + + +def save_config_snapshot(job_dir: str | os.PathLike[str], config_path: str | os.PathLike[str] = "config.yaml") -> None: + src = Path(config_path) + dst = Path(job_dir) / CONFIG_SNAPSHOT_FILE + if src.exists(): + shutil.copy2(src, dst) + else: + dst.write_text("{}\n", encoding="utf-8") + + +def load_job(job_dir: str | os.PathLike[str]) -> dict[str, Any]: + with open(Path(job_dir) / JOB_FILE, "r", encoding="utf-8") as file: + data = yaml.load(file) or {} + data["path"] = str(Path(job_dir)) + return data + + +def save_job(job_dir: str | os.PathLike[str], data: dict[str, Any]) -> None: + data = dict(data) + data["path"] = str(Path(job_dir)) + data["updated_at"] = data.get("updated_at") or _now() + with open(Path(job_dir) / JOB_FILE, "w", encoding="utf-8") as file: + yaml.dump(data, file) + + +def update_job(job_dir: str | os.PathLike[str], **updates: Any) -> dict[str, Any]: + job = load_job(job_dir) + job.update(updates) + job["updated_at"] = _now() + save_job(job_dir, job) + return job + + +def archive_job(job_dir: str | os.PathLike[str]) -> dict[str, Any]: + return update_job(job_dir, status="archived") + + +def list_jobs(workspace_root: str | os.PathLike[str] = DEFAULT_WORKSPACE_ROOT) -> list[dict[str, Any]]: + root = Path(workspace_root) + if not root.exists(): + return [] + jobs = [] + for child in sorted(root.iterdir(), reverse=True): + if child.is_dir() and (child / JOB_FILE).is_file(): + jobs.append(load_job(child)) + return jobs +``` + +- [ ] **Step 4: Run workspace tests** + +Run: + +```bash +/Users/oliverchow/VideoLingo/.venv/bin/python -m unittest tests.test_workspace -v +``` + +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add core/workspace.py tests/test_workspace.py +git commit -m "feat: add task workspace core" +``` + +--- + +### Task 2: Workspace-Aware Output Constants + +**Files:** +- Modify: `core/utils/models.py` +- Modify: `core/_10_gen_audio.py` +- Modify: `core/_11_merge_audio.py` +- Test: `tests/test_workspace.py` + +- [ ] **Step 1: Add a test for dynamic model paths** + +Append to `tests/test_workspace.py`: + +```python + def test_model_constants_resolve_against_active_workspace(self): + from core.utils import models + + job = workspace.create_job( + name="Video", + workspace_root=self.root / "jobs", + config_path=self.config_path, + ) + workspace.set_active_workspace(job["path"]) + + self.assertEqual( + os.fspath(models._2_CLEANED_CHUNKS), + str(Path(job["path"]) / "output" / "log" / "cleaned_chunks.xlsx"), + ) + self.assertEqual(str(models._AUDIO_TMP_DIR), str(Path(job["path"]) / "output" / "audio" / "tmp")) +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: + +```bash +/Users/oliverchow/VideoLingo/.venv/bin/python -m unittest tests.test_workspace.WorkspaceTests.test_model_constants_resolve_against_active_workspace -v +``` + +Expected: FAIL because constants are static strings. + +- [ ] **Step 3: Update `core/utils/models.py`** + +Replace static strings with: + +```python +from core.workspace import workspace_path + +_2_CLEANED_CHUNKS = workspace_path("output/log/cleaned_chunks.xlsx") +_3_1_SPLIT_BY_NLP = workspace_path("output/log/split_by_nlp.txt") +_3_2_SPLIT_BY_MEANING = workspace_path("output/log/split_by_meaning.txt") +_4_1_TERMINOLOGY = workspace_path("output/log/terminology.json") +_4_2_TRANSLATION = workspace_path("output/log/translation_results.xlsx") +_5_SPLIT_SUB = workspace_path("output/log/translation_results_for_subtitles.xlsx") +_5_REMERGED = workspace_path("output/log/translation_results_remerged.xlsx") +_8_1_AUDIO_TASK = workspace_path("output/audio/tts_tasks.xlsx") + +_OUTPUT_DIR = workspace_path("output") +_AUDIO_DIR = workspace_path("output/audio") +_RAW_AUDIO_FILE = workspace_path("output/audio/raw.mp3") +_VOCAL_AUDIO_FILE = workspace_path("output/audio/vocal.mp3") +_BACKGROUND_AUDIO_FILE = workspace_path("output/audio/background.mp3") +_AUDIO_REFERS_DIR = workspace_path("output/audio/refers") +_AUDIO_SEGS_DIR = workspace_path("output/audio/segs") +_AUDIO_TMP_DIR = workspace_path("output/audio/tmp") +``` + +Keep the existing `__all__` list unchanged. + +- [ ] **Step 4: Replace import-time audio templates** + +In `core/_10_gen_audio.py`, replace global `TEMP_FILE_TEMPLATE` and `OUTPUT_FILE_TEMPLATE` with helper functions: + +```python +def temp_file_path(number, line_index): + return os.path.join(str(_AUDIO_TMP_DIR), f"{number}_{line_index}_temp.wav") + + +def output_file_path(number, line_index): + return os.path.join(str(_AUDIO_SEGS_DIR), f"{number}_{line_index}.wav") +``` + +Use those helpers in `process_row()` and `merge_chunks()`. + +In `core/_11_merge_audio.py`, replace `OUTPUT_FILE_TEMPLATE` with: + +```python +def output_file_path(number, line_index): + return os.path.join(str(_AUDIO_SEGS_DIR), f"{number}_{line_index}.wav") +``` + +Use it in `get_audio_files()`. + +- [ ] **Step 5: Run tests** + +Run: + +```bash +/Users/oliverchow/VideoLingo/.venv/bin/python -m unittest tests.test_workspace -v +/Users/oliverchow/VideoLingo/.venv/bin/python -m compileall -q core +``` + +Expected: PASS and no compile errors. + +- [ ] **Step 6: Commit** + +```bash +git add core/utils/models.py core/_10_gen_audio.py core/_11_merge_audio.py tests/test_workspace.py +git commit -m "feat: resolve pipeline paths through active workspace" +``` + +--- + +### Task 3: Streamlit Workspace Selector And Output Helpers + +**Files:** +- Modify: `st.py` +- Modify: `core/_1_ytdlp.py` +- Modify: `core/st_utils/download_video_section.py` +- Modify: `core/st_utils/imports_and_utils.py` +- Modify: `core/utils/onekeycleanup.py` +- Modify: `core/utils/delete_retry_dubbing.py` +- Test: `tests/test_workspace.py` + +- [ ] **Step 1: Add tests for active output helpers** + +Append: + +```python + def test_find_video_files_uses_active_workspace_output(self): + from core._1_ytdlp import find_video_files + + job = workspace.create_job( + name="Video", + workspace_root=self.root / "jobs", + config_path=self.config_path, + ) + output_dir = Path(job["path"]) / "output" + (output_dir / "sample.mp4").write_bytes(b"video") + workspace.set_active_workspace(job["path"]) + + self.assertEqual(find_video_files(), str(output_dir / "sample.mp4")) +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: + +```bash +/Users/oliverchow/VideoLingo/.venv/bin/python -m unittest tests.test_workspace.WorkspaceTests.test_find_video_files_uses_active_workspace_output -v +``` + +Expected: FAIL because `find_video_files()` scans legacy `output/`. + +- [ ] **Step 3: Update `core/_1_ytdlp.py`** + +Import workspace helpers: + +```python +from core.workspace import output_path +``` + +Change defaults: + +```python +def download_video_ytdlp(url, save_path=None, resolution='1080'): + save_path = str(save_path or output_path()) +``` + +```python +def find_video_files(save_path=None): + save_path = str(save_path or output_path()) +``` + +Update the generated-output filter to use the resolved output path: + +```python +generated_prefix = os.path.join(save_path, "output") +video_files = [file for file in video_files if not os.path.basename(file).startswith("output")] +``` + +- [ ] **Step 4: Update upload/download UI paths** + +In `core/st_utils/download_video_section.py`, import `output_path` and replace `OUTPUT_DIR` usage with `str(output_path())` inside functions. `convert_audio_to_video()` should compute `output_dir = str(output_path())`. + +In `core/st_utils/imports_and_utils.py`, use `str(output_path())` for the subtitle zip source. + +In `core/utils/delete_retry_dubbing.py`, use `output_path()` for `dub.wav`, `output_dub.mp4`, and `audio/segs`. + +In `core/utils/onekeycleanup.py`, use `str(output_path())` as the source output directory and move files into `artifacts/history/` inside the active workspace when one is active. Keep `history/` as the legacy fallback. + +- [ ] **Step 5: Add Streamlit selector** + +In `st.py`, import: + +```python +from core import workspace +``` + +Add helper: + +```python +def workspace_section(): + st.sidebar.header(t("Task Workspace")) + jobs = workspace.list_jobs() + labels = [f"{job['name']} · {job['status']} · {job['id']}" for job in jobs] + selected = st.sidebar.selectbox(t("Continue task"), [""] + labels) + if selected: + job = jobs[labels.index(selected)] + workspace.set_active_workspace(job["path"]) + st.session_state["_active_workspace"] = job["path"] + elif st.session_state.get("_active_workspace"): + workspace.set_active_workspace(st.session_state["_active_workspace"]) + + new_name = st.sidebar.text_input(t("New task name"), value="") + if st.sidebar.button(t("New task"), key="new_workspace_task"): + job = workspace.create_job(name=new_name or "VideoLingo Task") + workspace.set_active_workspace(job["path"]) + st.session_state["_active_workspace"] = job["path"] + st.rerun() + + active = workspace.get_active_workspace() + if active: + job = workspace.load_job(active) + st.sidebar.caption(f"{job['name']} · {job['status']}") + if st.sidebar.button(t("Archive task"), key="archive_workspace_task"): + workspace.archive_job(active) + workspace.clear_active_workspace() + st.session_state.pop("_active_workspace", None) + st.rerun() +``` + +Call `workspace_section()` before `page_setting()` in the sidebar. Change `SUB_VIDEO` and `DUB_VIDEO` to calls inside functions: + +```python +sub_video = str(workspace.output_path("output_sub.mp4")) +dub_video = str(workspace.output_path("output_dub.mp4")) +``` + +- [ ] **Step 6: Run tests and compile** + +Run: + +```bash +/Users/oliverchow/VideoLingo/.venv/bin/python -m unittest tests.test_workspace -v +/Users/oliverchow/VideoLingo/.venv/bin/python -m compileall -q core st.py +``` + +Expected: PASS and no compile errors. + +- [ ] **Step 7: Commit** + +```bash +git add st.py core/_1_ytdlp.py core/st_utils/download_video_section.py core/st_utils/imports_and_utils.py core/utils/onekeycleanup.py core/utils/delete_retry_dubbing.py tests/test_workspace.py +git commit -m "feat: add Streamlit task workspace selection" +``` + +--- + +### Task 4: Batch Workspace Isolation + +**Files:** +- Modify: `batch/utils/batch_processor.py` +- Modify: `batch/utils/video_processor.py` +- Test: `tests/test_workspace.py` + +- [ ] **Step 1: Add batch workspace tests** + +Append: + +```python + def test_batch_workspace_path_can_be_stored_per_row(self): + import pandas as pd + from batch.utils.batch_processor import ensure_workspace_columns + + df = pd.DataFrame({"Video File": ["a.mp4"], "Status": [None]}) + updated = ensure_workspace_columns(df) + + self.assertIn("Workspace", updated.columns) + self.assertEqual(updated.at[0, "Workspace"], "") +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: + +```bash +/Users/oliverchow/VideoLingo/.venv/bin/python -m unittest tests.test_workspace.WorkspaceTests.test_batch_workspace_path_can_be_stored_per_row -v +``` + +Expected: FAIL because `ensure_workspace_columns` does not exist. + +- [ ] **Step 3: Update batch processor** + +Add to `batch/utils/batch_processor.py`: + +```python +from core import workspace + + +def ensure_workspace_columns(df): + if "Workspace" not in df.columns: + df["Workspace"] = "" + return df + + +def get_or_create_batch_workspace(row, index): + current = row.get("Workspace", "") + if isinstance(current, str) and current and os.path.exists(current): + return current + video_file = str(row["Video File"]) + job = workspace.create_job( + name=os.path.splitext(os.path.basename(video_file))[0] or f"Batch Task {index + 1}", + source_type="batch", + source_path=video_file if not video_file.startswith("http") else "", + source_url=video_file if video_file.startswith("http") else "", + ) + return job["path"] +``` + +In `process_batch()`, call `ensure_workspace_columns(df)` after reading Excel. Before `process_video`, set active workspace: + +```python +workspace_path = get_or_create_batch_workspace(row, index) +df.at[index, "Workspace"] = workspace_path +workspace.set_active_workspace(workspace_path) +``` + +Clear active workspace after each row in `finally`. + +- [ ] **Step 4: Update video processor** + +In `batch/utils/video_processor.py`, import `output_path`. Replace `OUTPUT_DIR = 'output'` usage in `prepare_output_folder()` and `process_input_file()` with `str(output_path())`. + +Remove restore-from-`batch/output/ERROR` behavior from `batch_processor.py` for workspace-aware retries. Failed work remains in its own workspace. + +- [ ] **Step 5: Run tests and compile** + +Run: + +```bash +/Users/oliverchow/VideoLingo/.venv/bin/python -m unittest tests.test_workspace -v +/Users/oliverchow/VideoLingo/.venv/bin/python -m compileall -q batch core +``` + +Expected: PASS and no compile errors. + +- [ ] **Step 6: Commit** + +```bash +git add batch/utils/batch_processor.py batch/utils/video_processor.py tests/test_workspace.py +git commit -m "feat: isolate batch tasks in workspaces" +``` + +--- + +### Task 5: Final Verification And Docs Note + +**Files:** +- Modify: `README.md` or `docs/pages/docs/start.en-US.md` + +- [ ] **Step 1: Add a short user-facing note** + +Add a short note near the quick start docs: + +```markdown +### Task Workspaces + +VideoLingo stores each new task under `workspace/jobs//` when launched from the Streamlit task selector. Each workspace contains a config snapshot and its own `output/` folder, so separate videos do not overwrite each other's intermediate files. +``` + +- [ ] **Step 2: Run full lightweight verification** + +Run: + +```bash +/Users/oliverchow/VideoLingo/.venv/bin/python -m unittest discover -s tests -v +/Users/oliverchow/VideoLingo/.venv/bin/python -m compileall -q core batch st.py +git status --short +``` + +Expected: tests pass, compile succeeds, status shows only intended files. + +- [ ] **Step 3: Commit** + +```bash +git add README.md docs/pages/docs/start.en-US.md +git commit -m "docs: document task workspace behavior" +``` + +--- + +## Self-Review + +Spec coverage: + +- Workspace creation, metadata, config snapshots: Task 1. +- Workspace-aware output paths and legacy fallback: Task 1 and Task 2. +- Streamlit new/continue/archive task UI: Task 3. +- Batch per-video workspace isolation: Task 4. +- Testing: Tasks 1 through 5. + +Scope check: + +- The plan does not add SQLite, multi-user execution, human review pages, cost tracking, or provider health dashboards. +- The plan keeps migration incremental and compatible with legacy `output/`. + +Placeholder scan: + +- No placeholder markers or vague future task markers are intentionally left in this plan. diff --git a/docs/superpowers/specs/2026-05-03-human-review-workbench-design.md b/docs/superpowers/specs/2026-05-03-human-review-workbench-design.md new file mode 100644 index 00000000..303e6628 --- /dev/null +++ b/docs/superpowers/specs/2026-05-03-human-review-workbench-design.md @@ -0,0 +1,244 @@ +# Human Review Workbench Design + +## Goal + +Add a human confirmation page for terminology, translation, subtitle segmentation, and TTS text inside each VideoLingo task workspace. + +The first version should turn the pipeline from a black-box run into a controllable review workflow without replacing the existing JSON, Excel, and SRT artifacts. The feature should help long-video users catch terminology mistakes, awkward translations, bad subtitle splits, and unnatural TTS text before final video generation. + +## Current Problem + +VideoLingo already generates useful intermediate files: + +- `output/log/terminology.json` +- `output/log/translation_results.xlsx` +- `output/log/translation_results_for_subtitles.xlsx` +- `output/log/translation_results_remerged.xlsx` +- `output/audio/tts_tasks.xlsx` +- final SRT files under `output/` and `output/audio/` + +However, the Streamlit flow currently runs major stages in sequence. Users can edit files manually, but the product does not tell them when to review, what file matters, or which later steps must be rerun after a correction. + +This is especially painful for long videos because one terminology mistake can propagate into many lines, and one bad subtitle split can affect both display subtitles and dubbing. + +## Recommended Approach + +Add a lightweight review workbench inside the active task workspace. + +The workbench exposes four review tabs: + +1. Terminology +2. Translation +3. Subtitles +4. TTS Text + +Each tab loads the existing artifact, shows it in an editable Streamlit table or form, lets the user save changes, and records whether the artifact has been confirmed. The pipeline inserts review checkpoints between generation steps, so users can confirm quality before continuing. + +This approach keeps the first version practical: + +- No database. +- No full timeline editor. +- No video-synchronized subtitle editing. +- No rewrite of the translation, subtitle, or TTS engines. + +## Scope For Version 1 + +Version 1 includes: + +- Review and edit terminology terms: `src`, `tgt`, and `note`. +- Review and edit translation rows: `Source` and `Translation`. +- Review and edit subtitle rows from the display subtitle split file. +- Review and edit audio/TTS task rows: `text` and `origin`, while preserving timing columns. +- Save edits back to the existing workspace artifact files. +- Backup the previous artifact before saving. +- Mark each review stage as `pending`, `confirmed`, or `stale`. +- Invalidate downstream review stages when an upstream artifact changes. +- Split the current Streamlit pipeline into smaller steps so review checkpoints can appear naturally. +- Provide buttons to confirm an artifact and continue the next stage. + +Version 1 does not include: + +- Per-word or waveform-level subtitle timing edits. +- Video preview synchronized with each subtitle row. +- Comments, assignments, or multi-user review. +- Side-by-side version diff UI. +- AI rewrite suggestions inside the review table. +- Batch review UI for many videos at once. + +Those can be added later once the artifact and checkpoint model is stable. + +## Review Stages + +### Terminology Review + +Runs after `_4_1_summarize.get_summary()` and before `_4_2_translate.translate_all()`. + +The page loads `output/log/terminology.json`, displays the `terms` list, and preserves other JSON keys such as `theme`. Users can add, delete, and edit terms. Confirming terminology means the translation step can use the reviewed glossary. + +Saving terminology marks translation, subtitles, and TTS text as `stale`. + +### Translation Review + +Runs after `_4_2_translate.translate_all()` and before `_5_split_sub.split_for_sub_main()`. + +The page loads `output/log/translation_results.xlsx`. Users can edit translation rows while keeping the source rows visible. Confirming translation means subtitle splitting can use the reviewed text. + +Saving translation marks subtitles and TTS text as `stale`. + +### Subtitle Review + +Runs after `_5_split_sub.split_for_sub_main()` and before `_6_gen_sub.align_timestamp_main()`. + +The page loads `output/log/translation_results_for_subtitles.xlsx` for display subtitles. Users can edit `Source` and `Translation` rows. If the display subtitle file and `output/log/translation_results_remerged.xlsx` have the same row count, saving subtitle edits updates matching rows in both files. If the row counts differ, saving updates only the display subtitle file and marks TTS text as `stale`, so the audio subtitle source is regenerated before dubbing continues. + +Confirming subtitles allows the app to regenerate SRT files through `_6_gen_sub.align_timestamp_main()`. + +Saving subtitles marks TTS text as `stale`. + +### TTS Text Review + +Runs after `_8_1_audio_task.gen_audio_task_main()` and before `_8_2_dub_chunks.gen_dub_chunks()`. + +The page loads `output/audio/tts_tasks.xlsx`. Users can edit the text that will be spoken by TTS, while timing fields remain visible but not the primary editing target. Confirming TTS text allows reference extraction, chunk generation, audio generation, and final dubbing to continue. + +Saving TTS text does not invalidate later review stages in version 1 because there is no later human review checkpoint. + +## Data Model + +Each task workspace gets: + +```text +workspace/jobs// + artifacts/ + reviews/ + review_status.yaml + backups/ + -. +``` + +`review_status.yaml` stores: + +```yaml +stages: + terminology: + status: "pending|confirmed|stale" + artifact: "output/log/terminology.json" + confirmed_at: "" + updated_at: "" + translation: + status: "pending|confirmed|stale" + artifact: "output/log/translation_results.xlsx" + confirmed_at: "" + updated_at: "" + subtitles: + status: "pending|confirmed|stale" + artifact: "output/log/translation_results_for_subtitles.xlsx" + confirmed_at: "" + updated_at: "" + tts_text: + status: "pending|confirmed|stale" + artifact: "output/audio/tts_tasks.xlsx" + confirmed_at: "" + updated_at: "" +``` + +The status file is workspace-local and should not affect other tasks. + +## Streamlit UX + +The sidebar continues to own task workspace selection. The main app adds a review workbench section after task selection and before final downloads. + +Recommended layout: + +- A compact status row showing the four review stages. +- Tabs for Terminology, Translation, Subtitles, and TTS Text. +- Each tab shows a file-missing state when the upstream step has not generated the artifact yet. +- Editable tables use `st.data_editor` for tabular files. +- Terminology review uses a table for `terms` plus a small read-only or editable area for `theme`. +- Buttons per tab: + - Save changes + - Confirm + - Mark as needs review + +The pipeline buttons should communicate the next required action. For example, after terminology is generated, the next visible action is to review terminology before translation continues. + +## Pipeline Flow + +The text workflow should be split into review-aware steps: + +1. ASR +2. sentence segmentation +3. terminology generation +4. terminology review +5. translation +6. translation review +7. subtitle split +8. subtitle review +9. SRT generation +10. subtitle burn-in + +The audio workflow should be split into: + +1. TTS task generation +2. TTS text review +3. dub chunk generation +4. reference audio extraction +5. TTS audio generation +6. full audio merge +7. final video merge + +For the first version, review checkpoints are handled by Streamlit UI state and the workspace status file. The background runner should stop after a generation step if the next checkpoint needs review. + +## Invalidation Rules + +When a user saves an upstream artifact, downstream stages become stale: + +- Terminology save invalidates translation, subtitles, and TTS text. +- Translation save invalidates subtitles and TTS text. +- Subtitle save invalidates TTS text. +- TTS text save invalidates no review stage. + +Stale does not delete files. It tells the user that later artifacts were generated from older content and should be regenerated or reconfirmed. + +## Error Handling + +If an artifact file is missing, the tab shows a clear waiting state and does not create a blank artifact silently. + +If a file cannot be parsed, the tab shows the parse error and leaves the file unchanged. + +Before saving, the workbench creates a backup copy in `artifacts/reviews/backups/`. If saving fails, the original artifact remains available. + +If a user edits table columns that are required by downstream code, the workbench validates required columns before writing. + +If no active task workspace exists, the review workbench should show a prompt to create or select a task first. + +## Testing + +Add focused tests for: + +- Creating a default `review_status.yaml`. +- Marking a stage confirmed. +- Saving terminology while preserving non-term JSON keys. +- Saving translation Excel and creating a backup. +- Invalidating downstream stages after upstream saves. +- Loading review artifacts through workspace-aware paths. + +Manual verification: + +- Create a task and run through terminology generation. +- Edit and confirm terminology, then continue translation. +- Edit and confirm translation, then continue subtitle split. +- Edit subtitle rows, regenerate SRT files, and confirm the SRT reflects edits. +- Generate TTS tasks, edit TTS text, continue dubbing, and confirm audio tasks use the edited text. + +## Implementation Notes + +Add a small review module rather than embedding all review logic in `st.py`. + +Suggested module responsibilities: + +- `core/review.py`: status file, artifact loading/saving, backups, invalidation rules. +- `core/st_utils/review_section.py`: Streamlit tabs and review controls. +- `st.py`: wire the review section and split workflow steps. + +The first implementation should keep artifact schemas close to the existing files so the current core pipeline remains the source of truth. diff --git a/docs/superpowers/specs/2026-05-03-task-workspace-design.md b/docs/superpowers/specs/2026-05-03-task-workspace-design.md new file mode 100644 index 00000000..9312f43c --- /dev/null +++ b/docs/superpowers/specs/2026-05-03-task-workspace-design.md @@ -0,0 +1,147 @@ +# Task Workspace Design + +## Goal + +Move VideoLingo from a single shared `output/` workflow toward isolated task workspaces. The first version should protect users from cross-run file/config pollution while preserving the current pipeline behavior. + +This is the foundation for later review pages, resumable retries, cost tracking, and provider health checks. + +## Current Problem + +The app writes most intermediate and final artifacts into global paths such as `output/`, `history/`, and `batch/output/`. Batch processing also temporarily mutates global `config.yaml` values. This makes it easy for one video run to affect another run, especially when users retry failed work, process multiple videos, or switch language/API settings. + +## Recommended Approach + +Use a lightweight workspace layer before adding a database. + +Each task gets a directory under: + +```text +workspace/jobs// +``` + +Initial layout: + +```text +workspace/jobs// + job.yaml + config.snapshot.yaml + output/ + logs/ + artifacts/ +``` + +`job.yaml` stores task metadata and status. `config.snapshot.yaml` stores the effective config at task creation time. `output/` mirrors the current global `output/` layout so the existing pipeline can be migrated gradually. + +## Scope For Version 1 + +Version 1 includes: + +- Create a new workspace for each Streamlit task. +- Continue an existing workspace from the UI. +- Archive a workspace without deleting it. +- Save a config snapshot per workspace. +- Route pipeline output paths through a small path helper instead of hardcoding new global paths. +- Keep compatibility with the existing `output/` directory during migration. +- Update batch mode so each video receives its own workspace and config snapshot. + +Version 1 does not include: + +- SQLite or a full task database. +- Multi-user permissions. +- Concurrent execution of multiple tasks in one Streamlit session. +- Human subtitle/terminology/translation review pages. +- Cost estimation and provider health dashboards. + +Those features should build on top of this workspace layer later. + +## Data Model + +`job.yaml`: + +```yaml +id: "20260503-153012-video-title" +name: "Video title" +source: + type: "upload|youtube|batch" + path: "" + url: "" +status: "created|running|paused|completed|failed|archived" +created_at: "2026-05-03T15:30:12+08:00" +updated_at: "2026-05-03T15:30:12+08:00" +current_stage: "" +error: "" +``` + +The schema is intentionally small. Later versions can add review checkpoints, token/cost totals, provider health, and retry history without changing the core directory layout. + +## Path Strategy + +Add a `core/workspace.py` module that owns: + +- Creating job IDs and directories. +- Reading/writing `job.yaml`. +- Saving config snapshots. +- Resolving paths such as `output/log/cleaned_chunks.xlsx` relative to the active workspace. +- Falling back to the legacy project root for old runs. + +Existing modules should avoid constructing fresh `output/...` strings when touched. The first migration target is `core/utils/models.py`, because many pipeline files import output constants from there. + +## Streamlit UX + +The first screen should become a simple task selector: + +- New task +- Continue task +- Archive task + +When a task is active, the app shows the current workspace name and status. Existing download buttons and preview behavior should continue to work because they read from the active workspace output directory. + +The UI should avoid exposing internal paths unless useful for debugging. + +## Batch UX + +Batch mode should create one workspace per input video. Each workspace receives: + +- The per-row source/target language values. +- A config snapshot. +- Its own output directory. +- Its own status and error message. + +Batch progress can still be written back to `batch/tasks_setting.xlsx`, but the workspace path should be stored so failed tasks can resume from the correct files. + +## Error Handling + +If a workspace cannot be created or loaded, the app should fail before starting the pipeline. + +If a task fails during processing, the active workspace remains intact and `job.yaml` records the failed stage and error message. Users can retry using the same workspace. + +If no active workspace exists, path helpers should default to legacy behavior so older scripts remain usable. + +## Testing + +Add focused tests for: + +- Creating a workspace and writing metadata. +- Saving config snapshots without mutating global config. +- Resolving output paths with and without an active workspace. +- Batch workspace creation per video row. +- Legacy fallback to `output/`. + +Manual verification: + +- Start a new Streamlit task and confirm files land under `workspace/jobs//output/`. +- Start a second task and confirm it does not read the first task's output. +- Run batch mode with two videos and confirm each receives separate workspace files. + +## Implementation Notes + +This should be done incrementally: + +1. Add workspace module and tests. +2. Add active workspace selection in Streamlit session state. +3. Migrate output constants in `core/utils/models.py`. +4. Update cleanup/download/history paths to use active workspace paths. +5. Update batch mode to create and use per-video workspaces. + +The migration should keep old `output/` behavior working until all pipeline modules have moved to workspace-aware paths. diff --git a/st.py b/st.py index 8cd967b1..52ae8582 100644 --- a/st.py +++ b/st.py @@ -1,8 +1,10 @@ import streamlit as st import os, sys, time from core.st_utils.imports_and_utils import * +from core.st_utils.review_section import review_section from core.st_utils.task_runner import TaskRunner from core import * +from core import review, workspace # SET PATH current_dir = os.path.dirname(os.path.abspath(__file__)) @@ -11,13 +13,18 @@ st.set_page_config(page_title="VideoLingo", page_icon="docs/logo.svg") -SUB_VIDEO = "output/output_sub.mp4" -DUB_VIDEO = "output/output_dub.mp4" - # ─── Task control UI (auto-refreshes every 1s while task is active) ─── +def _sub_video_path(): + return str(workspace.output_path("output_sub.mp4")) + + +def _dub_video_path(): + return str(workspace.output_path("output_dub.mp4")) + + @st.fragment(run_every=1) def _task_control_panel(runner_key: str): """Renders progress bar + pause/stop buttons. Auto-refreshes every 1s.""" @@ -92,9 +99,23 @@ def _task_control_panel(runner_key: str): # ─── Text processing ─── -def _get_text_steps(): - """Return the subtitle processing steps as (label, callable) list.""" - steps = [ +def _stage_confirmed(stage: str) -> bool: + try: + status = review.load_status() + return status["stages"][stage]["status"] == "confirmed" + except Exception: + return False + + +def _artifact_exists(stage: str) -> bool: + try: + return review.artifact_path(stage).exists() + except Exception: + return False + + +def _get_text_generation_steps(): + return [ (t("WhisperX word-level transcription"), _2_asr.transcribe), ( t("Sentence segmentation using NLP and LLM"), @@ -103,28 +124,29 @@ def _get_text_steps(): _3_2_split_meaning.split_sentences_by_meaning(), ), ), - ( - t("Summarization and multi-step translation"), - lambda: (_4_1_summarize.get_summary(), _4_2_translate.translate_all()), - ), - ( - t("Cutting and aligning long subtitles"), - lambda: ( - _5_split_sub.split_for_sub_main(), - _6_gen_sub.align_timestamp_main(), - ), - ), - ( - t("Merging subtitles into the video"), - _7_sub_into_vid.merge_subtitles_to_video, - ), + (t("Summarization and terminology generation"), _4_1_summarize.get_summary), + ] + + +def _get_translation_steps(): + return [(t("Multi-step translation"), _4_2_translate.translate_all)] + + +def _get_subtitle_split_steps(): + return [(t("Cutting and aligning long subtitles"), _5_split_sub.split_for_sub_main)] + + +def _get_subtitle_finalize_steps(): + return [ + (t("Generating timeline and subtitles"), _6_gen_sub.align_timestamp_main), + (t("Merging subtitles into the video"), _7_sub_into_vid.merge_subtitles_to_video), ] - return steps def text_processing_section(): st.header(t("b. Translate and Generate Subtitles")) runner = TaskRunner.get(st.session_state, "_text_runner") + sub_video = _sub_video_path() with st.container(border=True): st.markdown( @@ -134,29 +156,58 @@ def text_processing_section():

1. {t("WhisperX word-level transcription")}
2. {t("Sentence segmentation using NLP and LLM")}
- 3. {t("Summarization and multi-step translation")}
- 4. {t("Cutting and aligning long subtitles")}
+ 3. {t("Generate terminology for review")}
+ 4. {t("Review terminology, translation, and subtitles")}
5. {t("Generating timeline and subtitles")}
6. {t("Merging subtitles into the video")} """, unsafe_allow_html=True, ) - if not os.path.exists(SUB_VIDEO): + if not os.path.exists(sub_video): if runner.is_active: _task_control_panel("_text_runner") elif runner.is_done: _task_control_panel("_text_runner") + elif not _artifact_exists("terminology"): + if st.button( + t("Generate terminology for review"), + key="generate_terms_button", + ): + runner.start(_get_text_generation_steps()) + st.rerun() + elif not _stage_confirmed("terminology"): + st.info(t("Review and confirm terminology before translation.")) + elif not _artifact_exists("translation"): + if st.button( + t("Continue to translation"), + key="continue_translation_button", + ): + runner.start(_get_translation_steps()) + st.rerun() + elif not _stage_confirmed("translation"): + st.info(t("Review and confirm translation before subtitle splitting.")) + elif not _artifact_exists("subtitles"): + if st.button( + t("Generate subtitle split for review"), + key="continue_subtitle_split_button", + ): + runner.start(_get_subtitle_split_steps()) + st.rerun() + elif not _stage_confirmed("subtitles"): + st.info( + t("Review and confirm subtitles before final subtitle generation.") + ) else: if st.button( - t("Start Processing Subtitles"), key="text_processing_button" + t("Generate final subtitles and video"), + key="continue_subtitle_finalize_button", ): - steps = _get_text_steps() - runner.start(steps) + runner.start(_get_subtitle_finalize_steps()) st.rerun() else: if load_key("burn_subtitles"): - st.video(SUB_VIDEO) + st.video(sub_video) download_subtitle_zip_button(text=t("Download All Srt Files")) if st.button(t("Archive to 'history'"), key="cleanup_in_text_processing"): @@ -168,27 +219,24 @@ def text_processing_section(): # ─── Audio processing ─── -def _get_audio_steps(): - """Return the audio/dubbing processing steps as (label, callable) list.""" - steps = [ - ( - t("Generate audio tasks and chunks"), - lambda: ( - _8_1_audio_task.gen_audio_task_main(), - _8_2_dub_chunks.gen_dub_chunks(), - ), - ), +def _get_tts_task_steps(): + return [(t("Generate audio tasks"), _8_1_audio_task.gen_audio_task_main)] + + +def _get_audio_finalize_steps(): + return [ + (t("Generate audio chunks"), _8_2_dub_chunks.gen_dub_chunks), (t("Extract reference audio"), _9_refer_audio.extract_refer_audio_main), (t("Generate and merge audio files"), _10_gen_audio.gen_audio), (t("Merge full audio"), _11_merge_audio.merge_full_audio), (t("Merge final audio into video"), _12_dub_to_vid.merge_video_audio), ] - return steps def audio_processing_section(): st.header(t("c. Dubbing")) runner = TaskRunner.get(st.session_state, "_audio_runner") + dub_video = _dub_video_path() with st.container(border=True): st.markdown( @@ -197,24 +245,33 @@ def audio_processing_section(): {t("This stage includes the following steps:")}

1. {t("Generate audio tasks and chunks")}
- 2. {t("Extract reference audio")}
- 3. {t("Generate and merge audio files")}
- 4. {t("Merge final audio into video")} + 2. {t("Review and confirm TTS text")}
+ 3. {t("Extract reference audio")}
+ 4. {t("Generate and merge audio files")}
+ 5. {t("Merge final audio into video")} """, unsafe_allow_html=True, ) - if not os.path.exists(DUB_VIDEO): + if not os.path.exists(dub_video): if runner.is_active: _task_control_panel("_audio_runner") elif runner.is_done: _task_control_panel("_audio_runner") + elif not _artifact_exists("tts_text"): + if st.button( + t("Generate TTS text for review"), + key="generate_tts_text_button", + ): + runner.start(_get_tts_task_steps()) + st.rerun() + elif not _stage_confirmed("tts_text"): + st.info(t("Review and confirm TTS text before dubbing.")) else: if st.button( - t("Start Audio Processing"), key="audio_processing_button" + t("Generate dubbing"), key="continue_audio_finalize_button" ): - steps = _get_audio_steps() - runner.start(steps) + runner.start(_get_audio_finalize_steps()) st.rerun() else: st.success( @@ -223,7 +280,7 @@ def audio_processing_section(): ) ) if load_key("burn_subtitles"): - st.video(DUB_VIDEO) + st.video(dub_video) if st.button(t("Delete dubbing files"), key="delete_dubbing_files"): delete_dubbing_files() st.rerun() @@ -235,6 +292,42 @@ def audio_processing_section(): # ─── Main ─── +def workspace_section(): + st.header(t("Task Workspace")) + + if st.session_state.get("_active_workspace"): + try: + workspace.set_active_workspace(st.session_state["_active_workspace"]) + except FileNotFoundError: + st.session_state.pop("_active_workspace", None) + workspace.clear_active_workspace() + + jobs = workspace.list_jobs() + labels = [f"{job['name']} · {job['status']} · {job['id']}" for job in jobs] + selected = st.selectbox(t("Continue task"), [""] + labels) + if selected: + job = jobs[labels.index(selected)] + workspace.set_active_workspace(job["path"]) + st.session_state["_active_workspace"] = job["path"] + + new_name = st.text_input(t("New task name"), value="") + if st.button(t("New task"), key="new_workspace_task", width="stretch"): + job = workspace.create_job(name=new_name or "VideoLingo Task") + workspace.set_active_workspace(job["path"]) + st.session_state["_active_workspace"] = job["path"] + st.rerun() + + active = workspace.get_active_workspace() + if active: + job = workspace.load_job(active) + st.caption(f"{job['name']} · {job['status']}") + if st.button(t("Archive task"), key="archive_workspace_task", width="stretch"): + workspace.archive_job(active) + workspace.clear_active_workspace() + st.session_state.pop("_active_workspace", None) + st.rerun() + + def main(): logo_col, _ = st.columns([1, 1]) with logo_col: @@ -249,9 +342,11 @@ def main(): ) # add settings with st.sidebar: + workspace_section() page_setting() st.markdown(give_star_button, unsafe_allow_html=True) download_video_section() + review_section() text_processing_section() audio_processing_section() diff --git a/tests/test_review.py b/tests/test_review.py new file mode 100644 index 00000000..5c252475 --- /dev/null +++ b/tests/test_review.py @@ -0,0 +1,206 @@ +import tempfile +import unittest +from pathlib import Path + +from core import review, workspace + + +class ReviewTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name) + self.config_path = self.root / "config.yaml" + self.config_path.write_text("display_language: en\n", encoding="utf-8") + self.job = workspace.create_job( + name="Review Video", + workspace_root=self.root / "jobs", + config_path=self.config_path, + ) + workspace.set_active_workspace(self.job["path"]) + + def tearDown(self): + workspace.clear_active_workspace() + + def test_default_review_status_is_created_in_active_workspace(self): + status = review.load_status() + + status_path = ( + Path(self.job["path"]) + / "artifacts" + / "reviews" + / "review_status.yaml" + ) + self.assertTrue(status_path.is_file()) + self.assertEqual(status["stages"]["terminology"]["status"], "pending") + self.assertEqual( + status["stages"]["translation"]["artifact"], + "output/log/translation_results.xlsx", + ) + + def test_mark_confirmed_records_timestamp(self): + review.mark_confirmed("terminology") + status = review.load_status() + + self.assertEqual(status["stages"]["terminology"]["status"], "confirmed") + self.assertTrue(status["stages"]["terminology"]["confirmed_at"]) + + def test_save_terminology_preserves_theme_and_invalidates_downstream(self): + import json + + path = Path(self.job["path"]) / "output" / "log" + path.mkdir(parents=True, exist_ok=True) + terminology = path / "terminology.json" + terminology.write_text( + json.dumps( + { + "theme": "Physics lecture", + "terms": [{"src": "force", "tgt": "力", "note": "physics"}], + }, + ensure_ascii=False, + ), + encoding="utf-8", + ) + review.mark_confirmed("translation") + + review.save_terminology( + [{"src": "energy", "tgt": "能量", "note": "physics"}], + theme="Physics lecture", + ) + + saved = json.loads(terminology.read_text(encoding="utf-8")) + self.assertEqual(saved["theme"], "Physics lecture") + self.assertEqual(saved["terms"][0]["src"], "energy") + status = review.load_status() + self.assertEqual(status["stages"]["translation"]["status"], "stale") + backups = list( + ( + Path(self.job["path"]) + / "artifacts" + / "reviews" + / "backups" + ).glob("terminology-*.json") + ) + self.assertEqual(len(backups), 1) + + def test_save_table_creates_backup_and_validates_required_columns(self): + import pandas as pd + + path = Path(self.job["path"]) / "output" / "log" + path.mkdir(parents=True, exist_ok=True) + artifact = path / "translation_results.xlsx" + pd.DataFrame({"Source": ["hello"], "Translation": ["你好"]}).to_excel( + artifact, index=False + ) + + review.save_table( + "translation", + pd.DataFrame({"Source": ["hello"], "Translation": ["您好"]}), + required_columns=("Source", "Translation"), + ) + + saved = pd.read_excel(artifact) + self.assertEqual(saved.at[0, "Translation"], "您好") + backups = list( + ( + Path(self.job["path"]) + / "artifacts" + / "reviews" + / "backups" + ).glob("translation-*.xlsx") + ) + self.assertEqual(len(backups), 1) + + def test_save_table_rejects_missing_required_columns(self): + import pandas as pd + + with self.assertRaises(ValueError): + review.save_table( + "translation", + pd.DataFrame({"Source": ["hello"]}), + required_columns=("Source", "Translation"), + ) + + def test_save_subtitles_updates_remerged_when_row_counts_match(self): + import pandas as pd + + log_dir = Path(self.job["path"]) / "output" / "log" + log_dir.mkdir(parents=True, exist_ok=True) + pd.DataFrame({"Source": ["A"], "Translation": ["甲"]}).to_excel( + log_dir / "translation_results_for_subtitles.xlsx", + index=False, + ) + pd.DataFrame({"Source": ["A"], "Translation": ["甲"]}).to_excel( + log_dir / "translation_results_remerged.xlsx", + index=False, + ) + + review.save_subtitles(pd.DataFrame({"Source": ["A"], "Translation": ["乙"]})) + + display = pd.read_excel(log_dir / "translation_results_for_subtitles.xlsx") + remerged = pd.read_excel(log_dir / "translation_results_remerged.xlsx") + self.assertEqual(display.at[0, "Translation"], "乙") + self.assertEqual(remerged.at[0, "Translation"], "乙") + self.assertEqual(review.load_status()["stages"]["tts_text"]["status"], "stale") + + def test_save_subtitles_leaves_remerged_when_row_counts_differ(self): + import pandas as pd + + log_dir = Path(self.job["path"]) / "output" / "log" + log_dir.mkdir(parents=True, exist_ok=True) + pd.DataFrame({"Source": ["A"], "Translation": ["甲"]}).to_excel( + log_dir / "translation_results_for_subtitles.xlsx", + index=False, + ) + pd.DataFrame({"Source": ["A", "B"], "Translation": ["甲", "乙"]}).to_excel( + log_dir / "translation_results_remerged.xlsx", + index=False, + ) + + review.save_subtitles(pd.DataFrame({"Source": ["A"], "Translation": ["丙"]})) + + remerged = pd.read_excel(log_dir / "translation_results_remerged.xlsx") + self.assertEqual(len(remerged), 2) + self.assertEqual(remerged.at[0, "Translation"], "甲") + + def test_save_tts_text_preserves_timing_columns(self): + import pandas as pd + + audio_dir = Path(self.job["path"]) / "output" / "audio" + audio_dir.mkdir(parents=True, exist_ok=True) + pd.DataFrame( + { + "number": [1], + "start_time": ["00:00:00.000"], + "end_time": ["00:00:02.000"], + "duration": [2.0], + "text": ["old"], + "origin": ["source"], + } + ).to_excel(audio_dir / "tts_tasks.xlsx", index=False) + + review.save_table( + "tts_text", + pd.DataFrame( + { + "number": [1], + "start_time": ["00:00:00.000"], + "end_time": ["00:00:02.000"], + "duration": [2.0], + "text": ["new"], + "origin": ["source"], + } + ), + required_columns=( + "number", + "start_time", + "end_time", + "duration", + "text", + "origin", + ), + ) + + saved = pd.read_excel(audio_dir / "tts_tasks.xlsx") + self.assertEqual(saved.at[0, "text"], "new") + self.assertEqual(saved.at[0, "duration"], 2.0) diff --git a/tests/test_workspace.py b/tests/test_workspace.py new file mode 100644 index 00000000..60ebcba0 --- /dev/null +++ b/tests/test_workspace.py @@ -0,0 +1,163 @@ +import os +import tempfile +import unittest +from pathlib import Path + +from ruamel.yaml import YAML + +from core import workspace + + +class WorkspaceTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name) + self.config_path = self.root / "config.yaml" + self.config_path.write_text( + "display_language: en\n" + "api:\n" + " key: ''\n" + "allowed_video_formats:\n" + " - mp4\n", + encoding="utf-8", + ) + workspace.clear_active_workspace() + + def tearDown(self): + workspace.clear_active_workspace() + + def test_create_job_writes_metadata_and_config_snapshot(self): + job = workspace.create_job( + name="My Video!", + source_type="upload", + source_path="input.mp4", + workspace_root=self.root / "jobs", + config_path=self.config_path, + ) + + self.assertTrue(Path(job["path"]).is_dir()) + self.assertTrue((Path(job["path"]) / "output").is_dir()) + self.assertTrue((Path(job["path"]) / "job.yaml").is_file()) + self.assertTrue((Path(job["path"]) / "config.snapshot.yaml").is_file()) + self.assertEqual(job["name"], "My Video!") + self.assertEqual(job["source"]["type"], "upload") + + def test_active_workspace_resolves_output_path(self): + job = workspace.create_job( + name="Video", + workspace_root=self.root / "jobs", + config_path=self.config_path, + ) + workspace.set_active_workspace(job["path"]) + + resolved = workspace.output_path("log", "file.txt") + + self.assertEqual( + os.fspath(resolved), + str(Path(job["path"]) / "output" / "log" / "file.txt"), + ) + + def test_legacy_output_path_without_active_workspace(self): + workspace.clear_active_workspace() + + self.assertEqual( + os.fspath(workspace.output_path("log", "file.txt")), + "output/log/file.txt", + ) + + def test_archive_job_marks_status_without_deleting_files(self): + job = workspace.create_job( + name="Video", + workspace_root=self.root / "jobs", + config_path=self.config_path, + ) + + workspace.archive_job(job["path"]) + loaded = workspace.load_job(job["path"]) + + self.assertEqual(loaded["status"], "archived") + self.assertTrue(Path(job["path"]).exists()) + + def test_config_snapshot_is_independent_from_global_config(self): + job = workspace.create_job( + name="Video", + workspace_root=self.root / "jobs", + config_path=self.config_path, + ) + self.config_path.write_text( + "display_language: zh-CN\napi:\n key: changed\n", + encoding="utf-8", + ) + + yaml = YAML() + snapshot = yaml.load( + (Path(job["path"]) / "config.snapshot.yaml").read_text(encoding="utf-8") + ) + + self.assertEqual(snapshot["display_language"], "en") + self.assertEqual(snapshot["api"]["key"], "") + + def test_model_constants_resolve_against_active_workspace(self): + from core.utils import models + + job = workspace.create_job( + name="Video", + workspace_root=self.root / "jobs", + config_path=self.config_path, + ) + workspace.set_active_workspace(job["path"]) + + self.assertEqual( + os.fspath(models._2_CLEANED_CHUNKS), + str(Path(job["path"]) / "output" / "log" / "cleaned_chunks.xlsx"), + ) + self.assertEqual( + str(models._AUDIO_TMP_DIR), + str(Path(job["path"]) / "output" / "audio" / "tmp"), + ) + + def test_find_video_files_uses_active_workspace_output(self): + from core._1_ytdlp import find_video_files + + job = workspace.create_job( + name="Video", + workspace_root=self.root / "jobs", + config_path=self.config_path, + ) + output_dir = Path(job["path"]) / "output" + (output_dir / "sample.mp4").write_bytes(b"video") + workspace.set_active_workspace(job["path"]) + + self.assertEqual(find_video_files(), str(output_dir / "sample.mp4")) + + def test_config_utils_updates_active_workspace_snapshot(self): + from core.utils import config_utils + + job = workspace.create_job( + name="Video", + workspace_root=self.root / "jobs", + config_path=self.config_path, + ) + workspace.set_active_workspace(job["path"]) + + config_utils.update_key("display_language", "ja") + + yaml = YAML() + global_config = yaml.load(self.config_path.read_text(encoding="utf-8")) + snapshot = yaml.load( + (Path(job["path"]) / "config.snapshot.yaml").read_text(encoding="utf-8") + ) + self.assertEqual(global_config["display_language"], "en") + self.assertEqual(snapshot["display_language"], "ja") + self.assertEqual(config_utils.load_key("display_language"), "ja") + + def test_batch_workspace_path_can_be_stored_per_row(self): + import pandas as pd + from batch.utils.batch_processor import ensure_workspace_columns + + df = pd.DataFrame({"Video File": ["a.mp4"], "Status": [None]}) + updated = ensure_workspace_columns(df) + + self.assertIn("Workspace", updated.columns) + self.assertEqual(updated.at[0, "Workspace"], "")