Add shot/asset typing and owned task types to departments

Departments now carry a type (shot or asset) and own a list of task types (e.g. Animation: blocking/primary_pass/second_pass, Composite: first_pass/second_pass, plus a new Simulation department), additive to the existing flat Custom Task Type system. When a task's type belongs to a department, its department is derived and kept in sync server-side across create/update paths; Task Type is now editable in the Task Detail panel and Department options are filtered to the task's shot/asset scope.
2026-07-24 08:55:18 +08:00
parent 9bbd3df53c
commit 31d17780a4
11 changed files with 960 additions and 94 deletions
@@ -0,0 +1,107 @@
+#!/usr/bin/env python3
+"""
+Migration script to normalize projects.custom_departments from a flat list of
+strings (pre-department-type feature) to a list of objects:
+{"name": str, "type": "shot"|"asset", "task_types": [str]}.
+
+Usage:
+    python migrate_department_task_types.py
+"""
+
+import sqlite3
+import sys
+import json
+from pathlib import Path
+
+
+def get_database_path():
+    """Get the database path, trying multiple possible locations."""
+    possible_paths = [
+        "vfx_project_management.db",  # Primary database
+        "database.db",
+        "../vfx_project_management.db"
+    ]
+
+    for path in possible_paths:
+        if Path(path).exists():
+            return path
+
+    return "vfx_project_management.db"
+
+
+def migrate_database():
+    """Normalize any legacy plain-string custom_departments entries."""
+    db_path = get_database_path()
+    print(f"Using database: {db_path}")
+
+    conn = None
+    try:
+        conn = sqlite3.connect(db_path)
+        cursor = conn.cursor()
+
+        cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='projects'")
+        if not cursor.fetchone():
+            print("Projects table not found. Nothing to migrate.")
+            conn.close()
+            return
+
+        cursor.execute("SELECT id, custom_departments FROM projects")
+        rows = cursor.fetchall()
+
+        normalized_count = 0
+        for project_id, custom_departments_raw in rows:
+            if not custom_departments_raw:
+                continue
+
+            try:
+                custom_departments = json.loads(custom_departments_raw)
+            except (json.JSONDecodeError, TypeError):
+                continue
+
+            if not isinstance(custom_departments, list) or not custom_departments:
+                continue
+
+            needs_normalization = any(isinstance(d, str) for d in custom_departments)
+            if not needs_normalization:
+                continue
+
+            normalized = []
+            for d in custom_departments:
+                if isinstance(d, str):
+                    print(f"  Project {project_id}: normalizing legacy department '{d}' "
+                          f"(defaulting type='shot', task_types=['{d}'] - review if incorrect)")
+                    normalized.append({"name": d, "type": "shot", "task_types": [d]})
+                else:
+                    normalized.append(d)
+
+            cursor.execute(
+                "UPDATE projects SET custom_departments = ? WHERE id = ?",
+                (json.dumps(normalized), project_id)
+            )
+            normalized_count += 1
+
+        conn.commit()
+        print(f"Migration completed successfully! {normalized_count} project(s) normalized.")
+
+    except sqlite3.Error as e:
+        print(f"Database error: {e}")
+        if conn:
+            conn.rollback()
+        sys.exit(1)
+    except Exception as e:
+        print(f"Unexpected error: {e}")
+        if conn:
+            conn.rollback()
+        sys.exit(1)
+    finally:
+        if conn:
+            conn.close()
+
+
+if __name__ == "__main__":
+    print("VFX Project Management - Department Task Types Migration")
+    print("=" * 60)
+
+    migrate_database()
+
+    print("\nMigration completed successfully!")
@@ -1,6 +1,6 @@
 from fastapi import APIRouter, Depends, HTTPException, status
 from sqlalchemy.orm import Session
-from typing import List, Dict
+from typing import List, Dict, Optional

 from database import get_db
 from models.asset import Asset, AssetCategory
@@ -10,6 +10,7 @@ from models.user import User, UserRole
 from schemas.asset import AssetCreate, AssetUpdate, AssetResponse, AssetListResponse, TaskStatusInfo
 from schemas.task import TaskCreate
 from utils.auth import get_current_user_from_token, require_permission
+from utils.departments import find_owning_department
 from services.asset_soft_deletion import AssetSoftDeletionService

 router = APIRouter()
@@ -119,14 +120,14 @@ def get_all_asset_task_types(project_id: int, db: Session) -> List[str]:
    return STANDARD_ASSET_TASK_TYPES + custom_types


