alembic/versions/013_add_recipe_tables.py

@@ -0,0 +1,64 @@
+"""Phase 2 - 配方复用：recipes + recipe_items
+
+Revision ID: 013
+Revises: 012
+Create Date: 2026-06-29
+
+This migration creates two new tables:
+1. recipes — 配方主表
+2. recipe_items — 配方素材项表
+"""
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers
+revision = "013"
+down_revision = "012"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+
+    # ── 1. Create recipes table ──
+
+    conn.execute(sa.text("""
+        CREATE TABLE IF NOT EXISTS recipes (
+            id VARCHAR(36) PRIMARY KEY,
+            user_id VARCHAR(36) NOT NULL,
+            name VARCHAR(200) NOT NULL,
+            description TEXT NOT NULL DEFAULT '',
+            template_id VARCHAR(36) NOT NULL DEFAULT '',
+            generation_params JSONB NOT NULL DEFAULT '{}',
+            is_active BOOLEAN NOT NULL DEFAULT TRUE,
+            metadata JSONB NOT NULL DEFAULT '{}',
+            created_at TIMESTAMP NOT NULL DEFAULT NOW(),
+            updated_at TIMESTAMP NOT NULL DEFAULT NOW()
+        )
+    """))
+    conn.execute(sa.text(
+        "CREATE INDEX IF NOT EXISTS ix_recipes_user_id ON recipes(user_id)"
+    ))
+
+    # ── 2. Create recipe_items table ──
+
+    conn.execute(sa.text("""
+        CREATE TABLE IF NOT EXISTS recipe_items (
+            id VARCHAR(36) PRIMARY KEY,
+            recipe_id VARCHAR(36) NOT NULL,
+            item_type VARCHAR(20) NOT NULL,
+            item_id VARCHAR(36) NOT NULL,
+            position INTEGER NOT NULL DEFAULT 0,
+            metadata JSONB NOT NULL DEFAULT '{}'
+        )
+    """))
+    conn.execute(sa.text(
+        "CREATE INDEX IF NOT EXISTS ix_recipe_items_recipe_id ON recipe_items(recipe_id)"
+    ))
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    conn.execute(sa.text("DROP TABLE IF EXISTS recipe_items"))
+    conn.execute(sa.text("DROP TABLE IF EXISTS recipes"))

apps/api/app/api/router.py

@@ -6,6 +6,7 @@ from app.api.routes.chunked_upload import router as chunked_upload_router
 from app.api.routes.classification_jobs import router as classification_jobs_router
 from app.api.routes.duplication import router as duplication_router
 from app.api.routes.generated_videos import router as generated_videos_router
+from app.api.routes.recipes import router as recipes_router
 from app.api.routes.subscription import router as subscription_router
 from app.api.routes.titles import router as titles_router
 from app.api.routes.voices import router as voices_router
@@ -98,3 +99,8 @@ api_router.include_router(
     prefix="/subscription",
     tags=["Subscription"],
 )
+api_router.include_router(
+    recipes_router,
+    prefix="/recipes",
+    tags=["Recipe"],
+)

apps/api/app/api/routes/recipes.py

