257 lines
8.9 KiB
Python
257 lines
8.9 KiB
Python
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
|
|
|
|
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')})
|
|
Due Date: {item.get('dueDate', '')[:10] if item.get('dueDate') else 'N/A'}
|
|
Currency Code: {recurringPayment.get('currency', {}).get('currencyCode', 'N/A')}
|
|
Recurring Frequency: {frequency}
|
|
Next Strike Date: {recurringPayment.get("nextStrikeDate", "")[:10] if recurringPayment.get("nextStrikeDate", "") 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 for \"{recurringPayment.get('payee', 'N/A')}\""
|
|
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)
|