-def create_default_tasks_for_asset(asset: Asset, task_types: List[str], db: Session) -> List[Task]:
+def create_default_tasks_for_asset(asset: Asset, task_types: List[str], db: Session, project: Optional[Project] = None) -> List[Task]:
    """Create default tasks for an asset."""
    created_tasks = []
-    
+
    for task_type in task_types:
        # Create task name based on type
        task_name = f"{asset.name} - {task_type.title()}"
-        
+
        # Create the task
        db_task = Task(
            project_id=asset.project_id,
@@ -134,12 +135,13 @@ def create_default_tasks_for_asset(asset: Asset, task_types: List[str], db: Sess
            task_type=task_type,
            name=task_name,
            description=f"Default {task_type} task for {asset.name}",
-            status="not_started"
+            status="not_started",
+            department=find_owning_department(project, task_type) if project else None
        )
-        
+
        db.add(db_task)
        created_tasks.append(db_task)
-    
+
    return created_tasks


@@ -362,7 +364,7 @@ async def create_asset(
 ):
    """Create a new asset in a project with optional default tasks"""
    # Check project access
-    check_project_access(project_id, current_user, db)
+    project = check_project_access(project_id, current_user, db)
    
    # Check if asset name already exists in project (exclude soft deleted)
    existing_asset = db.query(Asset).filter(
@@ -408,7 +410,7 @@ async def create_asset(
            task_types = get_default_asset_task_types(asset.category)
        
        # Create the tasks
-        created_tasks = create_default_tasks_for_asset(db_asset, task_types, db)
+        created_tasks = create_default_tasks_for_asset(db_asset, task_types, db, project)
        task_count = len(created_tasks)
    
    db.commit()
@@ -553,13 +555,15 @@ async def create_asset_task(
    
    # Create the task
    task_name = f"{asset.name} - {task_type.title()}"
+    project = db.query(Project).filter(Project.id == asset.project_id).first()
    db_task = Task(
        project_id=asset.project_id,
        asset_id=asset.id,
        task_type=task_type,
        name=task_name,
        description=f"{task_type.title()} task for {asset.name}",
-        status="not_started"
+        status="not_started",
+        department=find_owning_department(project, task_type)
    )
    
    db.add(db_task)
@@ -1080,21 +1080,41 @@ async def delete_custom_task_type(

 # Department Management Endpoints

-# Standard departments (read-only)
-STANDARD_DEPARTMENTS = ["layout", "animation", "lighting", "composite", "modeling", "rigging", "surfacing"]
+# Standard departments (read-only): name, whether they apply to shots or assets,
+# and the task types they own.
+STANDARD_DEPARTMENTS = [
+    {"name": "layout", "type": "shot", "task_types": ["layout"]},
+    {"name": "animation", "type": "shot", "task_types": ["blocking", "primary_pass", "second_pass"]},
+    {"name": "simulation", "type": "shot", "task_types": ["simulation"]},
+    {"name": "lighting", "type": "shot", "task_types": ["lighting"]},
+    {"name": "composite", "type": "shot", "task_types": ["first_pass", "second_pass"]},
+    {"name": "modeling", "type": "asset", "task_types": ["modeling"]},
+    {"name": "rigging", "type": "asset", "task_types": ["rigging"]},
+    {"name": "surfacing", "type": "asset", "task_types": ["surfacing"]},
+]
+STANDARD_DEPARTMENT_NAMES = [d["name"] for d in STANDARD_DEPARTMENTS]
+
+
+def _find_custom_department(custom_departments: list, name: str):
+    """Find a custom department dict by name, or None."""
+    for department in custom_departments:
+        if department["name"] == name:
+            return department
+    return None


 def _build_all_departments_response(db_project: Project):
    """Helper function to build AllDepartmentsResponse"""
-    from schemas.department import AllDepartmentsResponse
+    from schemas.department import AllDepartmentsResponse, DepartmentInfo

    custom_departments = db_project.custom_departments or []
-    all_departments = STANDARD_DEPARTMENTS + custom_departments
+    standard_infos = [DepartmentInfo(**d) for d in STANDARD_DEPARTMENTS]
+    custom_infos = [DepartmentInfo(**d) for d in custom_departments]

    return AllDepartmentsResponse(
-        departments=all_departments,
-        standard_departments=STANDARD_DEPARTMENTS,
-        custom_departments=custom_departments
+        departments=standard_infos + custom_infos,
+        standard_departments=standard_infos,
+        custom_departments=custom_infos
    )


@@ -1144,19 +1164,23 @@ async def add_department(

    custom_departments = db_project.custom_departments or []

-    if department_create.department in STANDARD_DEPARTMENTS:
+    if department_create.department in STANDARD_DEPARTMENT_NAMES:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=f"Department '{department_create.department}' is a standard department and cannot be added as custom"
        )

-    if department_create.department in custom_departments:
+    if _find_custom_department(custom_departments, department_create.department):
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=f"Department '{department_create.department}' already exists"
        )

-    custom_departments.append(department_create.department)
+    custom_departments.append({
+        "name": department_create.department,
+        "type": department_create.department_type,
+        "task_types": department_create.task_types
+    })

    db_project.custom_departments = custom_departments
    flag_modified(db_project, 'custom_departments')
@@ -1209,29 +1233,28 @@ async def update_department(
        )

    custom_departments = db_project.custom_departments or []
+    existing = _find_custom_department(custom_departments, department_update.old_name)

-    if department_update.old_name not in custom_departments:
+    if not existing:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Custom department '{department_update.old_name}' not found"
        )

-    if department_update.new_name in STANDARD_DEPARTMENTS:
+    if department_update.new_name in STANDARD_DEPARTMENT_NAMES:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=f"Department '{department_update.new_name}' is a standard department"
        )

-    if department_update.new_name in custom_departments and department_update.new_name != department_update.old_name:
+    if (department_update.new_name != department_update.old_name
+            and _find_custom_department(custom_departments, department_update.new_name)):
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=f"Department '{department_update.new_name}' already exists"
        )

-    custom_departments = [
-        department_update.new_name if d == department_update.old_name else d
-        for d in custom_departments
-    ]
+    existing["name"] = department_update.new_name

    db_project.custom_departments = custom_departments
    flag_modified(db_project, 'custom_departments')
@@ -1275,7 +1298,7 @@ async def delete_department(
    """Delete a custom department (blocked if any member or task is currently using it)"""
    from models.task import Task

-    if department in STANDARD_DEPARTMENTS:
+    if department in STANDARD_DEPARTMENT_NAMES:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Standard departments cannot be deleted"
@@ -1291,7 +1314,7 @@ async def delete_department(

    custom_departments = db_project.custom_departments or []

-    if department not in custom_departments:
+    if not _find_custom_department(custom_departments, department):
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Custom department '{department}' not found"
@@ -1318,7 +1341,7 @@ async def delete_department(
            }
        )

-    custom_departments.remove(department)
+    custom_departments = [d for d in custom_departments if d["name"] != department]

    db_project.custom_departments = custom_departments
    flag_modified(db_project, 'custom_departments')
@@ -1336,6 +1359,240 @@ async def delete_department(
    return _build_all_departments_response(db_project)


+@router.post("/{project_id}/departments/{department}/task-types", status_code=status.HTTP_201_CREATED)
+async def add_department_task_type(
+    project_id: int,
+    department: str,
+    task_type_data: dict,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(require_coordinator_or_admin)
+):
+    """Add a task type to a custom department"""
+    from schemas.department import DepartmentTaskTypeCreate
+
+    try:
+        task_type_create = DepartmentTaskTypeCreate(**task_type_data)
+    except Exception as e:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=str(e)
+        )
+
+    db_project = db.query(Project).filter(Project.id == project_id).first()
+
+    if not db_project:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Project not found"
+        )
+
+    if department in STANDARD_DEPARTMENT_NAMES:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Standard departments' task types are fixed and cannot be modified"
+        )
+
+    custom_departments = db_project.custom_departments or []
+    existing = _find_custom_department(custom_departments, department)
+
+    if not existing:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Custom department '{department}' not found"
+        )
+
+    if task_type_create.task_type in existing["task_types"]:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=f"Task type '{task_type_create.task_type}' already exists in department '{department}'"
+        )
+
+    existing["task_types"].append(task_type_create.task_type)
+
+    db_project.custom_departments = custom_departments
+    flag_modified(db_project, 'custom_departments')
+
+    try:
+        db.commit()
+        db.refresh(db_project)
+    except Exception as e:
+        db.rollback()
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to add department task type"
+        )
+
+    return _build_all_departments_response(db_project)
+
+
+@router.put("/{project_id}/departments/{department}/task-types/{task_type}")
+async def rename_department_task_type(
+    project_id: int,
+    department: str,
+    task_type: str,
+    update_data: dict,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(require_coordinator_or_admin)
+):
+    """Rename a task type within a custom department, cascading the rename to matching tasks"""
+    from schemas.department import DepartmentTaskTypeUpdate
+    from models.task import Task
+
+    try:
+        task_type_update = DepartmentTaskTypeUpdate(**update_data)
+    except Exception as e:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=str(e)
+        )
+
+    if task_type != task_type_update.old_name:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Task type in URL does not match old_name in request body"
+        )
+
+    db_project = db.query(Project).filter(Project.id == project_id).first()
+
+    if not db_project:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Project not found"
+        )
+
+    if department in STANDARD_DEPARTMENT_NAMES:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Standard departments' task types are fixed and cannot be modified"
+        )
+
+    custom_departments = db_project.custom_departments or []
+    existing = _find_custom_department(custom_departments, department)
+
+    if not existing:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Custom department '{department}' not found"
+        )
+
+    if task_type_update.old_name not in existing["task_types"]:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Task type '{task_type_update.old_name}' not found in department '{department}'"
+        )
+
+    if (task_type_update.new_name != task_type_update.old_name
+            and task_type_update.new_name in existing["task_types"]):
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=f"Task type '{task_type_update.new_name}' already exists in department '{department}'"
+        )
+
+    existing["task_types"] = [
+        task_type_update.new_name if t == task_type_update.old_name else t
+        for t in existing["task_types"]
+    ]
+
+    db_project.custom_departments = custom_departments
+    flag_modified(db_project, 'custom_departments')
+
+    # Cascade rename to tasks using this task type within this department
+    tasks_to_update = db.query(Task).filter(
+        Task.project_id == project_id,
+        Task.department == department,
+        Task.task_type == task_type_update.old_name
+    ).all()
+    for task in tasks_to_update:
+        task.task_type = task_type_update.new_name
+
+    try:
+        db.commit()
+        db.refresh(db_project)
+    except Exception as e:
+        db.rollback()
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to rename department task type"
+        )
+
+    return _build_all_departments_response(db_project)
+
+
+@router.delete("/{project_id}/departments/{department}/task-types/{task_type}")
+async def delete_department_task_type(
+    project_id: int,
+    department: str,
+    task_type: str,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(require_coordinator_or_admin)
+):
+    """Delete a task type from a custom department (blocked if any task is currently using it)"""
+    from models.task import Task
+
+    db_project = db.query(Project).filter(Project.id == project_id).first()
+
+    if not db_project:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Project not found"
+        )
+
+    if department in STANDARD_DEPARTMENT_NAMES:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Standard departments' task types are fixed and cannot be modified"
+        )
+
+    custom_departments = db_project.custom_departments or []
+    existing = _find_custom_department(custom_departments, department)
+
+    if not existing:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Custom department '{department}' not found"
+        )
+
+    if task_type not in existing["task_types"]:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Task type '{task_type}' not found in department '{department}'"
+        )
+
+    tasks_using_task_type = db.query(Task).filter(
+        Task.project_id == project_id,
+        Task.department == department,
+        Task.task_type == task_type
+    ).all()
+
+    if tasks_using_task_type:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail={
+                "error": f"Cannot delete task type '{task_type}' because it is currently in use",
+                "department": department,
+                "task_type": task_type,
+                "task_count": len(tasks_using_task_type)
+            }
+        )
+
+    existing["task_types"] = [t for t in existing["task_types"] if t != task_type]
+
+    db_project.custom_departments = custom_departments
+    flag_modified(db_project, 'custom_departments')
+
+    try:
+        db.commit()
+        db.refresh(db_project)
+    except Exception as e:
+        db.rollback()
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to delete department task type"
+        )
+
+    return _build_all_departments_response(db_project)
+
+
 # Submission Configuration Endpoints

