freedak/OpenMontage
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implementation spec: governance, decision intelligence, theme system, and E2E bug fixes

Browse Source 
Implements the 2026-04-02 transformation spec (Phases 1-8) and fixes all
critical bugs found during 5-pipeline E2E testing.

Governance & Decision Intelligence:
- Pipeline-specific stage order in checkpoint (replaces global STAGES list)
- Provider scoring engine (lib/scoring.py) with 7-dimension weighted ranking
- Decision log artifact enforced at proposal/idea stage across all 10 pipelines
- Delivery promise classifier prevents silent motion-to-still downgrades
- Structured shot language in scene_plan schema (camera, lens, lighting, DOF)
- Variation checker and slideshow risk scorer block samey output before render
- Creative intake, capability extension, and creative-intake meta skills
- Final self-review artifact with 5 mandatory checks before presenting output
- Source media review contract for user-supplied footage

Render & Theme System:
- Remotion AnimatedBackground now derives colors from playbook (no more hardcoded
  dark blue fintech gradient on every video)
- video_compose builds custom ThemeConfig from playbook YAML colors/fonts —
  custom playbooks flow through to Remotion automatically
- Explainer component wires theme to all child components (charts, cards, etc.)
- resolveAsset() handles absolute paths on Windows/Unix via file:// URIs
- RENDERER_FAMILY_MAP synced with actual Remotion compositions

Critical Bug Fixes:
- Windows npx subprocess: run_command() resolves .cmd wrappers via shutil.which()
- Silent renderer downgrade: Remotion failure now returns explicit error with
  options instead of silently falling back to FFmpeg
