#!/usr/bin/env python3 """ Vessel Committed Cost Analyzer - Final Production Script Analyzes committed costs for vessels using direct API calls with ThreadPoolExecutor Environment Variables (optional when using MCP tool): - MONGODB_URI: MongoDB connection string - MONGODB_DATABASE: Database name (default: syia-etl-dev) - EYESHARE_BASE_URL: EyeShare API base URL - EYESHARE_CLIENT_ID: EyeShare client ID - EYESHARE_CLIENT_SECRET: EyeShare client secret - EYESHARE_MODULE: EyeShare module (default: purchaseorder) When executed through MCP's generate_committed_cost_report tool, environment variables are automatically provided from MCP server configuration. """ import pandas as pd import json import time import logging import os import sys import argparse from datetime import datetime from typing import List, Dict, Tuple, Optional from concurrent.futures import ThreadPoolExecutor, as_completed import requests from dotenv import load_dotenv import pymongo from urllib.parse import quote_plus import numpy as np # Load environment variables load_dotenv() # Initialize logging - will be configured later after output_dir is known logger = logging.getLogger(__name__) class VesselCommittedCostAnalyzer: """ Complete vessel committed cost analyzer with direct API calls and ThreadPoolExecutor """ def __init__(self, vessel_code: str, end_date: str, output_dir: str = '.'): self.vessel_code = vessel_code.upper() self.end_date = end_date self.output_dir = os.path.abspath(output_dir) self.timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") # Configure logging to write to output directory self._setup_logging() # Configuration - requires environment variables self.config = { 'mongodb_uri': os.getenv('MONGODB_URI', 'mongodb://user:g0YI30j97s4E@db.syia.ai:27017/?authMechanism=SCRAM-SHA-1&authSource=syia-etl-dev'), 'mongodb_uri_dev': os.getenv('MONGODB_URI_DEV', 'mongodb://user:g0YI30j97s4E@db.syia.ai:27017/?authMechanism=SCRAM-SHA-1&authSource=syia-etl-dev'), 'mongodb_database': os.getenv('MONGODB_DATABASE', 'syia-etl-dev'), 'eyeshare_base_url': os.getenv('EYESHARE_BASE_URL', 'https://synergymarine.eye-share.com'), 'eyeshare_client_id': os.getenv('EYESHARE_CLIENT_ID', 'c1abd2d9-101a-4113-999c-ce7bb41ccc4f'), 'eyeshare_client_secret': os.getenv('EYESHARE_CLIENT_SECRET', 'fV13ZAXGN0k4GBrLortDgLPH8YBYUu59XiOLC7VqhDRtr1eRjbpJlBevq%2BydQKiXzw3bSnJQFJhBRXBJQfiCvQ%3D%3D'), 'eyeshare_module': os.getenv('EYESHARE_MODULE', 'purchaseorder') } # Tokens self.eyeshare_token = None self.mongodb_client = None # Validate configuration self._validate_config() def _setup_logging(self): """Setup logging to write to the output directory""" try: # Create logs directory in the output directory log_dir = os.path.join(self.output_dir, 'logs') os.makedirs(log_dir, exist_ok=True) # Generate log filename with timestamp log_filename = f"vessel_committed_cost_analyzer_{self.timestamp}.log" log_filepath = os.path.join(log_dir, log_filename) # Clear any existing handlers for handler in logger.handlers[:]: logger.removeHandler(handler) # Configure logging to write to both console and file in output directory logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(log_filepath), logging.StreamHandler() ], force=True # Override any existing configuration ) # Log the log file path logger.info(f"Log file created at: {log_filepath}") logger.info(f"Vessel: {self.vessel_code}, End Date: {self.end_date}, Output Dir: {self.output_dir}") except Exception as e: print(f"Warning: Could not setup logging in output directory: {e}") # Fallback to basic console logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') def _validate_config(self): """Validate required configuration""" required_keys = ['mongodb_uri', 'eyeshare_base_url', 'eyeshare_client_id', 'eyeshare_client_secret'] missing_keys = [key for key in required_keys if not self.config.get(key)] if missing_keys: error_msg = f"Missing required environment variables: {missing_keys}\n" error_msg += "Please ensure these are set in your environment or MCP server configuration." raise ValueError(error_msg) def _get_mongodb_client(self): """Get MongoDB client""" try: # Disable sessions to avoid permission issues client = pymongo.MongoClient( self.config['mongodb_uri'], connect=True, serverSelectionTimeoutMS=5000 ) # Test connection client.admin.command('ping') logger.info("MongoDB connection established successfully") return client except Exception as e: logger.error(f"Failed to connect to MongoDB: {e}") raise def _get_eyeshare_token(self) -> str: """Get EyeShare API token""" if self.eyeshare_token: return self.eyeshare_token try: logger.info("Getting EyeShare authentication token...") token_url = f"{self.config['eyeshare_base_url']}/auth/connect/token" # Use the exact payload format that works payload = f"scope=api&client_id={self.config['eyeshare_client_id']}&client_secret={self.config['eyeshare_client_secret']}&grant_type=client_credentials" headers = { 'Content-Type': 'application/x-www-form-urlencoded' } response = requests.post(token_url, headers=headers, data=payload, timeout=30) response.raise_for_status() token_info = response.json() self.eyeshare_token = token_info['access_token'] logger.info("EyeShare token obtained successfully") return self.eyeshare_token except Exception as e: logger.error(f"Failed to get EyeShare token: {e}") raise def fetch_operating_expenses_category(self) -> pd.DataFrame: """Fetch operating expenses category from MongoDB""" try: logger.info(f"Fetching operating expenses category") # Use MongoDB URI from configuration client = pymongo.MongoClient(self.config['mongodb_uri']) db = client[self.config['mongodb_database']] collection = db['budget_category_raw_data'] # Query for current and previous year query = {} # Use find() instead of aggregate() projection = { 'accountCode': 1, 'category': 1, '_id': 0 } cursor = collection.find(query, projection) data = list(cursor) if not data: logger.warning("No operating expenses category found") return pd.DataFrame() # Convert to DataFrame and format df = pd.DataFrame(data) df = df.rename(columns={ 'category': 'Category' }) logger.info(f"Retrieved and formatted {len(df)} operating expenses category records") client.close() return df except Exception as e: logger.error(f"Failed to fetch operating expenses category: {e}") raise finally: try: client.close() except: pass def fetch_purchase_orders(self) -> pd.DataFrame: """Fetch purchase orders from MongoDB""" try: logger.info(f"Fetching purchase orders for vessel {self.vessel_code} till {self.end_date}") # Use MongoDB URI from configuration client = pymongo.MongoClient(self.config['mongodb_uri']) db = client[self.config['mongodb_database']] collection = db['purchase_order'] # MongoDB query query = { 'vesselCode': self.vessel_code, 'purchaseOrderIssueDate': {'$lte': datetime.strptime(self.end_date, '%Y-%m-%d')}, 'purchaseOrderStatus':{'$in': ['OPEN', 'CLOSED', 'SHORT-CLOSED', 'UNDER-AMENDMENT', 'GRN RECEIVED']} } # Use find() instead of aggregate() to avoid permission issues projection = { 'vesselName': 1, 'purchaseOrderNo': 1, 'purchaseOrderIssueDate': 1, 'exchangeRate': 1, 'TCD': 1, 'purchaseOrderValue': 1, 'currencyCode': 1, 'vendorCode': 1, 'vendorName': 1, 'purchaseOrderStatus': 1, 'itemName': 1, 'quantity': 1, 'unitprice': 1, 'intaccountcode': 1, 'ItemLineNo': 1, 'vesselCode': 1, 'modifiedDate': 1, 'tcdForLine': 1, '_id': 0 } cursor = collection.find(query, projection) data = list(cursor) if not data: logger.warning("No purchase orders found") return pd.DataFrame() # Convert to DataFrame and format df = pd.DataFrame(data) df = df.rename(columns={ 'vesselName': 'Vessel Name', 'purchaseOrderNo': 'Purchase Order No', 'purchaseOrderIssueDate': 'Purchase Order Issue Date', 'exchangeRate': 'Exchange Rate', 'TCD': 'TCD', 'purchaseOrderValue': 'Purchase Order Value', 'currencyCode': 'Currency Code', 'vendorCode': 'Vendor Code', 'vendorName': 'Vendor Name', 'purchaseOrderStatus': 'Purchase Order Status', 'itemName': 'Item Name', 'quantity': 'Quantity', 'unitprice': 'Unit Price', 'intaccountcode': 'Int Account Code', 'ItemLineNo': 'Item Line No', 'vesselCode': 'Vessel Code', 'modifiedDate': 'Modified/Short Closed Date' }) # Calculate PO Amount USD df['PO Amount USD'] = df['Purchase Order Value'] * df.get('Exchange Rate', 1) logger.info(f"Retrieved and formatted {len(df)} purchase order records") client.close() return df except Exception as e: logger.error(f"Failed to fetch purchase orders: {e}") raise finally: try: client.close() except: pass def fetch_vessel_expenses(self) -> pd.DataFrame: """Fetch vessel expenses from MongoDB""" try: logger.info(f"Fetching vessel expenses for vessel {self.vessel_code} till {self.end_date}") # Use MongoDB URI from configuration client = pymongo.MongoClient(self.config['mongodb_uri']) db = client[self.config['mongodb_database']] collection = db['budget_expenses_raw_data'] # Query for current and previous year current_year = datetime.strptime(self.end_date, '%Y-%m-%d').year previous_year = current_year - 1 # Parse the date properly end_date_obj = datetime.strptime(self.end_date, '%Y-%m-%d') query = { 'vesselCode': self.vessel_code, 'postingDate': {'$lte': end_date_obj}, 'PONO': { '$ne': None }, 'eyeshareId': { '$ne': None }, 'accountNo': '2130-110' # Specific account filter } # Use find() instead of aggregate() projection = { 'baseAmount': 1, 'eyeshareId': 1, 'PONO': 1, 'postingDate': 1, '_id': 0 } cursor = collection.find(query, projection) data = list(cursor) if not data: logger.warning("No vessel expenses found") return pd.DataFrame() # Convert to DataFrame and format df = pd.DataFrame(data) df = df.rename(columns={ 'baseAmount': 'Base Amount', 'eyeshareId': 'EyeShare ID', 'postingDate': 'Posting Date', 'PONO': 'PO No' }) logger.info(f"Retrieved and formatted {len(df)} vessel expense records") client.close() return df except Exception as e: logger.error(f"Failed to fetch vessel expenses: {e}") raise finally: try: client.close() except: pass def fetch_vessel_expenses_previous_year(self) -> pd.DataFrame: """Fetch vessel expenses previous year from MongoDB""" try: logger.info(f"Fetching vessel expenses previous year for vessel {self.vessel_code} till {self.end_date}") # Use MongoDB URI from configuration client = pymongo.MongoClient(self.config['mongodb_uri']) db = client[self.config['mongodb_database']] collection = db['budget_expenses_previous_year_raw_data'] # Query for current and previous year current_year = datetime.strptime(self.end_date, '%Y-%m-%d').year previous_year = current_year - 1 # Parse the date properly end_date_obj = datetime.strptime(self.end_date, '%Y-%m-%d') query = { 'vesselCode': self.vessel_code, 'postingDate': {'$lte': end_date_obj}, 'PONO': { '$ne': None }, 'eyeshareId': { '$ne': None }, 'accountNo': '2130-110' # Specific account filter } # Use find() instead of aggregate() projection = { 'baseAmount': 1, 'eyeshareId': 1, 'PONO': 1, 'postingDate': 1, '_id': 0 } cursor = collection.find(query, projection) data = list(cursor) if not data: logger.warning("No vessel expenses found") return pd.DataFrame() # Convert to DataFrame and format df = pd.DataFrame(data) df = df.rename(columns={ 'baseAmount': 'Base Amount', 'eyeshareId': 'EyeShare ID', 'postingDate': 'Posting Date', 'PONO': 'PO No' }) logger.info(f"Retrieved and formatted {len(df)} vessel expense records") client.close() return df except Exception as e: logger.error(f"Failed to fetch vessel expenses: {e}") raise finally: try: client.close() except: pass def process_po_lines_with_direct_api(self, eyeshare_ids: List[str]) -> pd.DataFrame: """Process PO lines using direct API calls with ThreadPoolExecutor""" try: logger.info(f"Processing {len(eyeshare_ids)} EyeShare IDs using direct API calls with ThreadPoolExecutor") if not eyeshare_ids: logger.warning("No EyeShare IDs to process") return pd.DataFrame() # Get EyeShare token self._get_eyeshare_token() # Process in batches to avoid overwhelming the API batch_size =100 # Smaller batch size for better performance all_po_lines = [] total_batches = (len(eyeshare_ids) + batch_size - 1) // batch_size for batch_idx in range(total_batches): start_idx = batch_idx * batch_size end_idx = min(start_idx + batch_size, len(eyeshare_ids)) batch_ids = eyeshare_ids[start_idx:end_idx] logger.info(f"Processing batch {batch_idx + 1}/{total_batches} ({len(batch_ids)} EyeShare IDs)") batch_po_lines = self._process_batch_with_threadpool(batch_ids) all_po_lines.extend(batch_po_lines) # Small delay between batches time.sleep(0.5) if all_po_lines: df = pd.DataFrame(all_po_lines) logger.info(f"Successfully processed {len(df)} PO lines from {len(eyeshare_ids)} EyeShare IDs") return df else: logger.warning("No PO lines data retrieved from API calls") return pd.DataFrame() except Exception as e: logger.error(f"Failed to process PO lines with direct API: {e}") return pd.DataFrame() def _process_batch_with_threadpool(self, batch_ids: List[str]) -> List[Dict]: """Process a batch of EyeShare IDs using ThreadPoolExecutor""" try: def process_single_eyeshare_id(eyeshare_id: str) -> Dict: """Process a single EyeShare ID""" try: # Use the sample API endpoint format # eyeshare_id: The GUID representing the EyeShare entity ID (e.g., c53b3ce1-eb0a-4ad3-b9b5-5b16fd1f51cd) # companyCode: The vessel code (e.g., GEVI) url = f"{self.config['eyeshare_base_url']}/api/entity/{eyeshare_id}?module=purchaseorder&companyCode={self.vessel_code}" headers = { 'Authorization': f'Bearer {self.eyeshare_token}', 'Content-Type': 'application/x-www-form-urlencoded' } response = requests.request("GET", url, headers=headers, data={}, timeout=30) if response.status_code == 200: data = response.json() # Extract PO lines from response po_lines = [] # Helper function to filter required fields def filter_po_line_fields(line, eyeshare_id): """Extract only required fields from PO line""" return { 'PoNumber': line.get('PoNumber', ''), 'Line': line.get('Line', 0), 'InvoicedNow': line.get('InvoicedNow', 0.0), 'Invoiced': line.get('Invoiced', 0.0), 'CostCentre': line.get('CostCentre', ''), 'QtyWrittenOff': line.get('QtyWrittenOff', 0.0), 'eyeshareId': eyeshare_id } # Check if the response contains PO lines data and StateName is 'Transferred' state_name = data.get('State', {}).get('Name', '') # Check if the response contains PO lines data if state_name == 'Transferred' and isinstance(data, dict): # Check for PurchaseOrders with PoLines (based on sample structure) if 'PurchaseOrders' in data and data['PurchaseOrders']: for po in data['PurchaseOrders']: if 'PoLines' in po and po['PoLines']: for line in po['PoLines']: filtered_line = filter_po_line_fields(line, eyeshare_id) po_lines.append(filtered_line) # Check for PurchaseOrderLines (main field from API response) elif 'PurchaseOrderLines' in data and data['PurchaseOrderLines']: for line in data['PurchaseOrderLines']: filtered_line = filter_po_line_fields(line, eyeshare_id) po_lines.append(filtered_line) # Check for Lines field as backup elif 'Lines' in data and data['Lines']: for line in data['Lines']: filtered_line = filter_po_line_fields(line, eyeshare_id) po_lines.append(filtered_line) # Check for ImportedPurchaseOrderLines elif 'ImportedPurchaseOrderLines' in data and data['ImportedPurchaseOrderLines']: for line in data['ImportedPurchaseOrderLines']: filtered_line = filter_po_line_fields(line, eyeshare_id) po_lines.append(filtered_line) # Legacy support for PoLines elif 'PoLines' in data and data['PoLines']: for line in data['PoLines']: filtered_line = filter_po_line_fields(line, eyeshare_id) po_lines.append(filtered_line) elif isinstance(data, list): # If it's a list of PO lines directly for line in data: if isinstance(line, dict): filtered_line = filter_po_line_fields(line, eyeshare_id) po_lines.append(filtered_line) return { 'success': True, 'eyeshare_id': eyeshare_id, 'po_lines': po_lines, 'count': len(po_lines) } else: logger.debug(f"API error for EyeShare ID {eyeshare_id}: {response.status_code}") return { 'success': False, 'eyeshare_id': eyeshare_id, 'error': f"HTTP {response.status_code}" } except Exception as e: logger.debug(f"Error processing EyeShare ID {eyeshare_id}: {e}") return { 'success': False, 'eyeshare_id': eyeshare_id, 'error': str(e) } # Process with ThreadPoolExecutor all_po_lines = [] successful_count = 0 failed_count = 0 with ThreadPoolExecutor(max_workers=10) as executor: # Submit all tasks future_to_eyeshare_id = { executor.submit(process_single_eyeshare_id, eyeshare_id): eyeshare_id for eyeshare_id in batch_ids } # Process completed tasks for future in as_completed(future_to_eyeshare_id): result = future.result() if result['success']: all_po_lines.extend(result['po_lines']) successful_count += 1 if result['count'] > 0: logger.debug(f" āœ… {result['eyeshare_id']}: {result['count']} PO lines") else: failed_count += 1 logger.debug(f" āŒ {result['eyeshare_id']}: {result['error']}") logger.info(f" Batch completed: {successful_count} success, {failed_count} failed, {len(all_po_lines)} PO lines") return all_po_lines except Exception as e: logger.error(f"ThreadPool batch processing failed: {e}") return [] def fetch_invoice_data_for_pos(self, committed_costs_df: pd.DataFrame) -> pd.DataFrame: """Fetch invoice data from Eyeshare API for all unique Purchase Order Numbers with ThreadPoolExecutor""" try: if committed_costs_df.empty: logger.warning("No committed costs data to fetch invoice information") return pd.DataFrame() # Get unique Purchase Order Numbers - NO NORMALIZATION unique_pos = committed_costs_df['Purchase Order No'].dropna().unique().tolist() logger.info(f"Fetching invoice data for {len(unique_pos)} unique Purchase Orders") logger.info(f"Sample PO numbers to search: {unique_pos[:5]}") if not unique_pos: logger.warning("No Purchase Order Numbers found") return pd.DataFrame() # Get EyeShare token self._get_eyeshare_token() # Process POs with ThreadPoolExecutor all_invoice_data = [] def fetch_single_po_invoices(po_number: str) -> Dict: """Fetch invoice data for a single PO using POST method""" try: url = "https://synergymarine.eye-share.com/api/search?companyCode=India&module=invoice" payload = json.dumps({ "Page": { "Skip": 0, "Limit": 200 }, "Filters": { "Equality": [ { "Field": "PurchaseOrderNumbers", "StringValue": po_number } ], "Contains": [], "OpenInterval": [], "ClosedInterval": [], "Regex": [], "Arrays": [], "SubSet": [], "TextIndex": [], "StartsWith": [], "WildCard": [], "Features": { "ExtendedAccess": True, "View": "search" } }, "Sort": [], "CountOnly": False }) headers = { 'Authorization': f'Bearer {self.eyeshare_token}', 'Content-Type': 'application/json' } # Using POST method response = requests.post(url, headers=headers, data=payload, timeout=30) if response.status_code == 200: data = response.json() invoice_records = [] # Extract invoice data from response if 'Items' in data and data['Items']: for item in data['Items']: # Use actual PO number from response - NO NORMALIZATION actual_po_number = item.get('PurchaseOrderNumber', '') state_name = item.get('StateName', '') scan_no = item.get('Head', {}).get('ScanNo', '') period = pd.to_datetime(item['Head']['Period']).strftime('%Y-%b-%d') # Only add if we have a valid PO number from the response if actual_po_number: invoice_records.append({ 'Purchase Order No': actual_po_number, # Use exact PO from response 'StateName': state_name, 'ScanNo': str(scan_no) if scan_no else '', 'Period' : period }) return { 'success': True, 'po_number': po_number, 'invoice_data': invoice_records, 'count': len(invoice_records) } else: logger.debug(f"API error for PO {po_number}: {response.status_code}") return { 'success': False, 'po_number': po_number, 'error': f"HTTP {response.status_code}" } except Exception as e: logger.debug(f"Error fetching invoices for PO {po_number}: {e}") return { 'success': False, 'po_number': po_number, 'error': str(e) } # Process with ThreadPoolExecutor successful_count = 0 failed_count = 0 with ThreadPoolExecutor(max_workers=10) as executor: # Submit all tasks future_to_po = { executor.submit(fetch_single_po_invoices, po_number): po_number for po_number in unique_pos } # Process completed tasks for future in as_completed(future_to_po): result = future.result() if result['success']: all_invoice_data.extend(result['invoice_data']) successful_count += 1 if result['count'] > 0: logger.debug(f" āœ… {result['po_number']}: {result['count']} invoices") else: failed_count += 1 logger.debug(f" āŒ {result['po_number']}: {result['error']}") logger.info(f"Invoice fetch completed: {successful_count} success, {failed_count} failed, {len(all_invoice_data)} invoice records") if all_invoice_data: # Convert to DataFrame and group by PO to create comma-separated values invoice_df = pd.DataFrame(all_invoice_data) logger.info(f"Raw invoice data: {len(invoice_df)} records") logger.info(f"Unique POs in invoice data: {invoice_df['Purchase Order No'].nunique()}") # Group by Purchase Order No and create comma-separated StateName and ScanNo - EXACT MATCH grouped_invoice_df = invoice_df.groupby('Purchase Order No').agg({ 'StateName': lambda x: ','.join(f'({i})' for i in filter(None, x.unique())), 'ScanNo': lambda x: ','.join(f'({i})' for i in filter(None, x.astype(str).unique())), 'Period': lambda x: ','.join(f'({i})' for i in filter(None, x.astype(str).unique())) }).reset_index() logger.info(f"Processed {len(grouped_invoice_df)} unique POs with invoice data") logger.info(f"Sample grouped invoice data:\n{grouped_invoice_df.head()}") return grouped_invoice_df else: logger.warning("No invoice data retrieved from API calls") return pd.DataFrame() except Exception as e: logger.error(f"Failed to fetch invoice data: {e}") return pd.DataFrame() def calculate_committed_costs(self, po_lines_df: pd.DataFrame, purchase_orders_df: pd.DataFrame, operating_expenses_category_df: pd.DataFrame) -> Tuple[pd.DataFrame, Dict]: """Calculate committed costs from PO lines and purchase orders""" try: logger.info("Calculating committed costs...") if po_lines_df.empty: logger.warning("No PO lines data to calculate committed costs") return pd.DataFrame(), {} # Filter out records with InvoicedNow = 0 # po_lines_filtered = po_lines_df[po_lines_df['InvoicedNow'] > 0].copy() po_lines_filtered = po_lines_df.copy() logger.info(f"Filtered PO lines: {len(po_lines_filtered)} (removed InvoicedNow = 0)") if po_lines_filtered.empty: logger.warning("No PO lines with InvoicedNow > 0") return pd.DataFrame(), {} # Group by PoNumber and Line po_lines_grouped = po_lines_filtered.groupby(['PoNumber', 'Line']).agg({ 'InvoicedNow': 'sum', 'QtyWrittenOff': 'sum', 'eyeshareId': lambda x: ','.join(str(i) for i in x) }).reset_index() # Join with purchase orders merged_df = purchase_orders_df.merge( po_lines_grouped, left_on=['Purchase Order No', 'Item Line No'], right_on=['PoNumber', 'Line'], how='left' ) # Join with operating expenses category merged_df = merged_df.merge( operating_expenses_category_df, left_on='Int Account Code', right_on='accountCode', how='left' ).drop(columns=['accountCode']) # Calculate quantity differences and committed costs merged_df['Quantity Diff'] = merged_df['Quantity'] - (merged_df['QtyWrittenOff'] + merged_df['InvoicedNow']).fillna(0) # Calculate Qty Open in format "Qty Open (Total Quantity)" def format_qty_value(value): """Format quantity value: show as int if whole number, else as float with 2 decimal places""" if pd.isna(value): return "0" if float(value) == int(float(value)): return str(int(float(value))) else: return f"{float(value):.2f}" merged_df['Qty Open'] = merged_df['Quantity Diff'].apply(format_qty_value) + '(' + merged_df['Quantity'].apply(format_qty_value) + ')' merged_df = merged_df[merged_df['Quantity Diff'] > 0] merged_df = merged_df[~((merged_df['Purchase Order Status'] == 'SHORT-CLOSED') & (merged_df['Modified/Short Closed Date'] < self.end_date))] def calc_cc_unit_price(row): try: if pd.notna(row['tcdForLine']): # Avoid division by zero if row['Quantity Diff'] != 0: return (row['Quantity Diff'] * row['Unit Price']) + (row['tcdForLine'] / (row['Quantity'] / row['Quantity Diff'])) else: return row['Quantity Diff'] * row['Unit Price'] elif row['TCD'] != 0 and pd.notna(row['TCD']): # Avoid division by zero if row['Purchase Order Value'] != 0 and row['TCD'] != 0: return (row['Quantity Diff'] * row['Unit Price']) + ((row['Quantity Diff'] * row['Unit Price']) / (row['Purchase Order Value'] / row['TCD'])) else: return row['Quantity Diff'] * row['Unit Price'] else: return row['Quantity Diff'] * row['Unit Price'] except Exception as e: logger.error(f"Error in calc_cc_unit_price: {e}") return row['Quantity Diff'] * row['Unit Price'] merged_df['CC Unit Price'] = merged_df.apply(calc_cc_unit_price, axis=1) merged_df['CC Amount USD'] = merged_df['Exchange Rate'] * merged_df['CC Unit Price'] # Calculate totals total_cc_amount = merged_df['CC Amount USD'].sum() logger.info(f"Calculated committed costs for {len(merged_df)} records") logger.info(f"Total CC Amount USD: ${total_cc_amount:,.2f}") return merged_df, { 'total_cc_amount_usd': total_cc_amount, 'records_count': len(merged_df) } except Exception as e: logger.error(f"Failed to calculate committed costs: {e}") return pd.DataFrame(), {} def generate_reports(self, purchase_orders_df: pd.DataFrame, vessel_expenses_df: pd.DataFrame, po_lines_df: pd.DataFrame, committed_costs_df: pd.DataFrame, summary_stats: Dict) -> Dict[str, str]: """Generate comprehensive Excel reports""" try: logger.info("Generating comprehensive reports...") reports = {} # 1. Main Purchase Orders Report po_file = os.path.join(self.output_dir, f"{self.vessel_code}_Purchase_Orders_Final_{self.timestamp}.xlsx") purchase_orders_df.to_excel(po_file, index=False) reports['purchase_orders'] = po_file logger.info(f"Generated Purchase Orders report: {po_file}") # 2. Vessel Expenses Report ve_file = os.path.join(self.output_dir, f"{self.vessel_code}_Vessel_Expenses_Final_{self.timestamp}.xlsx") vessel_expenses_df.to_excel(ve_file, index=False) reports['vessel_expenses'] = ve_file logger.info(f"Generated Vessel Expenses report: {ve_file}") # 3. PO Lines Report (if data exists) if not po_lines_df.empty: po_lines_file = os.path.join(self.output_dir, f"{self.vessel_code}_PO_Lines_Final_{self.timestamp}.xlsx") po_lines_df.to_excel(po_lines_file, index=False) reports['po_lines'] = po_lines_file logger.info(f"Generated PO Lines report: {po_lines_file}") # 4. Committed Cost Report (if data exists) if not committed_costs_df.empty: cc_file = os.path.join(self.output_dir, f"{self.vessel_code}_Committed_Cost_Final_{self.timestamp}.xlsx") with pd.ExcelWriter(cc_file, engine='openpyxl') as writer: committed_costs_df.to_excel(writer, sheet_name='Committed Costs', index=False) # Summary sheet summary_df = pd.DataFrame([{ 'Metric': 'Total CC Amount USD', 'Value': f"${summary_stats.get('total_cc_amount_usd', 0):,.2f}" }, { 'Metric': 'Total Quantity Difference', 'Value': f"{summary_stats.get('total_quantity_diff', 0):,.2f}" }, { 'Metric': 'Total Records', 'Value': summary_stats.get('records_count', 0) }, { 'Metric': 'Report Generated', 'Value': datetime.now().strftime('%Y-%m-%d %H:%M:%S') }]) summary_df.to_excel(writer, sheet_name='Summary', index=False) reports['committed_cost'] = cc_file logger.info(f"Generated Committed Cost report: {cc_file}") # 5. Complete Analysis Report complete_file = os.path.join(self.output_dir, f"{self.vessel_code}_Complete_Committed_Cost_Analysis_{self.timestamp}.xlsx") with pd.ExcelWriter(complete_file, engine='openpyxl') as writer: # Committed Cost Summary Sheet (formerly Grouped Summary) if not committed_costs_df.empty: # Prepare grouped data grouped_df = committed_costs_df.copy() # Rename intaccountcode to Account Code if 'Int Account Code' in grouped_df.columns: grouped_df['Account Code'] = grouped_df['Int Account Code'] # Format Purchase Order Issue Date if 'Purchase Order Issue Date' in grouped_df.columns: grouped_df['Purchase Order Issue Date'] = pd.to_datetime(grouped_df['Purchase Order Issue Date']).dt.strftime('%Y-%b-%d') # Group by specified columns and sum CC Amount USD groupby_columns = [ 'Account Code', 'Vessel Name', 'Purchase Order No', 'Purchase Order Issue Date', 'Purchase Order Value', 'Currency Code', 'Vendor Name', 'Purchase Order Status', 'Vessel Code', 'Category', 'StateName', 'ScanNo', 'Period', 'PO Amount USD' ] # Select only columns that exist existing_groupby_cols = [col for col in groupby_columns if col in grouped_df.columns] # Perform groupby and sum CC Amount USD agg_dict = {'CC Amount USD': 'sum', 'CC Unit Price': 'sum'} # Add Qty Open aggregation if it exists if 'Qty Open' in grouped_df.columns: agg_dict['Qty Open'] = 'first' # Take first value since it should be the same for same PO line grouped_summary = grouped_df.groupby(existing_groupby_cols, as_index=False).agg(agg_dict) # Ensure proper column ordering: Account Code, Category, other columns, Qty Open, StateName, ScanNo, CC Amount USD if 'Account Code' in grouped_summary.columns: cols = ['Account Code'] if 'Category' in grouped_summary.columns: cols.append('Category') # Add middle columns, excluding Qty Open, PO Amount USD, StateName, ScanNo, CC Amount USD for now middle_cols = [col for col in grouped_summary.columns if col not in ('Account Code', 'Category', 'Qty Open', 'StateName', 'ScanNo', 'PO Amount USD', 'CC Amount USD')] cols += middle_cols # Add Qty Open before StateName and ScanNo if 'Qty Open' in grouped_summary.columns: cols.append('Qty Open') # Add StateName and ScanNo before PO Amount USD and CC Amount USD if 'StateName' in grouped_summary.columns: cols.append('StateName') if 'ScanNo' in grouped_summary.columns: cols.append('ScanNo') # Add PO Amount USD before CC Amount USD if 'PO Amount USD' in grouped_summary.columns: cols.append('PO Amount USD') # Format PO Amount USD column to two decimal places grouped_summary['PO Amount USD'] = grouped_summary['PO Amount USD'].map(lambda x: f"{x:.2f}" if pd.notnull(x) else "") # Add CC Amount USD last if 'CC Amount USD' in grouped_summary.columns: cols.append('CC Amount USD') grouped_summary = grouped_summary[cols] # Add total row total_row = pd.DataFrame({col: [''] for col in grouped_summary.columns}, index=[0]) total_row['Account Code'] = 'TOTAL' total_row['CC Amount USD'] = grouped_summary['CC Amount USD'].sum() grouped_summary = pd.concat([grouped_summary, total_row], ignore_index=True) grouped_summary = grouped_summary.rename(columns={ 'Period':'Posting Date', 'StateName':'Eyeshare Status' }) grouped_summary = grouped_summary[["Account Code","Category","Vessel Name","Vessel Code","Purchase Order No","Purchase Order Issue Date","Vendor Name","Purchase Order Status","Purchase Order Value","Currency Code", "Posting Date","Qty Open","Eyeshare Status","ScanNo","CC Unit Price","PO Amount USD","CC Amount USD"]] # Write committed cost summary as the first sheet grouped_summary.to_excel(writer, sheet_name='Committed Cost Summary', index=False) # Format the sheet worksheet = writer.sheets['Committed Cost Summary'] # Import openpyxl formatting classes from openpyxl.styles import Font, PatternFill, Alignment, Border, Side from openpyxl.utils import get_column_letter # Define styles header_font = Font(bold=True, color="FFFFFF") header_fill = PatternFill(start_color="366092", end_color="366092", fill_type="solid") header_alignment = Alignment(horizontal="center", vertical="center") # CC Amount USD column highlight highlight_fill = PatternFill(start_color="E6F3FF", end_color="E6F3FF", fill_type="solid") currency_format = '#,##0.00' # Total row style total_font = Font(bold=True) total_fill = PatternFill(start_color="D9D9D9", end_color="D9D9D9", fill_type="solid") # Border style thin_border = Border( left=Side(style='thin'), right=Side(style='thin'), top=Side(style='thin'), bottom=Side(style='thin') ) # Apply header formatting for col in range(1, len(grouped_summary.columns) + 1): cell = worksheet.cell(row=1, column=col) cell.font = header_font cell.fill = header_fill cell.alignment = header_alignment cell.border = thin_border # Find CC Amount USD column cc_amount_col = None for idx, col_name in enumerate(grouped_summary.columns, 1): if col_name == 'CC Amount USD': cc_amount_col = idx break # Apply formatting to data rows for row in range(2, len(grouped_summary) + 2): for col in range(1, len(grouped_summary.columns) + 1): cell = worksheet.cell(row=row, column=col) cell.border = thin_border # Highlight CC Amount USD column if col == cc_amount_col: cell.fill = highlight_fill cell.number_format = currency_format cell.alignment = Alignment(horizontal="right") # Format total row if row == len(grouped_summary) + 1: cell.font = total_font cell.fill = total_fill # Auto-adjust column widths for column in worksheet.columns: max_length = 0 column_letter = get_column_letter(column[0].column) for cell in column: try: if len(str(cell.value)) > max_length: max_length = len(str(cell.value)) except: pass adjusted_width = min(max_length + 2, 50) worksheet.column_dimensions[column_letter].width = adjusted_width # Freeze top row worksheet.freeze_panes = 'A2' # Data sheets purchase_orders_df.to_excel(writer, sheet_name='Purchase Orders', index=False) vessel_expenses_df.to_excel(writer, sheet_name='Vessel Expenses', index=False) if not po_lines_df.empty: po_lines_df.to_excel(writer, sheet_name='PO Lines', index=False) if not committed_costs_df.empty: committed_costs_df.to_excel(writer, sheet_name='Committed Costs', index=False) reports['complete_analysis'] = complete_file logger.info(f"Generated Complete Analysis report: {complete_file}") return reports except Exception as e: logger.error(f"Failed to generate reports: {e}") return {} def run_analysis(self) -> int: """Run the complete committed cost analysis""" print(f"šŸš€ Starting Vessel Committed Cost Analysis for {self.vessel_code} till {self.end_date}") start_time = time.time() try: # Step 1: Fetch purchase orders logger.info("Step 1: Fetching purchase orders...") purchase_orders_df = self.fetch_purchase_orders() # Step 2: Fetch vessel expenses logger.info("Step 2: Fetching vessel expenses...") vessel_expenses_df = self.fetch_vessel_expenses() # Step 2.1: Fetch vessel expenses previous year logger.info("Step 2.1: Fetching vessel expenses previous year...") vessel_expenses_previous_year_df = self.fetch_vessel_expenses_previous_year() vessel_expenses_df = pd.concat([vessel_expenses_df, vessel_expenses_previous_year_df]) # Step 3: Extract EyeShare IDs logger.info("Step 3: Extracting EyeShare IDs...") eyeshare_ids = vessel_expenses_df['EyeShare ID'].dropna().unique().tolist() logger.info(f"Found {len(eyeshare_ids)} unique EyeShare IDs") # Step 4: Process PO lines with direct API calls logger.info("Step 4: Processing PO lines with direct API calls...") po_lines_df = self.process_po_lines_with_direct_api(eyeshare_ids) # Step 4.1: Fetch operating expenses category logger.info("Step 4.1: Fetching operating expenses category...") operating_expenses_category_df = self.fetch_operating_expenses_category() # Step 5: Calculate committed costs logger.info("Step 5: Calculating committed costs...") committed_costs_df, summary_stats = self.calculate_committed_costs(po_lines_df, purchase_orders_df, operating_expenses_category_df) # Step 5.1: Fetch invoice data for POs in committed costs logger.info("Step 5.1: Fetching invoice data for Purchase Orders...") invoice_data_df = self.fetch_invoice_data_for_pos(committed_costs_df) # Step 5.2: Join invoice data with committed costs - EXACT MATCH if not invoice_data_df.empty and not committed_costs_df.empty: logger.info("Step 5.2: Joining invoice data with committed costs...") logger.info(f"Committed costs unique POs: {committed_costs_df['Purchase Order No'].nunique()}") logger.info(f"Invoice data unique POs: {invoice_data_df['Purchase Order No'].nunique()}") # Show sample PO numbers from both datasets for comparison logger.info(f"Sample committed costs POs: {list(committed_costs_df['Purchase Order No'].unique()[:5])}") logger.info(f"Sample invoice POs: {list(invoice_data_df['Purchase Order No'].unique()[:5])}") # EXACT MATCH - No normalization committed_costs_df = committed_costs_df.merge( invoice_data_df, on='Purchase Order No', how='left' ) # Fill NaN values with empty strings committed_costs_df['StateName'] = committed_costs_df['StateName'].fillna('') committed_costs_df['ScanNo'] = committed_costs_df['ScanNo'].fillna('') committed_costs_df['Period'] = committed_costs_df['Period'].fillna('') # Check how many records got matched matched_records = committed_costs_df[committed_costs_df['StateName'] != ''].shape[0] logger.info(f"Successfully joined invoice data: {matched_records}/{len(committed_costs_df)} records matched") else: # Add empty columns if no invoice data if not committed_costs_df.empty: committed_costs_df['StateName'] = '' committed_costs_df['ScanNo'] = '' committed_costs_df['Period'] = '' # Step 6: Generate reports logger.info("Step 6: Generating reports...") reports = self.generate_reports(purchase_orders_df, vessel_expenses_df, po_lines_df, committed_costs_df, summary_stats) # Final summary execution_time = time.time() - start_time # Create summary data for SIYA frontend summary_data = { "status": "completed", "vessel_code": self.vessel_code, "analysis_date": self.end_date, "execution_time_seconds": round(execution_time, 2), "statistics": { "purchase_orders": len(purchase_orders_df), "vessel_expenses": len(vessel_expenses_df), "eyeshare_ids": len(eyeshare_ids), "po_lines": len(po_lines_df), "committed_cost_records": summary_stats.get('records_count', 0), "total_cc_amount_usd": round(summary_stats.get('total_cc_amount_usd', 0), 2), "total_quantity_diff": round(summary_stats.get('total_quantity_diff', 0), 2) }, "generated_reports": [ { "type": report_type, "filename": os.path.basename(filename), "full_path": filename } for report_type, filename in reports.items() ], "timestamp": datetime.now().isoformat(), "output_directory": self.output_dir } # Save summary to JSON file for SIYA frontend summary_file = os.path.join(self.output_dir, f"{self.vessel_code}_Analysis_Summary_{self.timestamp}.json") try: with open(summary_file, 'w') as f: json.dump(summary_data, f, indent=2) logger.info(f"Summary data saved to: {summary_file}") except Exception as e: logger.error(f"Failed to save summary data: {e}") # Print the formatted summary (for console/logs) print("\n" + "="*80) print("šŸŽ‰ VESSEL COMMITTED COST ANALYSIS COMPLETED") print("="*80) print(f"Vessel Code: {self.vessel_code}") print(f"Analysis Date: {self.end_date}") print(f"Execution Time: {execution_time:.2f} seconds") print(f"Purchase Orders: {len(purchase_orders_df):,}") print(f"Vessel Expenses: {len(vessel_expenses_df):,}") print(f"EyeShare IDs: {len(eyeshare_ids):,}") print(f"PO Lines: {len(po_lines_df):,}") print(f"Committed Cost Records: {summary_stats.get('records_count', 0):,}") if summary_stats: print(f"Total CC Amount USD: ${summary_stats.get('total_cc_amount_usd', 0):,.2f}") print(f"Total Quantity Diff: {summary_stats.get('total_quantity_diff', 0):,.2f}") print("\nGenerated Reports:") for report_type, filename in reports.items(): print(f" āœ… {filename}") print("\nšŸŽ‰ ANALYSIS COMPLETED SUCCESSFULLY!") return 0 except Exception as e: logger.error(f"Analysis failed: {e}") print(f"\nāŒ ANALYSIS FAILED: {e}") return 1 finally: # Cleanup handled automatically by garbage collection pass def main(vessel_code=None, end_date=None): """Main function""" # If vessel_code is provided directly, use it (for Jupyter notebook or direct calls) if vessel_code is not None: # Use provided end_date or default to current date if end_date is None: end_date = datetime.now().strftime('%Y-%m-%d') try: analyzer = VesselCommittedCostAnalyzer(vessel_code, end_date) return analyzer.run_analysis() except Exception as e: print(f"āŒ Failed to initialize analyzer: {e}") return 1 # Otherwise, use argparse for command line usage parser = argparse.ArgumentParser( description='Vessel Committed Cost Analyzer', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: python vessel_committed_cost_analyzer.py BASC # Uses today's date python vessel_committed_cost_analyzer.py BASC --end-date 2025-07-20 # Specific date python vessel_committed_cost_analyzer.py BWET --end-date 2025-12-31 # Specific date """ ) parser.add_argument('vessel_code', help='Vessel code (e.g., BASC, BWET, GEVI, ACET)') parser.add_argument('--end-date', default=datetime.now().strftime('%Y-%m-%d'), help='End date for analysis (YYYY-MM-DD format, default: today)') parser.add_argument('--output-dir', default='.', help='Output directory for generated files (default: current directory)') args = parser.parse_args() try: analyzer = VesselCommittedCostAnalyzer(args.vessel_code, args.end_date, args.output_dir) return analyzer.run_analysis() except Exception as e: print(f"āŒ Failed to initialize analyzer: {e}") return 1 if __name__ == "__main__": # For command line usage, parse arguments exit(main())