@@ -0,0 +1,216 @@
+"""Recipe CRUD + use routes."""
+from __future__ import annotations
+
+from typing import List
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
+from sqlalchemy.orm import Session
+
+from app.auth import AuthenticatedUser, get_current_user
+from app.dependencies import get_db_session, get_user_repository
+from app.schemas.recipe import (
+    CreateRecipeRequest,
+    ListRecipesResponse,
+    RecipeItemResponse,
+    RecipeResponse,
+    UpdateRecipeRequest,
+    UseRecipeResponse,
+)
+from packages.adapters.sqlalchemy_impl.recipe_repository import SQLAlchemyRecipeRepository
+from packages.application.recipe.commands import (
+    CreateRecipeCommand,
+    RecipeItemCommand,
+    UpdateRecipeCommand,
+)
+from packages.application.recipe.use_cases import (
+    CreateRecipeUseCase,
+    DeleteRecipeUseCase,
+    FeatureDisabledError,
+    GetRecipeUseCase,
+    ListRecipesUseCase,
+    NotFoundError,
+    UpdateRecipeUseCase,
+    UseRecipeUseCase,
+)
+from packages.ports.user_repository import UserRepository
+
+router = APIRouter()
+
+
+def _get_recipe_repository(session: Session = Depends(get_db_session)) -> SQLAlchemyRecipeRepository:
+    return SQLAlchemyRecipeRepository(session)
+
+
+def _get_user_plan(user_id: str, user_repository: UserRepository) -> str:
+    user = user_repository.find_by_id(user_id)
+    if user is None:
+        return "free"
+    return getattr(user, "subscription_plan", "free") or "free"
+
+
+def _item_to_response(item) -> RecipeItemResponse:
+    return RecipeItemResponse(
+        id=item.id,
+        recipe_id=item.recipe_id,
+        item_type=item.item_type,
+        item_id=item.item_id,
+        position=item.position,
+        metadata=item.metadata_,
+    )
+
+
+def _to_response(recipe) -> RecipeResponse:
+    return RecipeResponse(
+        id=recipe.id,
+        user_id=recipe.user_id,
+        name=recipe.name,
+        description=recipe.description,
+        template_id=recipe.template_id,
+        generation_params=recipe.generation_params,
+        items=[_item_to_response(i) for i in getattr(recipe, "items", [])],
+        is_active=recipe.is_active,
+        metadata=recipe.metadata_,
+        created_at=recipe.created_at,
+        updated_at=recipe.updated_at,
+    )
+
+
+@router.get("", response_model=ListRecipesResponse)
+def list_recipes(
+    skip: int = Query(0, ge=0),
+    limit: int = Query(50, ge=1, le=200),
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    recipe_repository: SQLAlchemyRecipeRepository = Depends(_get_recipe_repository),
+) -> ListRecipesResponse:
+    user_id = authenticated_user.user.id
+    use_case = ListRecipesUseCase(recipe_repository)
+    recipes = use_case.execute(user_id, skip=skip, limit=limit)
+    total = recipe_repository.count_by_user(user_id)
+    return ListRecipesResponse(
+        items=[_to_response(r) for r in recipes],
+        total=total,
+    )
+
+
+@router.get("/{recipe_id}", response_model=RecipeResponse)
+def get_recipe(
+    recipe_id: str,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    recipe_repository: SQLAlchemyRecipeRepository = Depends(_get_recipe_repository),
+) -> RecipeResponse:
+    user_id = authenticated_user.user.id
+    use_case = GetRecipeUseCase(recipe_repository)
+    recipe = use_case.execute(recipe_id, user_id)
+    if recipe is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Recipe not found")
+    return _to_response(recipe)
+
+
+@router.post("", response_model=RecipeResponse, status_code=status.HTTP_201_CREATED)
+def create_recipe(
+    request: CreateRecipeRequest,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    recipe_repository: SQLAlchemyRecipeRepository = Depends(_get_recipe_repository),
+) -> RecipeResponse:
+    user_id = authenticated_user.user.id
+    command = CreateRecipeCommand(
+        user_id=user_id,
+        name=request.name,
+        description=request.description,
+        template_id=request.template_id,
+        generation_params=request.generation_params,
+        items=[
+            RecipeItemCommand(
+                item_type=ic.item_type,
+                item_id=ic.item_id,
+                position=ic.position,
+                metadata_=ic.metadata_,
+            )
+            for ic in request.items
+        ],
+        metadata_=request.metadata_,
+    )
+    use_case = CreateRecipeUseCase(recipe_repository)
+    recipe = use_case.execute(command)
+    return _to_response(recipe)
+
+
+@router.patch("/{recipe_id}", response_model=RecipeResponse)
+def update_recipe(
+    recipe_id: str,
+    request: UpdateRecipeRequest,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    recipe_repository: SQLAlchemyRecipeRepository = Depends(_get_recipe_repository),
+) -> RecipeResponse:
+    user_id = authenticated_user.user.id
+    command = UpdateRecipeCommand(
+        recipe_id=recipe_id,
+        user_id=user_id,
+        name=request.name,
+        description=request.description,
+        template_id=request.template_id,
+        generation_params=request.generation_params,
+        items=(
+            [
+                RecipeItemCommand(
+                    item_type=ic.item_type,
+                    item_id=ic.item_id,
+                    position=ic.position,
+                    metadata_=ic.metadata_,
+                )
+                for ic in request.items
+            ]
+            if request.items is not None
+            else None
+        ),
+        metadata_=request.metadata_,
+    )
+    use_case = UpdateRecipeUseCase(recipe_repository)
+    try:
+        recipe = use_case.execute(command)
+    except NotFoundError:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Recipe not found")
+    return _to_response(recipe)
+
+
+@router.delete("/{recipe_id}", status_code=status.HTTP_204_NO_CONTENT, response_class=Response)
+def delete_recipe(
+    recipe_id: str,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    recipe_repository: SQLAlchemyRecipeRepository = Depends(_get_recipe_repository),
+) -> Response:
+    user_id = authenticated_user.user.id
+    use_case = DeleteRecipeUseCase(recipe_repository)
+    deleted = use_case.execute(recipe_id, user_id)
+    if not deleted:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Recipe not found")
+    return Response(status_code=204)
+
+
+@router.post("/{recipe_id}/use", response_model=UseRecipeResponse)
+def use_recipe(
+    recipe_id: str,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    recipe_repository: SQLAlchemyRecipeRepository = Depends(_get_recipe_repository),
+    user_repository: UserRepository = Depends(get_user_repository),
+) -> UseRecipeResponse:
+    user_id = authenticated_user.user.id
+    plan_name = _get_user_plan(user_id, user_repository)
+    use_case = UseRecipeUseCase(recipe_repository)
+    try:
+        result = use_case.execute(recipe_id, user_id, user_plan=plan_name)
+    except FeatureDisabledError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail=str(exc),
+        )
+    except NotFoundError:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Recipe not found")
+
+    return UseRecipeResponse(
+        recipe=_to_response(result.recipe),
+        warnings=[
+            {"item_type": w.item_type, "item_id": w.item_id, "position": w.position}
+            for w in result.warnings
+        ],
+    )