- .env inline comment parsing strips trailing # comments from API keys
- concat_path UnboundLocalError in video_compose finally block
- audio_mixer and showcase_card capture=True kwarg bug
- Selector estimate_cost() calls fixed (_select_tool -> _select_best_tool)
- asset_manifest schema expanded with provider, license, subtype fields
- screen-demo subtitle_gen moved from required to optional tools
- Duration drift detection in post-render final review (>25% warns)
This commit is contained in:
calesthio
2026-04-03 09:35:09 -07:00
parent a7e5f7498b
commit 2cd36fa8e0
83 changed files with 6076 additions and 282 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/checkpoint.py
+146 -16
View File
@@ -16,7 +16,16 @@ import jsonschema
from schemas.artifacts import ARTIFACT_NAMES, validate_artifact
STAGES = ["research", "proposal", "idea", "script", "scene_plan", "assets", "edit", "compose", "publish"]
# All known stages across all pipelines (used only for artifact name lookup).
ALL_KNOWN_STAGES = frozenset([
"research", "proposal", "idea", "script", "scene_plan",
"assets", "edit", "compose", "publish",
])
# Backward-compatible alias — existing code / tests that import STAGES still work.
# New code should use get_pipeline_stages(pipeline_type) instead.
STAGES = ["research", "proposal", "idea", "script", "scene_plan",
"assets", "edit", "compose", "publish"]
CANONICAL_STAGE_ARTIFACTS = {
"research": "research_brief",
@@ -30,6 +39,40 @@ CANONICAL_STAGE_ARTIFACTS = {
"publish": "publish_log",
}
# Additional artifacts that may be produced alongside canonical ones.
# These are not stage-defining but are required by governance contracts.
SUPPLEMENTARY_ARTIFACTS = {
"source_media_review", # Required before first planning stage when user media exists
"final_review", # Required by compose stage before presenting to user
}
def get_pipeline_stages(pipeline_type: str | None) -> list[str]:
"""Return the ordered stage list for a specific pipeline.
Falls back to STAGES (deterministic canonical order) when pipeline_type
is not provided or the manifest cannot be loaded.
Previous versions used a set intersection here, which produced
nondeterministic ordering. The fallback now uses a stable list.
"""
if pipeline_type is None:
# Deterministic canonical fallback — sorted to ensure stable ordering
import logging
logging.getLogger(__name__).warning(
"get_pipeline_stages called without pipeline_type — "
"using canonical fallback order. Pass pipeline_type for correctness."
)
return list(STAGES)
try:
from lib.pipeline_loader import load_pipeline, get_stage_order
manifest = load_pipeline(pipeline_type)
return get_stage_order(manifest)
except (FileNotFoundError, Exception):
# Graceful fallback: return all known stages in canonical order
return list(STAGES)
CHECKPOINT_SCHEMA_PATH = (
Path(__file__).resolve().parent.parent
/ "schemas"
@@ -76,13 +119,26 @@ def _validate_artifacts_for_stage(
def validate_checkpoint(checkpoint: dict[str, Any]) -> None:
"""Validate checkpoint structure and canonical artifact payloads."""
"""Validate checkpoint structure and canonical artifact payloads.
Uses pipeline_type (if present) to resolve the valid stage list.
Falls back to ALL_KNOWN_STAGES when pipeline_type is absent.
"""
stage = checkpoint.get("stage")
status = checkpoint.get("status")
artifacts = checkpoint.get("artifacts")
pipeline_type = checkpoint.get("pipeline_type")
if not isinstance(stage, str) or stage not in STAGES:
raise CheckpointValidationError(f"Invalid stage: {stage!r}")
valid_stages = (
set(get_pipeline_stages(pipeline_type)) if pipeline_type
else ALL_KNOWN_STAGES
)
if not isinstance(stage, str) or stage not in valid_stages:
raise CheckpointValidationError(
f"Invalid stage: {stage!r} for pipeline {pipeline_type!r}. "
f"Valid stages: {sorted(valid_stages)}"
)
if not isinstance(status, str):
raise CheckpointValidationError(f"Invalid status: {status!r}")
if not isinstance(artifacts, dict):
@@ -100,6 +156,40 @@ def _checkpoint_path(pipeline_dir: Path, project_id: str, stage: str) -> Path:
return pipeline_dir / project_id / f"checkpoint_{stage}.json"
def _decision_log_path(pipeline_dir: Path, project_id: str) -> Path:
return pipeline_dir / project_id / "decision_log.json"
def _merge_decision_log(
pipeline_dir: Path, project_id: str, new_log: dict[str, Any]
) -> None:
"""Append new decisions to the project-level decision log.
Each stage may produce decisions. This function merges them into a
single cumulative file so reviewers and the bench can inspect the
full audit trail.
"""
path = _decision_log_path(pipeline_dir, project_id)
if path.exists():
with open(path) as f:
existing = json.load(f)
else:
existing = {
"version": "1.0",
"project_id": project_id,
"decisions": [],
}
existing_ids = {d["decision_id"] for d in existing.get("decisions", [])}
for decision in new_log.get("decisions", []):
if decision.get("decision_id") not in existing_ids:
existing["decisions"].append(decision)
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "w") as f:
json.dump(existing, f, indent=2)
def write_checkpoint(
pipeline_dir: Path,
project_id: str,
@@ -118,12 +208,20 @@ def write_checkpoint(
metadata: Optional[dict] = None,
) -> Path:
"""Write a checkpoint file for a pipeline stage."""
if stage not in STAGES:
raise ValueError(f"Invalid stage: {stage!r}. Must be one of {STAGES}")
valid_stages = (
set(get_pipeline_stages(pipeline_type)) if pipeline_type
else ALL_KNOWN_STAGES
)
if stage not in valid_stages:
raise ValueError(
f"Invalid stage: {stage!r} for pipeline {pipeline_type!r}. "
f"Valid stages: {sorted(valid_stages)}"
)
checkpoint = {
"version": "1.0",
"project_id": project_id,
"pipeline_type": pipeline_type or "unknown",
"stage": stage,
"status": status,
"timestamp": datetime.now(timezone.utc).isoformat(),
@@ -132,9 +230,6 @@ def write_checkpoint(
"human_approved": human_approved,
"artifacts": artifacts,
}
if pipeline_type is not None:
checkpoint["pipeline_type"] = pipeline_type
if style_playbook is not None:
checkpoint["style_playbook"] = style_playbook
if review is not None:
@@ -146,6 +241,26 @@ def write_checkpoint(
if metadata is not None:
checkpoint["metadata"] = metadata
# Merge decision_log: if this checkpoint carries new decisions,
# append them to the project-level decision log file, then write the
# reference back into relevant artifacts so downstream consumers can find it.
if "decision_log" in artifacts and isinstance(artifacts["decision_log"], dict):
_merge_decision_log(pipeline_dir, project_id, artifacts["decision_log"])
log_ref = str(_decision_log_path(pipeline_dir, project_id))
# Write decision_log_ref into proposal_packet and render_report
# artifacts if they are present in this checkpoint.
for artifact_key in ("proposal_packet", "render_report"):
if artifact_key in artifacts and isinstance(artifacts[artifact_key], dict):
plan_or_top = artifacts[artifact_key]
# proposal_packet stores it under production_plan
if artifact_key == "proposal_packet":
plan = plan_or_top.get("production_plan")
if isinstance(plan, dict):
plan["decision_log_ref"] = log_ref
else:
plan_or_top["decision_log_ref"] = log_ref
validate_checkpoint(checkpoint)
path = _checkpoint_path(pipeline_dir, project_id, stage)
@@ -191,20 +306,35 @@ def get_latest_checkpoint(
return checkpoint
def get_completed_stages(pipeline_dir: Path, project_id: str) -> list[str]:
"""Return list of stages that have a completed checkpoint."""
def get_completed_stages(
pipeline_dir: Path, project_id: str, pipeline_type: str | None = None
) -> list[str]:
"""Return list of stages that have a completed checkpoint.
When pipeline_type is provided, only checks stages defined in that
pipeline's manifest — preventing false positives from leftover
checkpoints of a different pipeline type.
"""
stages_to_check = get_pipeline_stages(pipeline_type)
completed = []
for stage in STAGES:
for stage in stages_to_check:
cp = read_checkpoint(pipeline_dir, project_id, stage)
if cp and cp.get("status") == "completed":
completed.append(stage)
return completed
def get_next_stage(pipeline_dir: Path, project_id: str) -> Optional[str]:
"""Determine the next stage to run based on completed checkpoints."""
completed = set(get_completed_stages(pipeline_dir, project_id))
for stage in STAGES:
def get_next_stage(
pipeline_dir: Path, project_id: str, pipeline_type: str | None = None
) -> Optional[str]:
"""Determine the next stage to run based on completed checkpoints.
Uses pipeline-specific stage order so that pipelines with different
stage sequences (e.g. cinematic vs explainer) progress correctly.
"""
stages = get_pipeline_stages(pipeline_type) if pipeline_type else STAGES
completed = set(get_completed_stages(pipeline_dir, project_id, pipeline_type))
for stage in stages:
if stage not in completed:
return stage
return None
lib/delivery_promise.py
+247
View File
@@ -0,0 +1,247 @@
"""Delivery promise classifier.
Before provider selection, classify what the production is actually promising
to deliver. This prevents the most damaging failure mode: silently downgrading
from motion-led to still-led without the user knowing.
The delivery promise is set at the proposal stage and locked. If the compose
stage can't honor it, the system must stop and ask — not silently substitute.
"""
from __future__ import annotations
from dataclasses import dataclass, asdict
from enum import Enum
from typing import Any
class PromiseType(Enum):
MOTION_LED = "motion_led"
SOURCE_LED = "source_led"
DATA_EXPLAINER = "data_explainer"
TEACHER_EXPLAINER = "teacher_explainer"
SCREEN_DEMO = "screen_demo"
AVATAR_PRESENTER = "avatar_presenter"
HYBRID = "hybrid"
LOCALIZATION = "localization"
# Rules per promise type — what is and isn't acceptable
PROMISE_RULES: dict[str, dict[str, Any]] = {
"motion_led": {
"still_fallback_allowed": False,
"requires_video_generation": True,
"min_motion_ratio": 0.7, # At least 70% of cuts must be real motion (video/animation, not Remotion slides)
"description": "Video's quality depends on real motion — generated video clips, footage, or animation.",
},
"source_led": {
"still_fallback_allowed": True,
"requires_video_generation": False,
"min_motion_ratio": 0.3,
"description": "User-provided footage is the primary medium. Generated assets fill gaps only.",
},
"data_explainer": {
"still_fallback_allowed": True,
"requires_video_generation": False,
"min_motion_ratio": 0.0,
"description": "Data visualization and explanation. Motion graphics preferred but images acceptable.",
},
"teacher_explainer": {
"still_fallback_allowed": True,
"requires_video_generation": False,
"min_motion_ratio": 0.0,
"description": "Educational content. Clarity and comprehension over spectacle.",
},
"screen_demo": {
"still_fallback_allowed": True,
"requires_video_generation": False,
"min_motion_ratio": 0.0,
"description": "Screen recording or product demo. Legibility over cinematic dressing.",
},
"avatar_presenter": {
"still_fallback_allowed": False,
"requires_video_generation": True,
"min_motion_ratio": 0.3,
"description": "AI avatar or talking head presentation. Requires video generation for presenter.",
},
"hybrid": {
"still_fallback_allowed": True,
"requires_video_generation": False,
"min_motion_ratio": 0.2,
"description": "Mix of source footage, generated content, and graphics.",
},
"localization": {
"still_fallback_allowed": True,
"requires_video_generation": False,
"min_motion_ratio": 0.0,
"description": "Translation/dubbing of existing video. Preserving source timing and clarity.",
},
}
@dataclass
class DeliveryPromise:
"""Classifies what the production promises to deliver."""
promise_type: PromiseType
motion_required: bool
source_required: bool
tone_mode: str # "cinematic", "educational", "corporate", "playful", "raw"
quality_floor: str # "draft", "presentable", "broadcast"
approved_fallback: str | None = None # "animatic", "still_led", or None
def to_dict(self) -> dict[str, Any]:
d = asdict(self)
d["promise_type"] = self.promise_type.value
return d
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "DeliveryPromise":
return cls(
promise_type=PromiseType(data["promise_type"]),
motion_required=data.get("motion_required", False),
source_required=data.get("source_required", False),
tone_mode=data.get("tone_mode", "corporate"),
quality_floor=data.get("quality_floor", "presentable"),
approved_fallback=data.get("approved_fallback"),
)
def get_rules(self) -> dict[str, Any]:
"""Get the enforcement rules for this promise type."""
return PROMISE_RULES.get(self.promise_type.value, {})
def validate_cuts(self, cuts: list[dict]) -> dict[str, Any]:
"""Validate a list of edit cuts against this delivery promise.
Returns a dict with 'valid', 'violations', and 'motion_ratio'.
"""
rules = self.get_rules()
violations = []
if not cuts:
return {"valid": False, "violations": ["No cuts provided"], "motion_ratio": 0.0}
# Count motion vs slide-grammar vs still cuts.
# Only real video/animation/avatar footage counts as motion.
# Remotion component scenes (text_card, chart, kpi_grid, etc.) are
# "animated slides" — they have transitions but are NOT real motion.
_SLIDE_GRAMMAR_TYPES = frozenset({
"text_card", "stat_card", "chart", "bar_chart",
"line_chart", "pie_chart", "kpi_grid", "comparison",
"progress", "callout",
})
_REAL_MOTION_TYPES = frozenset({"video", "animation", "avatar"})
motion_cuts = 0
slide_cuts = 0
still_cuts = 0
for cut in cuts:
source = cut.get("source", "")
cut_type = cut.get("type", "")
# Determine category for this cut
is_motion = False
is_slide = False
if source:
ext = source.rsplit(".", 1)[-1].lower() if "." in source else ""
if ext in ("mp4", "mov", "webm", "avi", "mkv"):
is_motion = True
if cut_type in _REAL_MOTION_TYPES:
is_motion = True
elif cut_type in _SLIDE_GRAMMAR_TYPES:
is_slide = True
if is_motion:
motion_cuts += 1
elif is_slide:
slide_cuts += 1
else:
still_cuts += 1
total = motion_cuts + slide_cuts + still_cuts
# Motion ratio is real motion vs everything — slide grammar does NOT count
motion_ratio = motion_cuts / total if total > 0 else 0.0
# Check motion requirement
min_ratio = rules.get("min_motion_ratio", 0.0)
if self.motion_required and motion_ratio < min_ratio:
violations.append(
f"Motion ratio {motion_ratio:.0%} is below minimum {min_ratio:.0%} "
f"for {self.promise_type.value}. "
f"{motion_cuts}/{total} cuts have real motion "
f"({slide_cuts} are animated slides which do not count as motion)."
)
# Check still fallback (slides + stills both count as non-motion)
non_motion = slide_cuts + still_cuts
if not rules.get("still_fallback_allowed", True) and non_motion > total * 0.5:
if self.approved_fallback != "still_led":
violations.append(
f"{self.promise_type.value} does not allow still-led fallback, "
f"but {non_motion}/{total} cuts are non-motion (stills + animated slides). "
f"User must approve 'still_led' fallback or provide motion content."
)
return {
"valid": len(violations) == 0,
"violations": violations,
"motion_ratio": motion_ratio,
"motion_cuts": motion_cuts,
"slide_cuts": slide_cuts,
"still_cuts": still_cuts,
}
def classify_from_brief(
pipeline_type: str,
user_intent: dict[str, Any],
) -> DeliveryPromise:
"""Classify delivery promise from pipeline type and user intent.
This provides a sensible default. The proposal-director should refine
it based on research and capability checks.
Args:
pipeline_type: Pipeline manifest name.
user_intent: Dict with keys like 'motion_required', 'has_footage',
'tone', 'quality', 'platform'.
"""
# Pipeline → default promise type mapping
pipeline_defaults: dict[str, PromiseType] = {
"cinematic": PromiseType.MOTION_LED,
"animated-explainer": PromiseType.DATA_EXPLAINER,
"animation": PromiseType.MOTION_LED,
"talking-head": PromiseType.AVATAR_PRESENTER,
"avatar-spokesperson": PromiseType.AVATAR_PRESENTER,
"screen-demo": PromiseType.SCREEN_DEMO,
"hybrid": PromiseType.HYBRID,
"localization-dub": PromiseType.LOCALIZATION,
"podcast-repurpose": PromiseType.SOURCE_LED,
"clip-factory": PromiseType.SOURCE_LED,
}
promise_type = pipeline_defaults.get(pipeline_type, PromiseType.HYBRID)
# Override with explicit user intent
if user_intent.get("motion_required") is False and promise_type == PromiseType.MOTION_LED:
promise_type = PromiseType.HYBRID
motion_required = user_intent.get("motion_required", promise_type in (
PromiseType.MOTION_LED, PromiseType.AVATAR_PRESENTER,
))
source_required = user_intent.get("has_footage", False)
if source_required and promise_type not in (PromiseType.SOURCE_LED, PromiseType.LOCALIZATION):
promise_type = PromiseType.SOURCE_LED
tone_mode = user_intent.get("tone", "corporate")
quality_floor = user_intent.get("quality", "presentable")
return DeliveryPromise(
promise_type=promise_type,
motion_required=motion_required,
source_required=source_required,
tone_mode=tone_mode,
quality_floor=quality_floor,
)
lib/pipeline_loader.py
+50
View File
@@ -85,3 +85,53 @@ def get_stage_review_focus(manifest: dict, stage_name: str) -> list[str]:
if stage["name"] == stage_name:
return stage.get("review_focus", [])
return []
# ---------------------------------------------------------------------------
# Capability-Extension Enforcement
# ---------------------------------------------------------------------------
class ExtensionNotPermitted(PermissionError):
"""Raised when a capability extension is used but not permitted by the pipeline."""
def check_extension_permitted(
manifest: dict,
extension_type: str,
) -> None:
"""Enforce that a capability extension is permitted by the pipeline manifest.
Args:
manifest: Loaded pipeline manifest dict.
extension_type: One of 'custom_scripts', 'custom_playbooks',
'custom_skills', 'custom_tools'.
Raises:
ExtensionNotPermitted: If the extension is not allowed.
"""
valid_extensions = {"custom_scripts", "custom_playbooks", "custom_skills", "custom_tools"}
if extension_type not in valid_extensions:
raise ValueError(
f"Unknown extension type {extension_type!r}. "
f"Valid types: {sorted(valid_extensions)}"
)
extensions = manifest.get("extensions", {})
if not extensions.get(extension_type, False):
raise ExtensionNotPermitted(
f"Pipeline {manifest.get('name', 'unknown')!r} does not permit "
f"{extension_type}. Set extensions.{extension_type}: true in the "
f"pipeline manifest to allow this."
)
def get_permitted_extensions(manifest: dict) -> dict[str, bool]:
"""Return the extension permission flags for a pipeline."""
defaults = {
"custom_scripts": False,
"custom_playbooks": False,
"custom_skills": False,
"custom_tools": False,
}
extensions = manifest.get("extensions", {})
return {k: extensions.get(k, v) for k, v in defaults.items()}
lib/playbook_generator.py
+225
View File
@@ -0,0 +1,225 @@
"""Custom playbook generator.
When none of the existing 4 playbooks match the production brief, the agent
can generate a custom playbook. This replaces the old behavior of forcing
everything through the closest preset.
The generator produces a schema-valid playbook YAML from a production context.
"""
from __future__ import annotations
import json
import yaml
from pathlib import Path
from typing import Any
import jsonschema
PLAYBOOK_SCHEMA_PATH = (
Path(__file__).resolve().parent.parent
/ "schemas" / "styles" / "playbook.schema.json"
)
STYLES_DIR = Path(__file__).resolve().parent.parent / "styles"
CUSTOM_STYLES_DIR = STYLES_DIR / "custom"
def _load_playbook_schema() -> dict:
with open(PLAYBOOK_SCHEMA_PATH) as f:
return json.load(f)
def load_existing_playbook(name: str) -> dict[str, Any]:
"""Load an existing playbook YAML by name."""
path = STYLES_DIR / f"{name}.yaml"
if not path.exists():
# Check custom directory
path = CUSTOM_STYLES_DIR / f"{name}.yaml"
if not path.exists():
raise FileNotFoundError(f"Playbook not found: {name}")
with open(path) as f:
return yaml.safe_load(f)
def list_playbooks() -> list[str]:
"""List all available playbook names (preset + custom)."""
names = [p.stem for p in STYLES_DIR.glob("*.yaml")]
if CUSTOM_STYLES_DIR.exists():
names.extend(p.stem for p in CUSTOM_STYLES_DIR.glob("*.yaml"))
return sorted(set(names))
def generate_playbook(
name: str,
context: dict[str, Any],
base_playbook: str | None = None,
) -> dict[str, Any]:
"""Generate a custom playbook from production context.
Args:
name: Name for the new playbook.
context: Dict with keys like:
- mood: str (e.g., "warm", "dark", "energetic")
- tone: str (e.g., "cinematic", "educational", "corporate")
- colors: dict with primary, accent, background, text (optional)
- fonts: dict with headings, body (optional)
- pace: str (optional)
- audience: str (optional)
base_playbook: Name of existing playbook to use as a starting point.
Returns:
Schema-valid playbook dict.
"""
# Start from base or create fresh
if base_playbook:
playbook = load_existing_playbook(base_playbook)
else:
playbook = _create_minimal_playbook(name, context)
# Override identity
playbook["identity"]["name"] = name
if context.get("mood"):
playbook["identity"]["mood"] = context["mood"]
if context.get("pace"):
playbook["identity"]["pace"] = context["pace"]
if context.get("tone"):
# Map tone to category
tone_to_category = {
"cinematic": "cinematic",
"educational": "minimalist",
"corporate": "motion-graphics",
"playful": "motion-graphics",
"raw": "cinematic",
}
playbook["identity"]["category"] = tone_to_category.get(
context["tone"], "custom"
)
# Override colors if provided
if context.get("colors"):
colors = context["colors"]
cp = playbook["visual_language"]["color_palette"]
if colors.get("primary"):
cp["primary"] = [colors["primary"]] if isinstance(colors["primary"], str) else colors["primary"]
if colors.get("accent"):
cp["accent"] = [colors["accent"]] if isinstance(colors["accent"], str) else colors["accent"]
if colors.get("background"):
cp["background"] = colors["background"]
if colors.get("text"):
cp["text"] = colors["text"]
# Override fonts if provided
if context.get("fonts"):
fonts = context["fonts"]
if fonts.get("headings"):
playbook["typography"]["headings"]["font"] = fonts["headings"]
if fonts.get("body"):
playbook["typography"]["body"]["font"] = fonts["body"]
return playbook
def _create_minimal_playbook(name: str, context: dict[str, Any]) -> dict[str, Any]:
"""Create a minimal but complete playbook from scratch."""
mood = context.get("mood", "professional")
tone = context.get("tone", "corporate")
# Sensible defaults based on mood
if mood in ("dark", "cinematic", "dramatic"):
bg = "#0F172A"
text = "#F8FAFC"
primary = ["#3B82F6"]
accent = ["#F59E0B"]
elif mood in ("warm", "intimate", "organic"):
bg = "#FFFBEB"
text = "#1C1917"
primary = ["#D97706"]
accent = ["#059669"]
elif mood in ("playful", "energetic", "bold"):
bg = "#FFFFFF"
text = "#1F2937"
primary = ["#7C3AED"]
accent = ["#EC4899"]
else: # professional, clean, neutral
bg = "#FFFFFF"
text = "#1F2937"
primary = ["#2563EB"]
accent = ["#F59E0B"]
return {
"identity": {
"name": name,
"category": "custom",
"mood": mood,
"pace": context.get("pace", "moderate"),
"best_for": f"Custom playbook for {tone} {mood} content",
},
"visual_language": {
"color_palette": {
"primary": primary,
"accent": accent,
"background": bg,
"text": text,
},
"composition": "balanced grid with breathing room",
"texture": "clean digital",
},
"typography": {
"headings": {"font": "Inter", "weight": 700},
"body": {"font": "Inter", "weight": 400},
},
"motion": {
"transitions": ["crossfade", "cut"],
"animation_style": "spring-based with moderate damping",
"pacing_rules": {
"min_scene_hold_seconds": 2.0,
"max_scene_hold_seconds": 6.0,
"text_card_hold_seconds": 3.5,
"stat_card_hold_seconds": 4.0,
"transition_duration_seconds": 0.4,
},
},
"audio": {
"voice_style": "clear, conversational, authoritative",
"music_mood": mood,
"music_volume": 0.15,
},
"asset_generation": {
"image_prompt_prefix": f"{mood} {tone} style",
"consistency_anchors": [f"{mood} color palette", f"{tone} visual language"],
},
"quality_rules": [
"Maintain color consistency across all scenes",
"Text must be legible on all backgrounds",
"Transitions should be purposeful, not decorative",
],
"chart_palette": primary + accent + ["#10B981", "#EF4444", "#8B5CF6"],
}
def save_playbook(
playbook: dict[str, Any],
project_name: str | None = None,
) -> Path:
"""Validate and save a playbook to the custom styles directory.
Args:
playbook: Schema-valid playbook dict.
project_name: Optional project name for the filename.
Returns:
Path to the saved YAML file.
"""
schema = _load_playbook_schema()
jsonschema.validate(instance=playbook, schema=schema)
name = project_name or playbook["identity"]["name"]
filename = name.lower().replace(" ", "-").replace("_", "-")
CUSTOM_STYLES_DIR.mkdir(parents=True, exist_ok=True)
path = CUSTOM_STYLES_DIR / f"{filename}.yaml"
with open(path, "w") as f:
yaml.dump(playbook, f, default_flow_style=False, allow_unicode=True)
return path
lib/scoring.py
+369
View File
@@ -0,0 +1,369 @@
"""Provider and production path scoring engine.
Replaces naive "first available provider" selection with weighted
multi-dimensional scoring. Every provider choice should be explainable —
not just "it was available."
Scores are normalized 0-1. Higher is better.
"""
from __future__ import annotations
from dataclasses import dataclass, asdict, field
from typing import Any
# ---------------------------------------------------------------------------
# Provider Score
# ---------------------------------------------------------------------------
@dataclass
class ProviderScore:
"""Scored evaluation of a provider against a specific task context."""
tool_name: str
provider: str
task_fit: float = 0.0 # 0-1: best fit for this exact asset class
output_quality: float = 0.0 # 0-1: expected fidelity for the brief
control: float = 0.0 # 0-1: reference/style directability
reliability: float = 0.0 # 0-1: runtime confidence
cost_efficiency: float = 0.0 # 0-1: quality per dollar
latency: float = 0.0 # 0-1: acceptable turnaround
continuity: float = 0.0 # 0-1: fits already locked decisions
@property
def weighted_score(self) -> float:
return (
self.task_fit * 0.30
+ self.output_quality * 0.20
+ self.control * 0.15
+ self.reliability * 0.15
+ self.cost_efficiency * 0.10
+ self.latency * 0.05
+ self.continuity * 0.05
)
def to_dict(self) -> dict[str, Any]:
d = asdict(self)
d["weighted_score"] = self.weighted_score
return d
def explain(self) -> str:
"""Human-readable explanation of this score."""
parts = [f"{self.tool_name} ({self.provider}): {self.weighted_score:.2f}"]
top = sorted(
[
("task_fit", self.task_fit, 0.30),
("output_quality", self.output_quality, 0.20),
("control", self.control, 0.15),
("reliability", self.reliability, 0.15),
("cost_efficiency", self.cost_efficiency, 0.10),
("latency", self.latency, 0.05),
("continuity", self.continuity, 0.05),
],
key=lambda x: x[1] * x[2],
reverse=True,
)
for name, val, weight in top[:3]:
parts.append(f" {name}={val:.2f} (w={weight})")
return "\n".join(parts)
# ---------------------------------------------------------------------------
# Production Path Score
# ---------------------------------------------------------------------------
@dataclass
class ProductionPathScore:
"""Scored evaluation of an entire production path."""
path_label: str
delivery_fit: float = 0.0
quality_fit: float = 0.0
capability_confidence: float = 0.0
fallback_integrity: float = 0.0
budget_fit: float = 0.0
speed_fit: float = 0.0
controllability: float = 0.0
consistency_fit: float = 0.0
@property
def weighted_score(self) -> float:
return (
self.delivery_fit * 0.25
+ self.quality_fit * 0.20
+ self.capability_confidence * 0.15
+ self.fallback_integrity * 0.10
+ self.budget_fit * 0.10
+ self.speed_fit * 0.08
+ self.controllability * 0.07
+ self.consistency_fit * 0.05
)
def to_dict(self) -> dict[str, Any]:
d = asdict(self)
d["weighted_score"] = self.weighted_score
return d
# ---------------------------------------------------------------------------
# Scoring Functions
# ---------------------------------------------------------------------------
def _keyword_overlap(set_a: set[str], set_b: set[str]) -> float:
"""Jaccard-like overlap score between two keyword sets."""
if not set_a or not set_b:
return 0.0
a = {s.lower().strip() for s in set_a}
b = {s.lower().strip() for s in set_b}
intersection = len(a & b)
union = len(a | b)
return intersection / union if union > 0 else 0.0
# Semantic synonym clusters: when intent says "cinematic" and tool says
# "film" or "movie", that's a match even without literal keyword overlap.
_SYNONYM_CLUSTERS: list[set[str]] = [
{"cinematic", "film", "movie", "trailer", "dramatic", "epic"},
{"explainer", "educational", "tutorial", "teaching", "lesson"},
{"corporate", "business", "professional", "enterprise"},
{"social", "tiktok", "instagram", "reels", "shorts", "viral"},
{"animation", "animated", "motion-graphics", "motion", "kinetic"},
{"realistic", "photorealistic", "lifelike", "natural"},
{"stock", "footage", "b-roll", "library"},
{"avatar", "presenter", "talking-head", "spokesperson"},
{"voiceover", "narration", "speech", "voice"},
{"music", "soundtrack", "background-music", "score", "ambient"},
]
def _expand_synonyms(words: set[str]) -> set[str]:
"""Expand a word set with synonyms from known clusters."""
expanded = set(words)
for cluster in _SYNONYM_CLUSTERS:
if expanded & cluster:
expanded |= cluster
return expanded
def _compute_task_fit(
best_for: set[str],
intent: str,
style_keywords: set[str],
) -> float:
"""Score how well a tool's best_for matches the task intent and style.
Uses synonym expansion so that semantic near-misses (e.g. "cinematic"
vs "film") still score well, not just literal keyword overlap.
"""
if not best_for:
return 0.3 # Unknown capability — modest default
intent_words = _expand_synonyms(set(intent.lower().split()))
best_for_words = set()
for desc in best_for:
best_for_words.update(desc.lower().split())
best_for_words = _expand_synonyms(best_for_words)
intent_score = _keyword_overlap(intent_words, best_for_words)
style_expanded = _expand_synonyms({kw.lower() for kw in style_keywords})
style_score = _keyword_overlap(style_expanded, best_for_words)
return min(1.0, intent_score * 0.7 + style_score * 0.3 + 0.1)
def _compute_control(supports: dict[str, Any]) -> float:
"""Score controllability from the supports dict.
Features are weighted by creative impact — controlnet and reference_image
are worth more than seed or aspect_ratio.
"""
# (feature_name, weight) — higher weight = more creative control
control_features = [
("controlnet", 2.0),
("reference_image", 1.8),
("style_transfer", 1.5),
("inpainting", 1.5),
("img2img", 1.3),
("negative_prompt", 1.0),
("custom_size", 0.8),
("aspect_ratio", 0.7),
("seed", 0.5),
]
if not supports:
return 0.3
total_weight = sum(w for _, w in control_features)
earned = sum(w for f, w in control_features if supports.get(f))
return min(1.0, earned / (total_weight * 0.5))
def _compute_cost_efficiency(
estimated_cost: float,
budget_remaining: float | None,
) -> float:
"""Score cost efficiency. Free is 1.0, over-budget is 0.0."""
if estimated_cost <= 0:
return 1.0
if budget_remaining is not None and budget_remaining <= 0:
return 0.0
if budget_remaining is not None:
ratio = estimated_cost / budget_remaining
if ratio > 0.5:
return 0.1
if ratio > 0.2:
return 0.5
return 0.8
# No budget info — use absolute cost heuristic
if estimated_cost < 0.05:
return 0.9
if estimated_cost < 0.20:
return 0.7
if estimated_cost < 1.00:
return 0.5
return 0.3
def _compute_continuity(
provider: str,
locked_providers: set[str],
) -> float:
"""Score how well this provider fits already-locked decisions."""
if not locked_providers:
return 0.5 # No prior context
if provider in locked_providers:
return 0.9 # Same provider = likely consistent style
return 0.4 # Different provider = possible style break
def score_provider(tool, task_context: dict[str, Any]) -> ProviderScore:
"""Score a provider against a task context.
Args:
tool: A BaseTool instance.
task_context: Dict with keys:
- intent (str): What the asset is for
- style_keywords (list[str]): Visual/audio style descriptors
- budget_remaining_usd (float|None): Remaining budget
- locked_providers (set[str]): Providers already chosen
- motion_required (bool): Whether motion is a hard requirement
- asset_type (str): "image", "video", "audio", "music", "voice"
"""
info = tool.get_info()
status = str(tool.get_status())
best_for = set(info.get("best_for", []))
intent = task_context.get("intent", "")
style_keywords = set(task_context.get("style_keywords", []))
task_fit = _compute_task_fit(best_for, intent, style_keywords)
# Reliability: uses historical success rate if available, else availability status.
hist_success = info.get("historical_success_rate") # 0.0-1.0 if tracked
if hist_success is not None:
reliability = float(hist_success)
elif status == "available":
# Stable tools get higher baseline than experimental ones
reliability = 0.95 if info.get("stability") == "production" else 0.8
elif status == "degraded":
reliability = 0.4
else:
reliability = 0.0
# Control: from supports dict
control = _compute_control(info.get("supports", {}))
# Cost efficiency
try:
estimated_cost = tool.estimate_cost(task_context)
except Exception:
estimated_cost = 0.0
cost_efficiency = _compute_cost_efficiency(
estimated_cost, task_context.get("budget_remaining_usd")
)
# Latency: uses measured p50 latency if available, else runtime class heuristic.
measured_p50 = info.get("latency_p50_seconds") # historical median
if measured_p50 is not None:
# Map measured latency to a 0-1 score (sub-second is best, >60s is worst)
if measured_p50 <= 1.0:
latency = 1.0
elif measured_p50 <= 10.0:
latency = 0.8
elif measured_p50 <= 30.0:
latency = 0.6
elif measured_p50 <= 60.0:
latency = 0.4
else:
latency = 0.2
else:
runtime = info.get("runtime", "api")
if runtime in ("local", "local_gpu"):
latency = 0.9
elif runtime == "hybrid":
latency = 0.6
else:
latency = 0.4
# Continuity
continuity = _compute_continuity(
info.get("provider", ""),
set(task_context.get("locked_providers", [])),
)
# Output quality: uses measured quality score if available (e.g. from
# user ratings or automated eval), else falls back to stability + tier.
measured_quality = info.get("quality_score") # 0.0-1.0 if tracked
if measured_quality is not None:
output_quality = float(measured_quality)
else:
stability = info.get("stability", "experimental")
tier = info.get("tier", "")
quality_map = {"production": 0.9, "beta": 0.7, "experimental": 0.4}
output_quality = quality_map.get(stability, 0.5)
# Tier bonus: generate-tier tools that are production-stable get a nudge
if tier == "generate" and stability == "production":
output_quality = min(1.0, output_quality + 0.05)
# Motion-required penalty: if task needs motion but tool is image-only
if task_context.get("motion_required") and task_context.get("asset_type") == "video":
cap = info.get("capability", "")
if "video" not in cap:
task_fit *= 0.2 # Heavy penalty
return ProviderScore(
tool_name=info.get("name", "unknown"),
provider=info.get("provider", "unknown"),
task_fit=min(1.0, task_fit),
output_quality=output_quality,
control=control,
reliability=reliability,
cost_efficiency=cost_efficiency,
latency=latency,
continuity=continuity,
)
def rank_providers(
tools: list,
task_context: dict[str, Any],
) -> list[ProviderScore]:
"""Rank a list of tools by weighted score for a given task context.
Returns scores sorted best-first.
"""
scores = [score_provider(t, task_context) for t in tools]
return sorted(scores, key=lambda s: s.weighted_score, reverse=True)
def format_ranking(rankings: list[ProviderScore], top_n: int = 5) -> str:
"""Format a ranking list for user presentation."""
lines = []
for i, r in enumerate(rankings[:top_n], 1):
lines.append(
f" {i}. {r.tool_name} ({r.provider}) — "
f"score: {r.weighted_score:.2f} "
f"[fit={r.task_fit:.1f} quality={r.output_quality:.1f} "
f"control={r.control:.1f} reliable={r.reliability:.1f} "
f"cost={r.cost_efficiency:.1f}]"
)
return "\n".join(lines)
lib/shot_prompt_builder.py
+166
View File
@@ -0,0 +1,166 @@
"""Shot prompt builder — converts structured shot language into provider-optimized prompts.
Uses a 5-layer framework based on professional cinematography prompting research:
Layer 1: Camera (lens, depth of field)
Layer 2: Movement (shot size, camera movement)
Layer 3: Subject (description + texture keywords)
Layer 4: Lighting (lighting key, color temperature)
Layer 5: Style (adapted from playbook, not verbatim)
This replaces the old approach of prepending a fixed playbook image_prompt_prefix
to every scene description, which made all scenes look the same.
"""
from __future__ import annotations
from typing import Any
# Mapping from shot_language enums to natural language for prompting
_SHOT_SIZE_PHRASES = {
"extreme_wide": "extreme wide shot showing vast environment",
"wide": "wide shot capturing full scene",
"medium_wide": "medium-wide shot framing subject with surroundings",
"medium": "medium shot from waist up",
"medium_close": "medium close-up from chest up",
"close_up": "close-up focusing on face or detail",
"extreme_close_up": "extreme close-up on fine detail",
"over_shoulder": "over-the-shoulder perspective",
"insert": "insert shot of specific detail",
"establishing": "establishing shot setting the location",
}
_MOVEMENT_PHRASES = {
"static": "locked-off static camera",
"pan_left": "smooth pan to the left",
"pan_right": "smooth pan to the right",
"tilt_up": "gentle tilt upward",
"tilt_down": "gentle tilt downward",
"dolly_in": "slow dolly in toward subject",
"dolly_out": "slow dolly out from subject",
"tracking_left": "tracking shot moving left alongside subject",
"tracking_right": "tracking shot moving right alongside subject",
"crane_up": "crane shot rising upward",
"crane_down": "crane shot descending",
"handheld": "handheld camera with natural movement",
"steadicam": "smooth steadicam following movement",
"whip_pan": "fast whip pan",
"orbital": "orbital camera circling subject",
"zoom_in": "slow zoom in",
"zoom_out": "slow zoom out",
"rack_focus": "rack focus shift between foreground and background",
}
_LIGHTING_PHRASES = {
"high_key": "bright high-key lighting, minimal shadows",
"low_key": "dramatic low-key lighting with deep shadows",
"natural": "natural ambient lighting",
"golden_hour": "warm golden hour sunlight",
"blue_hour": "cool blue hour twilight",
"tungsten_warm": "warm tungsten interior lighting",
"neon": "neon-lit with vibrant color spill",
"silhouette": "backlit silhouette",
"rim_lit": "rim lighting highlighting edges",
"volumetric": "volumetric light with visible rays",
"overcast_soft": "soft overcast diffused light",
}
_DOF_PHRASES = {
"shallow": "shallow depth of field with bokeh",
"medium": "medium depth of field",
"deep": "deep focus with everything sharp",
}
_COLOR_TEMP_PHRASES = {
"cool": "cool blue-toned color palette",
"neutral": "neutral balanced colors",
"warm": "warm amber-toned color palette",
"mixed": "mixed color temperatures for contrast",
}
def build_shot_prompt(
scene: dict[str, Any],
style_context: dict[str, Any] | None = None,
) -> str:
"""Convert a scene with structured shot language into a generation prompt.
Args:
scene: Scene dict from scene_plan (with shot_language, description,
texture_keywords, etc.)
style_context: Optional playbook-derived style info with keys like
'generation_prefix', 'visual_language', 'mood'.
Returns:
A natural-language prompt optimized for image/video generation.
"""
sl = scene.get("shot_language", {})
layers: list[str] = []
# Layer 1: Camera — lens and depth of field
camera_parts = []
if sl.get("lens_mm"):
camera_parts.append(f"{sl['lens_mm']}mm lens")
if sl.get("depth_of_field"):
camera_parts.append(_DOF_PHRASES.get(sl["depth_of_field"], ""))
if camera_parts:
layers.append(", ".join(filter(None, camera_parts)))
# Layer 2: Movement — shot size and camera movement
movement_parts = []
if sl.get("shot_size"):
movement_parts.append(_SHOT_SIZE_PHRASES.get(sl["shot_size"], sl["shot_size"]))
if sl.get("camera_movement") and sl["camera_movement"] != "static":
movement_parts.append(_MOVEMENT_PHRASES.get(sl["camera_movement"], sl["camera_movement"]))
if movement_parts:
layers.append(", ".join(movement_parts))
# Layer 3: Subject — the scene description + texture keywords
description = scene.get("description", "")
texture = scene.get("texture_keywords", [])
subject_parts = [description]
if texture:
subject_parts.append(", ".join(texture))
layers.append(". ".join(filter(None, subject_parts)))
# Layer 4: Lighting — lighting key and color temperature
lighting_parts = []
if sl.get("lighting_key"):
lighting_parts.append(_LIGHTING_PHRASES.get(sl["lighting_key"], sl["lighting_key"]))
if sl.get("color_temperature"):
lighting_parts.append(_COLOR_TEMP_PHRASES.get(sl["color_temperature"], ""))
if lighting_parts:
layers.append(", ".join(filter(None, lighting_parts)))
# Layer 5: Style — adapted from playbook (NOT verbatim prefix)
if style_context:
mood = style_context.get("mood", "")
visual_lang = style_context.get("visual_language", {})
style_hint = visual_lang.get("aesthetic", "") or mood
if style_hint:
layers.append(f"Style: {style_hint}")
return ". ".join(filter(None, layers))
def build_batch_prompts(
scenes: list[dict[str, Any]],
style_context: dict[str, Any] | None = None,
) -> list[dict[str, str]]:
"""Build prompts for all visual scenes in a scene plan.
Returns list of {scene_id, prompt} dicts.
"""
results = []
for scene in scenes:
# Skip non-visual scene types
scene_type = scene.get("type", "")
if scene_type in ("transition",):
continue
prompt = build_shot_prompt(scene, style_context)
results.append({
"scene_id": scene.get("id", "unknown"),
"prompt": prompt,
"hero_moment": scene.get("hero_moment", False),
})
return results
lib/slideshow_risk.py
+245
View File
@@ -0,0 +1,245 @@
"""Slideshow risk scorer.
Scores a video plan across 6 dimensions that reliably predict whether
the output will feel like a slideshow rather than directed video.
Each dimension is scored 0-5 (lower is better):
- repetition: same layouts/backgrounds/scene grammar recurring
- decorative_visuals: scenes decorate instead of communicate
- weak_motion: motion exists but has no narrative purpose
- weak_shot_intent: no explicit reason for framing or reveal rhythm
- typography_overreliance: too much of the video is text-first
- unsupported_cinematic_claims: cinematic label without structure
Verdict:
< 2.0: strong
< 3.0: acceptable
< 4.0: revise
>= 4.0: fail — should not proceed to compose
"""
from __future__ import annotations
from typing import Any
def score_slideshow_risk(
scenes: list[dict[str, Any]],
edit_decisions: dict[str, Any] | None = None,
renderer_family: str | None = None,
) -> dict[str, Any]:
"""Score slideshow risk across 6 dimensions.
Args:
scenes: Scene list from scene_plan artifact.
edit_decisions: Optional edit_decisions artifact for transition analysis.
renderer_family: Optional renderer family for cinematic claim verification.
Returns:
{
"average": float,
"verdict": str,
"dimensions": {dimension_name: {"score": float, "reason": str}},
}
"""
if not scenes:
return {
"average": 5.0,
"verdict": "fail",
"dimensions": {},
}
dimensions = {
"repetition": _score_repetition(scenes),
"decorative_visuals": _score_decorative(scenes),
"weak_motion": _score_weak_motion(scenes),
"weak_shot_intent": _score_weak_intent(scenes),
"typography_overreliance": _score_typography(scenes),
"unsupported_cinematic_claims": _score_cinematic_claims(scenes, renderer_family),
}
scores = [d["score"] for d in dimensions.values()]
average = sum(scores) / len(scores)
if average < 2.0:
verdict = "strong"
elif average < 3.0:
verdict = "acceptable"
elif average < 4.0:
verdict = "revise"
else:
verdict = "fail"
return {
"average": round(average, 2),
"verdict": verdict,
"dimensions": dimensions,
}
def _score_repetition(scenes: list[dict]) -> dict[str, Any]:
"""Score visual repetition across scenes."""
if len(scenes) < 3:
return {"score": 0.0, "reason": "Too few scenes to assess repetition"}
# Check for repeated scene types
from collections import Counter
types = Counter(s.get("type", "unknown") for s in scenes)
most_common_type, most_common_count = types.most_common(1)[0]
type_ratio = most_common_count / len(scenes)
# Check for repeated descriptions (crude similarity)
descriptions = [s.get("description", "").lower()[:50] for s in scenes]
unique_desc_ratio = len(set(descriptions)) / len(descriptions)
# Check shot size repetition
sizes = [s.get("shot_language", {}).get("shot_size", "none") for s in scenes]
size_ratio = Counter(sizes).most_common(1)[0][1] / len(scenes)
score = 0.0
reasons = []
if type_ratio > 0.7:
score += 2.0
reasons.append(f"Scene type '{most_common_type}' dominates at {type_ratio:.0%}")
if unique_desc_ratio < 0.6:
score += 1.5
reasons.append(f"Only {unique_desc_ratio:.0%} unique descriptions")
if size_ratio > 0.6:
score += 1.5
reasons.append(f"Same shot size in {size_ratio:.0%} of scenes")
return {"score": min(5.0, score), "reason": "; ".join(reasons) or "Good variety"}
def _score_decorative(scenes: list[dict]) -> dict[str, Any]:
"""Score whether scenes are decorative vs communicative."""
decorative_count = 0
for scene in scenes:
has_info_role = bool(scene.get("information_role"))
has_narrative_role = bool(scene.get("narrative_role"))
has_intent = bool(scene.get("shot_intent"))
# A scene with no role or intent is likely decorative
if not has_info_role and not has_narrative_role and not has_intent:
decorative_count += 1
ratio = decorative_count / len(scenes)
score = min(5.0, ratio * 5.0)
if ratio > 0.5:
reason = f"{decorative_count}/{len(scenes)} scenes have no stated purpose (no information_role, narrative_role, or shot_intent)"
elif ratio > 0.2:
reason = f"{decorative_count}/{len(scenes)} scenes lack stated purpose"
else:
reason = "Most scenes have clear communicative purpose"
return {"score": round(score, 1), "reason": reason}
def _score_weak_motion(scenes: list[dict]) -> dict[str, Any]:
"""Score whether camera movement is purposeful."""
total_moving = 0
purposeless_moving = 0
for scene in scenes:
sl = scene.get("shot_language", {})
movement = sl.get("camera_movement", "static")
if movement not in ("static", "unspecified", None):
total_moving += 1
# Movement without shot_intent suggests arbitrary motion
if not scene.get("shot_intent"):
purposeless_moving += 1
if total_moving == 0:
# No movement at all is fine for some styles, but scores moderate
return {"score": 1.5, "reason": "No camera movement defined (may be intentional for static style)"}
ratio = purposeless_moving / total_moving
score = min(5.0, ratio * 4.0)
if ratio > 0.5:
reason = f"{purposeless_moving}/{total_moving} moving shots lack shot_intent"
else:
reason = "Camera movement appears purposeful"
return {"score": round(score, 1), "reason": reason}
def _score_weak_intent(scenes: list[dict]) -> dict[str, Any]:
"""Score shot intent completeness."""
with_intent = sum(1 for s in scenes if s.get("shot_intent"))
ratio = with_intent / len(scenes)
# Invert: more intent = lower score
score = min(5.0, (1.0 - ratio) * 5.0)
if ratio < 0.3:
reason = f"Only {with_intent}/{len(scenes)} scenes have shot_intent — most shots lack purpose"
elif ratio < 0.6:
reason = f"{with_intent}/{len(scenes)} scenes have shot_intent"
else:
reason = "Strong shot intent coverage"
return {"score": round(score, 1), "reason": reason}
def _score_typography(scenes: list[dict]) -> dict[str, Any]:
"""Score text-first overreliance."""
text_scenes = sum(
1 for s in scenes
if s.get("type") in ("text_card", "stat_card", "kpi_grid")
)
ratio = text_scenes / len(scenes)
if ratio > 0.6:
score = 4.0
reason = f"{text_scenes}/{len(scenes)} scenes are text/stat cards — video feels like animated slides"
elif ratio > 0.4:
score = 2.5
reason = f"{text_scenes}/{len(scenes)} scenes are text-based — consider balancing with visual scenes"
elif ratio > 0.2:
score = 1.0
reason = "Balanced text and visual content"
else:
score = 0.0
reason = "Visual-first approach"
return {"score": score, "reason": reason}
def _score_cinematic_claims(
scenes: list[dict],
renderer_family: str | None,
) -> dict[str, Any]:
"""Score whether cinematic claims are backed by cinematic structure."""
is_cinematic = renderer_family and "cinematic" in renderer_family.lower()
if not is_cinematic:
return {"score": 0.0, "reason": "Not claiming cinematic treatment"}
issues = []
# Cinematic should have: varied shot sizes, intentional movement, hero moments
hero_count = sum(1 for s in scenes if s.get("hero_moment"))
if hero_count == 0:
issues.append("Claims cinematic but has no hero_moment defined")
has_movement = sum(
1 for s in scenes
if s.get("shot_language", {}).get("camera_movement", "static") != "static"
)
if has_movement < len(scenes) * 0.3:
issues.append(f"Claims cinematic but only {has_movement}/{len(scenes)} scenes have camera movement")
has_lighting = sum(
1 for s in scenes
if s.get("shot_language", {}).get("lighting_key")
)
if has_lighting < len(scenes) * 0.3:
issues.append(f"Claims cinematic but only {has_lighting}/{len(scenes)} scenes define lighting")
score = min(5.0, len(issues) * 1.8)
reason = "; ".join(issues) if issues else "Cinematic claims supported by structure"
return {"score": round(score, 1), "reason": reason}
lib/source_media_review.py
+395
View File
@@ -0,0 +1,395 @@
"""Source media review helper.
Standardizes inspection of user-supplied media files so pipelines stop
reinventing partial checks. Uses existing analysis tools (audio_probe,
frame_sampler, scene_detect, transcriber) to produce a normalized
source_media_review artifact.
The contract: if user-supplied media exists, source_media_review is
REQUIRED before the first planning stage that depends on creative
assumptions. Never claim a file was reviewed unless a real probe ran.
"""
from __future__ import annotations
import json
import logging
from pathlib import Path
from typing import Any, Optional
logger = logging.getLogger(__name__)
# Media type detection by extension
_VIDEO_EXTENSIONS = frozenset({".mp4", ".mov", ".webm", ".avi", ".mkv", ".m4v"})
_AUDIO_EXTENSIONS = frozenset({".mp3", ".wav", ".aac", ".flac", ".ogg", ".m4a", ".opus"})
_IMAGE_EXTENSIONS = frozenset({".jpg", ".jpeg", ".png", ".webp", ".bmp", ".tiff", ".svg"})
def detect_media_type(path: Path) -> Optional[str]:
"""Classify a file as video, audio, or image by extension."""
ext = path.suffix.lower()
if ext in _VIDEO_EXTENSIONS:
return "video"
if ext in _AUDIO_EXTENSIONS:
return "audio"
if ext in _IMAGE_EXTENSIONS:
return "image"
return None
def _probe_video(path: Path, tool_registry: Any) -> dict[str, Any]:
"""Probe a video file using audio_probe (ffprobe wrapper) and frame_sampler."""
result: dict[str, Any] = {"technical_probe": {}, "representative_frames": [], "quality_risks": []}
# Technical probe via audio_probe or ffprobe
try:
audio_probe = tool_registry.get_tool("audio_probe")
if audio_probe:
probe_result = audio_probe.execute({"input_path": str(path)})
if probe_result.success:
result["technical_probe"] = probe_result.data
except Exception as e:
logger.warning("audio_probe failed for %s: %s", path, e)
# If audio_probe didn't work, try ffprobe directly
if not result["technical_probe"]:
try:
import subprocess
cmd = [
"ffprobe", "-v", "quiet", "-print_format", "json",
"-show_format", "-show_streams", str(path),
]
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
if proc.returncode == 0:
probe_data = json.loads(proc.stdout)
fmt = probe_data.get("format", {})
streams = probe_data.get("streams", [])
video_stream = next((s for s in streams if s.get("codec_type") == "video"), {})
audio_stream = next((s for s in streams if s.get("codec_type") == "audio"), {})
result["technical_probe"] = {
"duration_seconds": float(fmt.get("duration", 0)),
"resolution": f"{video_stream.get('width', '?')}x{video_stream.get('height', '?')}",
"fps": _parse_fps(video_stream.get("r_frame_rate", "0/1")),
"codec": video_stream.get("codec_name", "unknown"),
"audio_codec": audio_stream.get("codec_name", ""),
"sample_rate": int(audio_stream.get("sample_rate", 0)) if audio_stream else 0,
"channels": int(audio_stream.get("channels", 0)) if audio_stream else 0,
"file_size_bytes": int(fmt.get("size", 0)),
"bitrate_kbps": round(int(fmt.get("bit_rate", 0)) / 1000, 1),
}
except Exception as e:
logger.warning("ffprobe failed for %s: %s", path, e)
result["quality_risks"].append(f"Could not probe file: {e}")
# Sample frames
try:
frame_sampler = tool_registry.get_tool("frame_sampler")
if frame_sampler:
duration = result["technical_probe"].get("duration_seconds", 0)
timestamps = _sample_timestamps(duration, count=4)
sample_result = frame_sampler.execute({
"input_path": str(path),
"timestamps": timestamps,
"output_dir": str(path.parent / ".source_review_frames"),
})
if sample_result.success:
result["representative_frames"] = sample_result.data.get("frame_paths", [])
except Exception as e:
logger.warning("frame_sampler failed for %s: %s", path, e)
# Quality risk assessment
probe = result["technical_probe"]
if probe:
res = probe.get("resolution", "")
if res and "x" in res:
try:
w, h = res.split("x")
if int(w) < 720 or int(h) < 480:
result["quality_risks"].append(f"Low resolution ({res}) — may appear pixelated in final output")
except ValueError:
pass
if probe.get("channels", 0) == 1:
result["quality_risks"].append("Mono audio — consider if stereo output is expected")
if probe.get("duration_seconds", 0) < 3:
result["quality_risks"].append("Very short clip (<3s) — limited usability")
return result
def _probe_audio(path: Path, tool_registry: Any) -> dict[str, Any]:
"""Probe an audio file using audio_probe."""
result: dict[str, Any] = {"technical_probe": {}, "quality_risks": []}
try:
audio_probe = tool_registry.get_tool("audio_probe")
if audio_probe:
probe_result = audio_probe.execute({"input_path": str(path)})
if probe_result.success:
result["technical_probe"] = probe_result.data
except Exception as e:
logger.warning("audio_probe failed for %s: %s", path, e)
# Fallback ffprobe
if not result["technical_probe"]:
try:
import subprocess
cmd = [
"ffprobe", "-v", "quiet", "-print_format", "json",
"-show_format", "-show_streams", str(path),
]
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
if proc.returncode == 0:
probe_data = json.loads(proc.stdout)
fmt = probe_data.get("format", {})
stream = next(
(s for s in probe_data.get("streams", []) if s.get("codec_type") == "audio"),
{},
)
result["technical_probe"] = {
"duration_seconds": float(fmt.get("duration", 0)),
"audio_codec": stream.get("codec_name", "unknown"),
"sample_rate": int(stream.get("sample_rate", 0)),
"channels": int(stream.get("channels", 0)),
"file_size_bytes": int(fmt.get("size", 0)),
"bitrate_kbps": round(int(fmt.get("bit_rate", 0)) / 1000, 1),
}
except Exception as e:
logger.warning("ffprobe failed for audio %s: %s", path, e)
result["quality_risks"].append(f"Could not probe audio: {e}")
return result
def _probe_image(path: Path) -> dict[str, Any]:
"""Probe an image file for basic metadata."""
result: dict[str, Any] = {"technical_probe": {}, "quality_risks": []}
try:
from PIL import Image
img = Image.open(path)
w, h = img.size
result["technical_probe"] = {
"resolution": f"{w}x{h}",
"file_size_bytes": path.stat().st_size,
"codec": img.format or "unknown",
}
if w < 640 or h < 480:
result["quality_risks"].append(f"Low resolution ({w}x{h}) — may need upscaling")
except ImportError:
# PIL not available — use file size as minimal probe
result["technical_probe"] = {
"file_size_bytes": path.stat().st_size,
}
except Exception as e:
result["quality_risks"].append(f"Could not probe image: {e}")
return result
def _transcribe_if_available(
path: Path, media_type: str, tool_registry: Any
) -> Optional[str]:
"""Attempt transcription for video/audio files."""
if media_type not in ("video", "audio"):
return None
try:
transcriber = tool_registry.get_tool("transcriber")
if transcriber and transcriber.get_status().value == "available":
result = transcriber.execute({"input_path": str(path)})
if result.success:
text = result.data.get("text", "")
if text:
# Return summary, not full transcript
words = text.split()
if len(words) > 100:
return f"{' '.join(words[:100])}... ({len(words)} words total)"
return text
except Exception as e:
logger.warning("Transcription failed for %s: %s", path, e)
return None
def review_source_media(
files: list[Path],
context: dict[str, Any],
tool_registry: Any = None,
) -> dict[str, Any]:
"""Review user-supplied media files and produce a source_media_review artifact.
Args:
files: Paths to user-supplied media files.
context: Dict with optional keys like 'pipeline_type', 'project_dir'.
tool_registry: The tool registry instance (for accessing analysis tools).
Returns:
Schema-valid source_media_review artifact dict.
Must never claim a file was reviewed unless a real probe/sampling/transcription ran.
"""
if tool_registry is None:
try:
from tools.tool_registry import registry
registry.ensure_discovered()
tool_registry = registry
except Exception:
pass
reviewed_files: list[dict[str, Any]] = []
all_implications: list[str] = []
summaries: list[str] = []
for file_path in files:
media_type = detect_media_type(file_path)
if media_type is None:
logger.warning("Skipping unrecognized file type: %s", file_path)
continue
if not file_path.exists():
logger.warning("File does not exist: %s", file_path)
continue
entry: dict[str, Any] = {
"path": str(file_path),
"media_type": media_type,
"reviewed": True,
}
# Probe based on media type
if media_type == "video":
probe_data = _probe_video(file_path, tool_registry)
elif media_type == "audio":
probe_data = _probe_audio(file_path, tool_registry)
else:
probe_data = _probe_image(file_path)
entry["technical_probe"] = probe_data.get("technical_probe", {})
entry["quality_risks"] = probe_data.get("quality_risks", [])
entry["representative_frames"] = probe_data.get("representative_frames", [])
# Attempt transcription for audio/video
transcript = _transcribe_if_available(file_path, media_type, tool_registry)
if transcript:
entry["transcript_summary"] = transcript
# Build content summary
probe = entry["technical_probe"]
if media_type == "video":
dur = probe.get("duration_seconds", 0)
res = probe.get("resolution", "unknown")
has_audio = bool(probe.get("audio_codec"))
entry["content_summary"] = (
f"Video file: {dur:.1f}s at {res}, "
f"{'with' if has_audio else 'without'} audio"
)
entry["usable_for"] = _infer_video_usability(probe, transcript)
elif media_type == "audio":
dur = probe.get("duration_seconds", 0)
entry["content_summary"] = f"Audio file: {dur:.1f}s, {probe.get('audio_codec', 'unknown')}"
entry["usable_for"] = _infer_audio_usability(probe, transcript)
else:
res = probe.get("resolution", "unknown")
entry["content_summary"] = f"Image file: {res}"
entry["usable_for"] = ["visual asset", "reference image"]
summaries.append(f"{file_path.name}: {entry['content_summary']}")
reviewed_files.append(entry)
# Derive planning implications from quality risks
for risk in entry.get("quality_risks", []):
all_implications.append(f"Quality risk in {file_path.name}: {risk}")
# Build overall summary
if not reviewed_files:
summary = "No user-supplied media files could be reviewed."
all_implications.append("No source media available — production is fully generated.")
else:
summary = "; ".join(summaries)
# Add media-type implications
has_video = any(f["media_type"] == "video" for f in reviewed_files)
has_audio = any(f["media_type"] == "audio" for f in reviewed_files)
has_images = any(f["media_type"] == "image" for f in reviewed_files)
if has_video:
all_implications.append("Source video available — consider source-led or hybrid production approach")
if has_audio and not has_video:
all_implications.append("Audio-only source — production needs visual assets to accompany audio")
if has_images and not has_video:
all_implications.append("Image-only source — motion must come from animation or video generation")
if not all_implications:
all_implications.append("No specific constraints identified from source media.")
return {
"version": "1.0",
"files": reviewed_files,
"summary": summary,
"planning_implications": all_implications,
}
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _parse_fps(fps_str: str) -> float:
"""Parse ffprobe fps string like '30/1' or '24000/1001'."""
try:
if "/" in fps_str:
num, den = fps_str.split("/")
return round(int(num) / max(int(den), 1), 2)
return float(fps_str)
except (ValueError, ZeroDivisionError):
return 0.0
def _sample_timestamps(duration: float, count: int = 4) -> list[float]:
"""Generate evenly-spaced sample timestamps for a given duration."""
if duration <= 0:
return [0.0]
if count <= 1:
return [duration / 2]
step = duration / (count + 1)
return [round(step * (i + 1), 2) for i in range(count)]
def _infer_video_usability(probe: dict, transcript: Optional[str]) -> list[str]:
"""Infer what a video file can be used for."""
uses = []
dur = probe.get("duration_seconds", 0)
if dur > 10:
uses.append("hero footage")
if dur > 3:
uses.append("b-roll")
if transcript:
uses.append("source dialogue")
if probe.get("audio_codec"):
uses.append("source audio")
return uses or ["short clip"]
def _infer_audio_usability(probe: dict, transcript: Optional[str]) -> list[str]:
"""Infer what an audio file can be used for."""
uses = []
dur = probe.get("duration_seconds", 0)
if transcript:
uses.append("narration source")
if dur > 30:
uses.append("background music candidate")
if dur > 5:
uses.append("sound effect or ambient")
return uses or ["audio clip"]
def has_user_media(project_dir: Path) -> bool:
"""Check if a project directory contains user-supplied media files."""
if not project_dir.exists():
return False
for ext_set in (_VIDEO_EXTENSIONS, _AUDIO_EXTENSIONS, _IMAGE_EXTENSIONS):
for ext in ext_set:
if list(project_dir.glob(f"*{ext}")):
return True
return False
lib/variation_checker.py
+169
View File
@@ -0,0 +1,169 @@
"""Scene plan variation checker.
Analyzes a scene plan for repetitive patterns that make videos feel
like slideshows. Catches problems before asset generation begins.
This is a structural check, not a creative judgment — it flags concrete
patterns that reliably produce generic-feeling output.
"""
from __future__ import annotations
from collections import Counter
from typing import Any
# Generic language patterns that signal lazy scene descriptions
GENERIC_PHRASES = {
"a person", "a beautiful", "modern", "futuristic", "cutting-edge",
"in today's world", "sleek design", "innovative", "state-of-the-art",
"next-generation", "revolutionary", "a professional", "dynamic",
"vibrant", "stunning", "breathtaking", "amazing", "incredible",
"powerful", "seamless", "elegant solution",
}
def check_scene_variation(scenes: list[dict[str, Any]]) -> dict[str, Any]:
"""Analyze a scene plan for repetitive patterns.
Returns:
{
"score": float (0-5, lower is better),
"verdict": "strong" | "acceptable" | "revise" | "fail",
"violations": list of specific issues,
"suggestions": list of improvement suggestions,
}
"""
if not scenes:
return {"score": 5.0, "verdict": "fail", "violations": ["No scenes to check"], "suggestions": []}
violations: list[str] = []
suggestions: list[str] = []
# --- Check 1: Shot size variety ---
shot_sizes = [
s.get("shot_language", {}).get("shot_size", "unspecified")
for s in scenes
]
size_counts = Counter(shot_sizes)
if len(scenes) >= 4:
most_common_size, most_common_count = size_counts.most_common(1)[0]
if most_common_count / len(scenes) > 0.5:
violations.append(
f"Shot size '{most_common_size}' used in {most_common_count}/{len(scenes)} scenes "
f"({most_common_count/len(scenes):.0%}). Vary shot sizes for visual interest."
)
suggestions.append("Mix wide establishing shots with close-ups for visual rhythm.")
# --- Check 2: Consecutive same-size shots ---
consecutive_same = 0
for i in range(1, len(shot_sizes)):
if shot_sizes[i] == shot_sizes[i-1] and shot_sizes[i] != "unspecified":
consecutive_same += 1
if consecutive_same >= 3:
violations.append(
f"{consecutive_same} consecutive same-size shots. "
f"Vary shot sizes between scenes for editorial rhythm."
)
# --- Check 3: Static shot overuse ---
movements = [
s.get("shot_language", {}).get("camera_movement", "unspecified")
for s in scenes
]
static_count = sum(1 for m in movements if m in ("static", "unspecified"))
if len(scenes) >= 4 and static_count / len(scenes) > 0.6:
violations.append(
f"{static_count}/{len(scenes)} scenes are static or unspecified movement. "
f"Add intentional camera movement to at least 40% of scenes."
)
suggestions.append("Consider dolly_in for emphasis, tracking for energy, or crane for scale.")
# --- Check 4: Lighting variety ---
lightings = {
s.get("shot_language", {}).get("lighting_key")
for s in scenes
if s.get("shot_language", {}).get("lighting_key")
}
if len(scenes) >= 4 and len(lightings) <= 1:
violations.append(
f"Only {len(lightings)} unique lighting setup(s) across {len(scenes)} scenes. "
f"Vary lighting to create mood shifts."
)
# --- Check 5: Hero moment exists and is visually distinct ---
hero_scenes = [s for s in scenes if s.get("hero_moment")]
if len(scenes) >= 4 and not hero_scenes:
violations.append(
"No hero_moment flagged. Every video should have at least one visual peak."
)
suggestions.append("Mark the most impactful scene as hero_moment=true.")
if hero_scenes:
for hero in hero_scenes:
hero_idx = scenes.index(hero)
hero_size = hero.get("shot_language", {}).get("shot_size")
# Check neighbors
for offset in (-1, 1):
neighbor_idx = hero_idx + offset
if 0 <= neighbor_idx < len(scenes):
neighbor_size = scenes[neighbor_idx].get("shot_language", {}).get("shot_size")
if hero_size and neighbor_size and hero_size == neighbor_size:
violations.append(
f"Hero scene '{hero.get('id')}' has same shot size as neighbor. "
f"Hero moments should be visually distinct from surrounding scenes."
)
# --- Check 6: Description specificity ---
generic_count = 0
for scene in scenes:
desc = scene.get("description", "").lower()
for phrase in GENERIC_PHRASES:
if phrase in desc:
generic_count += 1
break
if generic_count >= len(scenes) * 0.3:
violations.append(
f"{generic_count}/{len(scenes)} scenes use generic language. "
f"Replace vague descriptions with specific visual details."
)
suggestions.append(
"Instead of 'a beautiful cityscape', try 'rain-slicked Tokyo intersection "
"at night, neon reflections in puddles, pedestrians with translucent umbrellas'."
)
# --- Check 7: Texture keywords presence ---
textured = sum(1 for s in scenes if s.get("texture_keywords"))
if len(scenes) >= 4 and textured < len(scenes) * 0.3:
violations.append(
f"Only {textured}/{len(scenes)} scenes have texture_keywords. "
f"Add texture descriptors to visual scenes for richer generation prompts."
)
# --- Check 8: Shot intent completeness ---
intented = sum(1 for s in scenes if s.get("shot_intent"))
if len(scenes) >= 4 and intented < len(scenes) * 0.5:
violations.append(
f"Only {intented}/{len(scenes)} scenes have shot_intent. "
f"Every scene should explain WHY it exists in the video."
)
# --- Score ---
# Each violation category adds ~0.6 to score
score = min(5.0, len(violations) * 0.6)
if score < 2.0:
verdict = "strong"
elif score < 3.0:
verdict = "acceptable"
elif score < 4.0:
verdict = "revise"
else:
verdict = "fail"
return {
"score": round(score, 1),
"verdict": verdict,
"violations": violations,
"suggestions": suggestions,
}