@router.get("/{project_id}/submission-config", response_model=ProjectSubmissionConfig)
@@ -13,6 +13,7 @@ from schemas.shot import (
    BulkShotCreate, BulkShotResponse, TaskStatusInfo
 )
 from utils.auth import get_current_user_from_token, require_permission
+from utils.departments import find_owning_department
 from services.shot_soft_deletion import ShotSoftDeletionService

 router = APIRouter()
@@ -131,26 +132,27 @@ def get_all_shot_task_types(project_id: int, db: Session) -> List[str]:
    return STANDARD_SHOT_TASK_TYPES + custom_types


-def create_default_tasks_for_shot(shot: Shot, task_types: List[str], db: Session):
+def create_default_tasks_for_shot(shot: Shot, task_types: List[str], db: Session, project: Project = None):
    """Create default tasks for a shot."""
    created_tasks = []
-    
+
    for task_type in task_types:
        task_name = f"{shot.name}_{task_type}"
        task_description = f"{task_type.title()} task for shot {shot.name}"
-        
+
        task = Task(
            project_id=shot.project_id,
            episode_id=shot.episode_id,
            shot_id=shot.id,
            task_type=task_type,
            name=task_name,
-            description=task_description
+            description=task_description,
+            department=find_owning_department(project, task_type) if project else None
        )
