""" Search blueprint for Plutus payment processing application. Provides unified search functionality across SinglePayments and Payments tables. Supports search by Splynx_ID and Payment Intent. """ from flask import Blueprint, render_template, request, jsonify from flask_login import login_required from sqlalchemy import or_, and_ from models import SinglePayments, Payments, Users from app import db from permissions import helpdesk_required import re search_bp = Blueprint('search', __name__, url_prefix='/search') @search_bp.route('/') @helpdesk_required def search_page(): """Display the unified payment search page.""" return render_template('search/search.html') @search_bp.route('/api', methods=['GET']) @helpdesk_required def search_payments(): """ API endpoint for searching payments. Query parameters: - q: Search query (Splynx_ID or Payment Intent) - type: Search type ('all', 'splynx_id', 'payment_intent') - limit: Maximum results to return (default: 50) """ try: query = request.args.get('q', '').strip() search_type = request.args.get('type', 'all') limit = min(int(request.args.get('limit', 50)), 100) # Max 100 results if not query: return jsonify({ 'success': False, 'error': 'Search query is required', 'results': [] }) # Determine search strategy based on query content if search_type == 'all': search_type = detect_search_type(query) # Build search queries for both tables single_payments_query = build_single_payments_query(query, search_type, limit) payments_query = build_payments_query(query, search_type, limit) # Execute searches single_payments_results = single_payments_query.all() payments_results = payments_query.all() # Format results formatted_results = [] # Add single payments results for payment in single_payments_results: formatted_results.append(format_single_payment_result(payment)) # Add batch payments results for payment in payments_results: formatted_results.append(format_batch_payment_result(payment)) # Sort by creation date (newest first) formatted_results.sort(key=lambda x: x['created'], reverse=True) # Apply limit to combined results formatted_results = formatted_results[:limit] return jsonify({ 'success': True, 'results': formatted_results, 'total_found': len(formatted_results), 'search_query': query, 'search_type': search_type }) except Exception as e: return jsonify({ 'success': False, 'error': str(e), 'results': [] }), 500 def detect_search_type(query: str) -> str: """ Detect the type of search based on query content. Args: query: Search query string Returns: str: 'payment_intent' if looks like PI, 'splynx_id' if numeric, 'all' otherwise """ # Check if it looks like a payment intent ID if re.match(r'^pi_[a-zA-Z0-9]+$', query): return 'payment_intent' # Check if it's purely numeric (likely Splynx ID) if query.isdigit(): return 'splynx_id' # Default to searching all fields return 'all' def build_single_payments_query(query: str, search_type: str, limit: int): """Build SQLAlchemy query for SinglePayments table.""" base_query = db.session.query(SinglePayments, Users.FullName).outerjoin( Users, SinglePayments.Who == Users.id ) if search_type == 'splynx_id': try: splynx_id = int(query) return base_query.filter(SinglePayments.Splynx_ID == splynx_id).limit(limit) except ValueError: return base_query.filter(False) # No results for invalid numeric elif search_type == 'payment_intent': return base_query.filter(SinglePayments.Payment_Intent == query).limit(limit) else: # search_type == 'all' # Try to convert to int for Splynx_ID search conditions = [] if query.isdigit(): conditions.append(SinglePayments.Splynx_ID == int(query)) conditions.extend([ SinglePayments.Payment_Intent == query, SinglePayments.Stripe_Customer_ID.like(f'%{query}%') ]) return base_query.filter(or_(*conditions)).limit(limit) def build_payments_query(query: str, search_type: str, limit: int): """Build SQLAlchemy query for Payments table.""" base_query = db.session.query(Payments) if search_type == 'splynx_id': try: splynx_id = int(query) return base_query.filter(Payments.Splynx_ID == splynx_id).limit(limit) except ValueError: return base_query.filter(False) # No results for invalid numeric elif search_type == 'payment_intent': return base_query.filter(Payments.Payment_Intent == query).limit(limit) else: # search_type == 'all' conditions = [] if query.isdigit(): conditions.append(Payments.Splynx_ID == int(query)) conditions.extend([ Payments.Payment_Intent == query, Payments.Stripe_Customer_ID.like(f'%{query}%') ]) return base_query.filter(or_(*conditions)).limit(limit) def format_single_payment_result(payment_data) -> dict: """Format a SinglePayment result for API response.""" payment, user_name = payment_data return { 'id': payment.id, 'type': 'single', 'splynx_id': payment.Splynx_ID, 'stripe_customer_id': payment.Stripe_Customer_ID, 'payment_intent': payment.Payment_Intent, 'payment_method': payment.Payment_Method, 'amount': payment.Payment_Amount, 'success': payment.Success, 'error': payment.Error, 'refund': payment.Refund, 'refund_followup': payment.Refund_FollowUp, 'pi_followup': payment.PI_FollowUp, 'created': payment.Created.isoformat(), 'processed_by': user_name or 'Unknown', 'detail_url': f'/single-payment/detail/{payment.id}', 'splynx_url': f'https://billing.interphone.com.au/admin/customers/view?id={payment.Splynx_ID}' if payment.Splynx_ID else None } def format_batch_payment_result(payment) -> dict: """Format a batch Payment result for API response.""" return { 'id': payment.id, 'type': 'batch', 'batch_id': payment.PaymentBatch_ID, 'splynx_id': payment.Splynx_ID, 'stripe_customer_id': payment.Stripe_Customer_ID, 'payment_intent': payment.Payment_Intent, 'payment_method': payment.Payment_Method, 'amount': payment.Payment_Amount, 'success': payment.Success, 'error': payment.Error, 'refund': payment.Refund, 'refund_followup': payment.Refund_FollowUp, 'pi_followup': payment.PI_FollowUp, 'created': payment.Created.isoformat(), 'processed_by': 'Batch Process', 'detail_url': f'/payment/detail/{payment.id}', 'batch_url': f'/batch/{payment.PaymentBatch_ID}', 'splynx_url': f'https://billing.interphone.com.au/admin/customers/view?id={payment.Splynx_ID}' if payment.Splynx_ID else None }