apps/api/app/schemas/recipe.py

@@ -0,0 +1,84 @@
+"""Recipe API schemas."""
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any, Dict, List, Optional
+
+from pydantic import BaseModel, Field
+
+
+# ── Response ──
+
+class RecipeItemResponse(BaseModel):
+    id: str
+    recipe_id: str
+    item_type: str
+    item_id: str
+    position: int
+    metadata_: Dict[str, Any] = Field(default_factory=dict, alias="metadata")
+
+    class Config:
+        populate_by_name = True
+
+
+class RecipeResponse(BaseModel):
+    id: str
+    user_id: str
+    name: str
+    description: str = ""
+    template_id: str = ""
+    generation_params: Dict[str, Any] = Field(default_factory=dict)
+    items: List[RecipeItemResponse] = Field(default_factory=list)
+    is_active: bool = True
+    metadata_: Dict[str, Any] = Field(default_factory=dict, alias="metadata")
+    created_at: datetime
+    updated_at: datetime
+
+    class Config:
+        populate_by_name = True
+
+
+class ListRecipesResponse(BaseModel):
+    items: List[RecipeResponse]
+    total: int = 0
+
+
+class UseRecipeResponse(BaseModel):
+    recipe: RecipeResponse
+    warnings: List[Dict[str, Any]] = Field(default_factory=list)
+
+
+# ── Request ──
+
+class RecipeItemRequest(BaseModel):
+    item_type: str
+    item_id: str
+    position: int = 0
+    metadata_: Dict[str, Any] = Field(default_factory=dict, alias="metadata")
+
+    class Config:
+        populate_by_name = True
+
+
+class CreateRecipeRequest(BaseModel):
+    name: str
+    description: str = ""
+    template_id: str = ""
+    generation_params: Dict[str, Any] = Field(default_factory=dict)
+    items: List[RecipeItemRequest] = Field(default_factory=list)
+    metadata_: Dict[str, Any] = Field(default_factory=dict, alias="metadata")
+
+    class Config:
+        populate_by_name = True
+
+
+class UpdateRecipeRequest(BaseModel):
+    name: Optional[str] = None
+    description: Optional[str] = None
+    template_id: Optional[str] = None
+    generation_params: Optional[Dict[str, Any]] = None
+    items: Optional[List[RecipeItemRequest]] = None
+    metadata_: Optional[Dict[str, Any]] = Field(default=None, alias="metadata")
+
+    class Config:
+        populate_by_name = True

packages/adapters/sqlalchemy_impl/models.py

@@ -254,3 +254,29 @@ class DuplicationSegmentModel(Base):
     matched_end = Column(Float, nullable=False)
     similarity = Column(Float, nullable=False)
 
+
+class RecipeModel(Base):
+    __tablename__ = "recipes"
+
+    id = Column(String(36), primary_key=True)
+    user_id = Column(String(36), nullable=False, index=True)
+    name = Column(String(200), nullable=False)
+    description = Column(Text, nullable=False, default="")
+    template_id = Column(String(36), nullable=False, default="")
+    generation_params = Column(JSON, nullable=False, default=dict)
+    is_active = Column(Boolean, nullable=False, default=True)
+    extra_meta = Column('metadata', JSON, nullable=False, default=dict)
+    created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
+    updated_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
+
+
+class RecipeItemModel(Base):
+    __tablename__ = "recipe_items"
+
+    id = Column(String(36), primary_key=True)
+    recipe_id = Column(String(36), nullable=False, index=True)
+    item_type = Column(String(20), nullable=False)
+    item_id = Column(String(36), nullable=False)
+    position = Column(Integer, nullable=False, default=0)
+    extra_meta = Column('metadata', JSON, nullable=False, default=dict)
+

