""" CSV Import Service for processing invoice data from CSV files. Handles parsing, validation, and data transformation. """ import csv import logging from datetime import datetime from typing import List, Dict, Optional from pathlib import Path logger = logging.getLogger(__name__) class CSVImportService: """Service for importing invoice data from CSV files.""" # Company name to ID mapping for all branches COMPANY_MAPPING = { # Bunbury (branch2) "ELECTRICS BUNBURY": 2, # Residential "AC BUNBURY": 3, # Air "SOLAR BUNBURY": 4, # Solar "COMMERCIAL BUNBURY": 5, # Commercial # Busselton (branch1) "ELECTRICS BUSSELTON": 139, # Electrical "AC BUSSELTON": 172, # Air # Mandurah (branch3) "ELECTRICAL MANDURAH": 2, # Electrical "ELECTRICS MANDURAH": 2, # Electrical (alternate spelling - used in 2026 data) "AIR MANDURAH": 35, # Air } def __init__(self, branch_id: str = "branch2"): """ Initialize CSV import service. Args: branch_id: Branch identifier (default: branch2 for Bunbury) """ self.branch_id = branch_id def parse_currency(self, value: str) -> float: """ Parse currency string to float. Handles positive and negative values (credit notes). Args: value: Currency string (e.g., "$1,169.27" or "($133.48)" for negatives) Returns: Float value """ if not value or value.strip() == "": return 0.0 # Check if negative (wrapped in parentheses) is_negative = value.strip().startswith("(") and value.strip().endswith(")") # Remove $, commas, parentheses, and whitespace cleaned = value.replace("$", "").replace(",", "").replace("(", "").replace(")", "").strip() try: amount = float(cleaned) return -amount if is_negative else amount except ValueError: logger.warning(f"Could not parse currency value: {value}") return 0.0 def parse_date(self, date_str: str) -> Optional[datetime]: """ Parse date string in DD/MM/YYYY format. Args: date_str: Date string (e.g., "03/05/2021") Returns: datetime object or None if invalid """ if not date_str or date_str.strip() == "": return None try: # Parse DD/MM/YYYY format return datetime.strptime(date_str.strip(), "%d/%m/%Y") except ValueError: logger.warning(f"Could not parse date: {date_str}") return None def get_company_id(self, company_name: str) -> Optional[int]: """ Map company name to company ID. Args: company_name: Company name from CSV Returns: Company ID or None if not found """ return self.COMPANY_MAPPING.get(company_name.strip()) def parse_csv(self, csv_path: str) -> List[Dict]: """ Parse CSV file and extract invoice data. Args: csv_path: Path to CSV file Returns: List of invoice dictionaries """ invoices = [] skipped_count = 0 error_count = 0 logger.info(f"📋 Parsing CSV file: {csv_path}") try: with open(csv_path, 'r', encoding='utf-8') as f: # Skip first line (criteria line) f.readline() # Read CSV reader = csv.DictReader(f) for row_num, row in enumerate(reader, start=3): # Start at 3 (header is line 2) try: # Extract required fields company_name = row.get("Company Name", "").strip() invoice_no = row.get("Invoice No.", "").strip() invoice_date_str = row.get("Invoice Date", "").strip() subtotal_str = row.get("Sub-Total (Ex Tax)", "").strip() status = row.get("Status", "").strip() # Validate required fields if not company_name or not invoice_no or not invoice_date_str: skipped_count += 1 continue # Get company ID company_id = self.get_company_id(company_name) if company_id is None: logger.debug(f"Skipping row {row_num}: Unknown company '{company_name}'") skipped_count += 1 continue # Parse date date_issued = self.parse_date(invoice_date_str) if date_issued is None: logger.warning(f"Skipping row {row_num}: Invalid date '{invoice_date_str}'") skipped_count += 1 continue # Parse amount total_ex_tax = self.parse_currency(subtotal_str) # Determine if paid is_paid = "Fully-paid" in status if status else False # Create invoice dictionary invoice = { "invoice_no": invoice_no, "branch_id": self.branch_id, "company_id": company_id, "company_name": company_name, "date_issued": date_issued, "total_ex_tax": total_ex_tax, "is_paid": is_paid, "status": status, "customer": row.get("Customer", "").strip(), "job_no": row.get("Job No(s).", "").strip(), "job_name": row.get("Job Name", "").strip(), "site": row.get("Site", "").strip() } invoices.append(invoice) except Exception as e: logger.error(f"Error parsing row {row_num}: {e}") error_count += 1 continue except Exception as e: logger.error(f"❌ Error reading CSV file: {e}") raise logger.info(f"✅ Parsed {len(invoices)} invoices from CSV") if skipped_count > 0: logger.info(f"⏭️ Skipped {skipped_count} rows") if error_count > 0: logger.warning(f"⚠️ {error_count} errors encountered") return invoices def validate_invoice(self, invoice: Dict) -> bool: """ Validate invoice data. Args: invoice: Invoice dictionary Returns: True if valid, False otherwise """ required_fields = ["invoice_no", "branch_id", "company_id", "date_issued", "total_ex_tax"] for field in required_fields: if field not in invoice or invoice[field] is None: logger.warning(f"Invalid invoice: missing {field}") return False return True def get_statistics(self, invoices: List[Dict]) -> Dict: """ Get statistics about parsed invoices. Args: invoices: List of invoice dictionaries Returns: Statistics dictionary """ if not invoices: return { "total_count": 0, "by_company": {}, "total_revenue": 0.0, "date_range": None } # Count by company by_company = {} total_revenue = 0.0 dates = [] for inv in invoices: company_id = inv["company_id"] company_name = inv["company_name"] if company_id not in by_company: by_company[company_id] = { "name": company_name, "count": 0, "revenue": 0.0 } by_company[company_id]["count"] += 1 by_company[company_id]["revenue"] += inv["total_ex_tax"] total_revenue += inv["total_ex_tax"] dates.append(inv["date_issued"]) date_range = None if dates: date_range = { "earliest": min(dates).strftime("%Y-%m-%d"), "latest": max(dates).strftime("%Y-%m-%d") } return { "total_count": len(invoices), "by_company": by_company, "total_revenue": total_revenue, "date_range": date_range }