diff --git a/apps/api/app/api/routes/edit_plans.py b/apps/api/app/api/routes/edit_plans.py index 3240ca8ae..b6b7b71b6 100644 --- a/apps/api/app/api/routes/edit_plans.py +++ b/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 diff --git a/apps/worker/tasks.py b/apps/worker/tasks.py index 2ac298712..0cf3f993b 100644 --- a/apps/worker/tasks.py +++ b/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) diff --git a/apps/worker/video_processing/dedup.py b/apps/worker/video_processing/dedup.py index dbc69308a..068d0f858 100644 --- a/apps/worker/video_processing/dedup.py +++ b/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) diff --git a/apps/worker/worker_app/tasks/__init__.py b/apps/worker/worker_app/tasks/__init__.py index 83cab8b39..8440125db 100644 --- a/apps/worker/worker_app/tasks/__init__.py +++ b/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"] diff --git a/apps/worker/worker_app/tasks/edit_plan_generator.py b/apps/worker/worker_app/tasks/edit_plan_generator.py index d8b5848b8..54788492e 100644 --- a/apps/worker/worker_app/tasks/edit_plan_generator.py +++ b/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 diff --git a/apps/worker/worker_app/tasks/voice_extraction.py b/apps/worker/worker_app/tasks/voice_extraction.py index 552c49173..b3cae3ec9 100644 --- a/apps/worker/worker_app/tasks/voice_extraction.py +++ b/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) diff --git a/packages/domain/__init__.py b/packages/domain/__init__.py index 813af8a52..cf6e71499 100644 --- a/packages/domain/__init__.py +++ b/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", ] diff --git a/packages/domain/edit_plan.py b/packages/domain/edit_plan.py new file mode 100644 index 000000000..c87564b64 --- /dev/null +++ b/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 diff --git a/packages/shared/__init__.py b/packages/shared/__init__.py index 7dbba873f..4ff9ed5b2 100644 --- a/packages/shared/__init__.py +++ b/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", +] diff --git a/packages/shared/config.py b/packages/shared/config.py new file mode 100644 index 000000000..a612a043d --- /dev/null +++ b/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 diff --git a/packages/shared/storage.py b/packages/shared/storage.py new file mode 100644 index 000000000..8100f28fc --- /dev/null +++ b/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()