#!/usr/bin/env python3 """ Notion API Integration Module Handles all Notion database operations for MAWB and Entry management. """ import requests import json import logging import os import re import time import hashlib from typing import Dict, List, Optional, Any from datetime import datetime, timedelta from zoneinfo import ZoneInfo from dotenv import load_dotenv # Load environment variables from .env file load_dotenv() class NotionIntegration: def __init__(self, api_key: str, database_ids: Dict[str, str], cache_ttl: int = 300): """ Initialize Notion API integration with smart caching. Args: api_key: Notion API key database_ids: Dictionary mapping table names to database IDs cache_ttl: Cache time-to-live in seconds (default: 5 minutes) """ self.api_key = api_key self.database_ids = database_ids self.base_url = "https://api.notion.com/v1" self.headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", "Notion-Version": "2025-09-03" } # Smart caching system self._cache = {} self._cache_ttl = cache_ttl self._cache_stats = {"hits": 0, "misses": 0, "total_requests": 0} def _get_cache_key(self, method: str, endpoint: str, data: Dict = None) -> str: """Generate a unique cache key for the request.""" # Create a deterministic string from the request parameters key_data = f"{method}:{endpoint}" if data: # Sort the data to ensure consistent keys key_data += f":{json.dumps(data, sort_keys=True)}" return hashlib.md5(key_data.encode()).hexdigest() def _is_cacheable_request(self, method: str, endpoint: str) -> bool: """Determine if a request should be cached.""" # Only cache GET requests and POST data source queries (read operations) if method == "GET": return True if method == "POST" and "/data_sources/" in endpoint and "/query" in endpoint: return True return False def _get_from_cache(self, cache_key: str) -> Optional[Dict]: """Get data from cache if valid.""" if cache_key in self._cache: cached_data, timestamp = self._cache[cache_key] if time.time() - timestamp < self._cache_ttl: self._cache_stats["hits"] += 1 logging.debug(f"Cache hit for key: {cache_key[:8]}...") return cached_data else: # Remove expired entry del self._cache[cache_key] self._cache_stats["misses"] += 1 return None def _store_in_cache(self, cache_key: str, data: Dict) -> None: """Store data in cache with timestamp.""" self._cache[cache_key] = (data, time.time()) logging.debug(f"Cached response for key: {cache_key[:8]}...") def get_cache_stats(self) -> Dict[str, Any]: """Get cache performance statistics.""" total = self._cache_stats["hits"] + self._cache_stats["misses"] hit_rate = (self._cache_stats["hits"] / total * 100) if total > 0 else 0 mawb_cache_size = len(getattr(self, '_mawb_lookup_cache', {})) return { "cache_entries": len(self._cache), "cache_hits": self._cache_stats["hits"], "cache_misses": self._cache_stats["misses"], "hit_rate_percent": round(hit_rate, 2), "total_requests": self._cache_stats["total_requests"], "mawb_cache_entries": mawb_cache_size, "milestone_api_calls_saved": self._cache_stats.get("milestone_api_savings", 0) } def clear_cache(self) -> None: """Clear all cache entries.""" self._cache.clear() logging.info("Cache cleared") def clear_expired_cache(self) -> int: """Clear expired cache entries and return count cleared.""" current_time = time.time() expired_keys = [ key for key, (_, timestamp) in self._cache.items() if current_time - timestamp >= self._cache_ttl ] for key in expired_keys: del self._cache[key] if expired_keys: logging.debug(f"Cleared {len(expired_keys)} expired cache entries") return len(expired_keys) def invalidate_mawb_cache(self, mawb: str) -> int: """ Invalidate all cache entries related to a specific MAWB. Called when MAWB-related data is updated to ensure consistency. Args: mawb: MAWB number to invalidate cache for Returns: Number of cache entries invalidated """ mawb_related_keys = [] # Find all cache keys that contain the MAWB number for cache_key in self._cache.keys(): # Check if cache key contains MAWB-related queries cached_data, _ = self._cache[cache_key] cache_content = json.dumps(cached_data) if cached_data else "" if mawb in cache_content or mawb in cache_key: mawb_related_keys.append(cache_key) # Remove the identified keys for key in mawb_related_keys: del self._cache[key] if mawb_related_keys: logging.debug(f"Invalidated {len(mawb_related_keys)} cache entries for MAWB {mawb}") return len(mawb_related_keys) def invalidate_task_cache(self, task_type: str = None) -> int: """ Invalidate cache entries related to task queries. Called when task status is updated. Args: task_type: Optional task type filter (e.g., "Entry Submit") Returns: Number of cache entries invalidated """ task_related_keys = [] for cache_key in self._cache.keys(): cached_data, _ = self._cache[cache_key] cache_content = json.dumps(cached_data) if cached_data else "" # Check if this is a task-related query is_task_query = ( "mawb_task" in cache_content or "/databases/" in cache_key and "task" in cache_content.lower() ) if is_task_query: if task_type is None or task_type in cache_content: task_related_keys.append(cache_key) # Remove the identified keys for key in task_related_keys: del self._cache[key] if task_related_keys: logging.debug(f"Invalidated {len(task_related_keys)} task cache entries{f' for type {task_type}' if task_type else ''}") return len(task_related_keys) def _make_request(self, method: str, endpoint: str, data: Dict = None) -> Dict: """Make API request to Notion with smart caching and retry logic.""" self._cache_stats["total_requests"] += 1 # Check cache for read operations if self._is_cacheable_request(method, endpoint): cache_key = self._get_cache_key(method, endpoint, data) cached_result = self._get_from_cache(cache_key) if cached_result is not None: return cached_result url = f"{self.base_url}{endpoint}" # Retry configuration max_retries = 3 base_delay = 1.0 # seconds timeout = 60 # 60 seconds timeout for batch operations for attempt in range(max_retries): try: if method == "GET": response = requests.get(url, headers=self.headers, json=data, timeout=timeout) elif method == "POST": response = requests.post(url, headers=self.headers, json=data, timeout=timeout) elif method == "PATCH": response = requests.patch(url, headers=self.headers, json=data, timeout=timeout) response.raise_for_status() result = response.json() # Store in cache if cacheable if self._is_cacheable_request(method, endpoint): self._store_in_cache(cache_key, result) # Periodically clean expired cache entries if self._cache_stats["total_requests"] % 50 == 0: self.clear_expired_cache() return result except requests.exceptions.Timeout as e: if attempt < max_retries - 1: delay = base_delay * (2 ** attempt) logging.warning(f"Request timeout (attempt {attempt + 1}/{max_retries}), retrying in {delay}s: {e}") time.sleep(delay) else: logging.error(f"Request timeout after {max_retries} attempts: {e}") raise except requests.exceptions.HTTPError as e: # Retry on rate limit (429) or server errors (5xx) if e.response is not None: status_code = e.response.status_code if status_code == 429 or (500 <= status_code < 600): if attempt < max_retries - 1: delay = base_delay * (2 ** attempt) logging.warning(f"HTTP {status_code} error (attempt {attempt + 1}/{max_retries}), retrying in {delay}s") time.sleep(delay) continue # Don't retry on client errors (4xx except 429) logging.error(f"Notion API request failed: {e}") if hasattr(e.response, 'text'): logging.error(f"Response: {e.response.text}") raise except requests.exceptions.RequestException as e: # Retry on network errors if attempt < max_retries - 1: delay = base_delay * (2 ** attempt) logging.warning(f"Network error (attempt {attempt + 1}/{max_retries}), retrying in {delay}s: {e}") time.sleep(delay) else: logging.error(f"Network error after {max_retries} attempts: {e}") raise def _format_datetime_property(self, dt: datetime = None, timezone: str = None) -> Dict: """Format datetime for Notion date property with timezone conversion. Args: dt: datetime object (if None, uses current time) timezone: timezone string to convert to (e.g., 'US/Eastern', 'US/Pacific') Returns: Notion date property dict with datetime in specified timezone """ if dt is None: dt = datetime.now() # Convert to specified timezone if provided if timezone: try: target_tz = ZoneInfo(timezone) if dt.tzinfo is None: # If naive datetime, assume it's already in the target timezone dt = dt.replace(tzinfo=target_tz) else: # Convert from current timezone to target timezone dt = dt.astimezone(target_tz) logging.debug(f"Converted datetime to {timezone}: {dt.isoformat()}") except Exception as e: logging.warning(f"Invalid timezone '{timezone}': {e}, using datetime as-is") # For Notion API: use ISO 8601 format with timezone offset # Note: Notion displays timezone offsets as "GMT-7", "GMT-8", etc. in UI # This is controlled by Notion's display preferences, not our API format if dt.tzinfo is not None: # Use ISO format with timezone offset (e.g., 2025-09-11T12:22:00-07:00) date_property = { "date": { "start": dt.isoformat() # Notion will display as "GMT-7" in UI } } else: # Naive datetime - assume UTC date_property = { "date": { "start": dt.isoformat() } } return date_property def _format_text_property(self, text: str) -> Dict: """Format text for Notion rich text property.""" return { "rich_text": [ { "text": { "content": text } } ] } def _format_title_property(self, text: str) -> Dict: """Format text for Notion title property.""" return { "title": [ { "text": { "content": text } } ] } def _format_number_property(self, number) -> Dict: """Format number for Notion number property.""" return { "number": float(number) if number is not None else None } def _format_checkbox_property(self, checked: bool) -> Dict: """Format checkbox for Notion checkbox property.""" return { "checkbox": checked } def get_timezone_for_pod_code(self, pod_code: str) -> str: """Get timezone string for CBP port code. Args: pod_code: CBP port code (e.g., "4701", "2720") Returns: Timezone string (e.g., "US/Eastern") or default "US/Pacific" """ # Import constants here to avoid circular imports try: import sys import os sys.path.append(os.path.join(os.path.dirname(__file__), '..')) from utils.constants import POD_STATE_MAPPINGS, POD_TIMEZONE_MAPPINGS # Create reverse mapping: port code -> POD PORT_CODE_TO_POD = {code: pod for pod, (code, _) in POD_STATE_MAPPINGS.items()} # Get POD from port code pod = PORT_CODE_TO_POD.get(pod_code) logging.info(f"TIMEZONE DEBUG: Available port codes: {list(PORT_CODE_TO_POD.keys())}") logging.info(f"TIMEZONE DEBUG: Looking up port code '{pod_code}' -> POD '{pod}'") if pod: timezone = POD_TIMEZONE_MAPPINGS.get(pod, "US/Pacific") logging.info(f"TIMEZONE DEBUG: Mapped port code {pod_code} -> POD {pod} -> timezone {timezone}") return timezone else: logging.warning(f"TIMEZONE DEBUG: Unknown port code '{pod_code}', using default timezone US/Pacific") logging.info(f"TIMEZONE DEBUG: Available mappings: {POD_STATE_MAPPINGS}") return "US/Pacific" except Exception as e: logging.error(f"Error mapping port code to timezone: {e}") return "US/Pacific" def format_datetime_with_timezone(self, dt: datetime, timezone: str) -> Dict: """Format datetime with specific timezone for Notion. Args: dt: datetime object timezone: timezone string (e.g., 'US/Eastern') Returns: Notion date property dict """ return self._format_datetime_property(dt, timezone) def _get_data_source_id(self, database_name: str) -> str: """ Get the data source ID for a database using 2025-09-03 API. Args: database_name: Name of the database Returns: Data source ID """ database_id = self.database_ids.get(database_name) if not database_id: raise ValueError(f"Database ID not configured for: {database_name}") # Check cache first cache_key = f"data_source_{database_name}" if hasattr(self, '_data_source_cache') and cache_key in self._data_source_cache: return self._data_source_cache[cache_key] try: # Get database info to extract data source ID response = self._make_request("GET", f"/databases/{database_id}") data_sources = response.get("data_sources", []) if data_sources: data_source_id = data_sources[0]["id"] else: # Fallback: use database_id as data_source_id data_source_id = database_id # Cache the result if not hasattr(self, '_data_source_cache'): self._data_source_cache = {} self._data_source_cache[cache_key] = data_source_id logging.debug(f"Found data source ID for {database_name}: {data_source_id}") return data_source_id except Exception as e: logging.warning(f"Could not get data source ID for {database_name}, using database_id: {e}") return database_id def _make_database_query(self, database_name: str, query_data: Dict) -> Dict: """ Make a database query using the 2025-09-03 API format. Args: database_name: Name of the database query_data: Query payload Returns: Query response """ try: # Get the correct data source ID data_source_id = self._get_data_source_id(database_name) # Use data sources endpoint with POST method endpoint = f"/data_sources/{data_source_id}/query" logging.debug(f"Using data source endpoint for {database_name}: {data_source_id}") return self._make_request("POST", endpoint, query_data) except Exception as e: logging.error(f"Database query failed for {database_name}: {e}") raise def find_mawb_with_filters(self, mawb: str, customer: str = None, exists_only: bool = False) -> Optional[Dict]: """ Unified MAWB lookup with optional customer filtering. Args: mawb: MAWB number customer: Optional customer filter exists_only: If True, return True/False, if False return full record Returns: MAWB record dict if found (exists_only=False), True/False (exists_only=True), or None """ try: database_id = self.database_ids.get("mawb") if not database_id: logging.error("MAWB database ID not configured") return False if exists_only else None # Build filter - always include MAWB filter filters = [{ "property": "mawb", "title": { "equals": mawb } }] # Add customer filter if provided if customer: filters.append({ "property": "customer", "rich_text": { "equals": customer } }) # Build query data if len(filters) == 1: query_data = {"filter": filters[0]} else: query_data = {"filter": {"and": filters}} response = self._make_database_query("mawb", query_data) results = response.get("results", []) if exists_only: return len(results) > 0 else: return results[0] if results else None except Exception as e: logging.error(f"Error finding MAWB record: {e}") return False if exists_only else None def find_mawb_record_by_number(self, mawb: str) -> Optional[Dict]: """ Find MAWB record by MAWB number (backward compatibility wrapper). Args: mawb: MAWB number Returns: MAWB record dict if found, None otherwise """ return self.find_mawb_with_filters(mawb, customer=None, exists_only=False) def find_mawb_record_by_number_cached(self, mawb: str) -> Optional[Dict]: """ Enhanced MAWB lookup with specialized caching for milestone processing. Args: mawb: MAWB number Returns: MAWB record dict if found, None otherwise """ cache_key = f"mawb_lookup:{mawb}" # Initialize cache if it doesn't exist if not hasattr(self, '_mawb_lookup_cache'): self._mawb_lookup_cache = {} # Check specialized MAWB cache first cached_result = self._mawb_lookup_cache.get(cache_key) if cached_result and (time.time() - cached_result[1]) < 300: # 5-minute TTL self._cache_stats["hits"] += 1 logging.debug(f"MAWB lookup cache hit for: {mawb}") return cached_result[0] # Cache miss - use existing method self._cache_stats["misses"] += 1 result = self.find_mawb_record_by_number(mawb) # Cache the result if result: self._mawb_lookup_cache[cache_key] = (result, time.time()) logging.debug(f"MAWB lookup cached for: {mawb}") # Clean up old cache entries periodically if len(self._mawb_lookup_cache) > 100: self._cleanup_mawb_cache() return result def _cleanup_mawb_cache(self): """Clean up expired MAWB cache entries.""" current_time = time.time() expired_keys = [ key for key, (_, timestamp) in self._mawb_lookup_cache.items() if current_time - timestamp > 300 ] for key in expired_keys: del self._mawb_lookup_cache[key] logging.debug(f"Cleaned up {len(expired_keys)} expired MAWB cache entries") def check_mawb_exists(self, mawb: str, customer: str) -> bool: """ Check if MAWB already exists in Notion database (optimized wrapper). Args: mawb: MAWB number customer: Customer name Returns: True if MAWB exists, False otherwise """ return self.find_mawb_with_filters(mawb, customer=customer, exists_only=True) def create_mawb_record(self, mawb: str, customer: str, entry_type: str, pod: str, house_bill: str = None, cartons: int = None, pieces: int = None, cwt: float = None, eta: datetime = None, awb_received: datetime = None, manifest_received: datetime = None, password_received: datetime = None, entry_line: int = None) -> bool: """ Create new MAWB record in Notion. Args: mawb: MAWB number customer: Customer name entry_type: Type of entry pod: Port of Discharge house_bill: House bill number (optional) cartons: Number of cartons (optional) pieces: Number of pieces (optional) cwt: Chargeable weight from AWB (optional) eta: Estimated time of arrival (optional) awb_received: AWB received timestamp (optional) manifest_received: Manifest received timestamp (optional) password_received: Password received timestamp (optional) entry_line: Number of entry lines in manifest (optional) Returns: True if successful, False otherwise """ try: database_id = self.database_ids.get("mawb") if not database_id: logging.error("MAWB database ID not configured") return False properties = { "mawb": self._format_title_property(mawb), "customer": self._format_text_property(customer), "entry_type": self._format_text_property(entry_type), "pod": self._format_text_property(pod), "date_processed": self._format_datetime_property(), "last_updated": self._format_datetime_property() } # Add optional fields if house_bill: properties["house_bill"] = self._format_text_property(house_bill) if cartons is not None: properties["cartons"] = self._format_number_property(cartons) if pieces is not None: properties["pieces"] = self._format_number_property(pieces) if cwt is not None: properties["cwt"] = self._format_number_property(cwt) if eta: properties["eta"] = self._format_datetime_property(eta) if awb_received: properties["awb_received"] = self._format_datetime_property(awb_received, "US/Pacific") if manifest_received: properties["manifest_received"] = self._format_datetime_property(manifest_received, "US/Pacific") if password_received: properties["password_received"] = self._format_datetime_property(password_received, "US/Pacific") if entry_line is not None: properties["entry_line"] = self._format_number_property(entry_line) # Get the actual data source ID for page creation data_source_id = self._get_data_source_id("mawb") data = { "parent": { "type": "data_source_id", "data_source_id": data_source_id }, "properties": properties } self._make_request("POST", "/pages", data) logging.info(f"Created MAWB record: {mawb} for {customer}") return True except Exception as e: logging.error(f"Error creating MAWB record: {e}") return False def update_mawb_status(self, mawb: str, customer: str, **updates) -> bool: """ Update MAWB record status fields. Args: mawb: MAWB number customer: Customer name **updates: Fields to update (awb_received, manifest_received, password_received, cwt, etc.) Returns: True if successful, False otherwise """ try: # First find the record database_id = self.database_ids.get("mawb") if not database_id: logging.error("MAWB database ID not configured") return False query_data = { "filter": { "and": [ { "property": "mawb", "title": { "equals": mawb } }, { "property": "customer", "rich_text": { "equals": customer } } ] } } response = self._make_database_query("mawb", query_data) results = response.get("results", []) if not results: logging.error(f"MAWB record not found: {mawb}") return False page_id = results[0]["id"] # Format updates for Notion properties = {"last_updated": self._format_datetime_property()} for key, value in updates.items(): if isinstance(value, datetime): # Apply timezone for specific field types if key in ['awb_received', 'manifest_received', 'password_received']: properties[key] = self._format_datetime_property(value, "US/Pacific") else: properties[key] = self._format_datetime_property(value) elif isinstance(value, str): properties[key] = self._format_text_property(value) elif isinstance(value, (int, float)): properties[key] = self._format_number_property(value) elif isinstance(value, bool): properties[key] = self._format_checkbox_property(value) update_data = { "properties": properties } self._make_request("PATCH", f"/pages/{page_id}", update_data) logging.info(f"Updated MAWB {mawb} status: {updates}") return True except Exception as e: logging.error(f"Error updating MAWB status: {e}") return False def create_entry_record(self, mawb: str, entry_number: str, ams_submit_date: datetime = None, entry_submit_date: datetime = None, cbp_hold_line: str = None, cbp_hold_date: datetime = None, cbp_release_date: datetime = None, cpsc_review_line: str = None, cpsc_review_date: datetime = None, cpsc_release_date: datetime = None, estimated_review_date: datetime = None, auto_link_milestone_tasks: bool = True) -> bool: """ Create new Entry record in Notion with automatic MAWB relation linking. Args: mawb: MAWB number entry_number: Entry number ams_submit_date: AMS submit date (optional) entry_submit_date: Entry submit date (optional) cbp_hold_line: CBP hold line (optional) cbp_hold_date: CBP hold date (optional) cbp_release_date: CBP release date (optional) cpsc_review_line: CPSC review line (optional) cpsc_review_date: CPSC review date (optional) cpsc_release_date: CPSC release date (optional) estimated_review_date: Estimated review date from CPSC tracker (optional) auto_link_milestone_tasks: Whether to automatically link to milestone tasks (default True) Returns: True if successful, False otherwise """ try: database_id = self.database_ids.get("entry_record") if not database_id: logging.error("Entry record database ID not configured") return False # Find the MAWB record to link to mawb_record = self.find_mawb_record_by_number(mawb) # Extract POD for timezone determination pod_timezone = "US/Pacific" # default if mawb_record: try: mawb_properties = mawb_record.get("properties", {}) if "pod" in mawb_properties and mawb_properties["pod"].get("rich_text"): pod = mawb_properties["pod"]["rich_text"][0].get("text", {}).get("content", "") if pod: pod_timezone = self.get_timezone_for_pod_code(pod) logging.info(f"Using POD timezone {pod_timezone} for MAWB {mawb} entries") except Exception as e: logging.warning(f"Error extracting POD from MAWB {mawb}: {e}, using default timezone") properties = { "mawb": self._format_title_property(mawb), "entry_number": self._format_text_property(entry_number) } # Add MAWB relation if found if mawb_record: properties["mawb_reference"] = { "relation": [{"id": mawb_record["id"]}] } logging.info(f"Linking entry {entry_number} to MAWB record {mawb}") else: logging.warning(f"MAWB record not found for {mawb}, creating entry without relation") # Add optional fields with appropriate timezone handling if ams_submit_date: properties["ams_submit_date"] = self._format_datetime_property(ams_submit_date, pod_timezone) if entry_submit_date: properties["entry_submit_date"] = self._format_datetime_property(entry_submit_date, pod_timezone) if cbp_hold_line: properties["cbp_hold_line"] = self._format_text_property(cbp_hold_line) if cbp_hold_date: properties["cbp_hold_date"] = self._format_datetime_property(cbp_hold_date, pod_timezone) if cbp_release_date: properties["cbp_release_date"] = self._format_datetime_property(cbp_release_date, pod_timezone) if cpsc_review_line: properties["cpsc_review_line"] = self._format_text_property(cpsc_review_line) if cpsc_review_date: properties["cpsc_review_date"] = self._format_datetime_property(cpsc_review_date, "US/Eastern") if cpsc_release_date: properties["cpsc_release_date"] = self._format_datetime_property(cpsc_release_date, "US/Eastern") if estimated_review_date: properties["estimated_review_date"] = self._format_datetime_property(estimated_review_date, "US/Eastern") # Get the actual data source ID for page creation data_source_id = self._get_data_source_id("entry_record") data = { "parent": { "type": "data_source_id", "data_source_id": data_source_id }, "properties": properties } self._make_request("POST", "/pages", data) logging.info(f"Created entry record: {entry_number} for MAWB {mawb}") # Auto-link to milestone and clearance tasks if enabled if auto_link_milestone_tasks: try: # Get both milestone and clearance tasks milestone_tasks = self.get_mawb_tasks_by_action_and_mawb(mawb, "Entry Milestone Update") clearance_tasks = self.get_mawb_tasks_by_action_and_mawb(mawb, "Clearance Notice") milestone_success = True clearance_success = True # Link milestone tasks to existing field if milestone_tasks: milestone_task_ids = [task["id"] for task in milestone_tasks] milestone_success = self.create_entry_mawb_task_relation(entry_number, milestone_task_ids) # Link clearance tasks to new field if clearance_tasks: clearance_task_ids = [task["id"] for task in clearance_tasks] clearance_success = self.create_entry_clearance_task_relation(entry_number, clearance_task_ids) # Log results milestone_count = len(milestone_tasks) if milestone_tasks else 0 clearance_count = len(clearance_tasks) if clearance_tasks else 0 if milestone_success and clearance_success: logging.info(f"Auto-linked entry {entry_number} to {milestone_count} milestone tasks and {clearance_count} clearance tasks") else: failed_types = [] if not milestone_success and milestone_count > 0: failed_types.append("milestone") if not clearance_success and clearance_count > 0: failed_types.append("clearance") if failed_types: logging.warning(f"Failed to auto-link entry {entry_number} to {', '.join(failed_types)} tasks") if milestone_count == 0 and clearance_count == 0: logging.debug(f"No milestone or clearance tasks found for MAWB {mawb} to link to entry {entry_number}") except Exception as link_error: logging.error(f"Error auto-linking tasks for entry {entry_number}: {link_error}") return True except Exception as e: logging.error(f"Error creating entry record: {e}") return False def create_entry_records_batch(self, entries_data: List[Dict]) -> Dict[str, int]: """ Create multiple entry records in batch for improved performance. Args: entries_data: List of entry data dictionaries, each containing: - mawb: MAWB number - entry_number: Entry number - Optional fields: ams_submit_date, entry_submit_date, etc. Returns: Dict with statistics: {"created": count, "failed": count, "total": count} """ stats = {"created": 0, "failed": 0, "total": len(entries_data)} if not entries_data: logging.info("No entries to create in batch") return stats try: database_id = self.database_ids.get("entry_record") if not database_id: logging.error("Entry record database ID not configured") stats["failed"] = stats["total"] return stats # Pre-fetch MAWB records for all entries to avoid repeated lookups # Note: Caller is responsible for checking existence and separating create vs update unique_mawbs = list(set(entry.get("mawb", "") for entry in entries_data)) mawb_records_map = {} for mawb in unique_mawbs: if mawb: mawb_record = self.find_mawb_record_by_number(mawb) if mawb_record: mawb_records_map[mawb] = mawb_record["id"] logging.info(f"Pre-fetched {len(mawb_records_map)} MAWB records for batch entry creation") # Create entries in batches (Notion API doesn't support true batch creation, # but we can optimize by reducing redundant MAWB lookups) for entry_data in entries_data: try: mawb = entry_data.get("mawb", "") entry_number = entry_data.get("entry_number", "") if not mawb or not entry_number: logging.error(f"Missing required fields in entry data: {entry_data}") stats["failed"] += 1 continue properties = { "mawb": self._format_title_property(mawb), "entry_number": self._format_text_property(entry_number) } # Add MAWB relation if found in pre-fetched map if mawb in mawb_records_map: properties["mawb_reference"] = { "relation": [{"id": mawb_records_map[mawb]}] } # Add optional fields optional_fields = [ "ams_submit_date", "entry_submit_date", "entry_created", "entry_date", "cbp_hold_line", "cbp_hold_date", "cbp_release_date", "cpsc_review_line", "cpsc_review_date", "cpsc_release_date", "estimated_review_date" ] for field in optional_fields: value = entry_data.get(field) if value: if "date" in field or field == "entry_created": # Apply field-specific timezone logic timezone = None source_timezone = None entry_number = entry_data.get("entry_number", "unknown") if field == 'entry_date': # entry_date should always be in Pacific timezone to preserve the date value timezone = "US/Pacific" source_timezone = "US/Pacific" elif field in ['estimated_review_date']: timezone = "US/Eastern" elif field in ['cpsc_review_date', 'cpsc_release_date']: timezone = "US/Eastern" elif field in ['entry_submit_date', 'entry_created', 'cbp_hold_date', 'cbp_release_date']: # CBP fields should use POD timezone from entry_data, default to Pacific timezone = entry_data.get("_timezone", "US/Pacific") else: # Other date fields use provided timezone or no conversion timezone = entry_data.get("_timezone") logging.info(f"BATCH TIMEZONE DEBUG: Entry {entry_number}, field {field}, timezone: {timezone}") properties[field] = self._format_datetime_property(value, timezone, source_timezone=source_timezone) else: properties[field] = self._format_text_property(str(value)) # Get the actual data source ID for page creation data_source_id = self._get_data_source_id("entry_record") data = { "parent": { "type": "data_source_id", "data_source_id": data_source_id }, "properties": properties } self._make_request("POST", "/pages", data) stats["created"] += 1 # Auto-link to milestone and clearance tasks if enabled if entry_data.get("auto_link_milestone_tasks", True): try: # Get both milestone and clearance tasks milestone_tasks = self.get_mawb_tasks_by_action_and_mawb(mawb, "Entry Milestone Update") clearance_tasks = self.get_mawb_tasks_by_action_and_mawb(mawb, "Clearance Notice") milestone_success = True clearance_success = True # Link milestone tasks to existing field if milestone_tasks: milestone_task_ids = [task["id"] for task in milestone_tasks] milestone_success = self.create_entry_mawb_task_relation(entry_number, milestone_task_ids) # Link clearance tasks to new field if clearance_tasks: clearance_task_ids = [task["id"] for task in clearance_tasks] clearance_success = self.create_entry_clearance_task_relation(entry_number, clearance_task_ids) # Log results milestone_count = len(milestone_tasks) if milestone_tasks else 0 clearance_count = len(clearance_tasks) if clearance_tasks else 0 if milestone_success and clearance_success: if milestone_count > 0 or clearance_count > 0: logging.info(f"Auto-linked entry {entry_number} to {milestone_count} milestone tasks and {clearance_count} clearance tasks") else: failed_types = [] if not milestone_success and milestone_count > 0: failed_types.append("milestone") if not clearance_success and clearance_count > 0: failed_types.append("clearance") if failed_types: logging.warning(f"Failed to auto-link entry {entry_number} to {', '.join(failed_types)} tasks") except Exception as link_error: logging.error(f"Error auto-linking tasks for entry {entry_number}: {link_error}") except Exception as e: logging.error(f"Error creating entry {entry_data.get('entry_number', 'unknown')}: {e}") stats["failed"] += 1 logging.info(f"Batch entry creation completed: {stats['created']} created, {stats['failed']} failed out of {stats['total']} total") return stats except Exception as e: logging.error(f"Error in batch entry creation: {e}") stats["failed"] = stats["total"] return stats def create_mawb_tasks_by_type(self, mawb: str, customer: str, entry_type: str) -> bool: """ Create multiple MAWB task records based on customer and entry type. Args: mawb: MAWB number customer: Customer name (SHEIN, BELL) entry_type: Entry type (T86, T01, T11) Returns: True if all tasks created successfully, False otherwise """ try: # Find the MAWB record to link to mawb_record = self.find_mawb_record_by_number(mawb) if not mawb_record: logging.error(f"MAWB record not found for {mawb}") return False # Define tasks based on customer and entry type tasks = [] if customer == "SHEIN" and entry_type == "T01": tasks = [ "Entry Submit", "Clearance Notice", "Customs Document Upload", "Entry Milestone Update", "MAWB Milestone Update" ] elif entry_type in ["T86", "T11"]: # All customers for T86/T11 tasks = [ "AMS Submit", "ACAS Notify", "Clearance Notice" ] else: logging.warning(f"No tasks defined for {customer} {entry_type}") return True # Not an error, just no tasks to create # Create individual task records database_id = self.database_ids.get("mawb_task") if not database_id: logging.error("MAWB task database ID not configured") return False success_count = 0 for task_action in tasks: try: properties = { "mawb": self._format_title_property(mawb), "action": self._format_text_property(task_action), "done": self._format_checkbox_property(False), "entry_type": self._format_text_property(entry_type), "customer": self._format_text_property(customer), "mawb_reference": { "relation": [{"id": mawb_record["id"]}] } } # Get the actual data source ID for page creation data_source_id = self._get_data_source_id("mawb_task") data = { "parent": { "type": "data_source_id", "data_source_id": data_source_id }, "properties": properties } self._make_request("POST", "/pages", data) success_count += 1 logging.info(f"Created task record: {mawb} - {task_action}") except Exception as e: logging.error(f"Error creating task record {mawb} - {task_action}: {e}") if success_count == len(tasks): logging.info(f"Successfully created all {len(tasks)} task records for MAWB {mawb}") return True else: logging.warning(f"Created {success_count}/{len(tasks)} task records for MAWB {mawb}") return False except Exception as e: logging.error(f"Error creating MAWB task records: {e}") return False def create_mawb_task_record(self, mawb: str, acas_feedback: bool = False, bol_update: bool = False, correction_7501: bool = False, clearance_notice: bool = False, entry_milestone_update: bool = False, customs_doc_upload: bool = False, warehouse_in: datetime = None, warehouse_out: datetime = None, created_date: datetime = None, cargo_ready: datetime = None) -> bool: """ Create new MAWB task record in Notion. Args: mawb: MAWB number acas_feedback: ACAS feedback completed (default: False) bol_update: BOL update completed (default: False) correction_7501: 7501 correction completed (default: False) clearance_notice: Clearance notice sent (default: False) entry_milestone_update: Entry milestone updated (default: False) customs_doc_upload: Customs document uploaded (default: False) warehouse_in: Warehouse in timestamp (optional) warehouse_out: Warehouse out timestamp (optional) Returns: True if successful, False otherwise """ try: database_id = self.database_ids.get("mawb_task") if not database_id: logging.error("MAWB task database ID not configured") return False properties = { "mawb": self._format_title_property(mawb), "acas_feedback": self._format_checkbox_property(acas_feedback), "bol_update": self._format_checkbox_property(bol_update), "7501_correction": self._format_checkbox_property(correction_7501), "clearance_notice": self._format_checkbox_property(clearance_notice), "entry_milestone_update": self._format_checkbox_property(entry_milestone_update), "customs_doc_upload": self._format_checkbox_property(customs_doc_upload), "created_date": self._format_datetime_property(created_date or datetime.now()) } # Add optional timestamp fields if warehouse_in: properties["warehouse_in"] = self._format_datetime_property(warehouse_in) if warehouse_out: properties["warehouse_out"] = self._format_datetime_property(warehouse_out) if cargo_ready: properties["cargo_ready"] = self._format_datetime_property(cargo_ready) # Get the actual data source ID for page creation data_source_id = self._get_data_source_id("mawb_task") data = { "parent": { "type": "data_source_id", "data_source_id": data_source_id }, "properties": properties } self._make_request("POST", "/pages", data) logging.info(f"Created MAWB task record for: {mawb}") return True except Exception as e: logging.error(f"Error creating MAWB task record: {e}") return False def create_mawb_milestone_record(self, mawb: str, last_mile_carrier: str, cartons: int = 0, pallets: int = 0, gwt: float = None, cwt: float = None, warehouse_in: datetime = None, cargo_ready: datetime = None, warehouse_out: datetime = None, etd: datetime = None, eta: datetime = None, cartons_left_at_airline: int = 0, cartons_left_at_warehouse: int = 0) -> bool: """ Create new MAWB milestone record in Notion. Args: mawb: MAWB number (title field) last_mile_carrier: Last mile service provider name cartons: Number of cartons for this carrier pallets: Number of pallets (default 0) gwt: Gross weight (manual input by user) cwt: Chargeable weight (manual input by user) warehouse_in: Timestamp when cargo enters warehouse cargo_ready: Timestamp when cargo is ready warehouse_out: Timestamp when cargo leaves warehouse etd: Estimated time of departure eta: Estimated time of arrival cartons_left_at_airline: Cartons remaining at airline (default 0) cartons_left_at_warehouse: Cartons remaining at warehouse (default 0) Returns: True if successful, False otherwise """ try: database_id = self.database_ids.get("mawb_milestone") if not database_id: logging.error("MAWB milestone database ID not configured") return False # Find the MAWB record to create relation mawb_record = self.find_mawb_record_by_number(mawb) mawb_record_id = mawb_record.get("id") if mawb_record else None properties = { "mawb": self._format_title_property(mawb), "last_mile_carrier": self._format_rich_text_property(last_mile_carrier), "cartons": self._format_number_property(cartons), "pallets": self._format_number_property(pallets), "cartons_left_at_airline": self._format_number_property(cartons_left_at_airline), "cartons_left_at_warehouse": self._format_number_property(cartons_left_at_warehouse) } # Add optional fields if gwt is not None: properties["gwt"] = self._format_number_property(gwt) if cwt is not None: properties["cwt"] = self._format_number_property(cwt) if warehouse_in: properties["warehouse_in"] = self._format_datetime_property(warehouse_in) if cargo_ready: properties["cargo_ready"] = self._format_datetime_property(cargo_ready) if warehouse_out: properties["warehouse_out"] = self._format_datetime_property(warehouse_out) if etd: properties["etd"] = self._format_datetime_property(etd) if eta: properties["eta"] = self._format_datetime_property(eta) # Add relation to MAWB record if found if mawb_record_id: properties["mawb_reference"] = { "relation": [{"id": mawb_record_id}] } # Add relation to Clearance Notice tasks if they exist try: clearance_tasks = self.get_mawb_tasks_by_action_and_mawb(mawb, "Clearance Notice") if clearance_tasks: clearance_task_ids = [{"id": task["id"]} for task in clearance_tasks] properties["clearance_notice_task_reference"] = { "relation": clearance_task_ids } logging.debug(f"Linked milestone record to {len(clearance_task_ids)} clearance tasks for MAWB {mawb}") except Exception as e: logging.warning(f"Could not link clearance tasks to milestone for MAWB {mawb}: {e}") # Get the actual data source ID for page creation data_source_id = self._get_data_source_id("mawb_milestone") data = { "parent": { "type": "data_source_id", "data_source_id": data_source_id }, "properties": properties } self._make_request("POST", "/pages", data) logging.info(f"Created MAWB milestone record for: {mawb} - {last_mile_carrier}") return True except Exception as e: logging.error(f"Error creating MAWB milestone record: {e}") return False def update_mawb_milestone_record(self, mawb: str, last_mile_carrier: str, **updates) -> bool: """ Update MAWB milestone record. Args: mawb: MAWB number last_mile_carrier: Last mile carrier name **updates: Milestone fields to update Returns: True if successful, False otherwise """ try: database_id = self.database_ids.get("mawb_milestone") if not database_id: logging.error("MAWB milestone database ID not configured") return False # Find the specific milestone record query_data = { "filter": { "and": [ { "property": "mawb", "title": { "equals": mawb } }, { "property": "last_mile_carrier", "rich_text": { "equals": last_mile_carrier } } ] } } response = self._make_database_query("mawb_milestone", query_data) results = response.get("results", []) if not results: logging.error(f"MAWB milestone record not found: {mawb} - {last_mile_carrier}") return False page_id = results[0]["id"] # Format updates for Notion properties = {} for key, value in updates.items(): if isinstance(value, datetime): properties[key] = self._format_datetime_property(value) elif isinstance(value, (int, float)): properties[key] = self._format_number_property(value) elif isinstance(value, str): properties[key] = self._format_rich_text_property(value) elif isinstance(value, bool): properties[key] = self._format_checkbox_property(value) update_data = { "properties": properties } self._make_request("PATCH", f"/pages/{page_id}", update_data) logging.info(f"Updated MAWB milestone {mawb} - {last_mile_carrier}: {updates}") return True except Exception as e: logging.error(f"Error updating MAWB milestone record: {e}") return False def create_mawb_milestone_records_batch(self, milestone_records: List[Dict]) -> int: """ Create multiple milestone records efficiently with reduced API calls. Args: milestone_records: List of milestone record data dictionaries, each containing: - properties: Formatted Notion properties dict Returns: Number of successfully created records """ database_id = self.database_ids.get("mawb_milestone") if not database_id: logging.error("MAWB milestone database ID not configured") return 0 if not milestone_records: return 0 # Pre-fetch clearance tasks for all MAWBs to optimize API calls unique_mawbs = set() for record_data in milestone_records: properties = record_data.get("properties", {}) mawb_prop = properties.get("mawb", {}).get("title", []) if mawb_prop: mawb = mawb_prop[0].get("text", {}).get("content", "") if mawb: unique_mawbs.add(mawb) # Batch fetch clearance tasks for all MAWBs mawb_clearance_tasks = {} for mawb in unique_mawbs: try: clearance_tasks = self.get_mawb_tasks_by_action_and_mawb(mawb, "Clearance Notice") if clearance_tasks: mawb_clearance_tasks[mawb] = [{"id": task["id"]} for task in clearance_tasks] except Exception as e: logging.warning(f"Could not fetch clearance tasks for MAWB {mawb}: {e}") # Add clearance task relations to milestone records for record_data in milestone_records: properties = record_data.get("properties", {}) mawb_prop = properties.get("mawb", {}).get("title", []) if mawb_prop: mawb = mawb_prop[0].get("text", {}).get("content", "") if mawb in mawb_clearance_tasks: properties["clearance_notice_task_reference"] = { "relation": mawb_clearance_tasks[mawb] } if mawb_clearance_tasks: logging.info(f"Pre-fetched clearance tasks for {len(mawb_clearance_tasks)} MAWBs in batch milestone creation") success_count = 0 # Process in chunks to respect API limits and prevent overwhelming chunk_size = 10 for i in range(0, len(milestone_records), chunk_size): chunk = milestone_records[i:i + chunk_size] try: for record_data in chunk: # Get the actual data source ID for page creation data_source_id = self._get_data_source_id("mawb_milestone") data = { "parent": { "type": "data_source_id", "data_source_id": data_source_id }, "properties": record_data["properties"] } response = self._make_request("POST", "/pages", data) if response: success_count += 1 # Small delay to prevent rate limiting time.sleep(0.1) except Exception as e: logging.error(f"Error in batch milestone creation chunk: {e}") continue # Track API efficiency gains individual_calls_saved = len(milestone_records) - (len(milestone_records) // chunk_size + (len(milestone_records) % chunk_size > 0)) if "milestone_api_savings" not in self._cache_stats: self._cache_stats["milestone_api_savings"] = 0 self._cache_stats["milestone_api_savings"] += individual_calls_saved logging.info(f"Batch created {success_count}/{len(milestone_records)} milestone records (saved {individual_calls_saved} API calls)") return success_count def update_mawb_task(self, mawb: str, **updates) -> bool: """ Update MAWB task record. Args: mawb: MAWB number **updates: Task fields to update (boolean or datetime values) Returns: True if successful, False otherwise """ try: # First find the record database_id = self.database_ids.get("mawb_task") if not database_id: logging.error("MAWB task database ID not configured") return False query_data = { "filter": { "property": "mawb", "title": { "equals": mawb } } } response = self._make_database_query("mawb_task", query_data) results = response.get("results", []) if not results: logging.error(f"MAWB task record not found: {mawb}") return False page_id = results[0]["id"] # Format updates for Notion properties = {} for key, value in updates.items(): if isinstance(value, datetime): properties[key] = self._format_datetime_property(value) elif isinstance(value, bool): properties[key] = self._format_checkbox_property(value) update_data = { "properties": properties } self._make_request("PATCH", f"/pages/{page_id}", update_data) logging.info(f"Updated MAWB task {mawb}: {updates}") return True except Exception as e: logging.error(f"Error updating MAWB task: {e}") return False def get_all_mawb_records(self) -> List[Dict]: """ Get all MAWB records from Notion database. Returns: List of MAWB records with their properties """ try: database_id = self.database_ids.get("mawb") if not database_id: logging.error("MAWB database ID not configured") return [] query_data = { "page_size": 100 } all_results = [] has_more = True start_cursor = None while has_more: if start_cursor: query_data["start_cursor"] = start_cursor response = self._make_database_query("mawb", query_data) results = response.get("results", []) all_results.extend(results) has_more = response.get("has_more", False) start_cursor = response.get("next_cursor") return all_results except Exception as e: logging.error(f"Error getting MAWB records: {e}") return [] def get_entries_by_mawb_list(self, mawb_list: List[str]) -> List[Dict]: """ Get all entry records for a list of MAWBs using batched queries to handle large lists. Args: mawb_list: List of MAWB numbers Returns: List of entry records with their properties """ try: database_id = self.database_ids.get("entry_record") if not database_id: logging.error("Entry record database ID not configured") return [] if not mawb_list: return [] # Split large lists into smaller batches to avoid API limits BATCH_SIZE = 50 # Reasonable batch size for OR queries all_entries = [] for i in range(0, len(mawb_list), BATCH_SIZE): batch = mawb_list[i:i + BATCH_SIZE] batch_num = i//BATCH_SIZE + 1 total_batches = (len(mawb_list) + BATCH_SIZE - 1)//BATCH_SIZE logging.info(f"Processing entry batch {batch_num}/{total_batches} ({len(batch)} MAWBs)...") # Build query for this batch if len(batch) == 1: query_filter = { "property": "mawb", "title": {"equals": batch[0].strip()} } else: query_filter = { "or": [ { "property": "mawb", "title": {"equals": mawb.strip()} } for mawb in batch ] } query_data = { "filter": query_filter, "page_size": 100 } # Handle pagination for this batch has_more = True next_cursor = None while has_more: if next_cursor: query_data["start_cursor"] = next_cursor else: query_data.pop("start_cursor", None) response = self._make_database_query("entry_record", query_data) if not response: logging.warning(f"No response for entry batch {i//BATCH_SIZE + 1}") break results = response.get("results", []) all_entries.extend(results) has_more = response.get("has_more", False) next_cursor = response.get("next_cursor") logging.info(f"Retrieved {len(all_entries)} entries for {len(mawb_list)} MAWBs using {(len(mawb_list) + BATCH_SIZE - 1)//BATCH_SIZE} batches") return all_entries except Exception as e: logging.error(f"Error getting entries by MAWB list: {e}") return [] def get_mawb_info_by_list(self, mawb_list: List[str]) -> Dict[str, Dict]: """ Get MAWB information for a list of MAWBs. Args: mawb_list: List of MAWB numbers Returns: Dictionary mapping MAWB number to MAWB record data """ try: database_id = self.database_ids.get("mawb") if not database_id: logging.error("MAWB database ID not configured") return {} mawb_info = {} for mawb in mawb_list: query_data = { "filter": { "property": "mawb", "title": { "equals": mawb.strip() } } } response = self._make_database_query("mawb", query_data) results = response.get("results", []) if results: # Take the first match mawb_record = results[0] properties = mawb_record.get("properties", {}) # Extract key information pod = "" if "pod" in properties and properties["pod"].get("rich_text"): pod = properties["pod"]["rich_text"][0].get("text", {}).get("content", "") customer = "" if "customer" in properties and properties["customer"].get("rich_text"): customer = properties["customer"]["rich_text"][0].get("text", {}).get("content", "") entry_type = "" if "entry_type" in properties and properties["entry_type"].get("rich_text"): entry_type = properties["entry_type"]["rich_text"][0].get("text", {}).get("content", "") mawb_info[mawb.strip()] = { "pod": pod, "customer": customer, "entry_type": entry_type, "properties": properties } logging.info(f"Retrieved MAWB info for {len(mawb_info)} out of {len(mawb_list)} MAWBs") return mawb_info except Exception as e: logging.error(f"Error getting MAWB info by list: {e}") return {} def get_milestone_records_bulk(self, mawb_list: List[str]) -> List[Dict]: """ Get milestone records for a list of MAWBs using batched queries to handle large lists. Args: mawb_list: List of MAWB numbers to query Returns: List of milestone records for all requested MAWBs """ try: database_id = self.database_ids.get("mawb_milestone") if not database_id: logging.error("MAWB milestone database ID not configured") return [] if not mawb_list: return [] # Split large lists into smaller batches to avoid API limits BATCH_SIZE = 50 # Reasonable batch size for OR queries all_records = [] for i in range(0, len(mawb_list), BATCH_SIZE): batch = mawb_list[i:i + BATCH_SIZE] batch_num = i//BATCH_SIZE + 1 total_batches = (len(mawb_list) + BATCH_SIZE - 1)//BATCH_SIZE logging.info(f"Processing milestone batch {batch_num}/{total_batches} ({len(batch)} MAWBs)...") # Build query for this batch if len(batch) == 1: query_filter = { "property": "mawb", "title": {"equals": batch[0].strip()} } else: query_filter = { "or": [ { "property": "mawb", "title": {"equals": mawb.strip()} } for mawb in batch ] } query_data = { "filter": query_filter, "page_size": 100 } # Handle pagination for this batch has_more = True next_cursor = None while has_more: if next_cursor: query_data["start_cursor"] = next_cursor else: query_data.pop("start_cursor", None) response = self._make_database_query("mawb_milestone", query_data) if not response: logging.warning(f"No response for milestone batch {i//BATCH_SIZE + 1}") break results = response.get("results", []) all_records.extend(results) has_more = response.get("has_more", False) next_cursor = response.get("next_cursor") logging.info(f"Retrieved {len(all_records)} milestone records for {len(mawb_list)} MAWBs using {(len(mawb_list) + BATCH_SIZE - 1)//BATCH_SIZE} batches") return all_records except Exception as e: logging.error(f"Error getting milestone records bulk: {e}") return [] def get_entries_for_cpsc_tracking(self) -> List[Dict]: """ Get entry records that need CPSC tracking (have cpsc_review_date but no cpsc_review_line). Only queries entries created in the last 3 days to improve performance. Returns: List of entry records that need CPSC tracking """ try: database_id = self.database_ids.get("entry_record") if not database_id: logging.error("Entry record database ID not configured") return [] # Calculate date 3 days ago (include today) from datetime import datetime, timedelta three_days_ago = (datetime.now() - timedelta(days=3)).strftime("%Y-%m-%d") logging.info(f"Querying entries created on or after {three_days_ago}") # Query entries with cpsc_review_date AND created in last 3 days # Filter manually in Python for cpsc_review_line due to Notion API is_empty issues query_data = { "filter": { "and": [ { "property": "cpsc_review_date", "date": { "is_not_empty": True } }, { "timestamp": "created_time", "created_time": { "on_or_after": three_days_ago } } ] }, "page_size": 100 } all_entries = [] has_more = True start_cursor = None while has_more: if start_cursor: query_data["start_cursor"] = start_cursor response = self._make_database_query("entry_record", query_data) results = response.get("results", []) all_entries.extend(results) has_more = response.get("has_more", False) start_cursor = response.get("next_cursor") # Filter manually in Python - only entries with empty cpsc_review_line filtered_entries = [] for entry in all_entries: try: properties = entry.get("properties", {}) cpsc_line_prop = properties.get("cpsc_review_line", {}) cpsc_line_text = cpsc_line_prop.get("rich_text", []) # Check if cpsc_review_line is truly empty if not cpsc_line_text or len(cpsc_line_text) == 0: # No rich_text array means empty filtered_entries.append(entry) else: # Check if the text content is empty content = cpsc_line_text[0].get("text", {}).get("content", "") if content.strip() == "": filtered_entries.append(entry) except Exception as e: logging.warning(f"Error checking entry {entry.get('id', 'unknown')}: {e}") continue logging.info(f"Found {len(filtered_entries)} entries that need CPSC tracking (out of {len(all_entries)} created in last 3 days with cpsc_review_date)") return filtered_entries except Exception as e: logging.error(f"Error getting entries for CPSC tracking: {e}") return [] def get_entries_by_ids(self, entry_ids: List[str]) -> List[Dict]: """ Get specific entry records by their IDs. This is much more efficient for parallel processing than querying all entries. Args: entry_ids: List of Notion page IDs to retrieve Returns: List of entry record dictionaries """ try: if not entry_ids: return [] entries = [] logging.info(f"Fetching {len(entry_ids)} specific entries by ID...") # Fetch each entry individually by ID for entry_id in entry_ids: try: response = self._make_request("GET", f"/pages/{entry_id}", data=None) if response: entries.append(response) except Exception as e: logging.warning(f"Error fetching entry {entry_id}: {e}") continue logging.info(f"Successfully fetched {len(entries)} entries") return entries except Exception as e: logging.error(f"Error getting entries by IDs: {e}") return [] def update_missing_mawb_references(self) -> Dict[str, int]: """ Update missing MAWB references in entry records. Finds all entry records with blank mawb_reference relation field and links them to their corresponding MAWB record. Returns: Dict with statistics: {'processed': int, 'updated': int, 'errors': int} """ stats = {'processed': 0, 'updated': 0, 'errors': 0} try: database_id = self.database_ids.get("entry_record") if not database_id: logging.error("Entry record database ID not configured") return stats # Get all entry records (we'll filter manually for empty mawb_reference) query_data = {"page_size": 100} all_entries = [] has_more = True start_cursor = None while has_more: if start_cursor: query_data["start_cursor"] = start_cursor response = self._make_database_query("entry_record", query_data) results = response.get("results", []) all_entries.extend(results) has_more = response.get("has_more", False) start_cursor = response.get("next_cursor") logging.info(f"Retrieved {len(all_entries)} total entry records for MAWB reference check") # Filter for entries with empty mawb_reference entries_to_update = [] for entry in all_entries: try: properties = entry.get("properties", {}) mawb_ref_prop = properties.get("mawb_reference", {}) mawb_relation = mawb_ref_prop.get("relation", []) # Check if mawb_reference relation is empty if not mawb_relation or len(mawb_relation) == 0: # Get the MAWB number from the entry mawb_prop = properties.get("mawb", {}) mawb_title = mawb_prop.get("title", []) if mawb_title and len(mawb_title) > 0: mawb_number = mawb_title[0].get("text", {}).get("content", "").strip() if mawb_number: entries_to_update.append({ 'entry_id': entry.get("id"), 'mawb_number': mawb_number }) except Exception as e: logging.warning(f"Error checking entry {entry.get('id', 'unknown')}: {e}") stats['errors'] += 1 continue logging.info(f"Found {len(entries_to_update)} entries with missing MAWB references") # Update each entry with missing reference for entry_info in entries_to_update: try: stats['processed'] += 1 entry_id = entry_info['entry_id'] mawb_number = entry_info['mawb_number'] # Find the corresponding MAWB record mawb_record = self.find_mawb_record_by_number(mawb_number) if not mawb_record: logging.warning(f"No MAWB record found for MAWB number: {mawb_number}") stats['errors'] += 1 continue # Update the entry record with MAWB reference update_data = { "properties": { "mawb_reference": { "relation": [{"id": mawb_record["id"]}] } } } response = self._make_request("PATCH", f"/pages/{entry_id}", update_data) if response: stats['updated'] += 1 logging.info(f"Updated MAWB reference for entry with MAWB {mawb_number}") else: stats['errors'] += 1 except Exception as e: logging.error(f"Error updating entry {entry_info.get('entry_id', 'unknown')}: {e}") stats['errors'] += 1 continue logging.info(f"MAWB reference update completed: {stats['updated']} updated, {stats['errors']} errors out of {stats['processed']} processed") return stats except Exception as e: logging.error(f"Error updating missing MAWB references: {e}") stats['errors'] = stats['processed'] # Mark all as errors if batch fails return stats def update_entry_record_cpsc_data(self, mawb: str, entry_number: str, cpsc_review_line: str = None, estimated_review_date: datetime = None) -> bool: """ Update entry record with CPSC tracking data. Args: mawb: MAWB number entry_number: Entry number cpsc_review_line: CPSC review line (optional) estimated_review_date: Estimated review date (optional) Returns: True if successful, False otherwise """ try: # First find the record database_id = self.database_ids.get("entry_record") if not database_id: logging.error("Entry record database ID not configured") return False query_data = { "filter": { "and": [ { "property": "mawb", "title": { "equals": mawb } }, { "property": "entry_number", "rich_text": { "equals": entry_number } } ] } } response = self._make_database_query("entry_record", query_data) results = response.get("results", []) if not results: logging.error(f"Entry record not found: {mawb} - {entry_number}") return False page_id = results[0]["id"] # Format updates for Notion properties = {} if cpsc_review_line: properties["cpsc_review_line"] = self._format_text_property(cpsc_review_line) if estimated_review_date: properties["estimated_review_date"] = self._format_datetime_property(estimated_review_date, "US/Eastern") if not properties: logging.warning(f"No CPSC data to update for {mawb} - {entry_number}") return True update_data = { "properties": properties } self._make_request("PATCH", f"/pages/{page_id}", update_data) logging.info(f"Updated entry record {mawb} - {entry_number} with CPSC data") return True except Exception as e: logging.error(f"Error updating entry record CPSC data: {e}") return False def check_entry_exists(self, entry_number: str) -> bool: """ Check if an entry record exists by entry number. Args: entry_number: Entry number to check Returns: True if entry exists, False otherwise """ try: database_id = self.database_ids.get("entry_record") if not database_id: logging.error("Entry record database ID not configured") return False query_data = { "filter": { "property": "entry_number", "rich_text": { "equals": entry_number } } } response = self._make_database_query("entry_record", query_data) results = response.get("results", []) return len(results) > 0 except Exception as e: logging.error(f"Error checking if entry exists: {e}") return False def check_entry_field_populated(self, entry_number: str, field_name: str) -> bool: """ Check if a specific field is populated for an entry record. Args: entry_number: Entry number to check field_name: Name of the field to check Returns: True if field is populated, False otherwise """ try: database_id = self.database_ids.get("entry_record") if not database_id: logging.error("Entry record database ID not configured") return False query_data = { "filter": { "property": "entry_number", "rich_text": { "equals": entry_number } } } response = self._make_database_query("entry_record", query_data) results = response.get("results", []) if not results: logging.warning(f"Entry {entry_number} not found") return False # Check if the field is populated properties = results[0].get("properties", {}) field_data = properties.get(field_name, {}) # Handle different field types if "date" in field_data: return field_data["date"] is not None elif "rich_text" in field_data: rich_text_list = field_data["rich_text"] return len(rich_text_list) > 0 and rich_text_list[0].get("text", {}).get("content", "").strip() != "" elif "number" in field_data: return field_data["number"] is not None elif "checkbox" in field_data: return field_data["checkbox"] is not None else: logging.warning(f"Unknown field type for {field_name}") return False except Exception as e: logging.error(f"Error checking entry field: {e}") return False def update_entry_record_field(self, entry_number: str, field_name: str, field_value) -> bool: """ Update a specific field for an entry record. Args: entry_number: Entry number to update field_name: Name of the field to update field_value: Value to set (datetime, string, number, or boolean) Returns: True if successful, False otherwise """ try: database_id = self.database_ids.get("entry_record") if not database_id: logging.error("Entry record database ID not configured") return False # First find the entry record with retry for newly created records query_data = { "filter": { "property": "entry_number", "rich_text": { "equals": entry_number } } } page_id = None max_retries = 3 for attempt in range(max_retries): response = self._make_database_query("entry_record", query_data) results = response.get("results", []) if results: page_id = results[0]["id"] break elif attempt < max_retries - 1: # Wait briefly before retrying for newly created records import time logging.info(f"Entry record {entry_number} not found on attempt {attempt + 1}, retrying...") time.sleep(1) else: logging.error(f"Entry record not found after {max_retries} attempts: {entry_number}") return False # Format the field value based on type properties = {} # Special handling for known fields if isinstance(field_value, datetime): # Apply timezone for specific field types if field_name in ['awb_received', 'manifest_received', 'password_received']: properties[field_name] = self._format_datetime_property(field_value, "US/Pacific") elif field_name == 'entry_date': # entry_date should always be in Pacific timezone to preserve the date value properties[field_name] = self._format_datetime_property(field_value, "US/Pacific", source_timezone="US/Pacific") elif field_name == 'estimated_review_date': # Only estimated_review_date defaults to Eastern timezone properties[field_name] = self._format_datetime_property(field_value, "US/Eastern") elif field_name in ['cpsc_review_date', 'cpsc_release_date']: # CPSC fields use Eastern timezone properties[field_name] = self._format_datetime_property(field_value, "US/Eastern") elif field_name in ['entry_submit_date', 'entry_created', 'cbp_hold_date', 'cbp_release_date']: # CBP fields should use POD timezone - need to get POD from entry's MAWB try: # Get the entry record to find its MAWB entry_record = results[0] # We already have this from the query above mawb_relation = entry_record.get("properties", {}).get("mawb_reference", {}).get("relation", []) pod_timezone = "US/Pacific" # default if mawb_relation: # Get the MAWB record to extract POD mawb_id = mawb_relation[0]["id"] mawb_record = self._make_request("GET", f"/pages/{mawb_id}") if mawb_record: mawb_properties = mawb_record.get("properties", {}) if "pod" in mawb_properties and mawb_properties["pod"].get("rich_text"): pod = mawb_properties["pod"]["rich_text"][0].get("text", {}).get("content", "") if pod: pod_timezone = self.get_timezone_for_pod_code(pod) logging.info(f"Using POD timezone {pod_timezone} for CBP field {field_name} in entry {entry_number}") properties[field_name] = self._format_datetime_property(field_value, pod_timezone) except Exception as e: logging.warning(f"Error getting POD timezone for CBP field {field_name}: {e}, using Pacific default") properties[field_name] = self._format_datetime_property(field_value, "US/Pacific") else: # Other fields use no timezone conversion properties[field_name] = self._format_datetime_property(field_value) elif isinstance(field_value, str): properties[field_name] = self._format_text_property(field_value) elif isinstance(field_value, (int, float)): properties[field_name] = self._format_number_property(field_value) elif isinstance(field_value, bool): properties[field_name] = self._format_checkbox_property(field_value) else: logging.error(f"Unsupported field value type: {type(field_value)}") return False update_data = { "properties": properties } self._make_request("PATCH", f"/pages/{page_id}", update_data) logging.info(f"Updated entry {entry_number} field {field_name}") return True except Exception as e: logging.error(f"Error updating entry record field: {e}") return False def update_entry_record(self, entry_number: str, **kwargs) -> bool: """ Update an entry record with multiple fields. Args: entry_number: Entry number to update **kwargs: Field name-value pairs to update Returns: True if successful, False otherwise """ try: database_id = self.database_ids.get("entry_record") if not database_id: logging.error("Entry record database ID not configured") return False # Find the entry record query_data = { "filter": { "property": "entry_number", "rich_text": { "equals": entry_number } } } response = self._make_database_query("entry_record", query_data) results = response.get("results", []) if not results: logging.warning(f"Entry {entry_number} not found") return False page_id = results[0]["id"] # Build properties to update properties = {} # Extract timezone information if available pod_timezone = kwargs.get('_timezone') for field_name, field_value in kwargs.items(): # Skip internal fields that shouldn't be sent to Notion if field_name.startswith('_'): continue if field_value is not None: if isinstance(field_value, datetime): # Apply timezone for specific field types if field_name in ['awb_received', 'manifest_received', 'password_received']: properties[field_name] = self._format_datetime_property(field_value, "US/Pacific") elif field_name == 'entry_date': # entry_date should always be in Pacific timezone to preserve the date value properties[field_name] = self._format_datetime_property(field_value, "US/Pacific", source_timezone="US/Pacific") elif field_name == 'estimated_review_date': # Only estimated_review_date defaults to Eastern timezone properties[field_name] = self._format_datetime_property(field_value, "US/Eastern") elif field_name in ['cpsc_review_date', 'cpsc_release_date']: # CPSC fields use Eastern timezone (or POD timezone if explicitly provided) timezone = pod_timezone if pod_timezone else "US/Eastern" properties[field_name] = self._format_datetime_property(field_value, timezone) elif field_name in ['entry_submit_date', 'entry_created', 'cbp_hold_date', 'cbp_release_date']: # CBP fields should use POD timezone - get it from entry's MAWB if not provided if pod_timezone: properties[field_name] = self._format_datetime_property(field_value, pod_timezone) else: # Need to get POD timezone from entry's MAWB relation try: entry_record = results[0] # We already have this from the query above mawb_relation = entry_record.get("properties", {}).get("mawb_reference", {}).get("relation", []) cbp_timezone = "US/Pacific" # default if mawb_relation: # Get the MAWB record to extract POD mawb_id = mawb_relation[0]["id"] mawb_record = self._make_request("GET", f"/pages/{mawb_id}") if mawb_record: mawb_properties = mawb_record.get("properties", {}) if "pod" in mawb_properties and mawb_properties["pod"].get("rich_text"): pod = mawb_properties["pod"]["rich_text"][0].get("text", {}).get("content", "") if pod: cbp_timezone = self.get_timezone_for_pod_code(pod) logging.info(f"Using POD timezone {cbp_timezone} for CBP field {field_name} in entry {entry_number}") properties[field_name] = self._format_datetime_property(field_value, cbp_timezone) except Exception as e: logging.warning(f"Error getting POD timezone for CBP field {field_name}: {e}, using Pacific default") properties[field_name] = self._format_datetime_property(field_value, "US/Pacific") else: # Other fields use no timezone conversion properties[field_name] = self._format_datetime_property(field_value) elif isinstance(field_value, str): properties[field_name] = self._format_text_property(field_value) elif isinstance(field_value, (int, float)): properties[field_name] = {"number": field_value} elif isinstance(field_value, bool): properties[field_name] = {"checkbox": field_value} else: logging.warning(f"Unsupported field type for {field_name}: {type(field_value)}") continue if not properties: logging.info(f"No properties to update for entry {entry_number}") return True # Update the record update_data = {"properties": properties} response = self._make_request("PATCH", f"/pages/{page_id}", update_data) if response: logging.info(f"Updated entry {entry_number} with fields: {list(properties.keys())}") return True else: logging.error(f"Failed to update entry {entry_number}") return False except Exception as e: logging.error(f"Error updating entry record: {e}") return False def get_entry_records_with_incomplete_milestone_tasks(self) -> List[Dict]: """ Get entry records that have related mawb_task with action='Entry Milestone Update' and done=unchecked. Uses the new relation properties: entry_record.mawb_task_reference and mawb_task.entry_record_reference Returns: List of entry records with incomplete milestone tasks """ try: database_id = self.database_ids.get("entry_record") if not database_id: logging.error("Entry record database ID not configured") return [] # Note: Direct filtering on related properties in Notion API can be complex # We'll use a two-step approach: # 1. Get all mawb_task records with action='Entry Milestone Update' and done=false # 2. Get entry_records that reference those tasks # Step 1: Get incomplete milestone tasks mawb_task_db_id = self.database_ids.get("mawb_task") if not mawb_task_db_id: logging.error("MAWB task database ID not configured") return [] milestone_task_query = { "filter": { "and": [ { "property": "action", "rich_text": { "equals": "Entry Milestone Update" } }, { "property": "done", "checkbox": { "equals": False } } ] }, "page_size": 100 } # Handle pagination for milestone tasks incomplete_milestone_tasks = [] has_more = True start_cursor = None while has_more: if start_cursor: milestone_task_query["start_cursor"] = start_cursor milestone_response = self._make_database_query("mawb_task", milestone_task_query) tasks = milestone_response.get("results", []) incomplete_milestone_tasks.extend(tasks) has_more = milestone_response.get("has_more", False) start_cursor = milestone_response.get("next_cursor") if start_cursor: milestone_task_query["start_cursor"] = start_cursor else: milestone_task_query.pop("start_cursor", None) if not incomplete_milestone_tasks: logging.info("No incomplete Entry Milestone Update tasks found") return [] # Extract task IDs incomplete_task_ids = [task["id"] for task in incomplete_milestone_tasks] logging.info(f"Found {len(incomplete_task_ids)} incomplete Entry Milestone Update tasks") # Step 2: Get entry_records that reference these tasks all_entry_results = [] # Query entry_records for each incomplete task ID with pagination for task_id in incomplete_task_ids: entry_query = { "filter": { "property": "mawb_task_reference", "relation": { "contains": task_id } }, "page_size": 100 } # Handle pagination for each task's entry records has_more = True start_cursor = None while has_more: if start_cursor: entry_query["start_cursor"] = start_cursor entry_response = self._make_database_query("entry_record", entry_query) entry_results = entry_response.get("results", []) # Add task info to each entry record for context for entry in entry_results: entry["_incomplete_milestone_task_id"] = task_id all_entry_results.extend(entry_results) has_more = entry_response.get("has_more", False) start_cursor = entry_response.get("next_cursor") if start_cursor: entry_query["start_cursor"] = start_cursor else: entry_query.pop("start_cursor", None) # Remove duplicates (same entry might be linked to multiple incomplete tasks) seen_entries = set() unique_entries = [] for entry in all_entry_results: entry_id = entry["id"] if entry_id not in seen_entries: seen_entries.add(entry_id) unique_entries.append(entry) logging.info(f"Found {len(unique_entries)} unique entry records with incomplete Entry Milestone Update tasks") return unique_entries except Exception as e: logging.error(f"Error querying entry records with incomplete milestone tasks: {e}") return [] def get_unique_mawbs_with_unchecked_milestones(self) -> str: """ Get unique MAWB numbers from entry records that have unchecked milestone tasks. Returns: Comma-separated string of unique MAWB numbers """ try: # Get entry records with incomplete milestone tasks entries_with_incomplete_tasks = self.get_entry_records_with_incomplete_milestone_tasks() if not entries_with_incomplete_tasks: logging.info("No entries found with unchecked milestone tasks") return "" # Extract unique MAWB numbers unique_mawbs = set() for entry in entries_with_incomplete_tasks: properties = entry.get("properties", {}) # Extract MAWB from title property if "mawb" in properties and properties["mawb"].get("title"): mawb = properties["mawb"]["title"][0].get("text", {}).get("content", "") if mawb.strip(): unique_mawbs.add(mawb.strip()) # Convert to sorted list for consistent output mawb_list = sorted(list(unique_mawbs)) logging.info(f"Found {len(mawb_list)} unique MAWBs with unchecked milestone tasks") # Return as comma-separated string return ", ".join(mawb_list) except Exception as e: logging.error(f"Error getting unique MAWBs with unchecked milestones: {e}") return "" def create_entry_mawb_task_relation(self, entry_number: str, mawb_task_ids: List[str]) -> bool: """ Create relation between entry_record and mawb_task records using the new relation properties. Args: entry_number: Entry number to find the record mawb_task_ids: List of mawb_task page IDs to relate Returns: True if successful, False otherwise """ import time try: database_id = self.database_ids.get("entry_record") if not database_id: logging.error("Entry record database ID not configured") return False # Find the entry record with retry mechanism for newly created records query_data = { "filter": { "property": "entry_number", "rich_text": { "equals": entry_number } } } page_id = None max_retries = 3 for attempt in range(max_retries): response = self._make_database_query("entry_record", query_data) results = response.get("results", []) if results: page_id = results[0]["id"] break elif attempt < max_retries - 1: # Wait briefly before retrying for newly created records logging.info(f"Entry record {entry_number} not found on attempt {attempt + 1}, retrying...") time.sleep(1) else: logging.error(f"Entry record not found after {max_retries} attempts: {entry_number}") return False # Create relation data for mawb_task_reference mawb_task_relations = [] for task_id in mawb_task_ids: mawb_task_relations.append({"id": task_id}) # Update the entry record with mawb_task relations update_data = { "properties": { "mawb_task_reference": { "relation": mawb_task_relations } } } self._make_request("PATCH", f"/pages/{page_id}", update_data) logging.info(f"Created mawb_task relations for entry {entry_number}: {len(mawb_task_ids)} tasks") return True except Exception as e: logging.error(f"Error creating entry-mawb_task relation: {e}") return False def create_entry_clearance_task_relation(self, entry_number: str, clearance_task_ids: List[str]) -> bool: """ Create relation between entry_record and mawb_task records for clearance tasks using clearance_notice_task_reference. Args: entry_number: Entry number to find the record clearance_task_ids: List of clearance task page IDs to relate Returns: True if successful, False otherwise """ import time try: database_id = self.database_ids.get("entry_record") if not database_id: logging.error("Entry record database ID not configured") return False # Find the entry record with retry mechanism for newly created records query_data = { "filter": { "property": "entry_number", "rich_text": { "equals": entry_number } } } page_id = None max_retries = 3 for attempt in range(max_retries): response = self._make_database_query("entry_record", query_data) results = response.get("results", []) if results: page_id = results[0]["id"] break elif attempt < max_retries - 1: # Wait briefly before retrying for newly created records logging.info(f"Entry record {entry_number} not found on attempt {attempt + 1}, retrying...") time.sleep(1) else: logging.error(f"Entry record not found after {max_retries} attempts: {entry_number}") return False # Create relation data for clearance_notice_task_reference clearance_task_relations = [] for task_id in clearance_task_ids: clearance_task_relations.append({"id": task_id}) # Update the entry record with clearance task relations update_data = { "properties": { "clearance_notice_task_reference": { "relation": clearance_task_relations } } } self._make_request("PATCH", f"/pages/{page_id}", update_data) logging.info(f"Created clearance_task relations for entry {entry_number}: {len(clearance_task_ids)} tasks") return True except Exception as e: logging.error(f"Error creating entry-clearance_task relation: {e}") return False def get_mawb_tasks_by_action_and_mawb(self, mawb: str, action: str = "Entry Milestone Update") -> List[Dict]: """ Get mawb_task records by MAWB and action. Args: mawb: MAWB number action: Task action to filter by Returns: List of mawb_task records """ try: database_id = self.database_ids.get("mawb_task") if not database_id: logging.error("MAWB task database ID not configured") return [] # Find MAWB record first to get its ID mawb_record = self.find_mawb_record_by_number(mawb) if not mawb_record: logging.warning(f"MAWB record not found for: {mawb}") return [] mawb_id = mawb_record["id"] query_data = { "filter": { "and": [ { "property": "mawb_reference", "relation": { "contains": mawb_id } }, { "property": "action", "rich_text": { "equals": action } } ] } } response = self._make_database_query("mawb_task", query_data) results = response.get("results", []) logging.info(f"Found {len(results)} mawb_task records for MAWB {mawb} with action {action}") return results except Exception as e: logging.error(f"Error querying mawb_task records: {e}") return [] def check_mawb_task_by_action(self, mawb: str, action: str) -> bool: """ Find and check (mark as done) unchecked mawb_task for specific MAWB and action. Args: mawb: MAWB number action: Task action (e.g., "Entry Milestone Update", "MAWB Milestone Update") Returns: True if task was found and successfully checked, False otherwise """ try: database_id = self.database_ids.get("mawb_task") if not database_id: logging.error("MAWB task database ID not configured") return False # Query for unchecked task with specific MAWB and action query_data = { "filter": { "and": [ { "property": "mawb", "title": {"equals": mawb} }, { "property": "action", "rich_text": {"equals": action} }, { "property": "done", "checkbox": {"equals": False} } ] } } response = self._make_database_query("mawb_task", query_data) if not response: logging.warning(f"No response when querying tasks for MAWB {mawb} action {action}") return False results = response.get("results", []) if not results: logging.debug(f"No unchecked {action} task found for MAWB {mawb}") return False if len(results) > 1: logging.warning(f"Multiple unchecked {action} tasks found for MAWB {mawb}, updating first one") # Update the first matching task task_id = results[0]["id"] update_data = { "properties": { "done": { "checkbox": True } } } update_response = self._make_request("PATCH", f"/pages/{task_id}", update_data) if update_response and update_response.get("id"): logging.info(f"Successfully checked {action} task for MAWB {mawb}") return True else: logging.error(f"Failed to update {action} task for MAWB {mawb}") return False except Exception as e: logging.error(f"Error checking {action} task for MAWB {mawb}: {e}") return False def check_mawb_expected_entry_count_match(self, mawb: str) -> bool: """ Check if MAWB record's expected_entry_count formula field equals "Match". Args: mawb: MAWB number Returns: True if expected_entry_count equals "Match", False otherwise """ try: mawb_record = self.find_mawb_record_by_number(mawb) if not mawb_record: logging.warning(f"MAWB record not found for: {mawb}") return False properties = mawb_record.get("properties", {}) expected_count_prop = properties.get("expected_entry_count", {}) # Formula fields in Notion return different types based on the result if "formula" in expected_count_prop: formula_result = expected_count_prop["formula"] # Check if it's a string result if "string" in formula_result and formula_result["string"]: result_value = formula_result["string"].strip() is_match = result_value.lower() == "match" logging.info(f"MAWB {mawb} expected_entry_count: '{result_value}' -> Match: {is_match}") return is_match # Check if it's a number result elif "number" in formula_result: logging.info(f"MAWB {mawb} expected_entry_count returned number: {formula_result['number']}") return False # Check if it's boolean result elif "boolean" in formula_result: logging.info(f"MAWB {mawb} expected_entry_count returned boolean: {formula_result['boolean']}") return False else: logging.warning(f"MAWB {mawb} expected_entry_count formula returned unknown type: {formula_result}") return False else: logging.warning(f"MAWB {mawb} expected_entry_count field is not a formula or is empty") return False except Exception as e: logging.error(f"Error checking MAWB expected entry count match for {mawb}: {e}") return False def check_mawb_entry_submit_task_done(self, mawb: str) -> bool: """ Check if MAWB task with action "Entry Submit" has done field checked. Args: mawb: MAWB number Returns: True if Entry Submit task is marked as done, False otherwise """ try: entry_submit_tasks = self.get_mawb_tasks_by_action_and_mawb(mawb, "Entry Submit") if not entry_submit_tasks: logging.info(f"No Entry Submit task found for MAWB {mawb}") return False # Check if any Entry Submit task is marked as done for task in entry_submit_tasks: properties = task.get("properties", {}) done_prop = properties.get("done", {}) if "checkbox" in done_prop: is_done = done_prop["checkbox"] if is_done: logging.info(f"MAWB {mawb} Entry Submit task is marked as done") return True else: logging.info(f"MAWB {mawb} Entry Submit task is not done") else: logging.warning(f"MAWB {mawb} Entry Submit task missing done checkbox field") return False except Exception as e: logging.error(f"Error checking MAWB Entry Submit task status for {mawb}: {e}") return False def update_entry_submit_task_checkbox(self, mawb: str) -> bool: """ Update MAWB task with action "Entry Submit" to mark done field as checked. Only updates if the checkbox is currently unchecked. Args: mawb: MAWB number Returns: True if task was updated or already done, False if error or not found """ try: entry_submit_tasks = self.get_mawb_tasks_by_action_and_mawb(mawb, "Entry Submit") if not entry_submit_tasks: logging.warning(f"No Entry Submit task found for MAWB {mawb}") return False # Update all Entry Submit tasks that are not done updated_count = 0 for task in entry_submit_tasks: task_id = task.get("id") properties = task.get("properties", {}) done_prop = properties.get("done", {}) if "checkbox" in done_prop: is_done = done_prop["checkbox"] if not is_done: # Update the task to mark as done update_data = { "properties": { "done": self._format_checkbox_property(True) } } self._make_request("PATCH", f"/pages/{task_id}", update_data) logging.info(f"Updated MAWB {mawb} Entry Submit task to done=true") updated_count += 1 else: logging.debug(f"MAWB {mawb} Entry Submit task already marked as done") else: logging.warning(f"MAWB {mawb} Entry Submit task missing done checkbox field") return True except Exception as e: logging.error(f"Error updating MAWB Entry Submit task checkbox for {mawb}: {e}") return False def validate_mawb_ready_for_entry_submission(self, mawb: str) -> Dict[str, Any]: """ Validate if MAWB is ready for entry submission by checking: 1. expected_entry_count formula field equals "Match" 2. Entry Submit task is marked as done Args: mawb: MAWB number Returns: Dictionary with validation results: { 'ready': bool, 'expected_count_match': bool, 'entry_submit_done': bool, 'mawb': str, 'validation_message': str } """ try: result = { 'ready': False, 'expected_count_match': False, 'entry_submit_done': False, 'mawb': mawb, 'validation_message': '' } # Check if MAWB record exists mawb_record = self.find_mawb_record_by_number(mawb) if not mawb_record: result['validation_message'] = f"MAWB record not found: {mawb}" logging.warning(result['validation_message']) return result # Check expected_entry_count match result['expected_count_match'] = self.check_mawb_expected_entry_count_match(mawb) # Check Entry Submit task completion result['entry_submit_done'] = self.check_mawb_entry_submit_task_done(mawb) # Determine if MAWB is ready result['ready'] = result['expected_count_match'] and result['entry_submit_done'] # Create validation message if result['ready']: result['validation_message'] = f"MAWB {mawb} is ready for entry submission: expected count matches and Entry Submit task is complete" logging.info(result['validation_message']) else: issues = [] if not result['expected_count_match']: issues.append("expected entry count does not match") if not result['entry_submit_done']: issues.append("Entry Submit task is not completed") result['validation_message'] = f"MAWB {mawb} is NOT ready for entry submission: {', '.join(issues)}" logging.warning(result['validation_message']) return result except Exception as e: error_msg = f"Error validating MAWB readiness for {mawb}: {e}" logging.error(error_msg) return { 'ready': False, 'expected_count_match': False, 'entry_submit_done': False, 'mawb': mawb, 'validation_message': error_msg } def get_mawbs_ready_for_entry_submission(self, mawb_list: List[str] = None, limit: int = None) -> List[Dict[str, Any]]: """ Get list of MAWBs that are ready for entry submission. Args: mawb_list: Optional list of specific MAWBs to check. If None, checks all MAWBs. limit: Optional limit on number of MAWBs to check (for performance) Returns: List of validation results for MAWBs that are ready for submission """ try: if mawb_list is None: # Get all MAWB records all_mawb_records = self.get_all_mawb_records() mawb_list = [] for record in all_mawb_records: properties = record.get("properties", {}) mawb_prop = properties.get("mawb", {}) if "title" in mawb_prop and mawb_prop["title"]: mawb_number = mawb_prop["title"][0].get("text", {}).get("content", "").strip() if mawb_number: mawb_list.append(mawb_number) # Apply limit if specified if limit and len(mawb_list) > limit: mawb_list = mawb_list[:limit] logging.info(f"Limited check to first {limit} MAWBs for performance") logging.info(f"Checking readiness for {len(mawb_list)} MAWBs from database") else: logging.info(f"Checking readiness for {len(mawb_list)} specified MAWBs") ready_mawbs = [] total_checked = 0 for mawb in mawb_list: total_checked += 1 validation_result = self.validate_mawb_ready_for_entry_submission(mawb) if validation_result['ready']: ready_mawbs.append(validation_result) logging.info(f"Found {len(ready_mawbs)} MAWBs ready for entry submission out of {total_checked} checked") return ready_mawbs except Exception as e: logging.error(f"Error getting MAWBs ready for entry submission: {e}") return [] def get_mawbs_ready_for_entry_submission_batch(self, batch_size: int = 50, max_total: int = 200) -> List[Dict[str, Any]]: """ Optimized batch validation for large MAWB datasets with performance controls. Args: batch_size: Number of MAWBs to process in each batch max_total: Maximum total MAWBs to check (prevents runaway processing) Returns: List of validation results for MAWBs that are ready for submission """ try: logging.info(f"Starting batch MAWB validation (batch_size={batch_size}, max_total={max_total})") # Get MAWB records with formula field pre-filtered if possible all_mawb_records = self.get_all_mawb_records() if not all_mawb_records: logging.info("No MAWB records found") return [] # Extract MAWB data for batch processing mawb_data = [] for record in all_mawb_records: try: properties = record.get("properties", {}) mawb_prop = properties.get("mawb", {}) if "title" in mawb_prop and mawb_prop["title"]: mawb_number = mawb_prop["title"][0].get("text", {}).get("content", "").strip() if mawb_number: # Pre-check expected_entry_count to avoid unnecessary task queries expected_count_match = self._check_expected_count_from_record(record) mawb_data.append({ 'mawb': mawb_number, 'record': record, 'expected_count_match': expected_count_match }) # Apply max_total limit if len(mawb_data) >= max_total: logging.info(f"Reached max_total limit of {max_total} MAWBs") break except Exception as e: logging.warning(f"Error processing MAWB record: {e}") continue logging.info(f"Processing {len(mawb_data)} MAWBs in batches of {batch_size}") # Process in batches ready_mawbs = [] total_processed = 0 for i in range(0, len(mawb_data), batch_size): batch = mawb_data[i:i + batch_size] batch_number = (i // batch_size) + 1 total_batches = (len(mawb_data) + batch_size - 1) // batch_size logging.info(f"Processing batch {batch_number}/{total_batches} ({len(batch)} MAWBs)") # Process batch batch_ready = self._process_mawb_validation_batch(batch) ready_mawbs.extend(batch_ready) total_processed += len(batch) logging.info(f"Batch {batch_number} completed: {len(batch_ready)} ready out of {len(batch)}") logging.info(f"Batch validation completed: {len(ready_mawbs)} ready MAWBs out of {total_processed} processed") return ready_mawbs except Exception as e: logging.error(f"Error in batch MAWB validation: {e}") return [] def _check_expected_count_from_record(self, mawb_record: Dict) -> bool: """ Check expected_entry_count from pre-loaded MAWB record (optimization). Args: mawb_record: Pre-loaded MAWB record from Notion Returns: True if expected_entry_count equals "Match", False otherwise """ try: properties = mawb_record.get("properties", {}) expected_count_prop = properties.get("expected_entry_count", {}) if "formula" in expected_count_prop: formula_result = expected_count_prop["formula"] if "string" in formula_result and formula_result["string"]: result_value = formula_result["string"].strip() return result_value.lower() == "match" return False except Exception as e: logging.warning(f"Error checking expected count from record: {e}") return False def _process_mawb_validation_batch(self, batch: List[Dict]) -> List[Dict[str, Any]]: """ Process a batch of MAWBs for validation. Args: batch: List of MAWB data dictionaries Returns: List of ready MAWBs from this batch """ ready_mawbs = [] try: # First, filter by expected_count_match (already pre-checked) candidates = [item for item in batch if item['expected_count_match']] if not candidates: logging.debug(f"No candidates with matching expected count in batch of {len(batch)}") return ready_mawbs logging.debug(f"Checking Entry Submit tasks for {len(candidates)} candidates out of {len(batch)} in batch") # Check Entry Submit tasks for candidates only for item in candidates: try: mawb = item['mawb'] # Check Entry Submit task completion entry_submit_done = self.check_mawb_entry_submit_task_done(mawb) if entry_submit_done: # MAWB is ready for submission ready_result = { 'ready': True, 'expected_count_match': True, 'entry_submit_done': True, 'mawb': mawb, 'validation_message': f"MAWB {mawb} is ready for entry submission: expected count matches and Entry Submit task is complete" } ready_mawbs.append(ready_result) logging.debug(f"MAWB {mawb} is ready for submission") else: logging.debug(f"MAWB {mawb} has matching count but Entry Submit task not done") except Exception as e: logging.warning(f"Error validating MAWB {item.get('mawb', 'unknown')} in batch: {e}") continue return ready_mawbs except Exception as e: logging.error(f"Error processing validation batch: {e}") return ready_mawbs def get_entries_for_mawbs(self, mawb_list: List[str]) -> Dict[str, List[Dict]]: """ Get entries organized by MAWB for interactive editing. Args: mawb_list: List of MAWB numbers Returns: Dictionary mapping MAWB number to list of entry records with flattened data """ try: database_id = self.database_ids.get("entry_record") if not database_id: logging.error("Entry record database ID not configured") return {} entries_by_mawb = {} for mawb in mawb_list: query_data = { "filter": { "property": "mawb", "title": { "equals": mawb.strip() } }, "page_size": 100 } all_entries = [] has_more = True start_cursor = None while has_more: if start_cursor: query_data["start_cursor"] = start_cursor response = self._make_database_query("entry_record", query_data) results = response.get("results", []) # Flatten the entries for easier editing for entry in results: flattened = self._flatten_entry_for_editing(entry) if flattened: all_entries.append(flattened) has_more = response.get("has_more", False) start_cursor = response.get("next_cursor") entries_by_mawb[mawb.strip()] = all_entries logging.info(f"Retrieved {len(all_entries)} entries for MAWB {mawb}") return entries_by_mawb except Exception as e: logging.error(f"Error getting entries for MAWBs: {e}") return {} def _flatten_entry_for_editing(self, entry_record: Dict) -> Optional[Dict]: """ Flatten entry record for easier editing in DataFrame. Args: entry_record: Raw Notion entry record Returns: Flattened entry data or None if error """ try: properties = entry_record.get("properties", {}) flattened = { "id": entry_record.get("id", "") # Keep ID for updates } # Extract MAWB (title field) if "mawb" in properties and properties["mawb"].get("title"): flattened["mawb"] = properties["mawb"]["title"][0].get("text", {}).get("content", "") else: flattened["mawb"] = "" # Extract Entry Number (rich_text field) if "entry_number" in properties and properties["entry_number"].get("rich_text"): flattened["entry_number"] = properties["entry_number"]["rich_text"][0].get("text", {}).get("content", "") else: flattened["entry_number"] = "" # Extract date fields date_fields = [ "entry_submit_date", "cbp_hold_date", "cbp_release_date", "cpsc_review_date", "cpsc_release_date" ] for field in date_fields: if field in properties and properties[field].get("date"): date_str = properties[field]["date"].get("start") if date_str: try: # Convert to user-friendly format for Excel editing dt = datetime.fromisoformat(date_str.replace('Z', '+00:00')) # Format as MM/DD/YYYY HH:MM AM/PM (Excel-friendly) flattened[field] = dt.strftime('%m/%d/%Y %I:%M %p') except: flattened[field] = date_str else: flattened[field] = "" else: flattened[field] = "" # Extract text fields text_fields = ["cbp_hold_line", "cpsc_review_line"] for field in text_fields: if field in properties and properties[field].get("rich_text"): flattened[field] = properties[field]["rich_text"][0].get("text", {}).get("content", "") else: flattened[field] = "" return flattened except Exception as e: logging.error(f"Error flattening entry record: {e}") return None def update_entries_from_dataframe(self, df) -> Dict[str, int]: """ Update or create entry records from DataFrame. Args: df: DataFrame with entry data (must include 'mawb' and 'entry_number' columns) Returns: Dictionary with statistics: {'updated': int, 'errors': int, 'total': int} """ stats = {'updated': 0, 'errors': 0, 'total': 0} try: database_id = self.database_ids.get("entry_record") if not database_id: logging.error("Entry record database ID not configured") return stats # Process each row for index, row in df.iterrows(): stats['total'] += 1 try: # Validate required fields mawb = str(row.get('mawb', '')).strip() entry_number = str(row.get('entry_number', '')).strip() if not mawb or not entry_number: logging.warning(f"Row {index + 1}: Missing MAWB or Entry Number") stats['errors'] += 1 continue # Check if entry exists (use ID if available, otherwise search) entry_id = None if 'id' in row and str(row['id']).strip(): entry_id = str(row['id']).strip() else: # Search for existing entry existing_entry = self._find_entry_by_mawb_and_number(mawb, entry_number) if existing_entry: entry_id = existing_entry.get("id") # Prepare properties for update/create properties = {} # Handle date fields date_fields = [ "entry_submit_date", "cbp_hold_date", "cbp_release_date", "cpsc_review_date", "cpsc_release_date" ] for field in date_fields: if field in row and row[field] is not None: date_val = row[field] # Handle pandas NaT and invalid date strings import pandas as pd if pd.isna(date_val): continue date_str = str(date_val).strip() if not date_str or date_str.lower() in ['nan', 'nat', 'none']: continue if hasattr(date_val, 'isoformat'): # It's already a datetime object properties[field] = self._format_datetime_property(date_val) else: # Try to parse the date string (support multiple formats) try: if date_str: # Try different date formats dt = None date_formats = [ '%m/%d/%Y %I:%M %p', # MM/DD/YYYY HH:MM AM/PM (our export format) '%m/%d/%Y %H:%M', # MM/DD/YYYY HH:MM (24-hour) '%m/%d/%Y', # MM/DD/YYYY (date only) '%Y-%m-%d %H:%M:%S', # YYYY-MM-DD HH:MM:SS '%Y-%m-%dT%H:%M:%S', # ISO format without timezone '%Y-%m-%d', # YYYY-MM-DD (date only) ] for fmt in date_formats: try: dt = datetime.strptime(date_str, fmt) break except ValueError: continue if dt is None: # Fallback to ISO parsing dt = datetime.fromisoformat(date_str.replace('Z', '')) properties[field] = self._format_datetime_property(dt) except Exception as e: logging.warning(f"Could not parse date '{date_val}' for field {field}: {e}") # Handle text fields text_fields = ["cbp_hold_line", "cpsc_review_line"] for field in text_fields: if field in row and row[field] is not None and str(row[field]).strip(): properties[field] = self._format_text_property(str(row[field]).strip()) # Update or create entry if entry_id: # Update existing entry if properties: # Only update if there are changes update_data = {"properties": properties} self._make_request("PATCH", f"/pages/{entry_id}", update_data) logging.info(f"Updated entry {mawb} - {entry_number}") else: logging.info(f"No changes for entry {mawb} - {entry_number}") else: # Create new entry properties["mawb"] = self._format_title_property(mawb) properties["entry_number"] = self._format_text_property(entry_number) # Get the actual data source ID for page creation data_source_id = self._get_data_source_id("entry_record") create_data = { "parent": { "type": "data_source_id", "data_source_id": data_source_id }, "properties": properties } response = self._make_request("POST", "/pages", create_data) logging.info(f"Created new entry {mawb} - {entry_number}") stats['updated'] += 1 except Exception as e: logging.error(f"Error processing row {index + 1} ({mawb} - {entry_number}): {e}") stats['errors'] += 1 continue logging.info(f"Batch update completed: {stats['updated']} successful, {stats['errors']} errors out of {stats['total']} total") return stats except Exception as e: logging.error(f"Error in batch update from DataFrame: {e}") stats['errors'] = stats['total'] # Mark all as errors if batch fails return stats def _find_entry_by_mawb_and_number(self, mawb: str, entry_number: str) -> Optional[Dict]: """ Find entry record by MAWB and entry number. Args: mawb: MAWB number entry_number: Entry number Returns: Entry record or None if not found """ try: database_id = self.database_ids.get("entry_record") if not database_id: return None # Search with compound filter query_data = { "filter": { "and": [ { "property": "mawb", "title": { "equals": mawb.strip() } }, { "property": "entry_number", "rich_text": { "equals": entry_number.strip() } } ] }, "page_size": 1 } response = self._make_database_query("entry_record", query_data) results = response.get("results", []) return results[0] if results else None except Exception as e: logging.error(f"Error finding entry by MAWB and number: {e}") return None def get_mawb_records_with_entry_count_data(self) -> List[Dict]: """ Get MAWB records with expected_entry_count data pre-loaded. Returns records with embedded formula results to avoid re-fetching. """ try: database_id = self.database_ids.get("mawb") if not database_id: logging.error("MAWB database ID not configured") return [] # Fetch all records with pagination but include formula fields query_data = { "page_size": 100, # Request specific properties to reduce payload size "filter": { "property": "mawb", "title": { "is_not_empty": True } } } # Use pagination efficiently all_results = [] has_more = True start_cursor = None while has_more: if start_cursor: query_data["start_cursor"] = start_cursor response = self._make_database_query("mawb", query_data) results = response.get("results", []) all_results.extend(results) has_more = response.get("has_more", False) start_cursor = response.get("next_cursor") logging.info(f"Retrieved {len(all_results)} MAWB records with entry count data") return all_results except Exception as e: logging.error(f"Error getting MAWB records with entry count data: {e}") return [] def get_entry_submit_tasks_batch(self, mawb_ids: List[str]) -> List[Dict]: """ Efficiently query Entry Submit tasks for multiple MAWBs in batches. """ try: database_id = self.database_ids.get("mawb_task") if not database_id: logging.error("MAWB task database ID not configured") return [] # Notion API has limitations on OR filters, so we may need to batch BATCH_SIZE = 10 # Notion API limit for OR conditions all_tasks = [] for i in range(0, len(mawb_ids), BATCH_SIZE): batch_ids = mawb_ids[i:i + BATCH_SIZE] # Create OR filter for this batch or_conditions = [] for mawb_id in batch_ids: or_conditions.append({ "property": "mawb_reference", "relation": {"contains": mawb_id} }) query_data = { "filter": { "and": [ { "property": "action", "rich_text": {"equals": "Entry Submit"} }, { "or": or_conditions } ] } } response = self._make_database_query("mawb_task", query_data) batch_tasks = response.get("results", []) all_tasks.extend(batch_tasks) logging.info(f"Retrieved {len(all_tasks)} Entry Submit tasks for {len(mawb_ids)} MAWBs") return all_tasks except Exception as e: logging.error(f"Error getting Entry Submit tasks batch: {e}") return [] def update_entry_submit_task_checkbox_by_id(self, task_id: str) -> bool: """ Update Entry Submit task checkbox directly by task ID (no lookup required). """ try: update_data = { "properties": { "done": {"checkbox": True} } } response = self._make_request("PATCH", f"/pages/{task_id}", update_data) return response is not None except Exception as e: logging.error(f"Error updating Entry Submit task {task_id}: {e}") return False def _check_expected_count_from_record(self, record: Dict) -> bool: """ Check if a MAWB record's expected_entry_count formula field equals "Match". Uses the record data directly instead of making additional API calls. """ try: properties = record.get("properties", {}) expected_count_prop = properties.get("expected_entry_count", {}) # Formula fields in Notion return different types based on the result if "formula" in expected_count_prop: formula_result = expected_count_prop["formula"] # Check if it's a string result if "string" in formula_result and formula_result["string"]: result_value = formula_result["string"].strip() return result_value.lower() == "match" # Check if it's a number result elif "number" in formula_result: # If it's a number, it's not "Match" return False # Check if it's null/empty else: return False return False except Exception as e: logging.error(f"Error checking expected count from record: {e}") return False def get_all_entry_records_paginated(self) -> List[Dict]: """ Get all entry records using pagination for large datasets. Returns: List of all entry records """ try: database_id = self.database_ids.get("entry_record") if not database_id: logging.error("Entry record database ID not configured") return [] all_records = [] query_data = {"page_size": 100} has_more = True next_cursor = None while has_more: if next_cursor: query_data["start_cursor"] = next_cursor else: query_data.pop("start_cursor", None) response = self._make_database_query("entry_record", query_data) if not response: break results = response.get("results", []) all_records.extend(results) has_more = response.get("has_more", False) next_cursor = response.get("next_cursor") logging.info(f"Retrieved {len(all_records)} entry records total") return all_records except Exception as e: logging.error(f"Error getting all entry records: {e}") return [] def check_entries_exist_batch(self, entry_numbers: List[str]) -> Dict[str, bool]: """ Check if multiple entry numbers exist in batch for improved performance. Args: entry_numbers: List of entry numbers to check Returns: Dict mapping entry numbers to existence (True/False) Raises: Exception: If API calls fail, raises exception to prevent duplicate creation """ database_id = self.database_ids.get("entry_record") if not database_id: error_msg = "Entry record database ID not configured" logging.error(error_msg) raise ValueError(error_msg) if not entry_numbers: return {} # Split large lists into smaller batches to avoid API limits BATCH_SIZE = 50 existence_map = {} for i in range(0, len(entry_numbers), BATCH_SIZE): batch = entry_numbers[i:i + BATCH_SIZE] # Build OR query for this batch if len(batch) == 1: query_filter = { "property": "entry_number", "rich_text": {"equals": batch[0].strip()} } else: query_filter = { "or": [ { "property": "entry_number", "rich_text": {"equals": entry.strip()} } for entry in batch ] } query_data = { "filter": query_filter, "page_size": 100 } try: response = self._make_database_query("entry_record", query_data) results = response.get("results", []) # Mark found entries as existing found_entries = set() for result in results: entry_rich_text = result.get("properties", {}).get("entry_number", {}).get("rich_text", []) if entry_rich_text: entry_number = entry_rich_text[0].get("text", {}).get("content", "").strip() found_entries.add(entry_number) # Update existence map for this batch for entry in batch: existence_map[entry.strip()] = entry.strip() in found_entries except Exception as e: # CRITICAL: Do not default to False on API failures # This would cause duplicate entries to be created error_msg = f"Failed to check entry existence for batch {i//BATCH_SIZE + 1}: {e}" logging.error(error_msg) logging.error(f"Batch entries: {batch}") raise RuntimeError(error_msg) from e logging.info(f"Batch existence check completed: {sum(existence_map.values())}/{len(entry_numbers)} entries exist") return existence_map def get_entries_by_numbers(self, entry_numbers: List[str]) -> List[Dict]: """ Get entry records by entry numbers using batched queries. Args: entry_numbers: List of entry numbers to retrieve Returns: List of entry records """ try: database_id = self.database_ids.get("entry_record") if not database_id: logging.error("Entry record database ID not configured") return [] if not entry_numbers: return [] # Split large lists into smaller batches BATCH_SIZE = 50 all_records = [] for i in range(0, len(entry_numbers), BATCH_SIZE): batch = entry_numbers[i:i + BATCH_SIZE] # Build OR query for this batch if len(batch) == 1: query_filter = { "property": "entry_number", "rich_text": {"equals": batch[0].strip()} } else: query_filter = { "or": [ { "property": "entry_number", "rich_text": {"equals": entry.strip()} } for entry in batch ] } query_data = { "filter": query_filter, "page_size": 100 } response = self._make_database_query("entry_record", query_data) if response: results = response.get("results", []) all_records.extend(results) logging.info(f"Retrieved {len(all_records)} entry records by numbers") return all_records except Exception as e: logging.error(f"Error getting entries by numbers: {e}") return [] def update_entry_records_batch(self, updates: List[Dict]) -> Dict[str, int]: """ Update multiple entry records in batch for improved performance. Args: updates: List of update dictionaries containing: - entry_number: Entry number to update - updates: Dict of field updates Returns: Dict with statistics: {'updated': count, 'failed': count, 'total': count} """ import concurrent.futures import time stats = {'updated': 0, 'failed': 0, 'total': len(updates)} if not updates: logging.info("No records to update in batch") return stats logging.info(f"Starting batch update of {len(updates)} entry records...") def update_single_record(update_data): """Update a single record - for concurrent execution.""" try: entry_number = update_data.get('entry_number') field_updates = update_data.get('updates', {}) if not entry_number: logging.error(f"Missing entry number in update data: {update_data}") return False success = self.update_entry_record(entry_number, **field_updates) if success: logging.info(f"Batch updated {entry_number}: {list(field_updates.keys())}") else: logging.error(f"Failed to update {entry_number}") return success except Exception as e: logging.error(f"Error in batch update for {update_data.get('entry_number', 'unknown')}: {e}") return False # Process updates concurrently (5 at a time to avoid overwhelming the API) max_concurrent = 5 batch_size = max_concurrent for i in range(0, len(updates), batch_size): batch = updates[i:i + batch_size] batch_num = i // batch_size + 1 total_batches = (len(updates) + batch_size - 1) // batch_size logging.info(f"Processing batch {batch_num}/{total_batches} ({len(batch)} records)...") # Use ThreadPoolExecutor for concurrent API calls with concurrent.futures.ThreadPoolExecutor(max_workers=max_concurrent) as executor: # Submit all tasks in this batch futures = [executor.submit(update_single_record, update_data) for update_data in batch] # Wait for all tasks to complete and collect results for future in concurrent.futures.as_completed(futures): try: success = future.result(timeout=30) # 30 second timeout per update if success: stats['updated'] += 1 else: stats['failed'] += 1 except concurrent.futures.TimeoutError: logging.error("Update timed out after 30 seconds") stats['failed'] += 1 except Exception as e: logging.error(f"Unexpected error in batch update: {e}") stats['failed'] += 1 # Small delay between batches to be respectful to the API if i + batch_size < len(updates): time.sleep(0.5) logging.info(f"Batch update completed: {stats['updated']} updated, {stats['failed']} failed out of {stats['total']} total") return stats def get_milestone_records_bulk(self, mawb_list: List[str]) -> List[Dict]: """ Get milestone records for a list of MAWBs using batched queries to handle large lists. Args: mawb_list: List of MAWB numbers to query (empty list gets all records) Returns: List of milestone records for all requested MAWBs """ try: database_id = self.database_ids.get("mawb_milestone") if not database_id: logging.error("MAWB milestone database ID not configured") return [] # If empty list provided, get all records if not mawb_list: query_data = {"page_size": 100} all_records = [] has_more = True next_cursor = None while has_more: if next_cursor: query_data["start_cursor"] = next_cursor else: query_data.pop("start_cursor", None) response = self._make_database_query("mawb_milestone", query_data) if not response: logging.warning("No response for milestone query") break results = response.get("results", []) all_records.extend(results) has_more = response.get("has_more", False) next_cursor = response.get("next_cursor") logging.info(f"Retrieved {len(all_records)} milestone records (all records)") return all_records # Split large lists into smaller batches to avoid API limits BATCH_SIZE = 50 # Reasonable batch size for OR queries all_records = [] for i in range(0, len(mawb_list), BATCH_SIZE): batch = mawb_list[i:i + BATCH_SIZE] batch_num = i//BATCH_SIZE + 1 total_batches = (len(mawb_list) + BATCH_SIZE - 1)//BATCH_SIZE logging.info(f"Processing milestone batch {batch_num}/{total_batches} ({len(batch)} MAWBs)...") # Build query for this batch if len(batch) == 1: query_filter = { "property": "mawb", "title": {"equals": batch[0].strip()} } else: query_filter = { "or": [ { "property": "mawb", "title": {"equals": mawb.strip()} } for mawb in batch ] } query_data = { "filter": query_filter, "page_size": 100 } # Handle pagination for this batch has_more = True next_cursor = None while has_more: if next_cursor: query_data["start_cursor"] = next_cursor else: query_data.pop("start_cursor", None) response = self._make_database_query("mawb_milestone", query_data) if not response: logging.warning(f"No response for milestone batch {i//BATCH_SIZE + 1}") break results = response.get("results", []) all_records.extend(results) has_more = response.get("has_more", False) next_cursor = response.get("next_cursor") logging.info(f"Retrieved {len(all_records)} milestone records for {len(mawb_list)} MAWBs using {(len(mawb_list) + BATCH_SIZE - 1)//BATCH_SIZE} batches") return all_records except Exception as e: logging.error(f"Error getting milestone records bulk: {e}") return [] def update_mawb_milestone_records_batch(self, updates: List[Dict]) -> Dict[str, int]: """ Update multiple MAWB milestone records in batch for improved performance. Args: updates: List of update dictionaries containing: - mawb: MAWB number - last_mile_carrier: Carrier name - updates: Dict of field updates Returns: Dict with statistics: {'updated': count, 'failed': count, 'total': count} """ import concurrent.futures import time stats = {'updated': 0, 'failed': 0, 'total': len(updates)} if not updates: logging.info("No records to update in batch") return stats logging.info(f"Starting batch update of {len(updates)} milestone records...") def update_single_record(update_data): """Update a single record - for concurrent execution.""" try: mawb = update_data.get('mawb') carrier = update_data.get('last_mile_carrier') field_updates = update_data.get('updates', {}) if not mawb or not carrier: logging.error(f"Missing MAWB or carrier in update data: {update_data}") return False success = self.update_mawb_milestone_record(mawb, carrier, **field_updates) if success: logging.info(f"Batch updated {mawb}-{carrier}: {list(field_updates.keys())}") else: logging.error(f"Failed to update {mawb}-{carrier}") return success except Exception as e: logging.error(f"Error in batch update for {update_data.get('mawb', 'unknown')}: {e}") return False # Process updates concurrently (5 at a time to avoid overwhelming the API) max_concurrent = 5 batch_size = max_concurrent for i in range(0, len(updates), batch_size): batch = updates[i:i + batch_size] batch_num = i // batch_size + 1 total_batches = (len(updates) + batch_size - 1) // batch_size logging.info(f"Processing batch {batch_num}/{total_batches} ({len(batch)} records)...") # Use ThreadPoolExecutor for concurrent API calls with concurrent.futures.ThreadPoolExecutor(max_workers=max_concurrent) as executor: # Submit all tasks in this batch futures = [executor.submit(update_single_record, update_data) for update_data in batch] # Wait for all tasks to complete and collect results for future in concurrent.futures.as_completed(futures): try: success = future.result(timeout=30) # 30 second timeout per update if success: stats['updated'] += 1 else: stats['failed'] += 1 except concurrent.futures.TimeoutError: logging.error("Update timed out after 30 seconds") stats['failed'] += 1 except Exception as e: logging.error(f"Unexpected error in batch update: {e}") stats['failed'] += 1 # Small delay between batches to be respectful to the API if i + batch_size < len(updates): time.sleep(0.5) logging.info(f"Batch update completed: {stats['updated']} updated, {stats['failed']} failed out of {stats['total']} total") return stats def check_entry_fields_batch(self, entry_field_pairs: List[Dict[str, str]]) -> Dict[str, Dict[str, bool]]: """ Check if specific fields are populated for multiple entries in batch for improved performance. Args: entry_field_pairs: List of dictionaries with 'entry_number' and 'field_name' keys Example: [ {'entry_number': '12345', 'field_name': 'cpsc_review_date'}, {'entry_number': '12346', 'field_name': 'cpsc_release_date'} ] Returns: Dict mapping entry numbers to field status dictionaries Example: { '12345': {'cpsc_review_date': True}, '12346': {'cpsc_release_date': False} } """ try: database_id = self.database_ids.get("entry_record") if not database_id: logging.error("Entry record database ID not configured") return {} if not entry_field_pairs: return {} # Extract unique entry numbers for batch lookup entry_numbers = list(set([pair['entry_number'] for pair in entry_field_pairs])) # Get all entries in batch entry_records = self.get_entries_by_numbers(entry_numbers) # Create lookup map for quick access entry_lookup = {} for record in entry_records: entry_title = record.get("properties", {}).get("entry_number", {}).get("title", []) if entry_title: entry_number = entry_title[0].get("text", {}).get("content", "").strip() entry_lookup[entry_number] = record # Build results by checking each requested field results = {} for pair in entry_field_pairs: entry_number = pair['entry_number'] field_name = pair['field_name'] # Initialize entry result if not exists if entry_number not in results: results[entry_number] = {} # Check if entry exists if entry_number not in entry_lookup: logging.warning(f"Entry {entry_number} not found") results[entry_number][field_name] = False continue # Check if the field is populated properties = entry_lookup[entry_number].get("properties", {}) field_data = properties.get(field_name, {}) # Handle different field types is_populated = False if "date" in field_data: is_populated = field_data["date"] is not None elif "rich_text" in field_data: rich_text_list = field_data["rich_text"] is_populated = len(rich_text_list) > 0 and rich_text_list[0].get("text", {}).get("content", "").strip() != "" elif "number" in field_data: is_populated = field_data["number"] is not None elif "checkbox" in field_data: is_populated = field_data["checkbox"] is not None else: logging.warning(f"Unknown field type for {field_name}") is_populated = False results[entry_number][field_name] = is_populated logging.info(f"Batch field check completed for {len(entry_field_pairs)} entry-field pairs across {len(entry_numbers)} entries") return results except Exception as e: logging.error(f"Error checking entry fields in batch: {e}") return {} def format_entry_number(self, entry_number: str) -> str: """ Format entry number by removing prefixes and formatting consistently. Args: entry_number: Raw entry number (e.g., "9G3-0000251-7") Returns: Formatted entry number (e.g., "00002517") """ try: # Remove common prefixes and clean formatting cleaned = entry_number.strip() # Remove "9G3-" prefix if present if cleaned.upper().startswith("9G3-"): cleaned = cleaned[4:] # Remove all dashes and spaces cleaned = cleaned.replace("-", "").replace(" ", "") # If it ends with a single digit (check digit), keep it # Format: 00002517 (8 digits total) if cleaned.isdigit(): return cleaned return cleaned except Exception as e: logging.warning(f"Error formatting entry number '{entry_number}': {e}") return entry_number.strip() def format_master_bl(self, master_bl: str) -> str: """ Format Master B/L number consistently. Args: master_bl: Raw Master B/L number Returns: Formatted Master B/L number """ try: # Basic cleaning - remove extra whitespace cleaned = master_bl.strip() # Add dash after first 3 characters if not already present if len(cleaned) > 3 and cleaned[3] != '-': cleaned = cleaned[:3] + '-' + cleaned[3:] return cleaned except Exception as e: logging.warning(f"Error formatting Master B/L '{master_bl}': {e}") return master_bl def get_mawb_milestone_for_dwell_report(self, target_date: datetime = None) -> List[Dict]: """ Get MAWB milestone records for dwell report generation. Main filter: warehouse_out IS NULL (cargo still in warehouse) Args: target_date: Date for reference (defaults to current date) Returns: List of milestone records where warehouse_out is null """ if target_date is None: target_date = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) try: database_id = self.database_ids.get("mawb_milestone") if not database_id: logging.error("MAWB milestone database ID not configured") return [] # Query for records where warehouse_out is null (cargo still in warehouse) query_data = { "filter": { "property": "warehouse_out", "date": { "is_empty": True } } } logging.info(f"Querying milestone records where warehouse_out IS NULL") response = self._make_database_query("mawb_milestone", query_data) if not response: logging.warning("No response from milestone dwell report query") return [] results = response.get("results", []) logging.info(f"Retrieved {len(results)} milestone records for dwell report (warehouse_out IS NULL)") return results except Exception as e: logging.error(f"Error getting milestone records for dwell report: {e}") return [] def _get_all_milestone_records(self) -> List[Dict]: """ Get ALL milestone records without any date filtering. Returns: List of all milestone records """ try: database_id = self.database_ids.get("mawb_milestone") if not database_id: logging.error("MAWB milestone database ID not configured") return [] # Query with pagination to retrieve all records (no filters) query_data: Dict[str, Any] = {"page_size": 100} all_results: List[Dict] = [] has_more = True start_cursor = None while has_more: if start_cursor: query_data["start_cursor"] = start_cursor response = self._make_database_query("mawb_milestone", query_data) if not response: logging.warning("No response from milestone database query") break results = response.get("results", []) all_results.extend(results) has_more = response.get("has_more", False) start_cursor = response.get("next_cursor") logging.info(f"Retrieved {len(all_results)} total milestone records (paginated)") return all_results except Exception as e: logging.error(f"Error getting all milestone records: {e}") return [] def get_rollup_property_value(self, page_id: str, property_name: str) -> Dict: """ Retrieve rollup property value from a page. Args: page_id: The ID of the page property_name: Name of the rollup property Returns: Dictionary containing rollup property data """ try: # Note: property_name here should be the property ID from the page's properties dict endpoint = f"/pages/{page_id}/properties/{property_name}" response = self._make_request("GET", endpoint) if response: logging.debug(f"Retrieved rollup property {property_name} for page {page_id}") return response else: logging.warning(f"No response for rollup property {property_name} on page {page_id}") return {} except Exception as e: logging.error(f"Error retrieving rollup property {property_name} for page {page_id}: {e}") return {} def _extract_rollup_checkbox(self, prop_response: Dict) -> Optional[bool]: """Extract a boolean checkbox value from a rollup property response.""" try: if not prop_response: return None # New Notion API may return either a top-level rollup or paginated results if isinstance(prop_response, dict) and prop_response.get('type') == 'rollup': rollup = prop_response.get('rollup', {}) if rollup.get('type') == 'array': for item in rollup.get('array', []): if item.get('type') == 'checkbox': val = item.get('checkbox') if isinstance(val, bool): return val if item.get('type') == 'formula': formula = item.get('formula', {}) if formula.get('type') == 'boolean': return bool(formula.get('boolean')) elif rollup.get('type') == 'number': # Treat any positive number as True num = rollup.get('number') return bool(num) if num is not None else None if isinstance(prop_response, dict) and 'results' in prop_response: # Paginated property results for item in prop_response.get('results', []): if item.get('type') == 'checkbox': val = item.get('checkbox') if isinstance(val, bool): return val if item.get('type') == 'formula': formula = item.get('formula', {}) if formula.get('type') == 'boolean': return bool(formula.get('boolean')) return None except Exception: return None def create_dwell_report_record(self, date: datetime, pod: str, last_mile_carrier: str, volume: int, pallets_remaining: int = 0, volume_remaining_on_dock: int = 0, comments: str = "") -> bool: """ Create a dwell report record in Notion. Args: date: Report date pod: Port of discharge last_mile_carrier: Last mile carrier/product volume: Total volume (sum of parcels) pallets_remaining: Number of pallets remaining on dock volume_remaining_on_dock: Volume remaining on dock (sum of parcels where cargo ready but warehouse out is blank) comments: Comment about clearance status Returns: True if successful, False otherwise """ try: # Get the actual data source ID for page creation data_source_id = self._get_data_source_id("dwell_report") # Dwell report Date should be current date at 09:00 US/Pacific # Build a timezone-aware datetime at 9 AM Pacific for the provided date pacific = ZoneInfo("US/Pacific") date_pacific_9am = datetime( year=date.year, month=date.month, day=date.day, hour=9, minute=0, second=0, tzinfo=pacific, ) properties = { "Date": self._format_datetime_property(date_pacific_9am), "Broker Name": self._format_title_property("APEX"), "Port": self._format_text_property(pod), "LM Carrier/Product": self._format_text_property(last_mile_carrier), "Volume": self._format_number_property(volume), "# of Pallets Remaining on Dock": self._format_number_property(pallets_remaining), "Volume Remaining On Dock": self._format_number_property(volume_remaining_on_dock), "Comments": self._format_text_property(comments) } data = { "parent": { "type": "data_source_id", "data_source_id": data_source_id }, "properties": properties } result = self._make_request("POST", "/pages", data) if result: logging.info(f"Created dwell report record: {pod}-{last_mile_carrier} on {date.strftime('%Y-%m-%d')}") return True else: logging.error(f"Failed to create dwell report record: {pod}-{last_mile_carrier}") return False except Exception as e: logging.error(f"Error creating dwell report record: {e}") return False def generate_daily_dwell_report(self, target_date: datetime = None) -> Dict[str, int]: """ Generate daily dwell report by aggregating MAWB milestone data. Args: target_date: Date to generate report for (defaults to current date) Returns: Dictionary with generation statistics """ if target_date is None: target_date = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) try: logging.info(f"Generating dwell report for {target_date.strftime('%Y-%m-%d')}") # Get ALL milestone records (no date filtering at query level) # We'll apply different filters for different calculations all_milestone_records = self._get_all_milestone_records() if not all_milestone_records: logging.info("No milestone records found") return {"processed": 0, "created": 0, "errors": 0} # Group by POD and last_mile_carrier grouped_data = {} # Step 1: Process records for "remaining" calculations # Filter: cargo_ready IS NOT NULL AND warehouse_out IS NULL remaining_records = [] for record in all_milestone_records: properties = record.get("properties", {}) cargo_ready_prop = properties.get("cargo_ready", {}) cargo_ready_date = cargo_ready_prop.get("date", {}).get("start") if cargo_ready_prop.get("date") else None warehouse_out_prop = properties.get("warehouse_out", {}) warehouse_out_date = warehouse_out_prop.get("date", {}).get("start") if warehouse_out_prop.get("date") else None if cargo_ready_date is not None and warehouse_out_date is None: remaining_records.append(record) # Step 2: Process records for "volume" calculations # New logic: Sum parcels for all records where # warehouse_out IS NULL AND ETA is before today (strictly < target_date) volume_records = [] for record in all_milestone_records: properties = record.get("properties", {}) warehouse_out_prop = properties.get("warehouse_out", {}) warehouse_out_date = warehouse_out_prop.get("date", {}).get("start") if warehouse_out_prop.get("date") else None eta_prop = properties.get("eta", {}) eta_date_str = eta_prop.get("date", {}).get("start") if eta_prop.get("date") else None if warehouse_out_date is None and eta_date_str: try: eta_date = datetime.fromisoformat(eta_date_str.replace('Z', '+00:00')).replace(tzinfo=None) eta_date = eta_date.replace(hour=0, minute=0, second=0, microsecond=0) if eta_date < target_date: volume_records.append(record) except Exception as e: logging.warning(f"Error parsing ETA date {eta_date_str}: {e}") # Process remaining records for grouping for record in remaining_records: properties = record.get("properties", {}) # Extract POD and last mile carrier pod_prop = properties.get("pod", {}) pod = "" if pod_prop.get("select"): pod = pod_prop["select"].get("name", "") elif pod_prop.get("rich_text"): pod = pod_prop["rich_text"][0].get("text", {}).get("content", "") if pod_prop["rich_text"] else "" lm_carrier_prop = properties.get("last_mile_carrier", {}) lm_carrier = "" if lm_carrier_prop.get("select"): lm_carrier = lm_carrier_prop["select"].get("name", "") elif lm_carrier_prop.get("rich_text"): lm_carrier = lm_carrier_prop["rich_text"][0].get("text", {}).get("content", "") if lm_carrier_prop["rich_text"] else "" # Skip records without POD or last mile carrier if not pod or not lm_carrier: continue # Create group key group_key = f"{pod}_{lm_carrier}" if group_key not in grouped_data: grouped_data[group_key] = { "pod": pod, "last_mile_carrier": lm_carrier, "volume": 0, "pallets_remaining": 0, "volume_remaining_on_dock": 0, "customs_released_pallets": 0, "customs_not_released_pallets": 0 } # Add pallets remaining on dock pallets_prop = properties.get("pallets", {}) if pallets_prop.get("number") is not None: grouped_data[group_key]["pallets_remaining"] += pallets_prop["number"] # Add volume remaining on dock (parcels) parcels_prop = properties.get("parcel", {}) if parcels_prop.get("number") is None: parcels_prop = properties.get("parcels", {}) if parcels_prop.get("number") is not None: grouped_data[group_key]["volume_remaining_on_dock"] += parcels_prop["number"] # Defer customs released/not released calculation to clearance rollup stage # Process volume records for grouping for record in volume_records: properties = record.get("properties", {}) # Extract POD and last mile carrier pod_prop = properties.get("pod", {}) pod = "" if pod_prop.get("select"): pod = pod_prop["select"].get("name", "") elif pod_prop.get("rich_text"): pod = pod_prop["rich_text"][0].get("text", {}).get("content", "") if pod_prop["rich_text"] else "" lm_carrier_prop = properties.get("last_mile_carrier", {}) lm_carrier = "" if lm_carrier_prop.get("select"): lm_carrier = lm_carrier_prop["select"].get("name", "") elif lm_carrier_prop.get("rich_text"): lm_carrier = lm_carrier_prop["rich_text"][0].get("text", {}).get("content", "") if lm_carrier_prop["rich_text"] else "" # Skip records without POD or last mile carrier if not pod or not lm_carrier: continue # Create group key group_key = f"{pod}_{lm_carrier}" if group_key not in grouped_data: grouped_data[group_key] = { "pod": pod, "last_mile_carrier": lm_carrier, "volume": 0, "pallets_remaining": 0, "volume_remaining_on_dock": 0, "customs_released_pallets": 0, "customs_not_released_pallets": 0 } # Add volume (parcels) parcels_prop = properties.get("parcel", {}) if parcels_prop.get("number") is None: parcels_prop = properties.get("parcels", {}) if parcels_prop.get("number") is not None: grouped_data[group_key]["volume"] += parcels_prop["number"] # Calculate customs released/not released pallets for comments using clearance rollup # Get all records with warehouse_out IS NULL to calculate totals per group all_warehouse_records = [] for record in all_milestone_records: properties = record.get("properties", {}) warehouse_out_prop = properties.get("warehouse_out", {}) warehouse_out_date = warehouse_out_prop.get("date", {}).get("start") if warehouse_out_prop.get("date") else None if warehouse_out_date is None: all_warehouse_records.append(record) # Build page_id -> clearance status map using rollup property clearance_status_map = {} for record in all_warehouse_records: page_id = record.get("id") props = record.get("properties", {}) rollup_meta = props.get("clearance_notice_task_status", {}) prop_id = rollup_meta.get("id") if page_id and prop_id: try: resp = self.get_rollup_property_value(page_id, prop_id) is_cleared = self._extract_rollup_checkbox(resp) if is_cleared is not None: clearance_status_map[page_id] = is_cleared except Exception: pass # Aggregate pallets by clearance status per group for record in all_warehouse_records: properties = record.get("properties", {}) # Extract POD and last mile carrier pod_prop = properties.get("pod", {}) pod = "" if pod_prop.get("select"): pod = pod_prop["select"].get("name", "") elif pod_prop.get("rich_text"): pod = pod_prop["rich_text"][0].get("text", {}).get("content", "") if pod_prop["rich_text"] else "" lm_carrier_prop = properties.get("last_mile_carrier", {}) lm_carrier = "" if lm_carrier_prop.get("select"): lm_carrier = lm_carrier_prop["select"].get("name", "") elif lm_carrier_prop.get("rich_text"): lm_carrier = lm_carrier_prop["rich_text"][0].get("text", {}).get("content", "") if lm_carrier_prop["rich_text"] else "" if not pod or not lm_carrier: continue group_key = f"{pod}_{lm_carrier}" if group_key not in grouped_data: # Initialize if not present (e.g., group had no remaining_records but has volume records) grouped_data[group_key] = { "pod": pod, "last_mile_carrier": lm_carrier, "volume": 0, "pallets_remaining": 0, "volume_remaining_on_dock": 0, "customs_released_pallets": 0, "customs_not_released_pallets": 0 } pallets_prop = properties.get("pallets", {}) pallets_count = pallets_prop.get("number", 0) if pallets_prop.get("number") is not None else 0 is_cleared = clearance_status_map.get(record.get("id")) if is_cleared is True: grouped_data[group_key]["customs_released_pallets"] += pallets_count elif is_cleared is False: grouped_data[group_key]["customs_not_released_pallets"] += pallets_count # Create dwell report records created_count = 0 error_count = 0 for group_key, group_data in grouped_data.items(): # Generate comments released_pallets = group_data["customs_released_pallets"] not_released_pallets = group_data["customs_not_released_pallets"] comments = f"{released_pallets} pallets customs released, {not_released_pallets} pallets not released yet" # Create the dwell report record success = self.create_dwell_report_record( date=target_date, pod=group_data["pod"], last_mile_carrier=group_data["last_mile_carrier"], volume=group_data["volume"], pallets_remaining=group_data["pallets_remaining"], volume_remaining_on_dock=group_data["volume_remaining_on_dock"], comments=comments ) if success: created_count += 1 else: error_count += 1 total_processed = len(remaining_records) + len(volume_records) logging.info(f"Dwell report generation completed: {created_count} created, {error_count} errors") return { "processed": total_processed, "created": created_count, "errors": error_count } except Exception as e: logging.error(f"Error generating daily dwell report: {e}") return {"processed": 0, "created": 0, "errors": 1} def get_all_dwell_report_records_paginated(self, page_size: int = 100) -> List[Dict]: """ Get all dwell report records from Notion database with pagination. Args: page_size: Number of records per page Returns: List of all dwell report records """ try: database_id = self.database_ids.get("dwell_report") if not database_id: logging.error("Dwell report database ID not configured") return [] query_data = {"page_size": page_size} all_results = [] has_more = True start_cursor = None while has_more: if start_cursor: query_data["start_cursor"] = start_cursor response = self._make_database_query("dwell_report", query_data) results = response.get("results", []) all_results.extend(results) has_more = response.get("has_more", False) start_cursor = response.get("next_cursor") logging.info(f"Retrieved {len(all_results)} total dwell report records") return all_results except Exception as e: logging.error(f"Error getting all dwell report records: {e}") return [] def expand_env_variables(value): """Recursively expand environment variables in string values like ${VAR_NAME}""" if isinstance(value, dict): return {key: expand_env_variables(val) for key, val in value.items()} elif isinstance(value, list): return [expand_env_variables(item) for item in value] elif isinstance(value, str): # Replace ${VAR_NAME} with environment variable value pattern = r'\$\{([^}]+)\}' matches = re.findall(pattern, value) for match in matches: env_value = os.getenv(match) if env_value: value = value.replace(f'${{{match}}}', env_value) else: logging.warning(f"Environment variable {match} not found") return value def get_all_entry_records_paginated(self, page_size: int = 100) -> List[Dict]: """ Get all entry records from Notion database with pagination. Args: page_size: Number of records per page Returns: List of all entry records """ try: database_id = self.database_ids.get("entry_record") if not database_id: logging.error("Entry record database ID not configured") return [] query_data = {"page_size": page_size} all_results = [] has_more = True start_cursor = None while has_more: if start_cursor: query_data["start_cursor"] = start_cursor response = self._make_database_query("entry_record", query_data) results = response.get("results", []) all_results.extend(results) has_more = response.get("has_more", False) start_cursor = response.get("next_cursor") logging.info(f"Retrieved {len(all_results)} total entry records") return all_results except Exception as e: logging.error(f"Error getting all entry records: {e}") return [] def format_entry_number(self, entry_number: str) -> str: """ Format entry number by removing prefixes and formatting consistently. Args: entry_number: Raw entry number (e.g., "9G3-0000251-7") Returns: Formatted entry number (e.g., "00002517") """ try: # Remove common prefixes and clean formatting cleaned = entry_number.strip() # Remove "9G3-" prefix if present if cleaned.upper().startswith("9G3-"): cleaned = cleaned[4:] # Remove all dashes and spaces cleaned = cleaned.replace("-", "").replace(" ", "") # If it ends with a single digit (check digit), keep it # Format: 00002517 (8 digits total) if cleaned.isdigit(): return cleaned return cleaned except Exception as e: logging.warning(f"Error formatting entry number '{entry_number}': {e}") return entry_number.strip() def format_master_bl(self, master_bl: str) -> str: """ Format Master B/L number consistently. Args: master_bl: Raw Master B/L number Returns: Formatted Master B/L number """ try: # Basic cleaning - remove extra whitespace return master_bl.strip() except Exception as e: logging.warning(f"Error formatting Master B/L '{master_bl}': {e}") return master_bl def create_entry_records_batch(self, entry_data_list: List[Dict]) -> Dict[str, int]: """ Create multiple entry records in batch. Args: entry_data_list: List of entry data dictionaries with required fields Returns: Dictionary with statistics: {'created': int, 'errors': int, 'total': int} """ stats = {'created': 0, 'errors': 0, 'total': 0} try: database_id = self.database_ids.get("entry_record") if not database_id: logging.error("Entry record database ID not configured") return stats # Pre-fetch MAWB records for linking mawb_numbers = list(set([entry.get('mawb', '') for entry in entry_data_list if entry.get('mawb')])) mawb_lookup = {} for mawb_num in mawb_numbers: mawb_record = self.find_mawb_record_by_number(mawb_num) if mawb_record: mawb_lookup[mawb_num] = mawb_record # Pre-fetch milestone tasks for auto-linking milestone_tasks_lookup = {} for mawb_num in mawb_numbers: tasks = self.get_mawb_tasks_by_action_and_mawb(mawb_num, "Entry Milestone Update") if tasks: milestone_tasks_lookup[mawb_num] = [task["id"] for task in tasks] # Process each entry for entry_data in entry_data_list: stats['total'] += 1 try: mawb = entry_data.get('mawb', '').strip() entry_number = entry_data.get('entry_number', '').strip() if not mawb or not entry_number: logging.warning(f"Missing required fields for entry: {entry_data}") stats['errors'] += 1 continue # Build properties properties = { "mawb": self._format_title_property(mawb), "entry_number": self._format_text_property(entry_number) } # Add MAWB relation if available if mawb in mawb_lookup: properties["mawb_reference"] = { "relation": [{"id": mawb_lookup[mawb]["id"]}] } # Add optional date fields date_fields = [ 'entry_submit_date', 'entry_created', 'entry_date', 'cbp_hold_date', 'cbp_release_date', 'cpsc_review_date', 'cpsc_release_date', 'estimated_review_date' ] for field in date_fields: if field in entry_data and entry_data[field]: if isinstance(entry_data[field], datetime): properties[field] = self._format_datetime_property(entry_data[field]) elif isinstance(entry_data[field], str): try: dt = datetime.fromisoformat(entry_data[field].replace('Z', '')) properties[field] = self._format_datetime_property(dt) except: logging.warning(f"Could not parse date '{entry_data[field]}' for field {field}") # Add optional text fields text_fields = ['cbp_hold_line', 'cpsc_review_line'] for field in text_fields: if field in entry_data and entry_data[field]: properties[field] = self._format_text_property(str(entry_data[field])) # Get the actual data source ID for page creation data_source_id = self._get_data_source_id("entry_record") # Create the entry create_data = { "parent": { "type": "data_source_id", "data_source_id": data_source_id }, "properties": properties } response = self._make_request("POST", "/pages", create_data) entry_id = response.get("id") # Auto-link to milestone tasks if mawb in milestone_tasks_lookup and milestone_tasks_lookup[mawb]: try: self.create_entry_mawb_task_relation(entry_number, milestone_tasks_lookup[mawb]) except Exception as link_error: logging.warning(f"Failed to auto-link milestone tasks for {entry_number}: {link_error}") stats['created'] += 1 logging.info(f"Created entry record: {mawb} - {entry_number}") except Exception as e: logging.error(f"Error creating entry record: {e}") stats['errors'] += 1 continue logging.info(f"Batch entry creation completed: {stats['created']} created, {stats['errors']} errors out of {stats['total']} total") return stats except Exception as e: logging.error(f"Error in batch entry creation: {e}") stats['errors'] = stats['total'] return stats def update_entry_records_batch(self, updates: List[Dict]) -> Dict[str, int]: """ Update multiple entry records in batch. Args: updates: List of update dictionaries with 'entry_number' and field updates Returns: Dictionary with statistics: {'updated': int, 'errors': int, 'total': int} """ stats = {'updated': 0, 'errors': 0, 'total': 0} try: database_id = self.database_ids.get("entry_record") if not database_id: logging.error("Entry record database ID not configured") return stats # Get entry numbers to update entry_numbers = [update.get('entry_number', '') for update in updates if update.get('entry_number')] # Get existing entries existing_entries = self.get_entries_by_numbers(entry_numbers) # Process each update for update in updates: stats['total'] += 1 try: entry_number = update.get('entry_number', '').strip() if not entry_number: logging.warning("Missing entry_number in update") stats['errors'] += 1 continue if entry_number not in existing_entries: logging.warning(f"Entry record not found for update: {entry_number}") stats['errors'] += 1 continue entry_record = existing_entries[entry_number] page_id = entry_record["id"] # Build update properties properties = {} # Handle date fields date_fields = [ 'entry_submit_date', 'entry_created', 'entry_date', 'cbp_hold_date', 'cbp_release_date', 'cpsc_review_date', 'cpsc_release_date', 'estimated_review_date' ] for field in date_fields: if field in update and update[field] is not None: if isinstance(update[field], datetime): properties[field] = self._format_datetime_property(update[field]) elif isinstance(update[field], str) and update[field].strip(): try: dt = datetime.fromisoformat(update[field].replace('Z', '')) properties[field] = self._format_datetime_property(dt) except: logging.warning(f"Could not parse date '{update[field]}' for field {field}") # Handle text fields text_fields = ['cbp_hold_line', 'cpsc_review_line'] for field in text_fields: if field in update and update[field] is not None: properties[field] = self._format_text_property(str(update[field])) # Only update if there are properties to update if properties: update_data = {"properties": properties} self._make_request("PATCH", f"/pages/{page_id}", update_data) stats['updated'] += 1 logging.info(f"Updated entry record: {entry_number}") else: logging.info(f"No changes for entry record: {entry_number}") except Exception as e: logging.error(f"Error updating entry record: {e}") stats['errors'] += 1 continue logging.info(f"Batch entry update completed: {stats['updated']} updated, {stats['errors']} errors out of {stats['total']} total") return stats except Exception as e: logging.error(f"Error in batch entry update: {e}") stats['errors'] = stats['total'] return stats def load_notion_config() -> Dict: """Load Notion configuration from customer_config.json with environment variable expansion""" try: with open('config/customer_config.json', 'r', encoding='utf-8') as f: config = json.load(f) notion_settings = config.get("notion_settings", {}) # Expand environment variables in all string values for key, value in notion_settings.items(): notion_settings[key] = expand_env_variables(value) return notion_settings except Exception as e: logging.error(f"Error loading Notion config: {e}") return {} # Example usage if __name__ == "__main__": # Load configuration notion_config = load_notion_config() if not notion_config.get("api_key") or not notion_config.get("database_ids"): print("Please configure Notion API key and database IDs in config/customer_config.json") exit(1) # Initialize Notion integration notion = NotionIntegration( api_key=notion_config["api_key"], database_ids=notion_config["database_ids"] ) # Example operations mawb = "TEST123456" customer = "SHEIN" # Check if MAWB exists if not notion.check_mawb_exists(mawb, customer): # Create MAWB record notion.create_mawb_record( mawb=mawb, customer=customer, entry_type="T86", pod="LAX", manifest_received=datetime.now() ) # Create task record notion.create_mawb_task_record(mawb=mawb) # Create entry records notion.create_entry_record(mawb=mawb, entry_number="E001") notion.create_entry_record(mawb=mawb, entry_number="E002") # Update MAWB status notion.update_mawb_status(mawb, customer, awb_received=datetime.now()) # Entry counts are now handled automatically by Notion rollup properties print(f"Notion integration test completed for MAWB: {mawb}")