"""Concrete PostgreSQL Equipment Repository implementing EquipmentRepositoryPort. Strict Zero-Tolerance Synthetic Data & Hexagonal Architecture COMPLIANT. """ from __future__ import annotations from datetime import date from decimal import Decimal from typing import Any from app.core.postgres.project_crud_repository import ProjectCrudRepository from app.modules.operations.domain.ports.equipment_repository_port import EquipmentRepositoryPort class PostgresEquipmentRepository(EquipmentRepositoryPort): """PostgreSQL database adapter for equipment and telematics logs.""" def __init__(self, db_repo: ProjectCrudRepository | None = None) -> None: self.db = db_repo or ProjectCrudRepository() def get_connection(self): """Context manager yielding a database connection (for backward compatibility).""" return self.db.get_connection() def record_fuel_log( self, equipment_id: str, fill_date: date, fuel_quantity_liters: Decimal, unit_price_vnd: Decimal, total_cost_vnd: Decimal, odometer_hours: Decimal, source_invoice_id: str | None = None, ) -> dict[str, Any]: """Record an equipment fuel fill entry in erp_equipment_fuel_logs.""" sql = """ INSERT INTO erp_equipment_fuel_logs ( equipment_id, fill_date, fuel_quantity_liters, unit_price_vnd, total_cost_vnd, odometer_hours, source_invoice_id ) VALUES ( %s, %s, %s, %s, %s, %s, %s ) RETURNING id, created_at; """ with self.db.get_connection() as conn, conn.cursor() as cur: cur.execute( sql, ( equipment_id, fill_date, fuel_quantity_liters, unit_price_vnd, total_cost_vnd, odometer_hours, source_invoice_id, ), ) res = cur.fetchone() conn.commit() if res: return { "id": str(res["id"]), "created_at": res["created_at"].strftime("%Y-%m-%d %H:%M:%S") if hasattr(res["created_at"], "strftime") else str(res["created_at"]), } return {} def get_equipment_by_id_or_code(self, identifier: str) -> dict[str, Any] | None: """Fetch equipment details by either UUID id or equipment_code.""" sql = """ SELECT e.*, p.project_name AS current_project_name FROM erp_equipment e LEFT JOIN projects p ON e.current_project_id = p.id WHERE e.id::text = %s OR e.equipment_code = %s; """ with self.db.get_connection() as conn, conn.cursor() as cur: cur.execute(sql, (identifier, identifier)) return cur.fetchone() def get_fuel_logs_by_equipment(self, equipment_id: str) -> list[dict[str, Any]]: """Fetch fuel logs for an equipment ordered by fill_date desc.""" sql = """ SELECT f.*, i.invoice_number, i.invoice_series FROM erp_equipment_fuel_logs f LEFT JOIN erp_invoices i ON f.source_invoice_id = i.id WHERE f.equipment_id::text = %s ORDER BY f.fill_date DESC, f.created_at DESC; """ with self.db.get_connection() as conn, conn.cursor() as cur: cur.execute(sql, (str(equipment_id),)) return cur.fetchall() or [] def get_total_operated_hours(self, equipment_id: str) -> Decimal: """Calculate total operational hours from ca máy logs.""" sql = """ SELECT SUM(hours_operated) AS total_hours FROM erp_equipment_logs WHERE equipment_id::text = %s; """ with self.db.get_connection() as conn, conn.cursor() as cur: cur.execute(sql, (str(equipment_id),)) res = cur.fetchone() if res and res.get("total_hours") is not None: return Decimal(str(res["total_hours"])) return Decimal("0.0000") def get_fleet_counts(self) -> dict[str, Any]: """Fetch summary counts for trucks, excavators, and active machines.""" sql = """ SELECT count(id) as total_equipment, count(id) FILTER (WHERE equipment_code LIKE 'EQ-TRUTH-TRU-%') as total_trucks, count(id) FILTER (WHERE equipment_code LIKE 'EQ-TRUTH-EXC-%') as total_excavators, count(id) FILTER (WHERE status = 'operating') as operating_count, count(id) FILTER (WHERE license_plate IS NOT NULL) as identified_plates FROM erp_equipment; """ with self.db.get_connection() as conn, conn.cursor() as cur: cur.execute(sql) return cur.fetchone() or {} def get_fleet_fuel_totals(self) -> dict[str, Any]: """Fetch summary totals of all fuel fills from GDT tax invoices.""" sql = """ SELECT count(id) as total_fills, sum(fuel_quantity_liters) as total_liters, sum(total_cost_vnd) as total_cost, avg(unit_price_vnd) as avg_price, min(fill_date) as start_date, max(fill_date) as latest_fill_date FROM erp_equipment_fuel_logs; """ with self.db.get_connection() as conn, conn.cursor() as cur: cur.execute(sql) return cur.fetchone() or {} def get_fleet_equipment_summary(self) -> list[dict[str, Any]]: """Fetch aggregated equipment list with fuel totals and odometer stats.""" sql = """ SELECT e.id, e.equipment_code, e.equipment_name, e.license_plate, e.fuel_norm_per_hour, e.status, count(fl.id) as fill_count, sum(fl.fuel_quantity_liters) as total_liters, sum(fl.total_cost_vnd) as total_cost, max(fl.odometer_hours) as max_odo, min(fl.odometer_hours) as min_odo FROM erp_equipment e LEFT JOIN erp_equipment_fuel_logs fl ON fl.equipment_id = e.id WHERE e.license_plate IS NOT NULL OR e.equipment_code LIKE 'EQ-TRUTH-EXC-%' GROUP BY e.id, e.equipment_code, e.equipment_name, e.license_plate, e.fuel_norm_per_hour, e.status ORDER BY total_liters DESC NULLS LAST; """ with self.db.get_connection() as conn, conn.cursor() as cur: cur.execute(sql) return cur.fetchall() or [] def get_active_equipment_for_telematics(self, limit: int = 10) -> list[dict[str, Any]]: """Fetch active equipment along with their actual fuel and hour stats for telematics dashboard.""" sql = """ SELECT e.id, e.equipment_code, e.equipment_name, e.license_plate, e.fuel_norm_per_hour, e.status, count(fl.id) as fill_count, COALESCE(sum(fl.fuel_quantity_liters), 0) as total_liters, COALESCE(max(fl.odometer_hours), 0) as max_odo, COALESCE(min(fl.odometer_hours), 0) as min_odo FROM erp_equipment e LEFT JOIN erp_equipment_fuel_logs fl ON fl.equipment_id = e.id WHERE e.status = 'operating' OR fl.id IS NOT NULL GROUP BY e.id, e.equipment_code, e.equipment_name, e.license_plate, e.fuel_norm_per_hour, e.status ORDER BY total_liters DESC NULLS LAST LIMIT %s; """ with self.db.get_connection() as conn, conn.cursor() as cur: cur.execute(sql, (limit,)) return cur.fetchall() or []