-        
+
        db.add(task)
        created_tasks.append(task)
-    
+
    return created_tasks


@@ -420,7 +422,8 @@ async def create_shot(
        all_task_types = get_all_shot_task_types(episode.project_id, db)
        # Use default standard types for now (can be customized via project settings)
        default_task_types = get_default_shot_task_types()
-        created_tasks = create_default_tasks_for_shot(db_shot, default_task_types, db)
+        project = db.query(Project).filter(Project.id == episode.project_id).first()
+        created_tasks = create_default_tasks_for_shot(db_shot, default_task_types, db, project)
        db.commit()
        task_count = len(created_tasks)
    
@@ -498,7 +501,8 @@ async def create_shots_bulk(
    
    created_shots = []
    total_tasks_created = 0
-    
+    project = db.query(Project).filter(Project.id == project_id).first()
+
    try:
        # Create all shots - validation already done above
        for i, shot_name in enumerate(shot_names_to_create):
@@ -525,7 +529,7 @@ async def create_shots_bulk(
            # Create default tasks if requested
            task_count = 0
            if bulk_shot.create_default_tasks:
-                created_tasks = create_default_tasks_for_shot(db_shot, task_types, db)
+                created_tasks = create_default_tasks_for_shot(db_shot, task_types, db, project)
                task_count = len(created_tasks)
                total_tasks_created += task_count
            
@@ -694,6 +698,7 @@ async def create_shot_task(
    
    # Create the task
    task_name = f"{shot.name} - {task_type.title()}"
+    project = db.query(Project).filter(Project.id == shot.project_id).first()
    db_task = Task(
        project_id=shot.project_id,
        episode_id=shot.episode_id,
@@ -701,7 +706,8 @@ async def create_shot_task(
        task_type=task_type,
        name=task_name,
        description=f"{task_type.title()} task for {shot.name}",
-        status="not_started"
+        status="not_started",
+        department=find_owning_department(project, task_type)
    )
    
    db.add(db_task)
@@ -25,6 +25,7 @@ from schemas.task import (
 from utils.auth import get_current_user_from_token, _get_user_from_db, require_role, require_permission, user_has_permission
 from utils.notifications import notification_service
 from utils.file_handler import file_handler
+from utils.departments import find_owning_department

 router = APIRouter()

@@ -415,10 +416,16 @@ async def create_task(
        # Validate the provided status
        if not validate_task_status(db, task.project_id, task_data['status']):
            raise HTTPException(
-                status_code=400, 
+                status_code=400,
                detail=f"Invalid status '{task_data['status']}' for this project"
            )
-    
+
+    # If the task type belongs to a department, that department wins over
+    # anything explicitly submitted for `department`.
+    owning_department = find_owning_department(project, task_data.get('task_type'))
+    if owning_department:
+        task_data['department'] = owning_department
+
    # Create task
    db_task = Task(**task_data)
    db.add(db_task)
@@ -829,7 +836,15 @@ async def update_task(
                status_code=400,
                detail=f"Invalid status '{update_data['status']}' for this project"
            )
-    
+
+    # If the task type is being changed to one owned by a department, that
+    # department wins over anything explicitly submitted for `department`.
+    if 'task_type' in update_data:
+        project = db.query(Project).filter(Project.id == task.project_id).first()
+        owning_department = find_owning_department(project, update_data['task_type'])
+        if owning_department:
+            update_data['department'] = owning_department
+
    # Update task
    for field, value in update_data.items():
        setattr(task, field, value)
@@ -2,23 +2,35 @@
 Pydantic schemas for department management
 """
 from pydantic import BaseModel, Field, validator
-from typing import List
+from typing import List, Literal
 import re

+DEPARTMENT_NAME_PATTERN = r'^[a-z0-9_]{2,50}$'
+

 class CustomDepartmentCreate(BaseModel):
    """Schema for creating a new custom department"""
    department: str = Field(..., min_length=2, max_length=50, description="Department name")
+    department_type: Literal["shot", "asset"] = Field(..., description="Whether this department applies to shots or assets")
+    task_types: List[str] = Field(default_factory=list, description="Task types owned by this department")

    @validator('department')
    def validate_department_name(cls, v):
        """Validate department name format"""
-        if not re.match(r'^[a-z0-9_]{2,50}$', v):
+        if not re.match(DEPARTMENT_NAME_PATTERN, v):
            raise ValueError(
                'Department name must be 2-50 characters, lowercase alphanumeric with underscores only'
            )
        return v

+    @validator('task_types', each_item=True)
+    def validate_task_type_name(cls, v):
+        if not re.match(DEPARTMENT_NAME_PATTERN, v):
+            raise ValueError(
+                'Task type name must be 2-50 characters, lowercase alphanumeric with underscores only'
+            )
+        return v
+

 class CustomDepartmentUpdate(BaseModel):
    """Schema for updating a custom department name"""
@@ -28,18 +40,52 @@ class CustomDepartmentUpdate(BaseModel):
    @validator('new_name')
    def validate_department_name(cls, v):
        """Validate department name format"""
-        if not re.match(r'^[a-z0-9_]{2,50}$', v):
+        if not re.match(DEPARTMENT_NAME_PATTERN, v):
            raise ValueError(
                'Department name must be 2-50 characters, lowercase alphanumeric with underscores only'
            )
        return v


+class DepartmentTaskTypeCreate(BaseModel):
+    """Schema for adding a task type to a custom department"""
+    task_type: str = Field(..., min_length=2, max_length=50, description="Task type name")
+
+    @validator('task_type')
+    def validate_task_type_name(cls, v):
+        if not re.match(DEPARTMENT_NAME_PATTERN, v):
+            raise ValueError(
+                'Task type name must be 2-50 characters, lowercase alphanumeric with underscores only'
+            )
+        return v
+
+
+class DepartmentTaskTypeUpdate(BaseModel):
+    """Schema for renaming a task type within a custom department"""
+    old_name: str = Field(..., description="Current task type name")
+    new_name: str = Field(..., min_length=2, max_length=50, description="New task type name")
+
+    @validator('new_name')
+    def validate_task_type_name(cls, v):
+        if not re.match(DEPARTMENT_NAME_PATTERN, v):
+            raise ValueError(
+                'Task type name must be 2-50 characters, lowercase alphanumeric with underscores only'
+            )
+        return v
+
+
+class DepartmentInfo(BaseModel):
+    """A single department: its name, whether it applies to shots or assets, and its owned task types"""
+    name: str
+    type: Literal["shot", "asset"]
+    task_types: List[str]
+
+
 class AllDepartmentsResponse(BaseModel):
    """Schema for response containing all departments (standard + custom)"""
-    departments: List[str] = Field(..., description="All departments")
-    standard_departments: List[str] = Field(..., description="Standard departments (read-only)")
-    custom_departments: List[str] = Field(..., description="Custom departments")
+    departments: List[DepartmentInfo] = Field(..., description="All departments")
+    standard_departments: List[DepartmentInfo] = Field(..., description="Standard departments (read-only)")
+    custom_departments: List[DepartmentInfo] = Field(..., description="Custom departments")


 class DepartmentInUseError(BaseModel):
@@ -48,3 +94,11 @@ class DepartmentInUseError(BaseModel):
    department: str = Field(..., description="Department that is in use")
    member_count: int = Field(..., description="Number of project members using this department")
    task_count: int = Field(..., description="Number of tasks using this department")
+
+
+class DepartmentTaskTypeInUseError(BaseModel):
+    """Schema for error when trying to delete a department task type in use"""
+    error: str = Field(..., description="Error message")
+    department: str = Field(..., description="Department the task type belongs to")
+    task_type: str = Field(..., description="Task type that is in use")
+    task_count: int = Field(..., description="Number of tasks using this task type")
@@ -0,0 +1,21 @@
+"""Shared helpers for resolving the department that owns a given task type."""
+from typing import Optional
+
+
+def find_owning_department(db_project, task_type: str) -> Optional[str]:
+    """Return the name of the department (standard or custom) whose task_types
+    list contains task_type for this project, or None if no department owns it."""
+    if not task_type:
+        return None
+
+    from routers.projects import STANDARD_DEPARTMENTS
+
+    for department in STANDARD_DEPARTMENTS:
+        if task_type in department["task_types"]:
+            return department["name"]
+
+    for department in (db_project.custom_departments or []):
+        if task_type in department.get("task_types", []):
+            return department["name"]
+
+    return None