import json import requests import mysql.connector import logging import smtplib from email.message import EmailMessage # Configure logging with timestamp, level, and message format logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" ) def login_api(): """Authenticate and get JWT token from login API.""" payload = {"username": API_USERNAME, "password": API_PASSWORD} headers = {"Content-Type": "application/json"} try: response = requests.post( f"{BASE_URL}/auth/login", json=payload, headers=headers ) response.raise_for_status() data = response.json().get("data", {}) jwt = data.get("token") if jwt: logging.info("API login successful.") return jwt else: logging.error("JWT token not found in login response.") return None except requests.RequestException as e: logging.error(f"Login API error: {e}") return None def select_tenant(jwt): """Select tenant and retrieve a new JWT token.""" headers = {"Authorization": f"Bearer {jwt}", "Content-Type": "application/json"} try: response = requests.post( f"{BASE_URL}/auth/select-tenant/{API_TENANT}", headers=headers ) response.raise_for_status() data = response.json().get("data", {}) jwt = data.get("token") if jwt: logging.info("Tenant selected successfully.") return jwt else: logging.error("JWT token not found in tenant selection response.") return None except requests.RequestException as e: logging.error(f"Select tenant error: {e}") return None def get_recurring_templates(cursor): """Retrieve recurring payment templates scheduled for today which are active.""" try: query = """ SELECT Id FROM RecurringPayments WHERE StatusId = %s AND NextStrikeDate IS NOT NULL AND DATE(NextStrikeDate) = CURDATE() AND DATE(EndDate) >= CURDATE() AND IsActive IS TRUE """ val = ("da462422-13b2-45cc-a175-910a225f6fc8",) cursor.execute(query, val) result = cursor.fetchall() id_list = [row[0] for row in result] logging.info(f"Found {len(id_list)} recurring templates to process.") return id_list except mysql.connector.Error as e: logging.error(f"MySQL query error: {e}") return [] def convert_recurring_templates(jwt, recurring_template_ids): """Convert recurring payment templates to payment requests via API.""" if not recurring_template_ids: logging.info("No recurring templates to process.") return [] payload = {"recurringTemplateIds": recurring_template_ids} headers = {"Authorization": f"Bearer {jwt}", "Content-Type": "application/json"} try: response = requests.post( f"{BASE_URL}/expense/recurring-payment/convert/payment-request", json=payload, headers=headers, ) response.raise_for_status() message = response.json().get("message", "Conversion successful.") data = response.json().get("data", []) logging.info(f"Conversion response: {message}") return data except requests.HTTPError as http_err: status_code = http_err.response.status_code try: error_message = http_err.response.json().get( "message", http_err.response.text ) except Exception: error_message = http_err.response.text logging.error(f"HTTP error {status_code}: {error_message}") return [] except requests.RequestException as e: logging.error(f"Request error: {e}") return [] except Exception as e: logging.error(f"Unexpected error during conversion: {e}") return [] def get_frequency(frequency_code): """Translate frequency code to human-readable frequency string.""" mapping = { 0: "Monthly", 1: "Quarterly", 2: "Half-Yearly", 3: "Yearly", 4: "Daily", 5: "Weekly", } return mapping.get(frequency_code, "Unknown") def send_email_notifications(email_config, recurring_payment_data): """Send email notifications for generated payment requests.""" SMPTSERVER = email_config.get("SMPTSERVER") PORT = email_config.get("PORT") SENDER_EMAIL = email_config.get("SENDER_EMAIL") SENDER_PASSWORD = email_config.get("SENDER_PASSWORD") if not recurring_payment_data: logging.info("No payment request data to send emails for.") return try: with smtplib.SMTP(SMPTSERVER, PORT) as smtp: smtp.ehlo() smtp.starttls() smtp.ehlo() smtp.login(SENDER_EMAIL, SENDER_PASSWORD) for item in recurring_payment_data: recurringPayment = item.get("recurringPayment", {}) emails = item.get("emails") if not emails: logging.warning("No recipient emails found for one of the payment requests.") continue next_strike_date = recurringPayment.get("nextStrikeDate", "") formatted_next_strike_date = next_strike_date[:10] if next_strike_date else "N/A" frequency = get_frequency(recurringPayment.get("frequency")) # Compose email content content = f""" Here are the details of the payment: - Title: {recurringPayment.get('title', 'N/A')} - Description: {recurringPayment.get('description', 'N/A')} - Payee: {recurringPayment.get('payee', 'N/A')} - Amount: {recurringPayment.get('currency', {}).get('symbol', '')} {recurringPayment.get('amount', 'N/A')} ({recurringPayment.get('currency', {}).get('currencyName', 'N/A')}) - Currency Code: {recurringPayment.get('currency', {}).get('currencyCode', 'N/A')} - Recurring Frequency: {frequency} - Next Strike Date: {formatted_next_strike_date} - Strike Date: {recurringPayment.get('strikeDate', '')[:10] if recurringPayment.get('strikeDate') else 'N/A'} - End Date: {recurringPayment.get('endDate', '')[:10] if recurringPayment.get('endDate') else 'N/A'} - Expense Category: {recurringPayment.get('expenseCategory', {}).get('name', 'N/A')} - Status: {recurringPayment.get('status', {}).get('name', 'N/A')} Please ensure timely processing. """ # Setup EmailMessage msg = EmailMessage() msg["Subject"] = f"Payment Request Generated from Recurring Payment Template \"{recurringPayment.get('title', '')}\"" msg["From"] = SENDER_EMAIL msg["To"] = emails msg.set_content(content) try: smtp.send_message(msg) logging.info(f"Email sent successfully to {emails}.") except Exception as email_err: logging.error(f"Failed to send email to {emails}: {email_err}") except Exception as e: logging.error(f"SMTP connection or login error: {e}") # --- Main execution logic --- if __name__ == "__main__": import sys # Load configuration from JSON file GLOBAL_CONFIG_PATH = "config.json" try: with open(GLOBAL_CONFIG_PATH, "r", encoding="utf-8") as f: config = json.load(f) logging.info("Configuration loaded successfully.") except Exception as e: logging.critical(f"Failed to load config: {e}") sys.exit(1) # Extract essential API config parameters API_CONFIG = config.get("API", {}) BASE_URL = API_CONFIG.get("BASE_URL") API_USERNAME = API_CONFIG.get("USERNAME") API_PASSWORD = API_CONFIG.get("PASSWORD") API_TENANT = API_CONFIG.get("TENANTID") # Initialize database connection MYSQL_CONFIG = config.get("MYSQL", {}) try: mydb = mysql.connector.connect( host=MYSQL_CONFIG.get("DB_HOST"), user=MYSQL_CONFIG.get("DB_USER"), password=MYSQL_CONFIG.get("DB_PASSWORD"), database=MYSQL_CONFIG.get("DB_NAME"), ) mycursor = mydb.cursor() logging.info("Database connection established.") except mysql.connector.Error as e: logging.critical(f"Database connection error: {e}") sys.exit(1) # Authenticate and get JWT token token = login_api() if not token: logging.critical("Login failed, aborting.") sys.exit(1) # Select tenant and get tenant-specific JWT token jwt_token = select_tenant(token) if not jwt_token: logging.critical("Tenant selection failed, aborting.") sys.exit(1) # Fetch recurring payment template IDs for processing recurring_ids = get_recurring_templates(mycursor) # Close DB resources gracefully mycursor.close() mydb.close() logging.info("Database connection closed.") # Convert recurring templates to payment requests via API response_data = convert_recurring_templates(jwt_token, recurring_ids) # Send email notifications for generated payment requests EMAIL_CONFIG = config.get("SMPT", {}) send_email_notifications(EMAIL_CONFIG, response_data)