"""Projects (Native PM Construction Platform Projects). Supports: - Project list with real-time statistics and scoping. - Creation, editing, archiving of native projects. - Project team members management. - Unique project code and UUID resolution. """ from __future__ import annotations import uuid from datetime import date from typing import Any from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel, Field from sqlalchemy import func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from app.authz import AuthContext, Perm, require_permission from app.db import get_db from app.errors import NotFoundError, ValidationError from app.models import ( Project, ProjectMember, ProjectStatus, Task, TaskRelation, TaskRelationType, ) from app.scheduling_service import ( calculate_schedule, check_for_cycle, rollup_parent_tasks, sync_project_lifecycle_dates, ) from app.serializers import _iso, project_dict, project_member_dict, task_relation_dict from app.utils.date_service import date_service from app.utils.jalali import to_jalali_str router = APIRouter(prefix="/projects", tags=["projects"]) ACTIVE_QC = ("pre_qc_pending", "exec_qc_pending", "ready") FAILED_QC = ("pre_qc_failed", "qc_failed_rework") view_projects = require_permission(Perm.projects_view) manage_projects = require_permission(Perm.projects_create) class ProjectCreateSchema(BaseModel): name: str = Field(..., min_length=1, max_length=255) description: str = "" location: str = "" project_manager: str = "" planned_start: str | None = None planned_end: str | None = None status: str = ProjectStatus.ACTIVE.value class ProjectUpdateSchema(BaseModel): name: str | None = None description: str | None = None location: str | None = None project_manager: str | None = None planned_start: str | None = None planned_end: str | None = None actual_start: str | None = None actual_end: str | None = None status: str | None = None progress: float | None = None class ProjectMemberCreateSchema(BaseModel): user_id: str role: str = "site_engineer" class ErpMappingSchema(BaseModel): erp_project_id: str | None = None class TaskRelationCreateSchema(BaseModel): from_task_id: str to_task_id: str relation_type: str = "FS" lag_days: int = 0 class TaskRelationUpdateSchema(BaseModel): """Edit an existing dependency. The endpoints (from/to tasks) are immutable — changing them is a different relation, so it must be deleted and recreated.""" relation_type: str | None = None lag_days: int | None = None async def _resolve_project(db: AsyncSession, ctx: AuthContext, project_id_or_op: str | int) -> Project: """Find a project by UUID and enforce scope.""" query = select(Project).options( selectinload(Project.members).selectinload(ProjectMember.user) ) try: val_uuid = uuid.UUID(str(project_id_or_op)) stmt = query.where(Project.id == val_uuid) except ValueError: raise NotFoundError("شناسه پروژه نامعتبر است") project = (await db.execute(stmt)).scalar_one_or_none() if not project: raise NotFoundError("پروژه مورد نظر یافت نشد") ctx.ensure_native_project(project.id) return project async def native_project_stats(db: AsyncSession, project_id: uuid.UUID) -> dict: """Task counters of one native project.""" today = date_service.today() rows = ( await db.execute( select(Task.qc_status, Task.planned_end, Task.status, func.count()) .where(Task.project_id == project_id) .group_by(Task.qc_status, Task.planned_end, Task.status) ) ).all() total = sum(r[3] for r in rows) completed = sum(r[3] for r in rows if r[2] == "COMPLETED") in_progress = sum(r[3] for r in rows if r[2] == "IN_PROGRESS") blocked = sum(r[3] for r in rows if r[2] == "BLOCKED") awaiting = sum(r[3] for r in rows if r[0] in ACTIVE_QC) failed = sum(r[3] for r in rows if r[0] in FAILED_QC) overdue = sum(r[3] for r in rows if r[2] != "COMPLETED" and r[1] and r[1] < today) return { "total": total, "completed": completed, "in_progress": in_progress, "blocked": blocked, "awaiting_qc": awaiting, "failed_qc": failed, "overdue": overdue, } @router.get("") async def list_projects( ctx: AuthContext = Depends(view_projects), db: AsyncSession = Depends(get_db), ): """List all projects visible to the user, with stats and team counts.""" stmt = select(Project).options(selectinload(Project.members)).order_by(Project.created_at.desc()) # Apply project scoping for supervisor / scoped roles (native projects = UUID) stmt = ctx.scope_native(stmt, Project.id) projects = (await db.execute(stmt)).scalars().all() results = [] for p in projects: stats = await native_project_stats(db, p.id) results.append(project_dict(p, member_count=len(p.members), task_stats=stats)) return {"data": results} @router.post("") async def create_project( payload: ProjectCreateSchema, ctx: AuthContext = Depends(manage_projects), db: AsyncSession = Depends(get_db), ): """Create a new native project.""" p_start = date_service.parse(payload.planned_start) if payload.planned_start else None p_end = date_service.parse(payload.planned_end) if payload.planned_end else None if p_start and p_end and p_end < p_start: raise ValidationError("تاریخ پایان نمی‌تواند قبل از تاریخ شروع باشد", field="planned_end") project = Project( name=payload.name.strip(), description=payload.description.strip(), location=payload.location.strip() if payload.location else "", project_manager=payload.project_manager.strip() if payload.project_manager else ctx.user.display_name, planned_start=p_start, planned_end=p_end, status=payload.status.upper(), ) db.add(project) await db.flush() # Automatically add the creator as a project manager member creator_member = ProjectMember( project_id=project.id, user_id=ctx.user.id, role="project_manager", ) db.add(creator_member) await db.commit() await db.refresh(project) stats = {"total": 0, "completed": 0, "in_progress": 0, "blocked": 0, "overdue": 0} res = project_dict(project, member_count=1, task_stats=stats) res["identifier"] = str(project.id) return {"data": res} @router.get("/{id}") async def get_project( id: str, ctx: AuthContext = Depends(view_projects), db: AsyncSession = Depends(get_db), ): """Get project details, member list, and overall stats.""" project = await _resolve_project(db, ctx, id) stats = await native_project_stats(db, project.id) data = project_dict(project, member_count=len(project.members), task_stats=stats) data["identifier"] = str(project.id) data["members"] = [project_member_dict(m) for m in project.members] return {"data": data} @router.patch("/{id}") async def update_project( id: str, payload: ProjectUpdateSchema, ctx: AuthContext = Depends(require_permission(Perm.projects_update)), db: AsyncSession = Depends(get_db), ): """Update project details, status, or dates.""" project = await _resolve_project(db, ctx, id) if payload.name is not None: project.name = payload.name.strip() if payload.description is not None: project.description = payload.description.strip() if payload.location is not None: project.location = payload.location.strip() if payload.project_manager is not None: project.project_manager = payload.project_manager.strip() if payload.status is not None: project.status = payload.status.upper() if payload.progress is not None: project.progress = max(0.0, min(100.0, payload.progress)) if payload.planned_start is not None: project.planned_start = date_service.parse(payload.planned_start) if payload.planned_end is not None: project.planned_end = date_service.parse(payload.planned_end) if payload.actual_start is not None: project.actual_start = date_service.parse(payload.actual_start) if payload.actual_end is not None: project.actual_end = date_service.parse(payload.actual_end) if project.planned_start and project.planned_end and project.planned_end < project.planned_start: raise ValidationError("تاریخ پایان نمی‌تواند قبل از تاریخ شروع باشد", field="planned_end") await db.commit() await db.refresh(project) stats = await native_project_stats(db, project.id) return {"data": project_dict(project, member_count=len(project.members), task_stats=stats)} @router.post("/{id}/members") async def add_project_member( id: str, payload: ProjectMemberCreateSchema, ctx: AuthContext = Depends(require_permission(Perm.projects_update)), db: AsyncSession = Depends(get_db), ): """Add a team member to the project.""" project = await _resolve_project(db, ctx, id) target_user_id = uuid.UUID(payload.user_id) # Check if already a member existing = await db.scalar( select(ProjectMember).where( ProjectMember.project_id == project.id, ProjectMember.user_id == target_user_id ) ) if existing: existing.role = payload.role else: member = ProjectMember( project_id=project.id, user_id=target_user_id, role=payload.role, ) db.add(member) await db.commit() # Return updated members list await db.refresh(project) project = await _resolve_project(db, ctx, project.id) return {"data": [project_member_dict(m) for m in project.members]} @router.delete("/{id}/members/{user_id}") async def remove_project_member( id: str, user_id: str, ctx: AuthContext = Depends(require_permission(Perm.projects_update)), db: AsyncSession = Depends(get_db), ): """Remove a team member from the project.""" project = await _resolve_project(db, ctx, id) target_user_id = uuid.UUID(user_id) member = await db.scalar( select(ProjectMember).where( ProjectMember.project_id == project.id, ProjectMember.user_id == target_user_id ) ) if member: await db.delete(member) await db.commit() project = await _resolve_project(db, ctx, project.id) return {"data": [project_member_dict(m) for m in project.members]} @router.patch("/{id}/erp-mapping") async def map_erp_project( id: str, payload: ErpMappingSchema, ctx: AuthContext = Depends(require_permission(Perm.projects_update)), db: AsyncSession = Depends(get_db), ): project = await _resolve_project(db, ctx, id) return {"data": {"id": str(project.id)}} # --- Relations & Dependencies --- @router.get("/{id}/relations") async def list_project_relations( id: str, ctx: AuthContext = Depends(view_projects), db: AsyncSession = Depends(get_db), ): """List all task dependencies in a project.""" project = await _resolve_project(db, ctx, id) stmt = ( select(TaskRelation) .options(selectinload(TaskRelation.from_task), selectinload(TaskRelation.to_task)) .where(TaskRelation.project_id == project.id) ) relations = (await db.execute(stmt)).scalars().all() return {"data": [task_relation_dict(r) for r in relations]} @router.post("/{id}/relations") async def create_task_relation( id: str, payload: TaskRelationCreateSchema, ctx: AuthContext = Depends(require_permission(Perm.tasks_create)), db: AsyncSession = Depends(get_db), ): """Create a new task dependency (FS, SS, FF, SF + lag). Enforces cycle check.""" project = await _resolve_project(db, ctx, id) from_id = uuid.UUID(payload.from_task_id) to_id = uuid.UUID(payload.to_task_id) if from_id == to_id: raise ValidationError("یک وظیفه نمی‌تواند پیش‌نیاز خودش باشد") # Verify tasks belong to this project from_task = await db.get(Task, from_id) to_task = await db.get(Task, to_id) if not from_task or from_task.project_id != project.id: raise ValidationError("وظیفه پیش‌نیاز در این پروژه یافت نشد") if not to_task or to_task.project_id != project.id: raise ValidationError("وظیفه وابسته در این پروژه یافت نشد") # Load existing relations to verify DAG existing_rels = ( await db.execute(select(TaskRelation.from_task_id, TaskRelation.to_task_id).where(TaskRelation.project_id == project.id)) ).all() existing_pairs = [(r[0], r[1]) for r in existing_rels] # Check for duplicate if (from_id, to_id) in existing_pairs: raise ValidationError("این وابستگی قبلاً ثبت شده است") # Cycle check if check_for_cycle(existing_pairs, from_id, to_id): raise ValidationError("این وابستگی باعث ایجاد چرخه (حلقه تکرار) در ساختار زمان‌بندی می‌شود") rtype = payload.relation_type.upper() if rtype not in ("FS", "SS", "FF", "SF"): raise ValidationError("نوع وابستگی باید یکی از مقادیر FS، SS، FF، SF باشد") relation = TaskRelation( project_id=project.id, from_task_id=from_id, to_task_id=to_id, relation_type=rtype, lag_days=payload.lag_days, ) db.add(relation) await db.commit() # Recalculate schedule cascade automatically all_tasks = (await db.execute(select(Task).where(Task.project_id == project.id))).scalars().all() all_relations = (await db.execute(select(TaskRelation).where(TaskRelation.project_id == project.id))).scalars().all() sched_result = calculate_schedule(all_tasks, all_relations, project.planned_start) # Update tasks with new early dates if changed for t in all_tasks: if t.id in sched_result["tasks_schedule"]: info = sched_result["tasks_schedule"][t.id] t.planned_start = info["planned_start"] t.planned_end = info["planned_end"] t.duration = info["duration"] await db.commit() await db.refresh(relation) return {"data": task_relation_dict(relation, from_task.title, to_task.title)} @router.patch("/{id}/relations/{relation_id}") async def update_task_relation( id: str, relation_id: str, payload: TaskRelationUpdateSchema, ctx: AuthContext = Depends(require_permission(Perm.tasks_create)), db: AsyncSession = Depends(get_db), ): """Edit a dependency's type and/or lag, then re-run the schedule cascade.""" project = await _resolve_project(db, ctx, id) rel = await db.scalar( select(TaskRelation) .options(selectinload(TaskRelation.from_task), selectinload(TaskRelation.to_task)) .where(TaskRelation.id == uuid.UUID(relation_id), TaskRelation.project_id == project.id) ) if rel is None: raise NotFoundError("وابستگی یافت نشد") if payload.relation_type is not None: rtype = payload.relation_type.upper() if rtype not in ("FS", "SS", "FF", "SF"): raise ValidationError("نوع وابستگی باید یکی از مقادیر FS، SS، FF، SF باشد") rel.relation_type = rtype if payload.lag_days is not None: rel.lag_days = payload.lag_days await db.commit() # Recalculate the schedule so the edit takes effect immediately all_tasks = (await db.execute(select(Task).where(Task.project_id == project.id))).scalars().all() all_relations = (await db.execute(select(TaskRelation).where(TaskRelation.project_id == project.id))).scalars().all() sched_result = calculate_schedule(all_tasks, all_relations, project.planned_start) for t in all_tasks: if t.id in sched_result["tasks_schedule"]: info = sched_result["tasks_schedule"][t.id] t.planned_start = info["planned_start"] t.planned_end = info["planned_end"] t.duration = info["duration"] await db.commit() await db.refresh(rel) return {"data": task_relation_dict(rel, rel.from_task.title, rel.to_task.title)} @router.delete("/{id}/relations/{relation_id}") async def delete_task_relation( id: str, relation_id: str, ctx: AuthContext = Depends(require_permission(Perm.tasks_delete)), db: AsyncSession = Depends(get_db), ): """Delete a task dependency.""" project = await _resolve_project(db, ctx, id) rel_uuid = uuid.UUID(relation_id) rel = await db.scalar( select(TaskRelation).where(TaskRelation.id == rel_uuid, TaskRelation.project_id == project.id) ) if rel: await db.delete(rel) await db.commit() return {"data": {"deleted": True}} # --- Interactive Gantt & Topological Scheduling --- @router.get("/{id}/gantt") async def get_project_gantt( id: str, ctx: AuthContext = Depends(view_projects), db: AsyncSession = Depends(get_db), ): """Return full Gantt chart dataset with CPM calculations, relations, and critical path.""" project = await _resolve_project(db, ctx, id) # 1. Fetch tasks, relations, milestones tasks = ( await db.execute( select(Task) .options(selectinload(Task.children)) .where(Task.project_id == project.id) .order_by(Task.planned_start.asc().nulls_last(), Task.created_at.asc()) ) ).scalars().all() relations = ( await db.execute( select(TaskRelation) .options(selectinload(TaskRelation.from_task), selectinload(TaskRelation.to_task)) .where(TaskRelation.project_id == project.id) ) ).scalars().all() # 2. Run CPM scheduling analysis sched = calculate_schedule(tasks, relations, project.planned_start) sched_info = sched["tasks_schedule"] crit_set = set(sched["critical_path"]) # 3. Format tasks formatted_tasks = [] for t in tasks: t_sched = sched_info.get(t.id, {}) start_dt = t_sched.get("planned_start") or t.planned_start end_dt = t_sched.get("planned_end") or t.planned_end formatted_tasks.append({ "id": str(t.id), "title": t.title, "subject": t.title, "parent_id": str(t.parent_id) if t.parent_id else None, "status": t.status, "priority": t.priority, "planned_start": _iso(start_dt), "planned_end": _iso(end_dt), "planned_start_jalali": to_jalali_str(start_dt), "planned_end_jalali": to_jalali_str(end_dt), "duration": t_sched.get("duration") or t.duration or 1, "progress": t.progress or 0, "is_critical": t.id in crit_set, "float_days": t_sched.get("float_days", 0), "qc_status": t.qc_status or "none", "assignee_name": t.assignee_name or (t.assignee.display_name if getattr(t, "assignee", None) else None), }) return { "data": { "project": { "id": str(project.id), "identifier": str(project.id), "name": project.name, "planned_start": _iso(project.planned_start), "planned_end": _iso(project.planned_end), "planned_start_jalali": to_jalali_str(project.planned_start), "planned_end_jalali": to_jalali_str(project.planned_end), "progress": project.progress, }, "tasks": formatted_tasks, "relations": [task_relation_dict(r) for r in relations], "critical_path": [str(tid) for tid in sched["critical_path"]], "project_finish": _iso(sched.get("project_finish")), "project_finish_jalali": to_jalali_str(sched.get("project_finish")), } } @router.post("/{id}/reschedule") async def reschedule_project( id: str, ctx: AuthContext = Depends(require_permission(Perm.projects_update)), db: AsyncSession = Depends(get_db), ): """Trigger full topological forward-pass recalculation and persist cascaded dates.""" project = await _resolve_project(db, ctx, id) tasks = (await db.execute(select(Task).where(Task.project_id == project.id))).scalars().all() relations = (await db.execute(select(TaskRelation).where(TaskRelation.project_id == project.id))).scalars().all() sched = calculate_schedule(tasks, relations, project.planned_start) sched_info = sched["tasks_schedule"] updated_count = 0 for t in tasks: if t.id in sched_info: info = sched_info[t.id] t.planned_start = info["planned_start"] t.planned_end = info["planned_end"] t.duration = info["duration"] updated_count += 1 # Roll up summary tasks rollups = rollup_parent_tasks(tasks) for pid, rdata in rollups.items(): parent_task = await db.get(Task, pid) if parent_task: if rdata["planned_start"]: parent_task.planned_start = rdata["planned_start"] if rdata["planned_end"]: parent_task.planned_end = rdata["planned_end"] if rdata.get("duration"): parent_task.duration = rdata["duration"] await sync_project_lifecycle_dates(db, project.id) await db.commit() return {"data": {"recalculated": True, "tasks_updated": updated_count}}