diff --git a/kb_pipeline/pipeline.py b/kb_pipeline/pipeline.py index 8088662..016a780 100644 --- a/kb_pipeline/pipeline.py +++ b/kb_pipeline/pipeline.py @@ -153,7 +153,25 @@ class Pipeline: # Check if step should be skipped if self.config.skip_completed and state.is_step_completed(step_name): - logger.info(f"Skipping completed step: {step_name}") + # Ask user whether to re-run the completed step + should_rerun = self._ask_rerun_step(step_name, content_dir) + + if should_rerun: + logger.info(f"Re-running completed step: {step_name}") + # Clear the step's state record so it can be re-executed + if step_name in state.steps: + del state.steps[step_name] + state_manager.save(state) + else: + logger.info(f"Skipping completed step: {step_name}") + # Scan existing files and populate results for skipped steps + self._populate_skipped_step_results(step_name, results, content_dir, parsed) + continue + + # Skip ASR for text-based content types + text_content_types = {"article", "answer", "zhuanlan", "question"} + if step_name == "asr" and parsed.content_type in text_content_types: + logger.info(f"Skipping ASR for text content type: {parsed.content_type}") continue processor = self.processors[step_name] @@ -180,6 +198,7 @@ class Pipeline: platform=parsed.platform, content_dir=str(content_dir), content_type=parsed.content_type, + source_url=url, ) elif step_name == "asr": # ASR needs video path from download step @@ -266,3 +285,102 @@ class Pipeline: results=results, content_dir=content_dir, ) + + def _ask_rerun_step(self, step_name: str, content_dir: Path) -> bool: + """Ask user whether to re-run a completed step. + + Args: + step_name: Name of the step to potentially re-run + content_dir: Content directory path + + Returns: + True if user wants to re-run, False to skip + """ + try: + from rich.console import Console + from rich.prompt import Confirm + + console = Console() + return Confirm.ask( + f"Step '{step_name}' is already completed. Re-run it?", + default=False + ) + except ImportError: + # Fallback to basic input if rich is not available + response = input(f"Step '{step_name}' is already completed. Re-run it? [y/N] ") + return response.lower() in ('y', 'yes') + + def _populate_skipped_step_results( + self, + step_name: str, + results: Dict, + content_dir: Path, + parsed: ParsedLink + ) -> None: + """Populate results for skipped steps by scanning existing files. + + This ensures downstream steps can find the outputs they need, + even when a step is skipped. + + Args: + step_name: Name of the skipped step + results: Results dictionary to populate + content_dir: Content directory path + parsed: Parsed link with platform and content_id + """ + if step_name == "asr": + # Scan for subtitle files generated by ASR + # Try both new naming convention (with _subtitle suffix) and old convention (without suffix) + base_name_new = f"{parsed.platform}_{parsed.content_id}_subtitle" + base_name_old = f"{parsed.platform}_{parsed.content_id}" + + json_path = content_dir / f"{base_name_new}.json" + if not json_path.exists(): + json_path = content_dir / f"{base_name_old}.json" + + srt_path = content_dir / f"{base_name_new}.srt" + if not srt_path.exists(): + srt_path = content_dir / f"{base_name_old}.srt" + + output_files = {} + if json_path.exists(): + output_files["json"] = str(json_path) + if srt_path.exists(): + output_files["srt"] = str(srt_path) + + if output_files: + results["asr"] = { + "success": True, + "error_message": None, + "output_files": output_files, + } + logger.info(f"Found existing ASR outputs: {output_files}") + else: + logger.warning(f"No existing ASR outputs found in {content_dir}") + results["asr"] = { + "success": False, + "error_message": "No existing subtitle files found", + "output_files": {}, + } + + elif step_name == "download": + # Scan for metadata file + metadata_path = content_dir / "metadata.json" + if metadata_path.exists(): + results["download"] = { + "success": True, + "error_message": None, + "metadata_file": str(metadata_path), + } + logger.info(f"Found existing download metadata: {metadata_path}") + + elif step_name == "analyze": + # Scan for analysis output + analysis_path = content_dir / "analysis.md" + if analysis_path.exists(): + results["analyze"] = { + "success": True, + "error_message": None, + "analysis_file": str(analysis_path), + } + logger.info(f"Found existing analysis: {analysis_path}")