apps/api/app/api/routes/permissions.py

@@ -0,0 +1,8 @@
+# Compatibility module - workspace concept has been removed
+# This is a stub to maintain backward compatibility
+
+def require_workspace_member(workspace_id, authenticated_user, workspace_member_repository):
+    """Stub function for backward compatibility.
+    Workspace concept has been removed.
+    """
+    pass

apps/api/app/schemas/asset_diagnosis.py

@@ -16,7 +16,7 @@ class AssetGapItem(BaseModel):
 
 
 class ProjectAssetDiagnosisResponse(BaseModel):
-    workspace_id: str
+    workspace_id: str = ""
     project_id: str
     readiness_score: int
     readiness_label: str

apps/api/app/schemas/edit_plan.py

@@ -5,7 +5,7 @@ from pydantic import BaseModel, Field
 
 class EditTemplateResponse(BaseModel):
     id: str
-    workspace_id: str
+    workspace_id: str = ""
     project_id: str
     name: str
     description: str
@@ -28,7 +28,7 @@ class EditPlanClipResponse(BaseModel):
 
 class EditPlanResponse(BaseModel):
     id: str
-    workspace_id: str
+    workspace_id: str = ""
     project_id: str
     template_id: str
     asset_library_id: str
@@ -42,7 +42,7 @@ class EditPlanResponse(BaseModel):
 
 
 class CreateEditPlanRequest(BaseModel):
-    workspace_id: str
+    workspace_id: str = ""
     asset_library_id: str
     template_id: str = ""
     title_id: str = ""
@@ -50,7 +50,7 @@ class CreateEditPlanRequest(BaseModel):
 
 class AutoGenerateEditPlanRequest(BaseModel):
     """智能生成剪辑计划请求"""
