You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
482 lines
17 KiB
482 lines
17 KiB
"""
|
|
Payment Repository Module
|
|
|
|
This module provides a repository pattern for payment-related database operations.
|
|
It abstracts database access and provides a clean interface for CRUD operations.
|
|
"""
|
|
|
|
import logging
|
|
import pymysql
|
|
import json
|
|
from datetime import datetime
|
|
from typing import List, Dict, Any, Optional, Union
|
|
from sqlalchemy.orm import Session
|
|
from collections import defaultdict
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class PaymentRepository:
|
|
"""
|
|
Repository for payment-related database operations.
|
|
|
|
This class provides methods for creating, reading, updating, and deleting
|
|
payment records in both the Payments and SinglePayments tables.
|
|
"""
|
|
|
|
def __init__(self, db_session: Session):
|
|
"""
|
|
Initialize the payment repository.
|
|
|
|
Args:
|
|
db_session: SQLAlchemy database session
|
|
"""
|
|
self.db = db_session
|
|
|
|
def create_payment_batch(self) -> Optional[int]:
|
|
"""
|
|
Create a new payment batch record.
|
|
|
|
Returns:
|
|
Batch ID if successful, None otherwise
|
|
"""
|
|
from models import PaymentBatch
|
|
|
|
try:
|
|
batch = PaymentBatch()
|
|
self.db.add(batch)
|
|
self.db.commit()
|
|
logger.info(f"Created payment batch with ID {batch.id}")
|
|
return batch.id
|
|
except Exception as e:
|
|
self.db.rollback()
|
|
logger.error(f"Failed to create payment batch: {e}")
|
|
return None
|
|
|
|
def add_payments_to_batch(self, customers: List[Dict[str, Any]], batch_id: int, invoices: Dict) -> Dict[str, int]:
|
|
"""
|
|
Add multiple payments to a batch atomically.
|
|
|
|
Args:
|
|
customers: List of customer payment data dictionaries
|
|
batch_id: Batch ID to associate payments with
|
|
|
|
Returns:
|
|
Dictionary with 'added' and 'failed' counts
|
|
"""
|
|
from models import Payments
|
|
|
|
result = {"added": 0, "failed": 0}
|
|
payments_to_add = []
|
|
|
|
try:
|
|
for cust in customers:
|
|
cust_id_str = str(cust['customer_id'])
|
|
cust_invoices = invoices.get(cust_id_str) # Returns None if no invoices
|
|
|
|
payment = Payments(
|
|
PaymentBatch_ID=batch_id,
|
|
Splynx_ID=cust['customer_id'],
|
|
Stripe_Customer_ID=cust['stripe_customer_id'],
|
|
Payment_Amount=float(cust['deposit']) * -1,
|
|
Stripe_Payment_Method=cust.get('stripe_pm', None),
|
|
PaymentPlan_ID=cust.get('paymentplan_id', None),
|
|
Invoices_to_Pay=cust_invoices
|
|
)
|
|
payments_to_add.append(payment)
|
|
self.db.add(payment)
|
|
|
|
self.db.commit()
|
|
result["added"] = len(payments_to_add)
|
|
logger.info(f"Successfully added {len(payments_to_add)} payments to batch {batch_id}")
|
|
except Exception as e:
|
|
self.db.rollback()
|
|
result["failed"] = len(payments_to_add)
|
|
logger.error(f"Failed to add payments to batch {batch_id}: {e}")
|
|
|
|
return result
|
|
|
|
def get_payments_by_batch(self, batch_id: int) -> List[Any]:
|
|
"""
|
|
Get all payments for a given batch.
|
|
|
|
Args:
|
|
batch_id: Batch ID
|
|
|
|
Returns:
|
|
List of Payment objects
|
|
"""
|
|
from models import Payments
|
|
|
|
try:
|
|
payments = self.db.query(Payments).filter(Payments.PaymentBatch_ID == batch_id).all()
|
|
logger.debug(f"Found {len(payments)} payments for batch {batch_id}")
|
|
return payments
|
|
except Exception as e:
|
|
logger.error(f"Error fetching payments for batch {batch_id}: {e}")
|
|
return []
|
|
|
|
def get_payment_by_id(self, payment_id: int, payment_type: str = "pay") -> Optional[Any]:
|
|
"""
|
|
Get a payment record by ID.
|
|
|
|
Args:
|
|
payment_id: Payment record ID
|
|
payment_type: "pay" for Payments, "singlepay" for SinglePayments
|
|
|
|
Returns:
|
|
Payment object if found, None otherwise
|
|
"""
|
|
from models import Payments, SinglePayments
|
|
|
|
try:
|
|
if payment_type == "pay":
|
|
return self.db.query(Payments).filter(Payments.id == payment_id).first()
|
|
elif payment_type == "singlepay":
|
|
return self.db.query(SinglePayments).filter(SinglePayments.id == payment_id).first()
|
|
else:
|
|
logger.error(f"Invalid payment type: {payment_type}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Error fetching payment {payment_id}: {e}")
|
|
return None
|
|
|
|
def get_pending_payment_intents(self, payment_type: str = "both") -> Dict[str, List[Any]]:
|
|
"""
|
|
Get all payments with pending payment intents requiring follow-up.
|
|
|
|
Args:
|
|
payment_type: "pay", "singlepay", or "both"
|
|
|
|
Returns:
|
|
Dictionary with payment lists by type
|
|
"""
|
|
from models import Payments, SinglePayments
|
|
|
|
result = {}
|
|
try:
|
|
if payment_type in ["pay", "both"]:
|
|
result["pay"] = self.db.query(Payments).filter(Payments.PI_FollowUp == True).all()
|
|
if payment_type in ["singlepay", "both"]:
|
|
result["singlepay"] = self.db.query(SinglePayments).filter(SinglePayments.PI_FollowUp == True).all()
|
|
|
|
total = sum(len(v) for v in result.values())
|
|
logger.info(f"Found {total} pending payment intents requiring follow-up")
|
|
return result
|
|
except Exception as e:
|
|
logger.error(f"Error fetching pending payment intents: {e}")
|
|
return {}
|
|
|
|
def get_pending_refunds(self, payment_type: str = "both") -> Dict[str, List[Any]]:
|
|
"""
|
|
Get all payments with pending refunds requiring follow-up.
|
|
|
|
Args:
|
|
payment_type: "pay", "singlepay", or "both"
|
|
|
|
Returns:
|
|
Dictionary with payment lists by type
|
|
"""
|
|
from models import Payments, SinglePayments
|
|
|
|
result = {}
|
|
try:
|
|
if payment_type in ["pay", "both"]:
|
|
result["pay"] = self.db.query(Payments).filter(Payments.Refund_FollowUp == True).all()
|
|
if payment_type in ["singlepay", "both"]:
|
|
result["singlepay"] = self.db.query(SinglePayments).filter(SinglePayments.Refund_FollowUp == True).all()
|
|
|
|
total = sum(len(v) for v in result.values())
|
|
logger.info(f"Found {total} pending refunds requiring follow-up")
|
|
return result
|
|
except Exception as e:
|
|
logger.error(f"Error fetching pending refunds: {e}")
|
|
return {}
|
|
|
|
def get_active_payment_plans(self) -> List[Any]:
|
|
"""
|
|
Get all active payment plans.
|
|
|
|
Returns:
|
|
List of PaymentPlans objects
|
|
"""
|
|
from models import PaymentPlans
|
|
|
|
try:
|
|
plans = self.db.query(PaymentPlans).filter(PaymentPlans.Enabled == True).all()
|
|
logger.info(f"Found {len(plans)} active payment plans")
|
|
return plans
|
|
except Exception as e:
|
|
logger.error(f"Error fetching active payment plans: {e}")
|
|
return []
|
|
|
|
def get_test_customer_data(self, payment_method: List) -> Union[List[Dict[str, Any]], bool]:
|
|
|
|
customers = [
|
|
{
|
|
"customer_id": 1222024,
|
|
"stripe_customer_id": "cus_SoNAgAbkbFo8ZY",
|
|
"deposit": -10,
|
|
"stripe_pm": "pm_1RskQIPfYyg6zE1Spbtf4pIa",
|
|
"paymentplan_id": None
|
|
},
|
|
{
|
|
"customer_id": 1222025,
|
|
"stripe_customer_id": "cus_SoMVPWxdYstYbr",
|
|
"deposit": -10,
|
|
"stripe_pm": "pm_1SmTzDPfYyg6zE1SZnZogJAz",
|
|
"paymentplan_id": None
|
|
},
|
|
{
|
|
"customer_id": 1222026,
|
|
"stripe_customer_id": "cus_SoMVQ6Xj2dIrCR",
|
|
"deposit": -10,
|
|
"stripe_pm": "pm_1RsjmdPfYyg6zE1SOp2vBKNT",
|
|
"paymentplan_id": None
|
|
},
|
|
{
|
|
"customer_id": 1222027,
|
|
"stripe_customer_id": "cus_SoMVqjH7zkc5Qe",
|
|
"deposit": -10,
|
|
"stripe_pm": "pm_1RsjmFPfYyg6zE1SS1DmqdvB",
|
|
"paymentplan_id": None
|
|
},
|
|
{
|
|
"customer_id": 1222028,
|
|
"stripe_customer_id": "cus_SoQiDcSrNRxbPF",
|
|
"deposit": -10,
|
|
"stripe_pm": "pm_1SmTblPfYyg6zE1SWkzAFekG",
|
|
"paymentplan_id": None
|
|
},
|
|
|
|
]
|
|
return customers
|
|
|
|
|
|
def query_mysql_customer_invoices(self, mysql_config: dict, limit: int = 10) -> Dict:
|
|
"""
|
|
Query customer invoice data from external MySQL database.
|
|
|
|
Args:
|
|
mysql_config: MySQL connection configuration
|
|
limit: Maximum number of results
|
|
|
|
Returns:
|
|
List of customer dictionaries or False if no results/error
|
|
"""
|
|
invoices_dict = {}
|
|
connection = None
|
|
try:
|
|
connection = pymysql.connect(
|
|
host=mysql_config['host'],
|
|
database=mysql_config['database'],
|
|
user=mysql_config['user'],
|
|
password=mysql_config['password'],
|
|
port=mysql_config['port'],
|
|
autocommit=False,
|
|
cursorclass=pymysql.cursors.DictCursor
|
|
)
|
|
|
|
logger.info(f"Connected to MySQL database: {mysql_config['database']} on {mysql_config['host']}")
|
|
|
|
query = """
|
|
SELECT
|
|
i.id,
|
|
i.customer_id
|
|
FROM invoices i
|
|
WHERE i.status = 'not_paid'
|
|
ORDER BY i.customer_id ASC
|
|
LIMIT %s
|
|
"""
|
|
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(query, (limit))
|
|
results = cursor.fetchall()
|
|
|
|
if results:
|
|
logger.info(f"Found {len(results)} customer invoices")
|
|
# Group invoice IDs by customer_id
|
|
invoices_dict = defaultdict(list)
|
|
for res in results:
|
|
#print(res['id'], res['customer_id'])
|
|
invoices_dict[res['customer_id']].append(str(res['id']))
|
|
|
|
#print(f"\nCustomer Invoice Data: {json.dumps(invoices_dict)}\n")
|
|
print(f"Cust 31: {invoices_dict[31]}")
|
|
# Convert to comma-separated strings
|
|
invoices_dict = {
|
|
str(cust_id): ",".join(inv_ids)
|
|
for cust_id, inv_ids in invoices_dict.items()
|
|
}
|
|
|
|
return invoices_dict
|
|
else:
|
|
logger.info(f"No customer invoices found")
|
|
return False
|
|
|
|
except pymysql.Error as e:
|
|
logger.error(f"MySQL Error: {e}")
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"Unexpected Error querying MySQL: {e}")
|
|
return False
|
|
finally:
|
|
if connection:
|
|
connection.close()
|
|
logger.debug("MySQL connection closed")
|
|
|
|
def query_mysql_customers(self, mysql_config: dict, payment_methods: List, deposit_threshold: float = 0, limit: int = 10) -> Union[List[Dict[str, Any]], bool]:
|
|
"""
|
|
Query customer billing data from external MySQL database.
|
|
|
|
Args:
|
|
mysql_config: MySQL connection configuration
|
|
payment_method: Payment method ID (2=Direct Debit, 3=Card, 9=Payment Plan)
|
|
deposit_threshold: Maximum deposit threshold
|
|
limit: Maximum number of results
|
|
|
|
Returns:
|
|
List of customer dictionaries or False if no results/error
|
|
"""
|
|
connection = None
|
|
try:
|
|
connection = pymysql.connect(
|
|
host=mysql_config['host'],
|
|
database=mysql_config['database'],
|
|
user=mysql_config['user'],
|
|
password=mysql_config['password'],
|
|
port=mysql_config['port'],
|
|
autocommit=False,
|
|
cursorclass=pymysql.cursors.DictCursor
|
|
)
|
|
|
|
logger.info(f"Connected to MySQL database: {mysql_config['database']} on {mysql_config['host']}")
|
|
|
|
placeholders = ','.join(['%s'] * len(payment_methods))
|
|
|
|
query = f"""
|
|
SELECT DISTINCT
|
|
cb.customer_id,
|
|
cb.deposit,
|
|
cb.payment_method,
|
|
pad.field_1 AS stripe_customer_id
|
|
FROM customer_billing cb
|
|
LEFT OUTER JOIN payment_account_data pad ON cb.customer_id = pad.customer_id
|
|
WHERE cb.payment_method IN ({placeholders})
|
|
AND cb.deposit < %s
|
|
AND pad.field_1 IS NOT NULL
|
|
AND NOT EXISTS (
|
|
SELECT 1
|
|
FROM invoices i
|
|
WHERE i.customer_id = cb.customer_id
|
|
AND i.status = 'pending'
|
|
)
|
|
GROUP BY cb.customer_id, cb.deposit, cb.payment_method, pad.field_1
|
|
ORDER BY cb.payment_method ASC, cb.customer_id ASC
|
|
LIMIT %s
|
|
"""
|
|
|
|
###### Multi-month billing runs
|
|
#query = """
|
|
#SELECT DISTINCT
|
|
# cb.customer_id,
|
|
# cb.deposit,
|
|
# cb.payment_method,
|
|
# pad.field_1 AS stripe_customer_id
|
|
#FROM customer_billing cb
|
|
#LEFT OUTER JOIN payment_account_data pad ON cb.customer_id = pad.customer_id
|
|
#WHERE cb.payment_method = %s
|
|
#AND cb.deposit < %s
|
|
#AND pad.field_1 IS NOT NULL
|
|
#AND NOT EXISTS (
|
|
# SELECT 1
|
|
# FROM invoices i
|
|
# WHERE i.customer_id = cb.customer_id
|
|
# AND i.status = 'pending'
|
|
#)
|
|
#AND EXISTS (
|
|
# SELECT 1
|
|
# FROM invoices i
|
|
# WHERE i.customer_id = cb.customer_id
|
|
# AND i.date_till = '2026-01-20'
|
|
#)
|
|
#GROUP BY cb.customer_id, cb.deposit, cb.payment_method, pad.field_1
|
|
#ORDER BY cb.payment_method ASC, cb.customer_id ASC
|
|
#LIMIT %s
|
|
#"""
|
|
|
|
## Need to fix "date_till"
|
|
#####################
|
|
|
|
|
|
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(query, (*payment_methods, deposit_threshold, limit))
|
|
results = cursor.fetchall()
|
|
|
|
if results:
|
|
logger.info(f"Found {len(results)} customers for payment method {payment_methods}")
|
|
return results
|
|
else:
|
|
logger.info(f"No customers found for payment method {payment_methods}")
|
|
return False
|
|
|
|
except pymysql.Error as e:
|
|
logger.error(f"MySQL Error: {e}")
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"Unexpected Error querying MySQL: {e}")
|
|
return False
|
|
finally:
|
|
if connection:
|
|
connection.close()
|
|
logger.debug("MySQL connection closed")
|
|
|
|
def update_payment(self, payment_id: int, updates: Dict[str, Any], payment_type: str = "pay") -> bool:
|
|
"""
|
|
Update a payment record with given fields.
|
|
|
|
Args:
|
|
payment_id: Payment ID
|
|
updates: Dictionary of field names and values to update
|
|
payment_type: "pay" or "singlepay"
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
try:
|
|
payment = self.get_payment_by_id(payment_id, payment_type)
|
|
if not payment:
|
|
logger.error(f"Payment {payment_id} not found")
|
|
return False
|
|
|
|
for key, value in updates.items():
|
|
if hasattr(payment, key):
|
|
setattr(payment, key, value)
|
|
else:
|
|
logger.warning(f"Payment object has no attribute '{key}'")
|
|
|
|
self.db.commit()
|
|
logger.debug(f"Updated payment {payment_id} with {len(updates)} fields")
|
|
return True
|
|
except Exception as e:
|
|
self.db.rollback()
|
|
logger.error(f"Failed to update payment {payment_id}: {e}")
|
|
return False
|
|
|
|
def commit(self):
|
|
"""Commit the current database session."""
|
|
try:
|
|
self.db.commit()
|
|
except Exception as e:
|
|
self.db.rollback()
|
|
logger.error(f"Failed to commit database session: {e}")
|
|
raise
|
|
|
|
def rollback(self):
|
|
"""Rollback the current database session."""
|
|
try:
|
|
self.db.rollback()
|
|
except Exception as e:
|
|
logger.error(f"Failed to rollback database session: {e}")
|
|
raise
|
|
|