PABPAT/TCI_Shield
0
1# ============================================================2# TRADE CREDIT INSURANCE -- DATABASE MODULE3# ============================================================4# 2 Tables:5# 1. tci_customers -- customer profile + policy + claims6# 2. tci_buyers -- buyer info + our experience7#8# ID Formats:9# Customer : CUST-{sequential, rolls over} e.g. CUST-000110# Policy : POL-{YEAR}-{4 digit seq} e.g. POL-2026-000111# Claim : CLM-{YEAR}-{4 digit seq} e.g. CLM-2026-000112# Buyer : REG-{CC}-{registration_number} e.g. REG-GB-1234567813#14# Status values align with backend/core/enums.py15# PolicyStatus : inforce, expired, canceled, declined16# ClaimStatus : pending, approved, rejected, paid17# ============================================================18 19import boto320import json21import logging22from datetime import datetime23from decimal import Decimal24from botocore.exceptions import ClientError25 26from backend.core.enums import PolicyStatus, ClaimStatus27 28logging.basicConfig(level=logging.INFO, format="%(levelname)s - %(message)s")29 30dynamodb = boto3.resource("dynamodb", region_name="us-east-1")31 32 33# ============================================================34# SECTION 1 -- ID GENERATORS35# ============================================================36 37def generate_customer_id() -> str:38 """Generates next sequential customer ID."""39 table = dynamodb.Table("tci_customers")40 try:41 response = table.scan(ProjectionExpression="customer_id")42 items = response.get("Items", [])43 if not items:44 return "CUST-0001"45 ids = [int(i["customer_id"].split("-")[1]) for i in items if "-" in i.get("customer_id", "")]46 next_num = max(ids) + 1 if ids else 147 return f"CUST-{str(next_num).zfill(4)}"48 except Exception:49 return f"CUST-{str(datetime.now().microsecond).zfill(4)}"50 51 52def generate_policy_id() -> str:53 """Generates next sequential policy ID for current year."""54 year = datetime.now().year55 table = dynamodb.Table("tci_customers")56 try:57 response = table.scan(ProjectionExpression="policies")58 all_ids = []59 for item in response.get("Items", []):60 for policy in item.get("policies", []):61 pid = policy.get("policy_id", "")62 if pid.startswith(f"POL-{year}-"):63 try:64 all_ids.append(int(pid.split("-")[-1]))65 except ValueError:66 pass67 next_num = max(all_ids) + 1 if all_ids else 168 return f"POL-{year}-{str(next_num).zfill(4)}"69 except Exception:70 return f"POL-{year}-0001"71 72 73def generate_claim_id() -> str:74 """Generates next sequential claim ID for current year."""75 year = datetime.now().year76 table = dynamodb.Table("tci_customers")77 try:78 response = table.scan(ProjectionExpression="policies")79 all_ids = []80 for item in response.get("Items", []):81 for policy in item.get("policies", []):82 for claim in policy.get("claims", []):83 cid = claim.get("claim_id", "")84 if cid.startswith(f"CLM-{year}-"):85 try:86 all_ids.append(int(cid.split("-")[-1]))87 except ValueError:88 pass89 next_num = max(all_ids) + 1 if all_ids else 190 return f"CLM-{year}-{str(next_num).zfill(4)}"91 except Exception:92 return f"CLM-{year}-0001"93 94 95def generate_buyer_id(country_code: str, registration_number: str) -> str:96 """Generates buyer ID in format REG-{CC}-{registration_number}."""97 return f"REG-{country_code.upper()}-{registration_number}"98 99 100# ============================================================101# SECTION 2 -- CUSTOMER OPERATIONS102# ============================================================103 104def save_customer(data: dict) -> str:105 """Saves a new customer. Returns customer_id."""106 table = dynamodb.Table("tci_customers")107 customer_id = generate_customer_id()108 now = datetime.now().isoformat()109 110 table.put_item(Item={111 "customer_id": customer_id,112 "business_name": data.get("business_name"),113 "registration_no": data.get("registration_no"),114 "industry": data.get("industry"),115 "trade_type": data.get("trade_type"),116 "annual_turnover": Decimal(str(data.get("annual_turnover", 0))),117 "country": data.get("country"),118 "contact_email": data.get("contact_email"),119 "policies": [],120 "created_at": now,121 "updated_at": now,122 })123 logging.info(f"Customer saved : {customer_id}")124 return customer_id125 126 127def get_customer(customer_id: str) -> dict:128 """Retrieves a customer by ID."""129 table = dynamodb.Table("tci_customers")130 response = table.get_item(Key={"customer_id": customer_id})131 return response.get("Item")132 133 134def add_policy_to_customer(customer_id: str, policy: dict):135 """Adds a policy object to a customer's policies list."""136 table = dynamodb.Table("tci_customers")137 table.update_item(138 Key={"customer_id": customer_id},139 UpdateExpression="SET policies = list_append(policies, :p), updated_at = :now",140 ExpressionAttributeValues={141 ":p": [policy],142 ":now": datetime.now().isoformat()143 }144 )145 logging.info(f"Policy added : {policy.get('policy_id')} to {customer_id}")146 147 148def add_claim_to_policy(customer_id: str, policy_id: str, claim: dict):149 """Adds a claim to a specific policy within a customer record."""150 customer = get_customer(customer_id)151 if not customer:152 logging.warning(f"Customer not found: {customer_id}")153 return154 155 policies = customer.get("policies", [])156 for i, policy in enumerate(policies):157 if policy.get("policy_id") == policy_id:158 policies[i].setdefault("claims", []).append(claim)159 break160 161 table = dynamodb.Table("tci_customers")162 table.update_item(163 Key={"customer_id": customer_id},164 UpdateExpression="SET policies = :p, updated_at = :now",165 ExpressionAttributeValues={166 ":p": policies,167 ":now": datetime.now().isoformat()168 }169 )170 logging.info(f"Claim added : {claim.get('claim_id')} to {policy_id}")171 172 173# ============================================================174# SECTION 3 -- BUYER OPERATIONS175# ============================================================176 177def save_buyer(data: dict) -> str:178 """Saves or updates a buyer record. Returns buyer_id."""179 table = dynamodb.Table("tci_buyers")180 buyer_id = generate_buyer_id(181 data.get("country_code", "XX"),182 data.get("registration_number", "UNKNOWN")183 )184 now = datetime.now().isoformat()185 186 try:187 table.put_item(Item={188 "buyer_id": buyer_id,189 "registration_number": data.get("registration_number", "UNKNOWN"),190 "country_code": data.get("country_code", "XX"),191 "business_name": data.get("business_name"),192 "industry": data.get("industry"),193 "status": data.get("status", "active"),194 "created_at": now,195 "updated_at": now,196 })197 logging.info(f"Buyer saved : {buyer_id}")198 except Exception as e:199 logging.error(f"Failed to save buyer: {e}")200 201 return buyer_id202 203 204def get_buyer(buyer_id: str) -> dict:205 """Retrieves a buyer by ID."""206 table = dynamodb.Table("tci_buyers")207 response = table.get_item(Key={"buyer_id": buyer_id})208 return response.get("Item")