packages/adapters/sqlalchemy_impl/recipe_repository.py

@@ -0,0 +1,179 @@
+"""SQLAlchemy implementation of RecipeRepository."""
+from __future__ import annotations
+
+from typing import List, Optional
+
+from sqlalchemy.orm import Session
+
+from packages.adapters.sqlalchemy_impl.models import RecipeModel, RecipeItemModel
+from packages.domain.recipe import Recipe, RecipeItem
+
+
+class SQLAlchemyRecipeRepository:
+    """SQLAlchemy 配方仓储"""
+
+    def __init__(self, session: Session) -> None:
+        self.session = session
+
+    def list_by_user(
+        self,
+        user_id: str,
+        *,
+        skip: int = 0,
+        limit: int = 50,
+    ) -> List[Recipe]:
+        models = (
+            self.session.query(RecipeModel)
+            .filter(
+                RecipeModel.user_id == user_id,
+                RecipeModel.is_active == True,
+            )
+            .order_by(RecipeModel.created_at.desc())
+            .offset(skip)
+            .limit(limit)
+            .all()
+        )
+        recipes = [self._model_to_entity(m) for m in models]
+        # Load items for each recipe
+        for recipe in recipes:
+            recipe.items = self.list_items(recipe.id)
+        return recipes
+
+    def get(self, recipe_id: str, user_id: str) -> Optional[Recipe]:
+        model = (
+            self.session.query(RecipeModel)
+            .filter(
+                RecipeModel.id == recipe_id,
+                RecipeModel.user_id == user_id,
+            )
+            .first()
+        )
+        if model is None:
+            return None
+        recipe = self._model_to_entity(model)
+        recipe.items = self.list_items(recipe.id)
+        return recipe
+
+    def create(self, recipe: Recipe) -> Recipe:
+        model = RecipeModel(
+            id=recipe.id,
+            user_id=recipe.user_id,
+            name=recipe.name,
+            description=recipe.description,
+            template_id=recipe.template_id,
+            generation_params=recipe.generation_params,
+            is_active=recipe.is_active,
+            extra_meta=recipe.metadata_,
+        )
+        self.session.add(model)
+        self.session.commit()
+        self.session.refresh(model)
+        result = self._model_to_entity(model)
+        result.items = recipe.items
+        return result
+
+    def update(self, recipe: Recipe) -> Recipe:
+        model = (
+            self.session.query(RecipeModel)
+            .filter(
+                RecipeModel.id == recipe.id,
+                RecipeModel.user_id == recipe.user_id,
+            )
+            .first()
+        )
+        if model is None:
+            raise ValueError(f"Recipe {recipe.id} not found")
+        model.name = recipe.name
+        model.description = recipe.description
+        model.template_id = recipe.template_id
+        model.generation_params = recipe.generation_params
+        model.is_active = recipe.is_active
+        model.extra_meta = recipe.metadata_
+        self.session.commit()
+        self.session.refresh(model)
+        result = self._model_to_entity(model)
+        result.items = recipe.items
+        return result
+
+    def delete(self, recipe_id: str, user_id: str) -> bool:
+        model = (
+            self.session.query(RecipeModel)
+            .filter(
+                RecipeModel.id == recipe_id,
+                RecipeModel.user_id == user_id,
+            )
+            .first()
+        )
+        if model is None:
+            return False
+        model.is_active = False
+        self.session.commit()
+        return True
+
+    def count_by_user(self, user_id: str, is_active: bool = True) -> int:
+        return (
+            self.session.query(RecipeModel)
+            .filter(
+                RecipeModel.user_id == user_id,
+                RecipeModel.is_active == is_active,
+            )
+            .count()
+        )
+
+    def list_items(self, recipe_id: str) -> List[RecipeItem]:
+        models = (
+            self.session.query(RecipeItemModel)
+            .filter(RecipeItemModel.recipe_id == recipe_id)
+            .order_by(RecipeItemModel.position)
+            .all()
+        )
+        return [self._item_model_to_entity(m) for m in models]
+
+    def create_items(self, items: List[RecipeItem]) -> List[RecipeItem]:
+        for item in items:
+            model = RecipeItemModel(
+                id=item.id,
+                recipe_id=item.recipe_id,
+                item_type=item.item_type,
+                item_id=item.item_id,
+                position=item.position,
+                extra_meta=item.metadata_,
+            )
+            self.session.add(model)
+        self.session.commit()
+        return items
+
+    def delete_items_by_recipe(self, recipe_id: str) -> int:
+        count = (
+            self.session.query(RecipeItemModel)
+            .filter(RecipeItemModel.recipe_id == recipe_id)
+            .delete()
+        )
+        self.session.commit()
+        return count
+
+    @staticmethod
+    def _model_to_entity(model: RecipeModel) -> Recipe:
+        return Recipe(
+            id=model.id,
+            user_id=model.user_id,
+            name=model.name,
+            description=model.description or "",
+            template_id=model.template_id or "",
+            generation_params=model.generation_params or {},
+            is_active=model.is_active,
+            metadata_=model.extra_meta or {},
+            created_at=model.created_at,
+            updated_at=model.updated_at,
+        )
+
+    @staticmethod
+    def _item_model_to_entity(model: RecipeItemModel) -> RecipeItem:
+        return RecipeItem(
+            id=model.id,
+            recipe_id=model.recipe_id,
+            item_type=model.item_type,
+            item_id=model.item_id,
+            position=model.position or 0,
+            metadata_=model.extra_meta or {},
+        )

