Fix pipeline consistency and smoke coverage

.github/workflows/ci.yml

@@ -20,19 +20,15 @@ jobs:
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
-          pip install pillow numpy
+          pip install -r requirements.txt
 
       - name: Python syntax check
         run: |
-          python -m py_compile core/segment_tts.py
-          python -m py_compile core/timeline_mapper.py
-          python -m py_compile core/concat_engine.py
-          python -m py_compile core/cta_resource.py
-          python -m py_compile core/frame_extractor.py
-          python -m py_compile harness/self_report.py
-          python -m py_compile harness/harness.py
-          python -m py_compile extensions/storyboard/storyboard_ai.py
-          python -m py_compile extensions/animations/animation_templates.py
+          python -m py_compile $(git ls-files '*.py')
+
+      - name: Import and routing smoke check
+        run: |
+          python scripts/smoke_imports.py
 
       - name: JSON validation
         run: |

core/concat_engine.py

@@ -730,7 +730,7 @@ def concat(
 
         # 验证输出
         output_duration = self._probe_duration(output_video)
-        expected_duration = sum(e["duration"] for e in self.timeline)
+        expected_duration = self.timeline[-1].get("end_time", sum(e["duration"] for e in self.timeline))
         print(f"[ConcatEngine] 输出完成: {output_video}")
         print(f"[ConcatEngine] 预期时长: {expected_duration:.2f}s, 实际时长: {output_duration:.2f}s, 差值: {abs(output_duration - expected_duration):.3f}s")
 

core/frame_extractor.py

@@ -108,7 +108,10 @@ def _check_emoji_blocks(self, img: Image.Image) -> Tuple[bool, str]:
         # 转换为二值，检测大的连通区域
         gray = np.array(img.convert("L"))
         # 边缘检测：如果图像有大量文字，边缘会很多；如果大面积是方框，边缘集中在方框边界
-        from scipy import ndimage
+        try:
+            from scipy import ndimage
+        except ImportError:
+            return True, "未安装 scipy，跳过 emoji 方块启发式检测"
         edges = ndimage.sobel(gray)
         edge_density = (edges > 20).sum() / edges.size
 
@@ -183,8 +186,10 @@ def extract_and_check(
             if self.config.frames_per_segment == 1:
                 timestamps = [start_time + duration / 2]
             else:
+                # Sample inside the segment instead of exactly on boundaries. End
+                # timestamps can land on EOF after concat/codec rounding.
                 timestamps = [
-                    start_time + duration * i / (self.config.frames_per_segment - 1)
+                    start_time + duration * (i + 1) / (self.config.frames_per_segment + 1)
                     for i in range(self.config.frames_per_segment)
                 ]
 

core/segment_tts.py

@@ -23,7 +23,7 @@
 from dataclasses import dataclass, field, asdict
 from enum import Enum
 from pathlib import Path
-from typing import List, Optional, Dict
+from typing import List, Optional, Dict, Tuple
 
 import edge_tts
 

core/timeline_mapper.py

@@ -135,7 +135,7 @@ def _resolve_media_path(self, segment_id: str, prompt_entry: dict, scenes: Dict[
         if segment_id in scenes:
             p = scenes[segment_id]
             media_type = "video" if p.suffix.lower() in (".mp4", ".mov") else "image"
-            return str(p.relative_to(Path.cwd())), media_type
+            return str(p.resolve().relative_to(Path.cwd())), media_type
 
         explicit_path = prompt_entry.get("media_path", prompt_entry.get("fallback_path", ""))
         if explicit_path and Path(explicit_path).exists():
@@ -193,7 +193,10 @@ def build_timeline(self, transitions: Optional[List[dict]] = None) -> List[Timel
                 notes=prompt_entry.get("notes", prompt_entry.get("description", "")),
             )
             timeline.append(entry)
-            current_time += duration
+            if trans:
+                current_time += duration - trans.get("duration", 0.0)
+            else:
+                current_time += duration
 
         self.timeline = timeline
         return timeline
@@ -203,7 +206,7 @@ def save_timeline(self) -> Path:
         data = {
             "generator": "md2video.timeline_mapper",
             "version": "1.1.0",
-            "total_duration": sum(e.duration for e in self.timeline),
+            "total_duration": self.timeline[-1].end_time if self.timeline else 0.0,
             "segment_count": len(self.timeline),
             "has_effects": any(
                 e.fade_in > 0 or e.fade_out > 0 or e.transition is not None

docs/tasks/2026-06-08-fix-pipeline-consistency.md

@@ -2,12 +2,13 @@
 
 ## Metadata
 - Task ID: `TC-2026-06-08-pipeline-consistency`
-- Status: `open`
+- Status: `implemented`
 - Created: `2026-06-08`
 - Owner: `unassigned`
 - Repo: `md2video`
 - Primary layer: `code`
 - Secondary layers: `docs`, `ci`, `runtime-validation`
+- Implemented: `2026-06-08`
 
 ## Objective
 Make the repository's documented article-to-video pipeline runnable enough for local development by fixing import/runtime blockers, aligning examples with current APIs, removing animation-router ambiguity, and tightening CI so these regressions are caught automatically.
@@ -98,3 +99,33 @@ PY
 
 ## Handoff Notes
 The smallest safe repair is to fix `core.segment_tts`, align `examples/example_pipeline.py`, and add import smoke coverage to CI. Animation routing is the main design choice: prefer one canonical router and leave the older import path as a compatibility layer if practical.
+
+## Implementation Notes
+- Added the missing `Tuple` import so `core.segment_tts` imports in a dependency-complete environment.
+- Updated the example TTS callback to accept `segment_type`.
+- Routed the example pipeline to the broader animation renderer and added safe default animation variables for rule-generated prompts.
+- Kept the legacy `extensions.animation_templates.base.render_animation` entrypoint as a compatibility wrapper for newer animation types.
+- Added `scripts/smoke_imports.py` and wired CI to install `requirements.txt` before import/routing smoke checks.
+- Fixed `TimelineMapper` relative media path resolution for relative `scenes_dir` paths.
+- Aligned `TimelineMapper` saved start/end times with transition overlap semantics used by `ConcatEngine`.
+- Made `FrameExtractor` degrade gracefully when optional `scipy` is unavailable.
+- Changed frame sampling to avoid exact segment boundaries and EOF-sensitive frames.
+- Made narration verification tolerate small codec delay and phase inversion by using max absolute Pearson correlation over a small lag window.
+- Corrected `ConcatEngine` expected-duration logging to use timeline end time when transitions overlap clips.
+
+## Validation Results
+- `python -m py_compile` passed for tracked Python files plus the new smoke script.
+- `scripts/smoke_imports.py` passed: core imports and rule-driven animation routing validated.
+- Offline E2E passed in a temporary working directory with 5 generated segments and no transitions:
+  - article -> storyboard inputs
+  - generated local sine narration MP3 files
+  - generated local ffmpeg test videos
+  - `TimelineMapper` triple consistency
+  - `ConcatEngine` final MP4
+  - `FrameExtractor` report
+  - `VideoComplianceHarness` with no L1 failures
+  - `scripts/verify_narration.py` passed 5/5 segment correlations
+- Offline E2E with storyboard-generated transitions passed in a temporary working directory with 6 generated segments:
+  - `ConcatEngine` exercised the filter_complex effect path
+  - `VideoComplianceHarness` reported no L1 failures
+  - `scripts/verify_narration.py` passed 6/6 segment correlations

examples/example_pipeline.py

@@ -32,10 +32,33 @@
 from core.cta_resource import generate_qr_cta, CTAResourceManager
 from harness.harness import VideoComplianceHarness
 from extensions.storyboard.storyboard_ai import storyboard_from_article
-from extensions.animation_templates.base import render_animation
+from extensions.animations.animation_templates import render_animation
 from extensions.prompt_templates.base import PromptTemplateLibrary
 
 
+def _animation_vars_with_defaults(animation_type: str, vars_dict: dict, text: str = "") -> dict:
+    """Fill minimal safe vars for rule-generated animation prompts."""
+    vars_dict = dict(vars_dict or {})
+    text = text.strip()
+
+    defaults = {
+        "animated_text": {"text": text or "核心信息"},
+        "bar_chart": {"data": [("Before", 1), ("After", 3)]},
+        "pie_chart": {"data": [("A", 40), ("B", 60)]},
+        "trend_line": {"points": [1, 2, 3, 5, 8]},
+        "comparison_split": {"left": "Before", "right": "After"},
+        "table_scroll": {"headers": ["Item", "Value"], "rows": [["A", "1"], ["B", "3"]]},
+        "bullet_list": {"items": [text] if text else ["要点一", "要点二"], "title": "核心要点"},
+        "calendar_highlight": {"year": 2026, "month": 6, "highlight_day": 8},
+        "quote_card": {"quote": text or "核心观点", "author": ""},
+    }
+
+    for key, value in defaults.get(animation_type, {}).items():
+        if not vars_dict.get(key):
+            vars_dict[key] = value
+    return vars_dict
+
+
 def step1_storyboard(article_path: str = "examples/example_article.md"):
     """步骤1：文章 → 分镜"""
     print("=" * 60)
@@ -63,7 +86,7 @@ def step2_generate_tts():
 
     gen = SegmentedTTSGenerator(output_dir="output")
     gen.split_by_semantic("", scene_hints=hints)
-    asyncio.run(gen.generate_all(progress_callback=lambda sid, dur: print(f"  {sid}: {dur:.2f}s")))
+    asyncio.run(gen.generate_all(progress_callback=lambda sid, dur, seg_type: print(f"  {sid} [{seg_type}]: {dur:.2f}s")))
     manifest = gen.save_manifest()
     print(f"✅ TTS 已保存: {manifest}")
     return manifest
@@ -83,6 +106,18 @@ def step3_generate_scenes(budget_limit: int = 500):
 
     with open("prompts.json", "r", encoding="utf-8") as f:
         prompts = json.load(f)
+    segment_meta = {}
+    segments_path = Path("output/segments.json")
+    if segments_path.exists():
+        with open(segments_path, "r", encoding="utf-8") as f:
+            segments_data = json.load(f)
+        segment_meta = {
+            s["id"]: {
+                "duration": float(s.get("duration", 5.0) or 5.0),
+                "text": s.get("text", ""),
+            }
+            for s in segments_data.get("segments", [])
+        }
 
     lib = PromptTemplateLibrary(budget_limit=budget_limit)
 
@@ -93,7 +128,10 @@ def step3_generate_scenes(budget_limit: int = 500):
             output_path = f"rebuild_animations/{p['id']}.mp4"
             print(f"  生成动画: {p['id']} -> {output_path}")
             try:
-                render_animation(anim_type, p.get("vars", {}), output_path)
+                meta = segment_meta.get(p["id"], {})
+                duration = meta.get("duration", p.get("duration", 5.0))
+                vars_dict = _animation_vars_with_defaults(anim_type, p.get("vars", {}), meta.get("text", ""))
+                render_animation(anim_type, vars_dict, duration=duration, output_path=output_path)
             except Exception as e:
                 print(f"  ⚠️ 动画生成失败: {e}")
         else:

extensions/animation_templates/base.py

@@ -277,7 +277,15 @@ def render_animation(template_id: str, params_dict: dict, output_path: str, conf
         config: 可选的全局配置
     """
     if template_id not in ANIMATION_REGISTRY:
-        raise ValueError(f"未知模板: {template_id}。可用: {list(ANIMATION_REGISTRY.keys())}")
+        from extensions.animations.animation_templates import render_animation as render_extended_animation
+
+        duration = params_dict.get("duration", config.duration if config else 5.0)
+        return render_extended_animation(
+            template_id,
+            params_dict,
+            duration=duration,
+            output_path=output_path,
+        )
 
     template_cls, params_cls = ANIMATION_REGISTRY[template_id]
     params = params_cls(**params_dict)

extensions/animations/animation_templates.py

@@ -40,6 +40,24 @@
 import numpy as np
 
 
+SUPPORTED_ANIMATION_TYPES = {
+    "animated_text",
+    "bar_chart",
+    "pie_chart",
+    "trend_line",
+    "comparison_split",
+    "table_scroll",
+    "bullet_list",
+    "calendar_highlight",
+    "quote_card",
+}
+
+
+def available_animation_types() -> List[str]:
+    """Return animation_type values accepted by AnimationRenderer.render_by_type."""
+    return sorted(SUPPORTED_ANIMATION_TYPES)
+
+
 @dataclass
 class AnimationTemplate:
     """模板定义"""
@@ -484,9 +502,9 @@ def render_calendar_highlight(
         start_y = 350
 
         # 动画参数
-        grid_appear_frame = int(total_frames * 0.3)
+        grid_appear_frame = max(1, int(total_frames * 0.3))
         highlight_start = int(total_frames * 0.5)
-        highlight_end = int(total_frames * 0.85)
+        highlight_end = max(highlight_start + 1, int(total_frames * 0.85))
 
         for frame_idx in range(total_frames):
             img = self._create_base_frame(bg_color)
@@ -601,8 +619,8 @@ def wrap_text(text: str, max_width: int, font) -> List[str]:
         card_w = self.width - 120
 
         # 动画参数：文字尽快出现，不让观众等
-        fade_in_end = int(total_frames * 0.03)  # 3%时间完成卡片渐入
-        quote_end = int(total_frames * 0.20)    # 20%时间完成文字显示
+        fade_in_end = max(1, int(total_frames * 0.03))  # 3%时间完成卡片渐入
+        quote_end = max(fade_in_end + 1, int(total_frames * 0.20))    # 20%时间完成文字显示
         hold_end = total_frames
 
         for frame_idx in range(total_frames):
@@ -730,7 +748,7 @@ def render_by_type(
         }
 
         if animation_type not in renderers:
-            raise ValueError(f"Unknown animation_type: {animation_type}. Available: {list(renderers.keys())}")
+            raise ValueError(f"Unknown animation_type: {animation_type}. Available: {available_animation_types()}")
 
         return renderers[animation_type]()
 

scripts/smoke_imports.py

@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+"""Import and routing smoke checks for md2video CI."""
+
+import importlib
+import json
+import sys
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO_ROOT))
+
+MODULES = [
+    "core.segment_tts",
+    "core.timeline_mapper",
+    "core.concat_engine",
+    "core.frame_extractor",
+    "core.cta_resource",
+    "extensions.storyboard.storyboard_ai",
+    "extensions.animations.animation_templates",
+    "extensions.animation_templates.base",
+    "harness.harness",
+    "harness.memory_loader",
+    "harness.self_report",
+]
+
+
+def check_imports() -> None:
+    for module in MODULES:
+        importlib.import_module(module)
+        print(f"IMPORT_OK {module}")
+
+
+def check_animation_routing() -> None:
+    rules_path = Path("rules/storyboard_rules.json")
+    with open(rules_path, "r", encoding="utf-8") as f:
+        rules = json.load(f)
+
+    from extensions.animations.animation_templates import available_animation_types
+
+    supported = set(available_animation_types())
+    missing = []
+    for segment_type, mapping in rules.get("segment_type_mapping", {}).items():
+        if mapping.get("visual_type") != "animation":
+            continue
+        animation_type = mapping.get("animation_type")
+        if animation_type not in supported:
+            missing.append(f"{segment_type}:{animation_type}")
+
+    if missing:
+        raise SystemExit(
+            "Unsupported animation_type values in rules/storyboard_rules.json: "
+            + ", ".join(missing)
+        )
+
+    print("ANIMATION_ROUTING_OK")
+
+
+def main() -> None:
+    check_imports()
+    check_animation_routing()
+
+
+if __name__ == "__main__":
+    main()