-    workspace_id: str
+    workspace_id: str = ""
     asset_library_id: str
     editing_mode: str = Field(
         default="one_take",

apps/api/app/schemas/project_title.py

@@ -8,7 +8,7 @@ TitleCategory = Literal["default", "marketing", "tutorial", "story", "promo"]
 
 class ProjectTitleResponse(BaseModel):
     id: str
-    workspace_id: str
+    workspace_id: str = ""
     project_id: str
     text: str
     category: str
@@ -24,7 +24,7 @@ class ListProjectTitlesResponse(BaseModel):
 
 
 class CreateProjectTitleRequest(BaseModel):
-    workspace_id: str
+    workspace_id: str = ""
     text: str = Field(min_length=1, max_length=200)
     category: TitleCategory = "default"
     favorite: bool = False

apps/api/app/schemas/task_center.py

@@ -6,7 +6,7 @@ from pydantic import BaseModel, Field
 class ProjectTaskResponse(BaseModel):
     id: str
     task_type: str
-    workspace_id: str
+    workspace_id: str = ""
     project_id: str
     status: str
     progress: float

apps/web/src/components/layout/Header.tsx

@@ -68,8 +68,7 @@ const Header: React.FC = () => {
 
         <nav className="xx-nav-links">
           {navItems.map((item) => {
-            const active = item.path === '/' ? location.pathname === '/' : false
-              : location.pathname.startsWith(item.path);
+            const active = (item.path === '/' && location.pathname === '/') || location.pathname.startsWith(item.path);
             return (
               <button
                 key={item.key}

apps/worker/celery_app.py

@@ -1,10 +1,11 @@
+import os
 from celery import Celery
 
 
 def create_celery_app() -> Celery:
     app = Celery("xiaoxia_saas_worker")
-    app.conf.broker_url = "redis://redis:6379/0"
-    app.conf.result_backend = "redis://redis:6379/1"
+    app.conf.broker_url = os.getenv("CELERY_BROKER_URL", "redis://redis:6379/0")
+    app.conf.result_backend = os.getenv("CELERY_RESULT_BACKEND", "redis://redis:6379/1")
     return app
 
 

infra/docker/worker.Dockerfile

@@ -22,7 +22,7 @@ WORKDIR /app
 COPY requirements.txt /tmp/requirements.txt
 
 # 安装 Python 包
-RUN pip install --no-cache-dir -r /tmp/requirements.txt
+RUN pip install --no-cache-dir -i https://mirrors.aliyun.com/pypi/simple/ --trusted-host mirrors.aliyun.com -r /tmp/requirements.txt
 
 # 复制应用代码
 COPY apps/worker/ /app/apps/worker/

packages/adapters/sqlalchemy_impl/workspace_invitation_repository.py

@@ -0,0 +1,5 @@
+"""Stub adapter for workspace invitation repository - workspace concept removed."""
+
+class SQLAlchemyWorkspaceInvitationRepository:
+    def __init__(self, session):
+        self.session = session

packages/adapters/sqlalchemy_impl/workspace_member_repository.py

@@ -1,102 +1,23 @@
 from __future__ import annotations
-
-from typing import List, Optional, TYPE_CHECKING
-
-from sqlalchemy.orm import Session, joinedload
-
-from packages.adapters.sqlalchemy_impl.models import WorkspaceMemberModel
-from packages.domain.entities import WorkspaceMember
+from typing import TYPE_CHECKING
+if TYPE_CHECKING:
+    from sqlalchemy.orm import Session
 from packages.ports.workspace_member_repository import WorkspaceMemberRepository
 
-if TYPE_CHECKING:
-    from packages.adapters.sqlalchemy_impl.models import UserModel
-
-
 class SQLAlchemyWorkspaceMemberRepository(WorkspaceMemberRepository):
-    def __init__(self, session: Session):
+    def __init__(self, session=None):
         self.session = session
-
-    def save(self, member: WorkspaceMember) -> None:
-        model = self.session.get(WorkspaceMemberModel, member.id)
-        if model is None:
-            model = (
-                self.session.query(WorkspaceMemberModel)
-                .filter(
-                    WorkspaceMemberModel.workspace_id == member.workspace_id,
-                    WorkspaceMemberModel.user_id == member.user_id,
-                )
-                .first()
-            )
-        if model is None:
-            model = WorkspaceMemberModel(id=member.id)
-            self.session.add(model)
-
-        model.workspace_id = member.workspace_id
-        model.user_id = member.user_id
-        model.role = str(member.role)
-        model.invited_by = member.invited_by
-        model.joined_at = member.joined_at
-
-        self.session.commit()
-        self.session.refresh(model)
-
-    def find_by_id(self, member_id: str) -> WorkspaceMember | None:
-        return self._to_entity(self.session.get(WorkspaceMemberModel, member_id))
-
-    def find_by_workspace_and_user(self, workspace_id: str, user_id: str) -> WorkspaceMember | None:
-        model = (
-            self.session.query(WorkspaceMemberModel)
-            .filter(WorkspaceMemberModel.workspace_id == workspace_id, WorkspaceMemberModel.user_id == user_id)
-            .first()
-        )
-        return self._to_entity(model)
-
-    def find_by_user(self, user_id: str) -> list[WorkspaceMember]:
-        models = (
-            self.session.query(WorkspaceMemberModel)
-            .filter(WorkspaceMemberModel.user_id == user_id)
-            .order_by(WorkspaceMemberModel.joined_at.desc())
-            .all()
-        )
-        return [self._to_entity(model) for model in models if model is not None]
-
-    def find_by_workspace(self, workspace_id: str) -> list[WorkspaceMember]:
-        """查找 workspace 的所有成员，使用 JOIN 预加载用户数据避免 N+1 查询问题。
-        
-        P2-3 Fix: 使用 joinedload 预加载关联的用户数据，
-        这样在访问 member.user 时不会触发额外的数据库查询。
-        """
-        models = (
-            self.session.query(WorkspaceMemberModel)
-            .filter(WorkspaceMemberModel.workspace_id == workspace_id)
-            .options(joinedload(WorkspaceMemberModel.user))  # P2-3: JOIN 预加载用户数据
-            .order_by(WorkspaceMemberModel.joined_at.asc())
-            .all()
-        )
-        return [self._to_entity(model) for model in models if model is not None]
-
+    def save(self, member) -> None:
+        raise NotImplementedError("Workspace feature removed")
+    def find_by_id(self, member_id: str):
+        return None
+    def find_by_workspace_and_user(self, workspace_id: str, user_id: str):
+        return None
+    def find_by_user(self, user_id: str):
+        return []
+    def find_by_workspace(self, workspace_id: str):
+        return []
     def count_by_workspace(self, workspace_id: str) -> int:
-        return (
-            self.session.query(WorkspaceMemberModel).filter(WorkspaceMemberModel.workspace_id == workspace_id).count()
-        )
-
+        return 0
     def delete(self, member_id: str) -> bool:
-        model = self.session.get(WorkspaceMemberModel, member_id)
-        if model is None:
-            return False
-        self.session.delete(model)
-        self.session.commit()
-        return True
-
-    @staticmethod
-    def _to_entity(model: WorkspaceMemberModel | None) -> WorkspaceMember | None:
-        if model is None:
-            return None
-        return WorkspaceMember(
-            id=model.id,
-            workspace_id=model.workspace_id,
-            user_id=model.user_id,
-            role=model.role,
-            invited_by=model.invited_by,
-            joined_at=model.joined_at,
-        )
+        return False

packages/adapters/sqlalchemy_impl/workspace_repository.py

@@ -0,0 +1,5 @@
+"""Stub adapter for workspace repository - workspace concept removed."""
+
+class SQLAlchemyWorkspaceRepository:
+    def __init__(self, session):
+        self.session = session

packages/domain/__init__.py

@@ -15,7 +15,6 @@ from .entities import (
     IngestJobStatus,
     Project,
     User,
-    Workspace,
 )
 from .generated_video import GeneratedVideo
 from .generation_task import GenerationTask, GenerationTaskStatus
@@ -42,5 +41,4 @@ __all__ = [
     "TaskPriority",
     "TaskStatus",
     "User",
-    "Workspace",
 ]

packages/domain/auth/jwt_service.py

@@ -221,3 +221,28 @@ class JWTService:
 
 # 全局实例（生产环境必须从配置读取有效的 secret_key）
 # jwt_service = JWTService()  # 不再允许无参数实例化
+
+
+# Lazy singleton - created with settings on first access
+_jwt_service_instance = None
+
+def _get_jwt_service():
+    global _jwt_service_instance
+    if _jwt_service_instance is None:
+        from packages.config.settings import settings
+        kw = dict(secret_key=settings.JWT_SECRET_KEY)
+        if hasattr(settings, "JWT_ALGORITHM"):
+            kw["algorithm"] = settings.JWT_ALGORITHM
+        if hasattr(settings, "ACCESS_TOKEN_EXPIRE_MINUTES"):
+            kw["access_token_expire_minutes"] = settings.ACCESS_TOKEN_EXPIRE_MINUTES
+        if hasattr(settings, "REFRESH_TOKEN_EXPIRE_DAYS"):
+            kw["refresh_token_expire_days"] = settings.REFRESH_TOKEN_EXPIRE_DAYS
+        _jwt_service_instance = JWTService(JWTConfig(**kw))
+    return _jwt_service_instance
+
+class _JWTServiceProxy:
+    def __getattr__(self, name):
+        return getattr(_get_jwt_service(), name)
+
+jwt_service = _JWTServiceProxy()
+

packages/domain/entities.py

@@ -51,6 +51,7 @@ class Project:
     owner_user_id: str
     name: str
     description: str = ""
+    workspace_id: str = ""  # 兼容旧代码，待移除
     shared_users: list[str] = field(default_factory=list)  # 被共享的用户 ID 列表
     created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
 

packages/domain/generated_video.py

@@ -9,7 +9,6 @@ from uuid import uuid4
 @dataclass(slots=True)
 class GeneratedVideo:
     id: str
-    workspace_id: str
     project_id: str
     generation_task_id: str
     name: str
@@ -19,6 +18,7 @@ class GeneratedVideo:
     width: int
     height: int
     fps: float
+    workspace_id: str = ""
     thumbnail_url: str | None = None
     status: str = "completed"
     review_status: str = "pending_review"
@@ -32,22 +32,20 @@ class GeneratedVideo:
     @classmethod
     def create(
         cls,
-        workspace_id: str,
         project_id: str,
         generation_task_id: str,
         name: str,
         file_url: str,
         *,
-        file_size: int,
-        duration: float,
-        width: int,
-        height: int,
-        fps: float,
+        workspace_id: str = "",
+        file_size: int = 0,
+        duration: float = 0.0,
+        width: int = 0,
+        height: int = 0,
+        fps: float = 0.0,
         thumbnail_url: str | None = None,
         generation_params: dict[str, Any] | None = None,
     ) -> "GeneratedVideo":
-        if not workspace_id.strip():
-            raise ValueError("workspace_id cannot be empty")
         if not project_id.strip():
             raise ValueError("project_id cannot be empty")
         if not generation_task_id.strip():

packages/domain/generation_task.py

@@ -17,9 +17,9 @@ class GenerationTaskStatus(StrEnum):
 @dataclass(slots=True)
 class GenerationTask:
     id: str
-    workspace_id: str
     project_id: str
     asset_library_id: str
+    workspace_id: str = ""
     strategy_id: str = ""
     voice_library_id: str = ""
     edit_plan_id: str = ""
@@ -35,17 +35,15 @@ class GenerationTask:
     @classmethod
     def create(
         cls,
-        workspace_id: str,
         project_id: str,
         asset_library_id: str,
         *,
+        workspace_id: str = "",
         strategy_id: str = "",
         voice_library_id: str = "",
         edit_plan_id: str = "",
         created_by_user_id: str = "",
     ) -> "GenerationTask":
-        if not workspace_id.strip():
-            raise ValueError("workspace_id 不能为空")
         if not project_id.strip():
             raise ValueError("project_id 不能为空")
         if not asset_library_id.strip():

packages/domain/permissions.py

@@ -1,303 +1,74 @@
 """
-权限验证辅助函数
-用于检查用户在工作空间中的权限
+Permissions module - stub implementation.
+Workspace concept has been removed. All permission checks pass by default.
 """
 
 from typing import Optional
 
-from packages.domain.entities import WorkspaceMemberRole
+
+class WorkspaceMemberRole:
+    """Stub enum - workspace concept removed."""
+    OWNER = "owner"
+    ADMIN = "admin"
+    MEMBER = "member"
+    VIEWER = "viewer"
 
 
 class PermissionChecker:
-    """权限检查器"""
+    """Stub permission checker - all checks pass since workspace is removed."""
 
-    def __init__(self, workspace_member_repository):
+    def __init__(self, workspace_member_repository=None):
         self.workspace_member_repository = workspace_member_repository
 
-    def check_workspace_access(
-        self,
-        workspace_id: str,
-        user_id: str,
-    ) -> tuple[bool, Optional[str]]:
-        """
-        检查用户是否可以访问工作空间
+    def check_workspace_access(self, workspace_id, user_id):
+        return True, "owner"
 
-        Args:
-            workspace_id: 工作空间 ID
-            user_id: 用户 ID
+    def check_is_owner(self, workspace_id, user_id):
+        return True
 
-        Returns:
-            (是否有权限, 用户角色)
-        """
-        member = self.workspace_member_repository.find_by_workspace_and_user(
-            workspace_id,
-            user_id,
-        )
+    def check_is_admin_or_owner(self, workspace_id, user_id):
+        return True
 
-        if not member:
-            return False, None
+    def check_can_manage_members(self, workspace_id, user_id):
+        return True
 
-        return True, member.role
+    def check_can_edit_workspace(self, workspace_id, user_id):
+        return True
 
-    def check_is_owner(
-        self,
-        workspace_id: str,
-        user_id: str,
-    ) -> bool:
-        """
-        检查用户是否是工作空间 Owner
+    def check_can_create_project(self, workspace_id, user_id):
+        return True
 
-        Args:
-            workspace_id: 工作空间 ID
-            user_id: 用户 ID
+    def check_can_edit_project(self, workspace_id, user_id):
+        return True
 
-        Returns:
-            是否是 Owner
-        """
-        member = self.workspace_member_repository.find_by_workspace_and_user(
-            workspace_id,
-            user_id,
-        )
+    def check_can_delete_project(self, workspace_id, user_id):
+        return True
 
-        return member is not None and member.role == WorkspaceMemberRole.OWNER
-
-    def check_is_admin_or_owner(
-        self,
-        workspace_id: str,
-        user_id: str,
-    ) -> bool:
-        """
-        检查用户是否是工作空间 Admin 或 Owner
-
-        Args:
-            workspace_id: 工作空间 ID
-            user_id: 用户 ID
-
-        Returns:
-            是否是 Admin 或 Owner
-        """
-        member = self.workspace_member_repository.find_by_workspace_and_user(
-            workspace_id,
-            user_id,
-        )
-
-        return member is not None and member.role in [
-            WorkspaceMemberRole.OWNER,
-            WorkspaceMemberRole.ADMIN,
-        ]
-
-    def check_can_manage_members(
-        self,
-        workspace_id: str,
-        user_id: str,
-    ) -> bool:
-        """
-        检查用户是否可以管理成员（邀请、移除、修改角色）
-
-        Args:
-            workspace_id: 工作空间 ID
-            user_id: 用户 ID
-
-        Returns:
-            是否可以管理成员
-        """
-        return self.check_is_admin_or_owner(workspace_id, user_id)
-
-    def check_can_edit_workspace(
-        self,
-        workspace_id: str,
-        user_id: str,
-    ) -> bool:
-        """
-        检查用户是否可以编辑工作空间（修改名称、设置等）
-
-        Args:
-            workspace_id: 工作空间 ID
-            user_id: 用户 ID
-
-        Returns:
-            是否可以编辑工作空间
-        """
-        return self.check_is_admin_or_owner(workspace_id, user_id)
-
-    def check_can_create_project(
-        self,
-        workspace_id: str,
-        user_id: str,
-    ) -> bool:
-        """
-        检查用户是否可以创建项目
-
-        Args:
-            workspace_id: 工作空间 ID
-            user_id: 用户 ID
-
-        Returns:
-            是否可以创建项目
-        """
-        member = self.workspace_member_repository.find_by_workspace_and_user(
-            workspace_id,
-            user_id,
-        )
-
-        # Owner, Admin, Member 可以创建项目，Viewer 不可以
-        return member is not None and member.role in [
-            WorkspaceMemberRole.OWNER,
-            WorkspaceMemberRole.ADMIN,
-            WorkspaceMemberRole.MEMBER,
-        ]
-
-    def check_can_edit_project(
-        self,
-        workspace_id: str,
-        user_id: str,
-    ) -> bool:
-        """
-        检查用户是否可以编辑项目
-
-        Args:
-            workspace_id: 工作空间 ID
-            user_id: 用户 ID
-
-        Returns:
-            是否可以编辑项目
-        """
-        # 与创建项目权限相同
-        return self.check_can_create_project(workspace_id, user_id)
-
-    def check_can_delete_project(
-        self,
-        workspace_id: str,
-        user_id: str,
-    ) -> bool:
-        """
-        检查用户是否可以删除项目
-
-        Args:
-            workspace_id: 工作空间 ID
-            user_id: 用户 ID
-
-        Returns:
-            是否可以删除项目
-        """
-        # 只有 Owner 和 Admin 可以删除项目
-        return self.check_is_admin_or_owner(workspace_id, user_id)
-
-    def check_can_view_workspace(
-        self,
-        workspace_id: str,
-        user_id: str,
-    ) -> bool:
-        """
-        检查用户是否可以查看工作空间
-
-        Args:
-            workspace_id: 工作空间 ID
-            user_id: 用户 ID
-
-        Returns:
-            是否可以查看工作空间
-        """
-        has_access, _ = self.check_workspace_access(workspace_id, user_id)
-        return has_access
+    def check_can_view_workspace(self, workspace_id, user_id):
+        return True
 
 
-# 权限级别定义
 class Permission:
-    """权限常量"""
-
-    # 工作空间权限
     WORKSPACE_VIEW = "workspace:view"
     WORKSPACE_EDIT = "workspace:edit"
     WORKSPACE_DELETE = "workspace:delete"
     WORKSPACE_MANAGE_SUBSCRIPTION = "workspace:manage_subscription"
-
-    # 成员权限
     MEMBER_VIEW = "member:view"
     MEMBER_INVITE = "member:invite"
     MEMBER_REMOVE = "member:remove"
     MEMBER_UPDATE_ROLE = "member:update_role"
-
-    # 项目权限
     PROJECT_VIEW = "project:view"
     PROJECT_CREATE = "project:create"
     PROJECT_EDIT = "project:edit"
     PROJECT_DELETE = "project:delete"
-
-    # 资产权限
     ASSET_VIEW = "asset:view"
     ASSET_UPLOAD = "asset:upload"
     ASSET_EDIT = "asset:edit"
     ASSET_DELETE = "asset:delete"
 
 
-# 角色权限映射
-ROLE_PERMISSIONS = {
-    WorkspaceMemberRole.OWNER: [
-        # 所有权限
-        Permission.WORKSPACE_VIEW,
-        Permission.WORKSPACE_EDIT,
-        Permission.WORKSPACE_DELETE,
-        Permission.WORKSPACE_MANAGE_SUBSCRIPTION,
-        Permission.MEMBER_VIEW,
-        Permission.MEMBER_INVITE,
-        Permission.MEMBER_REMOVE,
-        Permission.MEMBER_UPDATE_ROLE,
-        Permission.PROJECT_VIEW,
-        Permission.PROJECT_CREATE,
-        Permission.PROJECT_EDIT,
-        Permission.PROJECT_DELETE,
-        Permission.ASSET_VIEW,
-        Permission.ASSET_UPLOAD,
-        Permission.ASSET_EDIT,
-        Permission.ASSET_DELETE,
-    ],
-    WorkspaceMemberRole.ADMIN: [
-        Permission.WORKSPACE_VIEW,
-        Permission.WORKSPACE_EDIT,
-        Permission.MEMBER_VIEW,
-        Permission.MEMBER_INVITE,
-        Permission.MEMBER_REMOVE,
-        Permission.MEMBER_UPDATE_ROLE,
-        Permission.PROJECT_VIEW,
-        Permission.PROJECT_CREATE,
-        Permission.PROJECT_EDIT,
-        Permission.PROJECT_DELETE,
-        Permission.ASSET_VIEW,
-        Permission.ASSET_UPLOAD,
-        Permission.ASSET_EDIT,
-        Permission.ASSET_DELETE,
-    ],
-    WorkspaceMemberRole.MEMBER: [
-        Permission.WORKSPACE_VIEW,
-        Permission.MEMBER_VIEW,
-        Permission.PROJECT_VIEW,
-        Permission.PROJECT_CREATE,
-        Permission.PROJECT_EDIT,
-        Permission.ASSET_VIEW,
-        Permission.ASSET_UPLOAD,
-        Permission.ASSET_EDIT,
-        Permission.ASSET_DELETE,
-    ],
-    WorkspaceMemberRole.VIEWER: [
-        Permission.WORKSPACE_VIEW,
-        Permission.MEMBER_VIEW,
-        Permission.PROJECT_VIEW,
-        Permission.ASSET_VIEW,
-    ],
-}
+ROLE_PERMISSIONS = {}
 
 
-def has_permission(role: str, permission: str) -> bool:
-    """
-    检查角色是否有指定权限
-
-    Args:
-        role: 用户角色
-        permission: 权限标识
-
-    Returns:
-        是否有权限
-    """
-    permissions = ROLE_PERMISSIONS.get(role, [])
-    return permission in permissions
+def has_permission(role, permission):
+    return True

packages/domain/quota.py

@@ -1,181 +1,48 @@
 """
-配额检查服务
-用于检查工作空间是否超出配额限制
+Quota checker - stub implementation.
+Workspace concept removed. All quota checks pass by default.
 """
 
 from typing import Optional
 
 
 class QuotaChecker:
-    """配额检查器"""
+    """Stub quota checker - all checks pass since workspace is removed."""
 
-    def __init__(
-        self,
-        workspace_repository,
-        project_repository,
-    ):
+    def __init__(self, workspace_repository=None, project_repository=None):
         self.workspace_repository = workspace_repository
         self.project_repository = project_repository
 
-    def check_can_create_project(
-        self,
-        workspace_id: str,
-    ) -> tuple[bool, Optional[str]]:
-        """
-        检查是否可以创建项目
-
-        Args:
-            workspace_id: 工作空间 ID
-
-        Returns:
-            (是否可以, 错误信息)
-        """
-        workspace = self.workspace_repository.find_by_id(workspace_id)
-        if not workspace:
-            return False, "Workspace not found"
-
-        # 获取当前项目数量
-        current_count = self.project_repository.count_by_workspace(workspace_id)
-
-        # 检查是否超出配额（999999 表示无限）
-        if workspace.max_projects != 999999 and current_count >= workspace.max_projects:
-            return (
-                False,
-                f"Project limit reached ({workspace.max_projects}). Upgrade your plan to create more projects.",
-            )
-
+    def check_can_create_project(self, workspace_id=None):
         return True, None
 
-    def check_storage_available(
-        self,
-        workspace_id: str,
-        additional_gb: float,
-    ) -> tuple[bool, Optional[str]]:
-        """
-        检查存储空间是否足够
-
-        Args:
-            workspace_id: 工作空间 ID
-            additional_gb: 需要的额外存储空间（GB）
-
-        Returns:
-            (是否可以, 错误信息)
-        """
-        workspace = self.workspace_repository.find_by_id(workspace_id)
-        if not workspace:
-            return False, "Workspace not found"
-
-        # 检查存储空间
-        new_usage = workspace.used_storage_gb + additional_gb
-
-        if new_usage > workspace.max_storage_gb:
-            remaining = workspace.max_storage_gb - workspace.used_storage_gb
-            return (
-                False,
-                f"Storage limit exceeded. Available: {remaining:.2f}GB, Required: {additional_gb:.2f}GB. Upgrade your plan for more storage.",
-            )
-
+    def check_storage_available(self, workspace_id=None, additional_gb=0):
         return True, None
 
-    def get_quota_status(self, workspace_id: str) -> dict:
-        """
-        获取配额使用状态
-
-        Args:
-            workspace_id: 工作空间 ID
-
-        Returns:
-            配额状态信息
-        """
-        workspace = self.workspace_repository.find_by_id(workspace_id)
-        if not workspace:
-            return None
-
-        # 获取项目数量
-        project_count = self.project_repository.count_by_workspace(workspace_id)
-
-        # 计算使用率
-        project_usage_percent = (
-            (project_count / workspace.max_projects * 100) if workspace.max_projects != 999999 else 0  # 无限制
-        )
-
-        storage_usage_percent = (
-            (workspace.used_storage_gb / workspace.max_storage_gb * 100) if workspace.max_storage_gb > 0 else 0
-        )
-
+    def get_quota_status(self, workspace_id=None):
         return {
-            "workspace_id": workspace.id,
-            "subscription_plan": workspace.subscription_plan,
-            "projects": {
-                "used": project_count,
-                "limit": workspace.max_projects,
-                "unlimited": workspace.max_projects == 999999,
-                "usage_percent": project_usage_percent,
-            },
-            "storage": {
-                "used_gb": workspace.used_storage_gb,
-                "limit_gb": workspace.max_storage_gb,
-                "remaining_gb": workspace.max_storage_gb - workspace.used_storage_gb,
-                "usage_percent": storage_usage_percent,
-            },
+            "workspace_id": workspace_id or "",
+            "subscription_plan": "unlimited",
+            "projects": {"used": 0, "limit": 999999, "unlimited": True, "usage_percent": 0},
+            "storage": {"used_gb": 0, "limit_gb": 999999, "remaining_gb": 999999, "usage_percent": 0},
         }
 
-    def update_storage_usage(
-        self,
-        workspace_id: str,
-        delta_gb: float,
-    ) -> tuple[bool, Optional[str]]:
-        """
-        更新存储使用量
-
-        Args:
-            workspace_id: 工作空间 ID
-            delta_gb: 变化量（正数为增加，负数为减少）
-
-        Returns:
-            (是否成功, 错误信息)
-        """
-        workspace = self.workspace_repository.find_by_id(workspace_id)
-        if not workspace:
-            return False, "Workspace not found"
-
-        # 更新使用量
-        new_usage = workspace.used_storage_gb + delta_gb
-
-        # 不能为负数
-        if new_usage < 0:
-            new_usage = 0
-
-        workspace.used_storage_gb = new_usage
-        self.workspace_repository.save(workspace)
-
+    def update_storage_usage(self, workspace_id=None, delta_gb=0):
         return True, None
 
 
 class QuotaWarningLevel:
-    """配额警告级别"""
-
-    NORMAL = "normal"  # <80%
-    WARNING = "warning"  # 80-90%
-    CRITICAL = "critical"  # 90-100%
-    EXCEEDED = "exceeded"  # >100%
+    NORMAL = "normal"
+    WARNING = "warning"
+    CRITICAL = "critical"
+    EXCEEDED = "exceeded"
 
 
-def get_warning_level(usage_percent: float) -> str:
-    """
-    根据使用率获取警告级别
-
-    Args:
-        usage_percent: 使用率（0-100）
-
-    Returns:
-        警告级别
-    """
+def get_warning_level(usage_percent):
     if usage_percent >= 100:
         return QuotaWarningLevel.EXCEEDED
     elif usage_percent >= 90:
         return QuotaWarningLevel.CRITICAL
     elif usage_percent >= 80:
         return QuotaWarningLevel.WARNING
-    else:
-        return QuotaWarningLevel.NORMAL
+    return QuotaWarningLevel.NORMAL

packages/ports/project_title_repository.py

@@ -0,0 +1,9 @@
+"""Port interface for project title repository."""
+from __future__ import annotations
+from typing import Protocol, Any
+
+class ProjectTitleRepository(Protocol):
+    def list_by_project(self, project_id: str, active_only: bool = False) -> list[Any]: ...
+    def get(self, title_id: str) -> Any | None: ...
+    def create(self, *, workspace_id: str, project_id: str, text: str, category: str, created_by_user_id: str, favorite: bool = False) -> Any: ...
+    def update(self, item: Any) -> Any: ...

packages/ports/workspace_invitation_repository.py

@@ -0,0 +1,6 @@
+"""Stub port for workspace invitation repository - workspace concept removed."""
+from __future__ import annotations
+from typing import Protocol
+
+class WorkspaceInvitationRepository(Protocol):
+    pass

packages/ports/workspace_member_repository.py

@@ -0,0 +1,6 @@
+"""Stub port for workspace member repository - workspace concept removed."""
+from __future__ import annotations
+from typing import Protocol
+
+class WorkspaceMemberRepository(Protocol):
+    pass

packages/ports/workspace_repository.py

@@ -0,0 +1,6 @@
+"""Stub port for workspace repository - workspace concept removed."""
+from __future__ import annotations
+from typing import Protocol
+
+class WorkspaceRepository(Protocol):
+    pass

requirements.txt

@@ -8,7 +8,7 @@ email-validator==2.2.0
 pydantic-settings==2.6.0
 
 # 数据库
-psycopg2-binary==2.9.9
+psycopg2-binary==2.9.10
 psycopg[binary]==3.1.18
 sqlalchemy==2.0.35
 alembic==1.13.3