packages/application/recipe/commands.py

@@ -0,0 +1,36 @@
+"""Recipe commands."""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import List, Optional
+
+
+@dataclass
+class RecipeItemCommand:
+    item_type: str
+    item_id: str
+    position: int = 0
+    metadata_: dict = field(default_factory=dict)
+
+
+@dataclass
+class CreateRecipeCommand:
+    user_id: str
+    name: str
+    description: str = ""
+    template_id: str = ""
+    generation_params: dict = field(default_factory=dict)
+    items: List[RecipeItemCommand] = field(default_factory=list)
+    metadata_: dict = field(default_factory=dict)
+
+
+@dataclass
+class UpdateRecipeCommand:
+    recipe_id: str
+    user_id: str
+    name: Optional[str] = None
+    description: Optional[str] = None
+    template_id: Optional[str] = None
+    generation_params: Optional[dict] = None
+    items: Optional[List[RecipeItemCommand]] = None
+    metadata_: Optional[dict] = None

packages/application/recipe/use_cases.py

@@ -0,0 +1,184 @@
+"""Recipe use cases."""
+from __future__ import annotations
+
+import uuid
+from dataclasses import dataclass
+from typing import List, Optional
+
+from packages.adapters.sqlalchemy_impl.recipe_repository import SQLAlchemyRecipeRepository
+from packages.application.recipe.commands import (
+    CreateRecipeCommand,
+    RecipeItemCommand,
+    UpdateRecipeCommand,
+)
+from packages.domain.recipe import Recipe, RecipeItem
+from packages.infrastructure.feature_flags import FeatureScope, feature_flags
+
+
+class NotFoundError(Exception):
+    pass
+
+
+class FeatureDisabledError(Exception):
+    pass
+
+
+@dataclass
+class MissingAssetWarning:
+    """使用配方时缺失的素材警告"""
+    item_type: str
+    item_id: str
+    position: int
+
+
+class CreateRecipeUseCase:
+    def __init__(self, repository: SQLAlchemyRecipeRepository) -> None:
+        self.repository = repository
+
+    def execute(self, command: CreateRecipeCommand) -> Recipe:
+        recipe_id = uuid.uuid4().hex
+        recipe = Recipe(
+            id=recipe_id,
+            user_id=command.user_id,
+            name=command.name,
+            description=command.description,
+            template_id=command.template_id,
+            generation_params=command.generation_params,
+            metadata_=command.metadata_,
+        )
+        recipe = self.repository.create(recipe)
+
+        # Create items
+        if command.items:
+            items = [
+                RecipeItem(
+                    id=uuid.uuid4().hex,
+                    recipe_id=recipe.id,
+                    item_type=ic.item_type,
+                    item_id=ic.item_id,
+                    position=ic.position,
+                    metadata_=ic.metadata_,
+                )
+                for ic in command.items
+            ]
+            self.repository.create_items(items)
+            recipe.items = items
+
+        return recipe
+
+
+class ListRecipesUseCase:
+    def __init__(self, repository: SQLAlchemyRecipeRepository) -> None:
+        self.repository = repository
+
+    def execute(
+        self,
+        user_id: str,
+        *,
+        skip: int = 0,
+        limit: int = 50,
+    ) -> List[Recipe]:
+        return self.repository.list_by_user(user_id, skip=skip, limit=limit)
+
+
+class GetRecipeUseCase:
+    def __init__(self, repository: SQLAlchemyRecipeRepository) -> None:
+        self.repository = repository
+
+    def execute(self, recipe_id: str, user_id: str) -> Optional[Recipe]:
+        return self.repository.get(recipe_id, user_id)
+
+
+class UpdateRecipeUseCase:
+    def __init__(self, repository: SQLAlchemyRecipeRepository) -> None:
+        self.repository = repository
+
+    def execute(self, command: UpdateRecipeCommand) -> Recipe:
+        existing = self.repository.get(command.recipe_id, command.user_id)
+        if existing is None:
+            raise NotFoundError(f"Recipe {command.recipe_id} not found")
+
+        if command.name is not None:
+            existing.name = command.name
+        if command.description is not None:
+            existing.description = command.description
+        if command.template_id is not None:
+            existing.template_id = command.template_id
+        if command.generation_params is not None:
+            existing.generation_params = command.generation_params
+        if command.metadata_ is not None:
+            existing.metadata_ = command.metadata_
+
+        self.repository.update(existing)
+
+        # Replace items if provided
+        if command.items is not None:
+            self.repository.delete_items_by_recipe(existing.id)
+            items = [
+                RecipeItem(
+                    id=uuid.uuid4().hex,
+                    recipe_id=existing.id,
+                    item_type=ic.item_type,
+                    item_id=ic.item_id,
+                    position=ic.position,
+                    metadata_=ic.metadata_,
+                )
+                for ic in command.items
+            ]
+            self.repository.create_items(items)
+            existing.items = items
+        else:
+            existing.items = self.repository.list_items(existing.id)
+
+        return existing
+
+
+class DeleteRecipeUseCase:
+    def __init__(self, repository: SQLAlchemyRecipeRepository) -> None:
+        self.repository = repository
+
+    def execute(self, recipe_id: str, user_id: str) -> bool:
+        return self.repository.delete(recipe_id, user_id)
+
+
+@dataclass
+class UseRecipeResult:
+    """使用配方的结果"""
+    recipe: Recipe
+    warnings: List[MissingAssetWarning]
+
+
+class UseRecipeUseCase:
+    """使用配方 — 校验 Feature Flag + 检查素材可用性"""
+
+    def __init__(self, repository: SQLAlchemyRecipeRepository) -> None:
+        self.repository = repository
+
+    def execute(
+        self,
+        recipe_id: str,
+        user_id: str,
+        *,
+        user_plan: str = "free",
+    ) -> UseRecipeResult:
+        # 1. 校验 Feature Flag（仅 basic/premium 可用）
+        if not feature_flags.is_enabled(
+            FeatureScope.RECIPE_REUSE,
+            user_plan=user_plan,
+        ):
+            raise FeatureDisabledError(
+                "配方复用功能仅对基础版和高级版用户开放"
+            )
+
+        # 2. 获取配方
+        recipe = self.repository.get(recipe_id, user_id)
+        if recipe is None:
+            raise NotFoundError(f"Recipe {recipe_id} not found")
+
+        # 3. 校验引用的素材/标题/配音是否仍存在
+        warnings: List[MissingAssetWarning] = []
+        # Note: 实际项目中这里需要注入 asset/title/voice repository
+        # 来校验每个 item 是否仍然存在。当前版本返回空警告列表，
+        # 由调用方（路由层）决定是否传入额外的校验逻辑。
+
+        return UseRecipeResult(recipe=recipe, warnings=warnings)

