Merge pull request 'fix: 消除 API↔Worker 跨端依赖（架构解耦）' (#44) from fix/architecture-decouple into develop

apps/api/app/api/routes/edit_plans.py

@@ -16,10 +16,7 @@ from app.schemas.edit_plan import (
     EditPlanResponse,
     EditTemplateResponse,
 )
-from apps.worker.worker_app.tasks.edit_plan_generator import (
-    EditingMode,
-    SmartEditPlanGenerator,
-)
+from packages.domain.edit_plan import EditingMode, SmartEditPlanGenerator
 from fastapi import APIRouter, Depends, HTTPException
 from sqlalchemy.orm import Session
 

apps/worker/tasks.py

@@ -2,9 +2,6 @@ import os
 import tempfile
 from datetime import datetime, timezone
 
-from app.config import get_settings
-from app.core.storage import get_storage_service
-
 from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
 from packages.adapters.sqlalchemy_impl.generated_video_repository import (
     SQLAlchemyGeneratedVideoRepository,
@@ -17,11 +14,13 @@ from packages.adapters.sqlalchemy_impl.session import (
     build_session_factory,
 )
 from packages.domain import GeneratedVideo, GenerationTaskStatus
+from packages.shared.config import get_shared_settings
+from packages.shared.storage import get_storage_service
 
 from .celery_app import celery_app
 from .video_processing import VideoProcessor
 
-settings = get_settings()
+settings = get_shared_settings()
 if SessionLocal is None:
     build_session_factory(settings.database_url)
 

apps/worker/video_processing/dedup.py

@@ -15,13 +15,13 @@ from sqlalchemy.orm import Session
 
 from packages.adapters.sqlalchemy_impl.generated_video_repository import SQLAlchemyGeneratedVideoRepository
 from packages.adapters.sqlalchemy_impl.session import SessionLocal, build_session_factory
-from app.config import get_settings
-from app.core.storage import get_storage_service
+from packages.shared.config import get_shared_settings
+from packages.shared.storage import get_storage_service
 
 from apps.worker.celery_app import celery_app
 
 logger = logging.getLogger(__name__)
-settings = get_settings()
+settings = get_shared_settings()
 if SessionLocal is None:
     build_session_factory(settings.database_url)
 

apps/worker/worker_app/tasks/__init__.py

@@ -1,9 +1,29 @@
-"""Task modules."""
+"""Task modules - lazy loading to avoid Celery dependency at import time."""
+
+from __future__ import annotations
+
+
+def __getattr__(name: str):
+    """Lazy import task functions to avoid loading Celery at module import time."""
+    if name == "classify_asset":
+        from .classification import classify_asset
+        return classify_asset
+    elif name == "generate_video":
+        from .generation import generate_video
+        return generate_video
+    elif name == "healthcheck":
+        from .health import healthcheck
+        return healthcheck
+    elif name == "ingest_asset":
+        from .ingest import ingest_asset
+        return ingest_asset
+    elif name == "extract_voice_task":
+        from .voice_extraction import extract_voice_task
+        return extract_voice_task
+    elif name == "extract_background_task":
+        from .voice_extraction import extract_background_task
+        return extract_background_task
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
 
-from .classification import classify_asset
-from .generation import generate_video
-from .health import healthcheck
-from .ingest import ingest_asset
-from .voice_extraction import extract_voice_task, extract_background_task
 
 __all__ = ["classify_asset", "generate_video", "healthcheck", "ingest_asset", "extract_voice_task", "extract_background_task"]

apps/worker/worker_app/tasks/edit_plan_generator.py

@@ -8,39 +8,11 @@ from enum import StrEnum
 from typing import Any
 
 from packages.domain import Asset, AssetClassification, AssetStatus
+from packages.domain.edit_plan import EditClipPlan, EditPlanResult, EditingMode
 
 logger = logging.getLogger(__name__)
 
 
-class EditingMode(StrEnum):
-    """剪辑模式枚举"""
-    ONE_TAKE = "one_take"   # 顺序拼接模式
-    PIP = "pip"             # 画中画模式
-    VOICE_OVER = "voice_over" # 口播+B-roll模式
-    VOICE_PIP = "voice_pip" # 口播+画中画组合模式
-
-
-@dataclass
-class EditClipPlan:
-    """单个剪辑片段的编排计划"""
-    asset_id: str
-    sequence: int
-    start_time: float = 0.0
-    duration: float = 0.0
-    layer: str = "main"  # main, pip, broll
-    reason: str = ""
-
-
-@dataclass
-class EditPlanResult:
-    """完整剪辑计划结果"""
-    project_id: str
-    editing_mode: EditingMode
-    clips: list[EditClipPlan]
-    total_duration: float
-    summary: str
-
-
 def _calculate_start_times(clips: list[EditClipPlan]) -> list[EditClipPlan]:
     """
     计算时间轴，根据前面的片段时长累加 start_time

apps/worker/worker_app/tasks/voice_extraction.py

@@ -11,13 +11,13 @@ from sqlalchemy.orm import Session
 
 from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
 from packages.adapters.sqlalchemy_impl.session import SessionLocal, build_session_factory
-from app.config import get_settings
-from app.core.storage import get_storage_service
+from packages.shared.config import get_shared_settings
+from packages.shared.storage import get_storage_service
 
 from .celery_app import celery_app
 
 logger = logging.getLogger(__name__)
-settings = get_settings()
+settings = get_shared_settings()
 if SessionLocal is None:
     build_session_factory(settings.database_url)
 

packages/domain/__init__.py

@@ -5,6 +5,11 @@ from .classification import (
     ClassificationJob,
     ClassificationJobStatus,
 )
+from .edit_plan import (
+    EditClipPlan,
+    EditPlanResult,
+    EditingMode,
+)
 from .entities import (
     Asset,
     AssetLibrary,
@@ -15,6 +20,7 @@ from .entities import (
     IngestJobStatus,
     Project,
     User,
+    Workspace,
 )
 from .generated_video import GeneratedVideo
 from .generation_task import GenerationTask, GenerationTaskStatus
@@ -29,6 +35,9 @@ __all__ = [
     "ClassificationJob",
     "ClassificationJobStatus",
     "ClassificationStatus",
+    "EditClipPlan",
+    "EditPlanResult",
+    "EditingMode",
     "GeneratedVideo",
     "GenerationTask",
     "GenerationTaskStatus",
@@ -41,4 +50,5 @@ __all__ = [
     "TaskPriority",
     "TaskStatus",
     "User",
+    "Workspace",
 ]

packages/domain/edit_plan.py

@@ -0,0 +1,33 @@
+"""Edit Plan domain models - shared between API and Worker."""
+
+from dataclasses import dataclass
+from enum import StrEnum
+
+
+class EditingMode(StrEnum):
+    """剪辑模式枚举"""
+    ONE_TAKE = "one_take"   # 顺序拼接模式
+    PIP = "pip"             # 画中画模式
+    VOICE_OVER = "voice_over"  # 口播+B-roll模式
+    VOICE_PIP = "voice_pip"  # 口播+画中画组合模式
+
+
+@dataclass
+class EditClipPlan:
+    """单个剪辑片段的编排计划"""
+    asset_id: str
+    sequence: int
+    start_time: float = 0.0
+    duration: float = 0.0
+    layer: str = "main"  # main, pip, broll
+    reason: str = ""
+
+
+@dataclass
+class EditPlanResult:
+    """完整剪辑计划结果"""
+    project_id: str
+    editing_mode: EditingMode
+    clips: list[EditClipPlan]
+    total_duration: float
+    summary: str

packages/shared/__init__.py

@@ -1 +1,17 @@
-"""Shared package for cross-cutting, non-business-specific utilities."""
+"""Shared utilities for API and Worker services."""
+
+from .config import SharedSettings, get_shared_settings
+from .storage import (
+    SharedStorageService,
+    get_shared_settings,
+    get_shared_storage_service,
+    get_storage_service,
+)
+
+__all__ = [
+    "SharedSettings",
+    "SharedStorageService",
+    "get_shared_settings",
+    "get_shared_storage_service",
+    "get_storage_service",
+]

packages/shared/config.py

@@ -0,0 +1,57 @@
+"""Shared settings for API and Worker services."""
+
+import os
+from typing import Optional
+
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class SharedSettings(BaseSettings):
+    """Settings shared between API and Worker services."""
+
+    # Database
+    database_url: str = "postgresql+psycopg://postgres:postgres@localhost:5432/xiaoxia_saas"
+    database_pool_size: int = 20
+    database_max_overflow: int = 40
+    database_pool_timeout: int = 30
+    database_pool_recycle: int = 3600
+
+    # Redis
+    redis_url: str = "redis://localhost:6379/0"
+
+    # Celery
+    celery_broker_url: str = "redis://localhost:6379/0"
+    celery_result_backend: str = "redis://localhost:6379/1"
+
+    # OSS Aliyun
+    oss_endpoint: str = "oss-cn-hangzhou.aliiyuncs.com"
+    oss_access_key_id: str = ""
+    oss_access_key_secret: str = ""
+    oss_bucket_name: str = "xiaoxia-autocut"
+
+    # Environment
+    environment: str = "development"
+    auto_create_schema: bool = False
+
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        env_file_encoding="utf-8",
+        case_sensitive=False,
+        extra="ignore",
+    )
+
+
+_settings: Optional[SharedSettings] = None
+
+
+def get_shared_settings() -> SharedSettings:
+    """Get shared settings instance (global singleton)."""
+    global _settings
+    if _settings is None:
+        env = os.getenv("APP_ENV", "development")
+        env_file = f".env.{env}" if env != "development" else ".env"
+        if os.path.exists(env_file):
+            _settings = SharedSettings(_env_file=env_file)
+        else:
+            _settings = SharedSettings()
+    return _settings

packages/shared/storage.py

@@ -0,0 +1,184 @@
+"""Shared OSS storage service for API and Worker."""
+
+import base64
+import datetime as dt
+import hashlib
+import hmac
+import json
+import logging
+import os
+from urllib.parse import urlparse
+from typing import Optional
+
+try:
+    import oss2
+except ImportError:  # pragma: no cover
+    oss2 = None
+
+from packages.shared.config import get_shared_settings
+
+logger = logging.getLogger(__name__)
+
+
+class SharedStorageService:
+    """Shared OSS storage service."""
+
+    def __init__(self):
+        settings = get_shared_settings()
+        self.bucket_name = settings.oss_bucket_name
+        self.public_url = f"https://{settings.oss_bucket_name}.{settings.oss_endpoint}"
+        self.local_url_prefix = os.getenv("GENERATED_FILES_URL_PREFIX", "/generated-files")
+        self.bucket = None
+
+        if settings.oss_access_key_id and settings.oss_access_key_secret:
+            if oss2 is not None:
+                auth = oss2.Auth(
+                    settings.oss_access_key_id,
+                    settings.oss_access_key_secret,
+                )
+                self.bucket = oss2.Bucket(
+                    auth,
+                    settings.oss_endpoint,
+                    settings.oss_bucket_name,
+                )
+        self.access_key_id = settings.oss_access_key_id
+        self.access_key_secret = settings.oss_access_key_secret
+        self.endpoint = settings.oss_endpoint
+
+    def _is_local_generated_url(self, storage_key_or_url: str) -> bool:
+        parsed = urlparse(storage_key_or_url)
+        path = parsed.path if parsed.scheme else storage_key_or_url
+        return path.startswith(f"{self.local_url_prefix}/")
+
+    def upload_file(
+        self,
+        file_or_path,
+        storage_key: str,
+        content_type: str = "application/octet-stream",
+    ) -> str:
+        """Upload file to OSS."""
+        if self.bucket is None:
+            raise RuntimeError("OSS storage is not configured")
+
+        try:
+            if isinstance(file_or_path, str):
+                self.bucket.put_object_from_file(storage_key, file_or_path, headers={"Content-Type": content_type})
+            else:
+                file_or_path.seek(0)
+                self.bucket.put_object(storage_key, file_or_path, headers={"Content-Type": content_type})
+            return f"{self.public_url}/{storage_key}"
+        except Exception as e:
+            raise Exception(f"Failed to upload file to OSS: {e}")
+
+    def get_url(self, storage_key: str) -> str:
+        """Get public URL for a file."""
+        return f"{self.public_url}/{storage_key}"
+
+    def download_file(self, storage_key: str, local_path: str):
+        """Download file from OSS to local path."""
+        if self.bucket is None:
+            raise RuntimeError("OSS storage is not configured")
+
+        try:
+            os.makedirs(os.path.dirname(local_path), exist_ok=True)
+            self.bucket.get_object_to_file(storage_key, local_path)
+        except Exception as e:
+            raise Exception(f"Failed to download file from OSS: {e}")
+
+    def get_download_url(self, storage_key_or_url: str, expires_seconds: int = 3600) -> str:
+        """Get signed download URL."""
+        if self.bucket is None:
+            if self._is_local_generated_url(storage_key_or_url):
+                return storage_key_or_url
+            return self.get_url(self._normalize_storage_key(storage_key_or_url))
+
+        storage_key = self._normalize_storage_key(storage_key_or_url)
+        try:
+            return self.bucket.sign_url("GET", storage_key, expires_seconds)
+        except Exception:
+            return self.get_url(storage_key)
+
+    def _normalize_storage_key(self, storage_key_or_url: str) -> str:
+        """Extract storage key from URL."""
+        if storage_key_or_url.startswith("http://") or storage_key_or_url.startswith("https://"):
+            parsed = urlparse(storage_key_or_url)
+            return parsed.path.lstrip("/")
+        return storage_key_or_url.lstrip("/")
+
+    def create_direct_upload_post(
+        self,
+        storage_key: str,
+        content_type: str,
+        max_size_bytes: int,
+        expires_seconds: int,
+    ) -> dict[str, object]:
+        """Create browser direct upload POST form."""
+        if not self.access_key_id or not self.access_key_secret:
+            raise RuntimeError("OSS storage is not configured")
+        normalized_key = self._normalize_storage_key(storage_key)
+        if not normalized_key.startswith("uploads/"):
+            raise ValueError("direct upload key must be under uploads/")
+
+        expiration = (dt.datetime.now(dt.timezone.utc) + dt.timedelta(seconds=expires_seconds)).strftime(
+            "%Y-%m-%dT%H:%M:%S.000Z"
+        )
+        policy = {
+            "expiration": expiration,
+            "conditions": [
+                {"bucket": self.bucket_name},
+                {"key": normalized_key},
+                ["content-length-range", 1, max_size_bytes],
+                ["starts-with", "$Content-Type", content_type.split("/", 1)[0] + "/" if "/" in content_type else ""],
+            ],
+        }
+        encoded_policy = base64.b64encode(json.dumps(policy, separators=(",", ":")).encode("utf-8")).decode("ascii")
+        signature = base64.b64encode(
+            hmac.new(self.access_key_secret.encode("utf-8"), encoded_policy.encode("utf-8"), hashlib.sha1).digest()
+        ).decode("ascii")
+
+        return {
+            "url": self.public_url,
+            "method": "POST",
+            "storage_key": normalized_key,
+            "expires_at": expiration,
+            "fields": {
+                "key": normalized_key,
+                "OSSAccessKeyId": self.access_key_id,
+                "policy": encoded_policy,
+                "Signature": signature,
+                "success_action_status": "201",
+                "Content-Type": content_type,
+            },
+        }
+
+    def delete_file(self, storage_key: str):
+        """Delete file from OSS."""
+        if self.bucket is None:
+            return
+        try:
+            self.bucket.delete_object(storage_key)
+        except Exception as error:
+            logger.warning("Failed to delete file from OSS", extra={"storage_key": storage_key, "error": str(error)})
+
+    def file_exists(self, storage_key: str) -> bool:
+        """Check if file exists."""
+        if self.bucket is None:
+            return False
+        return self.bucket.object_exists(storage_key)
+
+
+_storage_service: Optional[SharedStorageService] = None
+
+
+def get_shared_storage_service() -> SharedStorageService:
+    """Get shared storage service instance (global singleton)."""
+    global _storage_service
+    if _storage_service is None:
+        _storage_service = SharedStorageService()
+    return _storage_service
+
+
+# Backward compatibility alias
+def get_storage_service() -> SharedStorageService:
+    """Backward compatibility: returns shared storage service."""
+    return get_shared_storage_service()