import random from typing import Optional from sqlalchemy import Delete, Select, and_, text from sqlalchemy.orm import selectinload from src.auth.service import CurrentUser from src.database.core import CollectorDbSession, DbSession from src.database.service import CommonParameters, search_filter_sort_paginate from .model import ScopeEquipmentPart from .schema import ScopeEquipmentActivityCreate, ScopeEquipmentActivityUpdate # async def get(*, db_session: DbSession, scope_equipment_activity_id: str) -> Optional[ScopeEquipmentActivity]: # """Returns a document based on the given document id.""" # result = await db_session.get(ScopeEquipmentActivity, scope_equipment_activity_id) # return result from typing import Optional, List, Dict, Any from sqlalchemy.ext.asyncio import AsyncSession as DbSession from sqlalchemy.sql import text import logging logger = logging.getLogger(__name__) # async def get_all( # db_session: CollectorDbSession, # location_tag: Optional[str] = None, # start_year: int = 2023, # end_year: Optional[int] = None, # parent_wonum: Optional[str] = None # ) -> List[Dict[str, Any]]: # """ # Retrieve overhaul spare parts consumption data. # Handles missing data, null parent WO, and query safety. # Args: # db_session: Async SQLAlchemy session # location_tag: Optional location filter # start_year: Year to start analysis (default 2023) # end_year: Optional year to end analysis (default start_year + 1) # parent_wonum: Parent work order number (required for context) # Returns: # List of dictionaries with spare part usage per overhaul WO. # """ # # --- 1. Basic validation --- # if not parent_wonum: # logger.warning("Parent WO number not provided. Returning empty result.") # return [] # if start_year < 1900 or (end_year and end_year < start_year): # raise ValueError("Invalid year range provided.") # if end_year is None: # end_year = start_year + 1 # # --- 2. Build SQL safely --- # base_query = """ # WITH filtered_wo AS ( # SELECT wonum, location_tag # FROM public.wo_max # WHERE worktype = 'OH' # AND xx_parent = :parent_wonum # """ # params = { # "parent_wonum": parent_wonum, # } # if location_tag: # base_query += " AND location_tag = :location_tag" # params["location_tag"] = location_tag # base_query += """ # ), # filtered_materials AS ( # SELECT wonum, itemnum, itemqty, inv_curbaltotal, inv_avgcost # FROM public.wo_max_material # WHERE wonum IN (SELECT wonum FROM filtered_wo) # ) # SELECT # fwo.location_tag AS location_tag, # fm.itemnum, # spl.description AS sparepart_name, # COALESCE(SUM(fm.itemqty), 0) AS parts_consumed_in_oh, # COALESCE(AVG(fm.inv_avgcost), 0) AS avgcost, # COALESCE(AVG(fm.inv_curbaltotal), 0) AS inv_curbaltotal # FROM filtered_wo fwo # INNER JOIN filtered_materials fm ON fwo.wonum = fm.wonum # LEFT JOIN public.maximo_sparepart_pr_po_line spl ON fm.itemnum = spl.item_num # GROUP BY fwo.location_tag, fm.itemnum, spl.description # ORDER BY fwo.location_tag, fm.itemnum; # """ # # --- 3. Execute query --- # try: # result = await db_session.execute(text(base_query), params) # rows = result.fetchall() # # Handle "no data found" # if not rows: # logger.info(f"No spare part data found for parent WO {parent_wonum}.") # return [] # # --- 4. Map results cleanly --- # equipment_parts = [] # for row in rows: # try: # equipment_parts.append({ # "location_tag": row.location_tag, # "itemnum": row.itemnum, # "sparepart_name": row.sparepart_name or "-", # "parts_consumed_in_oh": float(row.parts_consumed_in_oh or 0), # "avgcost": float(row.avgcost or 0), # "inv_curbaltotal": float(row.inv_curbaltotal or 0) # }) # except Exception as parse_err: # logger.error(f"Failed to parse row {row}: {parse_err}") # continue # Skip malformed rows # return equipment_parts # except Exception as e: # logger.exception(f"Database query failed: {e}") # raise RuntimeError("Failed to fetch overhaul spare parts data.") from e from typing import List, Dict, Any, Optional from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.sql import text async def get_all( db_session: AsyncSession, location_tag: Optional[str] = None, start_year: int = 2023, end_year: Optional[int] = None ) -> List[Dict[str, Any]]: """ Get overhaul spare parts consumption data with optimized query. Args: db_session: SQLAlchemy async database session location_tag: Optional filter for location (asset_location) start_year: Starting year (default: 2023) end_year: Ending year (default: start_year + 1) Returns: List of dictionaries with spare parts consumption data """ # Set default end year if end_year is None: end_year = start_year + 1 # Build query dynamically query_str = """ WITH filtered_wo AS ( SELECT DISTINCT wonum, asset_location, asset_unit FROM public.wo_maximo ma WHERE ma.xx_parent IN ('155026', '155027', '155029', '155030') """ params = {} # Optional filter for location if location_tag: query_str += " AND asset_location = :location_tag" params["location_tag"] = location_tag query_str += """ ), filtered_materials AS ( SELECT mat.wonum, mat.itemnum, mat.itemqty, mat.inv_curbaltotal AS inv_curbaltotal, mat.inv_avgcost AS inv_avgcost FROM public.wo_maximo_material AS mat WHERE mat.wonum IN (SELECT wonum FROM filtered_wo) ) SELECT fwo.asset_location AS location_tag, ft.itemnum, COALESCE(spl.description, 'Unknown') AS sparepart_name, AVG(ft.itemqty) AS total_parts_used, COALESCE(AVG(ft.inv_avgcost), 0) AS avg_cost, COALESCE(AVG(ft.inv_curbaltotal), 0) AS avg_inventory_balance FROM filtered_wo AS fwo INNER JOIN filtered_materials AS ft ON fwo.wonum = ft.wonum LEFT JOIN public.maximo_sparepart_pr_po_line AS spl ON ft.itemnum = spl.item_num GROUP BY fwo.asset_location, ft.itemnum, spl.description ORDER BY fwo.asset_location, ft.itemnum; """ try: result = await db_session.execute(text(query_str), params) rows = result.fetchall() equipment_parts = [] for row in rows: equipment_parts.append({ "location_tag": row.location_tag, "itemnum": row.itemnum, "sparepart_name": row.sparepart_name, "parts_consumed_in_oh": float(row.total_parts_used or 0), "avg_cost": float(row.avg_cost or 0), "inv_curbaltotal": float(row.avg_inventory_balance or 0), }) return equipment_parts except Exception as e: print(f"[get_all] Database query error: {e}") raise