from typing import List from fastapi import HTTPException from app.api.company.schemas import Company, Role, Seafarer, Vessel, VesselTypesResponse import requests class CompanyService: def __init__(self): self.api_url = "https://cms.marinerskills.com/api/get-all-company-list" self.security_key = {"SecurityKey": "FslKvipEQ3hT2PfdZla00hp"} self.vessels_api_url = "https://cms.marinerskills.com/api/get-vessel-based-on-company" self.vessels_security_key = "FslKvipEQ3hT2PfdZla00hp" self.seafarer_api_url = "https://cms.marinerskills.com/api/get-seafarer-based-on-vessel" self.seafarer_security_key = "FslKvipEQ3hT2PfdZla00hp" self.role_api_url = "https://cms.marinerskills.com/api/get-roles-based-on-company" self.role_security_key = "FslKvipEQ3hT2PfdZla00hp" def get_all_companies(self) -> List[Company]: try: response = requests.post(self.api_url, json=self.security_key, timeout=10) response.raise_for_status() data = response.json() # Optionally validate response format if not isinstance(data, list): raise ValueError("Unexpected API response format") return [Company(**c) for c in data] except requests.RequestException as e: raise HTTPException(status_code=502, detail=f"External API error: {e}") except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # def get_vessels_by_company(self, company_id: int) -> List[Vessel]: # try: # payload = {"CompanyID": str(company_id), "SecurityKey": self.vessels_security_key} # response = requests.post(self.vessels_api_url, json=payload, timeout=10) # response.raise_for_status() # data = response.json() # # Check if the response contains the vessel list # if isinstance(data, list): # vessels = data # elif isinstance(data, dict) and "data" in data and isinstance(data["data"], list): # vessels = data["data"] # else: # raise ValueError(f"Unexpected API response format: {data}") # if not vessels: # raise HTTPException(status_code=404, detail="No vessels found for this companyId") # return [Vessel(**v) for v in vessels] # except requests.RequestException as e: # raise HTTPException(status_code=502, detail=f"External API error: {e}") # except Exception as e: # raise HTTPException(status_code=500, detail=str(e)) def get_vessels_by_company(self, company_id: int) -> List[Vessel]: try: payload = {"CompanyID": str(company_id), "SecurityKey": self.vessels_security_key} response = requests.post(self.vessels_api_url, json=payload, timeout=10) response.raise_for_status() data = response.json() if isinstance(data, list): vessels = data elif isinstance(data, dict) and isinstance(data.get("data"), list): vessels = data["data"] else: raise ValueError(f"Unexpected API response: {data}") fixed = [] for v in vessels: fixed.append({ "VesselID": v.get("vesselid") or v.get("VesselID"), "VesselName": v.get("vesselname") or v.get("VesselName"), "VesselType": v.get("vesseltype") or v.get("VesselType"), "VesselStds": v.get("vesselstds") or v.get("VesselStds"), "CompanyID": v.get("companyid") or v.get("CompanyID"), }) return [Vessel(**f) for f in fixed] except requests.RequestException as e: raise HTTPException(status_code=502, detail=f"External API error: {e}") except Exception as e: raise HTTPException(status_code=500, detail=str(e)) def get_seafarers_by_vessel(self, vessel_id: int) -> List[Seafarer]: try: # Prepare payload for third-party API payload = {"VesselID": str(vessel_id), "SecurityKey": self.seafarer_security_key} response = requests.post(self.seafarer_api_url, json=payload, timeout=10) response.raise_for_status() data = response.json() # Handle different possible response formats if isinstance(data, list): seafarers = data elif isinstance(data, dict): if "data" in data and isinstance(data["data"], list): seafarers = data["data"] else: raise ValueError(f"Unexpected API response format: {data}") else: raise ValueError("Invalid response from external API") if not seafarers: raise HTTPException(status_code=404, detail="No seafarers found for this vesselId") # Convert each dict to a Seafarer model return [Seafarer(**s) for s in seafarers] except requests.RequestException as e: raise HTTPException(status_code=502, detail=f"External API error: {e}") except Exception as e: raise HTTPException(status_code=500, detail=str(e)) def get_roles(self, company_id: str) -> List[Role]: payload = { "CompanyID": company_id, "SecurityKey": self.role_security_key } try: response = requests.post(self.role_api_url, json=payload, timeout=30) response.raise_for_status() data = response.json() if not isinstance(data, list): raise ValueError("Unexpected API response format") return [Role(**item) for item in data] except requests.RequestException as e: raise HTTPException(status_code=502, detail=f"External API error: {e}") except Exception as e: raise HTTPException(status_code=500, detail=str(e)) def get_vessel_types(self, company_id: int) -> VesselTypesResponse: try: payload = {"CompanyID": str(company_id), "SecurityKey": self.vessels_security_key} response = requests.post(self.vessels_api_url, json=payload, timeout=10) response.raise_for_status() data = response.json() # Normalize API response if isinstance(data, list): vessels = data elif isinstance(data, dict) and "data" in data: vessels = data["data"] else: raise ValueError(f"Unexpected API response format: {data}") # Convert to Pydantic models vessel_models = [Vessel(**v) for v in vessels] # Extract unique vessel types vessel_types = list({ v.VesselType.strip() for v in vessel_models if v.VesselType and v.VesselType.strip() }) if not vessel_types: raise HTTPException( status_code=404, detail="No vessel types found for this companyId" ) return VesselTypesResponse(vessel_types=vessel_types) except requests.RequestException as e: raise HTTPException(status_code=502, detail=f"External API error: {e}") except Exception as e: raise HTTPException(status_code=500, detail=str(e))