packages/domain/recipe.py

@@ -0,0 +1,33 @@
+"""Recipe domain entities."""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import List
+
+
+@dataclass
+class RecipeItem:
+    """配方中的单个素材/标题/配音项"""
+    id: str
+    recipe_id: str
+    item_type: str  # asset / title / voice
+    item_id: str
+    position: int = 0
+    metadata_: dict = field(default_factory=dict)
+
+
+@dataclass
+class Recipe:
+    """配方 — 一次「一键生成」的完整参数组合"""
+    id: str
+    user_id: str
+    name: str
+    description: str = ""
+    template_id: str = ""
+    generation_params: dict = field(default_factory=dict)
+    items: List[RecipeItem] = field(default_factory=list)
+    is_active: bool = True
+    metadata_: dict = field(default_factory=dict)
+    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
+    updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

packages/infrastructure/feature_flags.py

@@ -112,6 +112,7 @@ class FeatureFlags:
                 name=FeatureScope.RECIPE_REUSE,
                 description="配方复用功能",
                 global_enabled=True,
+                plan_overrides={"free": False},  # 仅基础版和高级版可用
             ),
         ]
         for flag in defaults:

packages/ports/recipe_repository.py

@@ -0,0 +1,43 @@
+"""Recipe repository port."""
+from __future__ import annotations
+
+from typing import List, Optional, Protocol
+
+from packages.domain.recipe import Recipe, RecipeItem
+
+
+class RecipeRepository(Protocol):
+    """配方仓储接口"""
+
+    def list_by_user(
+        self,
+        user_id: str,
+        *,
+        skip: int = 0,
+        limit: int = 50,
+    ) -> List[Recipe]:
+        ...
+
+    def get(self, recipe_id: str, user_id: str) -> Optional[Recipe]:
+        ...
+
+    def create(self, recipe: Recipe) -> Recipe:
+        ...
+
+    def update(self, recipe: Recipe) -> Recipe:
+        ...
+
+    def delete(self, recipe_id: str, user_id: str) -> bool:
+        ...
+
+    def count_by_user(self, user_id: str, is_active: bool = True) -> int:
+        ...
+
+    def list_items(self, recipe_id: str) -> List[RecipeItem]:
+        ...
+
+    def create_items(self, items: List[RecipeItem]) -> List[RecipeItem]:
+        ...
+
+    def delete_items_by_recipe(self, recipe_id: str) -> int:
+        ...

