from datetime import datetime, timedelta from typing import Optional, Union from fastapi import HTTPException, status from sqlalchemy import Delete, Select, and_, desc, func, not_, or_ from sqlalchemy.dialects.postgresql import insert from sqlalchemy.orm import selectinload from src.auth.service import CurrentUser from src.database.core import DbSession, CollectorDbSession from src.utils import update_model from src.database.service import CommonParameters, search_filter_sort_paginate from src.overhaul_scope.model import OverhaulScope from src.standard_scope.enum import ScopeEquipmentType from src.standard_scope.model import EquipmentOHHistory from src.workorder.model import MasterWorkOrder from src.equipment_workscope_group.model import EquipmentWorkscopeGroup from src.workscope_group.model import MasterActivity from src.workscope_group_maintenance_type.model import WorkscopeOHType from src.overhaul_scope.model import MaintenanceType from src.overhaul_scope.service import get as get_overhaul from src.maximo.service import get_history_oh_wo from .model import MasterEquipment, MasterEquipmentTree, StandardScope from .schema import ScopeEquipmentCreate, ScopeEquipmentUpdate from uuid import UUID async def get_by_location_tag(*, db_session: DbSession, location_tag: str): query = ( Select(StandardScope) .filter(StandardScope.location_tag == location_tag) .options(selectinload(StandardScope.master_equipment)) ) result = await db_session.execute(query) return result.unique().scalars().one_or_none() async def get_all(*, common, oh_scope: Optional[str] = None): """Returns all documents.""" query = Select(StandardScope).options( selectinload(StandardScope.master_equipment) ) query = query.order_by(desc(StandardScope.created_at)).options(selectinload(StandardScope.master_equipment)) if oh_scope: query = ( query .outerjoin(StandardScope.oh_history) # Use outerjoin to handle None values .join(StandardScope.workscope_groups) .join(EquipmentWorkscopeGroup.workscope_group) .join(MasterActivity.oh_types) .join(WorkscopeOHType.oh_type) .join(StandardScope.master_equipment) .filter(MasterEquipment.location_tag.is_not(None)) .filter(MaintenanceType.name == oh_scope) # .filter( # (StandardScope.is_alternating_oh == False) | # (StandardScope.oh_history == None) | # (StandardScope.oh_history.has(EquipmentOHHistory.last_oh_type != selected_overhaul.maintenance_type.name)) # ).distinct() ) results = await common['db_session'].execute(query) items = results.scalars().all() return { "items": items, "total": len(items) } async def get_by_oh_session_id(*, db_session: DbSession, oh_session_id: UUID): overhaul = await get_overhaul(db_session=db_session, overhaul_session_id=oh_session_id) query = ( Select(StandardScope) .outerjoin(StandardScope.oh_history) # Use outerjoin to handle None values .join(StandardScope.workscope_groups) .join(EquipmentWorkscopeGroup.workscope_group) .join(MasterActivity.oh_types) .join(WorkscopeOHType.oh_type) .join(MasterEquipment, StandardScope.location_tag == MasterEquipment.location_tag) .filter(MaintenanceType.name == overhaul.maintenance_type.name).filter( (StandardScope.is_alternating_oh == False) | (StandardScope.oh_history == None) | (StandardScope.oh_history.has(EquipmentOHHistory.last_oh_type != overhaul.maintenance_type.name)) ).distinct() ) result = await db_session.execute(query) return result.scalars().all(), overhaul async def create(*, db_session: DbSession, scope_equipment_in: ScopeEquipmentCreate): """Creates a new document.""" # scope_equipment = StandardScope(**scope_equipment_in.model_dump()) assetnums = scope_equipment_in.assetnums results = [] removal_date = scope_equipment_in.removal_date if scope_equipment_in.type == ScopeEquipmentType.TEMP: # Search for the next or ongoing overhaul session for the given scope stmt = ( Select(OverhaulScope.end_date) .where( OverhaulScope.type == scope_equipment_in.scope_name, (OverhaulScope.start_date <= datetime.now()) & (OverhaulScope.end_date >= datetime.now()) # Ongoing | (OverhaulScope.start_date > datetime.now()), # Upcoming ) .order_by(OverhaulScope.start_date.asc()) .limit(1) ) result = await db_session.execute(stmt) removal_date = result.scalar_one_or_none() # If no overhaul found, set a default removal date or handle the error if removal_date is None: # Handle if no overhaul session is found, set default or raise an error removal_date = datetime.now() + timedelta( days=30 ) # Example: 30 days from now for assetnum in assetnums: stmt = insert(StandardScope).values( assetnum=assetnum, scope_overhaul=scope_equipment_in.scope_name, type=scope_equipment_in.type, removal_date=removal_date, ) stmt = stmt.on_conflict_do_nothing( index_elements=["assetnum", "scope_overhaul"] ) await db_session.execute(stmt) results.append(assetnum) # await db_session.commit() return results async def update( *, db_session: DbSession, scope_equipment: StandardScope, scope_equipment_in: ScopeEquipmentUpdate ): """Updates a document.""" data = scope_equipment_in.model_dump() # update_data = scope_equipment_in.model_dump(exclude_defaults=True) # update_model(scope_equipment, update_data) # await db_session.commit() return scope_equipment async def delete(*, db_session: DbSession, assetnum: str): """Deletes a document.""" # query = Delete(StandardScope).where(StandardScope.assetnum == assetnum) # await db_session.execute(query) # await db_session.commit() return assetnum async def get_by_oh_scope( *, db_session: DbSession, oh_scope: str ): pass query = (Select(StandardScope) .outerjoin(StandardScope.oh_history) # Use outerjoin to handle None values .join(StandardScope.workscope_groups) .join(EquipmentWorkscopeGroup.workscope_group) .join(MasterActivity.oh_types) .join(WorkscopeOHType.oh_type) .filter(MaintenanceType.name == oh_scope) # .filter( # (StandardScope.is_alternating_oh == False) | # (StandardScope.oh_history == None) | # (StandardScope.oh_history.has(EquipmentOHHistory.last_oh_type != selected_overhaul.maintenance_type.name)) # ).distinct() ) results = await db_session.execute(query) return results.scalars().all() async def get_all_master_equipment(*, common: CommonParameters, scope_name): equipments_scope = [ equip.location_tag for equip in await get_by_oh_scope( db_session=common.get("db_session"), oh_scope=scope_name ) ] query = Select(MasterEquipment).filter(MasterEquipment.location_tag.is_not(None)) # Only add not_in filter if there are items in equipments_scope if equipments_scope: query = query.filter(MasterEquipment.location_tag.not_in(equipments_scope)) results = await search_filter_sort_paginate(model=query, **common) return results async def get_equipment_level_by_no(*, db_session: DbSession, level: int): query = ( Select(MasterEquipment) .join(MasterEquipment.equipment_tree) .where(MasterEquipmentTree.level_no == level) ) result = await db_session.execute(query) return result.scalars().all() async def get_history_standard_scope_wo_service(*, db_session: DbSession, collector_db_session:CollectorDbSession, oh_session_id:UUID): planning_oh_data = await get_by_oh_session_id(db_session=db_session, oh_session_id=oh_session_id) planning_scopes = planning_oh_data[0] overhaul = planning_oh_data[1] results = await get_history_oh_wo( db_session=db_session, collector_db_session=collector_db_session, oh_session_id=oh_session_id, parent_wo_num=overhaul.wo_parent ) scope_cost_map = {scope.location_tag: scope.service_cost for scope in planning_scopes} for result in results: result["planning_service_cost"] = scope_cost_map.get(result["location_tag"], 0) return results