tests/unit/test_recipe_use_cases.py

@@ -0,0 +1,210 @@
+"""Recipe use cases unit tests."""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from unittest.mock import Mock
+
+import pytest
+
+from packages.application.recipe.commands import (
+    CreateRecipeCommand,
+    RecipeItemCommand,
+    UpdateRecipeCommand,
+)
+from packages.application.recipe.use_cases import (
+    CreateRecipeUseCase,
+    DeleteRecipeUseCase,
+    FeatureDisabledError,
+    GetRecipeUseCase,
+    ListRecipesUseCase,
+    NotFoundError,
+    UpdateRecipeUseCase,
+    UseRecipeUseCase,
+)
+from packages.domain.recipe import Recipe, RecipeItem
+
+
+def _make_recipe(**kwargs) -> Recipe:
+    defaults = dict(
+        id="recipe001",
+        user_id="user001",
+        name="测试配方",
+        description="描述",
+        template_id="tpl001",
+        generation_params={"mode": "one_take"},
+        items=[],
+        is_active=True,
+        metadata_={},
+        created_at=datetime.now(timezone.utc),
+        updated_at=datetime.now(timezone.utc),
+    )
+    defaults.update(kwargs)
+    return Recipe(**defaults)
+
+
+def _make_item(**kwargs) -> RecipeItem:
+    defaults = dict(
+        id="item001",
+        recipe_id="recipe001",
+        item_type="asset",
+        item_id="asset001",
+        position=0,
+        metadata_={},
+    )
+    defaults.update(kwargs)
+    return RecipeItem(**defaults)
+
+
+class TestCreateRecipeUseCase:
+    @pytest.fixture
+    def mock_repo(self):
+        repo = Mock()
+        repo.create = Mock(side_effect=lambda r: r)
+        repo.create_items = Mock(side_effect=lambda items: items)
+        return repo
+
+    def test_create_basic(self, mock_repo):
+        uc = CreateRecipeUseCase(mock_repo)
+        cmd = CreateRecipeCommand(
+            user_id="user001",
+            name="我的配方",
+            description="desc",
+            template_id="tpl001",
+            generation_params={"mode": "one_take"},
+        )
+        result = uc.execute(cmd)
+        assert result.name == "我的配方"
+        assert result.user_id == "user001"
+        mock_repo.create.assert_called_once()
+
+    def test_create_with_items(self, mock_repo):
+        uc = CreateRecipeUseCase(mock_repo)
+        cmd = CreateRecipeCommand(
+            user_id="user001",
+            name="带素材配方",
+            items=[
+                RecipeItemCommand(item_type="asset", item_id="a1", position=0),
+                RecipeItemCommand(item_type="title", item_id="t1", position=1),
+                RecipeItemCommand(item_type="voice", item_id="v1", position=2),
+            ],
+        )
+        result = uc.execute(cmd)
+        assert len(result.items) == 3
+        mock_repo.create_items.assert_called_once()
+        items_arg = mock_repo.create_items.call_args[0][0]
+        assert items_arg[0].item_type == "asset"
+        assert items_arg[1].item_type == "title"
+        assert items_arg[2].item_type == "voice"
+
+
+class TestListRecipesUseCase:
+    def test_list(self):
+        repo = Mock()
+        repo.list_by_user = Mock(return_value=[_make_recipe()])
+        uc = ListRecipesUseCase(repo)
+        result = uc.execute("user001", skip=0, limit=10)
+        assert len(result) == 1
+        repo.list_by_user.assert_called_once_with("user001", skip=0, limit=10)
+
+
+class TestGetRecipeUseCase:
+    def test_get_found(self):
+        repo = Mock()
+        repo.get = Mock(return_value=_make_recipe())
+        uc = GetRecipeUseCase(repo)
+        result = uc.execute("recipe001", "user001")
+        assert result is not None
+        assert result.id == "recipe001"
+
+    def test_get_not_found(self):
+        repo = Mock()
+        repo.get = Mock(return_value=None)
+        uc = GetRecipeUseCase(repo)
+        result = uc.execute("recipe999", "user001")
+        assert result is None
+
+
+class TestUpdateRecipeUseCase:
+    @pytest.fixture
+    def mock_repo(self):
+        repo = Mock()
+        repo.get = Mock(return_value=_make_recipe())
+        repo.update = Mock(side_effect=lambda r: r)
+        repo.list_items = Mock(return_value=[])
+        repo.delete_items_by_recipe = Mock(return_value=0)
+        repo.create_items = Mock(side_effect=lambda items: items)
+        return repo
+
+    def test_update_name(self, mock_repo):
+        uc = UpdateRecipeUseCase(mock_repo)
+        cmd = UpdateRecipeCommand(
+            recipe_id="recipe001",
+            user_id="user001",
+            name="新名字",
+        )
+        result = uc.execute(cmd)
+        assert result.name == "新名字"
+
+    def test_update_not_found(self):
+        repo = Mock()
+        repo.get = Mock(return_value=None)
+        uc = UpdateRecipeUseCase(repo)
+        cmd = UpdateRecipeCommand(recipe_id="xxx", user_id="user001", name="x")
+        with pytest.raises(NotFoundError):
+            uc.execute(cmd)
+
+    def test_update_replace_items(self, mock_repo):
+        uc = UpdateRecipeUseCase(mock_repo)
+        cmd = UpdateRecipeCommand(
+            recipe_id="recipe001",
+            user_id="user001",
+            items=[RecipeItemCommand(item_type="voice", item_id="v2", position=0)],
+        )
+        result = uc.execute(cmd)
+        mock_repo.delete_items_by_recipe.assert_called_once_with("recipe001")
+        mock_repo.create_items.assert_called_once()
+        assert len(result.items) == 1
+
+
+class TestDeleteRecipeUseCase:
+    def test_delete_success(self):
+        repo = Mock()
+        repo.delete = Mock(return_value=True)
+        uc = DeleteRecipeUseCase(repo)
+        assert uc.execute("recipe001", "user001") is True
+
+    def test_delete_not_found(self):
+        repo = Mock()
+        repo.delete = Mock(return_value=False)
+        uc = DeleteRecipeUseCase(repo)
+        assert uc.execute("recipe999", "user001") is False
+
+
+class TestUseRecipeUseCase:
+    def test_use_success_basic_plan(self):
+        repo = Mock()
+        repo.get = Mock(return_value=_make_recipe())
+        uc = UseRecipeUseCase(repo)
+        result = uc.execute("recipe001", "user001", user_plan="basic")
+        assert result.recipe.id == "recipe001"
+        assert result.warnings == []
+
+    def test_use_success_premium_plan(self):
+        repo = Mock()
+        repo.get = Mock(return_value=_make_recipe())
+        uc = UseRecipeUseCase(repo)
+        result = uc.execute("recipe001", "user001", user_plan="premium")
+        assert result.recipe.id == "recipe001"
+
+    def test_use_free_plan_forbidden(self):
+        repo = Mock()
+        uc = UseRecipeUseCase(repo)
+        with pytest.raises(FeatureDisabledError):
+            uc.execute("recipe001", "user001", user_plan="free")
+
+    def test_use_not_found(self):
+        repo = Mock()
+        repo.get = Mock(return_value=None)
+        uc = UseRecipeUseCase(repo)
+        with pytest.raises(NotFoundError):
+            uc.execute("recipe999", "user001", user_plan="basic")