from typing import List, Optional from datetime import datetime from sqlalchemy.orm import Session from sqlalchemy import cast, Integer, exists, and_, func, desc from src.domain.entities.merchant import Merchant, MerchantListItem, MerchantLocation from src.domain.interfaces.merchant_repository import MerchantRepositoryInterface from src.infrastructure.models.merchant_model import MerchantModel from src.infrastructure.models.merchant_category_model import MerchantCategoryModel from src.infrastructure.models.merchant_module_model import MerchantModuleModel from src.infrastructure.models.digital_sign_model import DigitalSignDetailsModel from src.infrastructure.models.merchant_bank_model import MerchantBankModel from src.infrastructure.models.merchant_offer_model import MerchantOfferModel from src.infrastructure.models.merchant_files_model import MerchantFilesModel from src.infrastructure.models.merchant_location_model import MerchantLocationModel def _to_entity(m: MerchantModel) -> Merchant: return Merchant( id=m.id, category_id=m.category_id, merchant_name=m.merchant_name, slug=m.slug, merchant_logo=m.merchant_logo, financial_partner=m.financial_partner, contact_number=m.contact_number, email=m.email, address=m.address, grade_id=m.grade_id, status=m.status, max_no_cases=m.max_no_cases, max_approval_value=m.max_approval_value, hypothecation_charges=m.hypothecation_charges, status_changed_by=m.status_changed_by, deleted_at=m.deleted_at, created_at=m.created_at, created_by=m.created_by, updated_at=m.updated_at, type_of_firm=m.type_of_firm, nature_of_business=m.nature_of_business, owner_name=m.owner_name, title=m.title, gst_number=m.gst_number, is_processing_fees=m.is_processing_fees, is_collected=m.is_collected, code=m.code, referred_by=m.referred_by, rcp=m.rcp, sales_head=m.sales_head, finance_head=m.finance_head, avg_ticket_size=m.avg_ticket_size, rcp_email=m.rcp_email, rcp_mobile=m.rcp_mobile, sales_head_email=m.sales_head_email, sales_head_mobile=m.sales_head_mobile, finance_head_email=m.finance_head_email, finance_head_mobile=m.finance_head_mobile, is_guest_active=m.is_guest_active, is_gst=m.is_gst, state=m.state, anchor_merchant_id=m.anchor_merchant_id, pincode=m.pincode, source=m.source, gender=m.gender, dob=m.dob, pan_card=m.pan_card, is_auto_abort=m.is_auto_abort, city=m.city, co_applicant_hyphen=m.co_applicant_hyphen, co_applicant=m.co_applicant, gst_is_defaulter=m.gst_is_defaulter, gst_is_any_delay=m.gst_is_any_delay, months_gst_filled=m.months_gst_filled, gst_data=m.gst_data, is_physical_verification=m.is_physical_verification, is_auto_abort_approved=m.is_auto_abort_approved, rm_id=m.rm_id, is_auto_doc=m.is_auto_doc, is_auto_approve=m.is_auto_approve, high_risk=m.high_risk, vulnerable_merchant=m.vulnerable_merchant, vulnerable_merchant_date=m.vulnerable_merchant_date, pre_approval=m.pre_approval, send_collection_report=m.send_collection_report, pending_adjustable_amount=m.pending_adjustable_amount, pending_collection_report=m.pending_collection_report, special_notes=m.special_notes, longitude=m.longitude, latitude=m.latitude, year_in_business=m.year_in_business, location_audit=m.location_audit, liability_percent=m.liability_percent, insurance_premium=m.insurance_premium, insurance_collection_method=m.insurance_collection_method, location_serviceable=m.location_serviceable, user_banking_reason=m.user_banking_reason, pre_status=m.pre_status, charge_subvention_remaining_amount=m.charge_subvention_remaining_amount, welcome_letter_sent=m.welcome_letter_sent, send_da_on_checklist=m.send_da_on_checklist, target_number_application=m.target_number_application, target_value_application=m.target_value_application, custom_grade_id=m.custom_grade_id, remove_esign_otp=m.remove_esign_otp, pay_data_sharing=m.pay_data_sharing, official_merchant_name=m.official_merchant_name, collect_advance_amount=m.collect_advance_amount, is_fldg_active=m.is_fldg_active, is_trade_advance=m.is_trade_advance, trade_advance_amount=m.trade_advance_amount, trade_advance_balance=m.trade_advance_balance, agreement_signed_with=m.agreement_signed_with, agreement_signed_by=m.agreement_signed_by, merchant_reg_type=m.merchant_reg_type, designation_of_sign_auth=m.designation_of_sign_auth, available_exposure=m.available_exposure, skip_appl_otp=m.skip_appl_otp, auto_approve_category=m.auto_approve_category, combined_sl_la=m.combined_sl_la, ips=m.ips, mis_frequency=m.mis_frequency, create_app_lead_on_map=m.create_app_lead_on_map, subv_inv_on_center=m.subv_inv_on_center, processing_fee_on_advance_amount=m.processing_fee_on_advance_amount, hold_liability=m.hold_liability, total_revenue_per_month=m.total_revenue_per_month, percentage_of_sales_in_emi=m.percentage_of_sales_in_emi, number_of_lenders=m.number_of_lenders, number_of_visits=m.number_of_visits, landmark=m.landmark, table_of_offers=m.table_of_offers, send_da=m.send_da, grade_name=m.grade_name, is_hypothecation_active=m.is_hypothecation_active, remove_ip_whitelist=m.remove_ip_whitelist, pf_surcharge_active=m.pf_surcharge_active, kw_capping=m.kw_capping, disb_post_installation=m.disb_post_installation, receive_application=m.receive_application, ) class SqlMerchantRepository(MerchantRepositoryInterface): """SQLAlchemy implementation of MerchantRepositoryInterface.""" def __init__(self, db: Session): self.db = db def get_list(self) -> List[MerchantListItem]: def _file_exists(file_type: str): return ( exists() .where(and_( MerchantFilesModel.merchant_id == MerchantModel.id, MerchantFilesModel.type.is_(None), MerchantFilesModel.file_type == file_type, )) .correlate(MerchantModel) ) has_module_subq = ( exists() .where(and_( MerchantModuleModel.merchant_id == MerchantModel.id, MerchantModuleModel.deleted_at.is_(None), )) .correlate(MerchantModel) ) agreement_signed_subq = ( exists() .where(and_( DigitalSignDetailsModel.application_id == MerchantModel.id, DigitalSignDetailsModel.type == "merchant", DigitalSignDetailsModel.isagreementsigned.isnot(None), )) .correlate(MerchantModel) ) has_active_offer_subq = ( exists() .where(and_( MerchantOfferModel.merchant_id == MerchantModel.id, MerchantOfferModel.status == 1, MerchantOfferModel.deleted_at.is_(None), )) .correlate(MerchantModel) ) has_inactive_offer_subq = ( exists() .where(and_( MerchantOfferModel.merchant_id == MerchantModel.id, MerchantOfferModel.status == 0, MerchantOfferModel.deleted_at.is_(None), )) .correlate(MerchantModel) ) has_bank_subq = ( exists() .where(MerchantBankModel.merchant_id == MerchantModel.id) .correlate(MerchantModel) ) rows = ( self.db.query( MerchantModel.id, MerchantModel.merchant_name, MerchantModel.slug, MerchantModel.status, MerchantModel.pre_status, MerchantModel.receive_application, MerchantModel.code, MerchantModel.contact_number, MerchantModel.is_guest_active, MerchantCategoryModel.name.label("category_name"), has_module_subq.label("has_module"), agreement_signed_subq.label("is_agreement_signed"), _file_exists("pan_card").label("has_pan_card"), _file_exists("equifax_report").label("has_equifax"), _file_exists("gst_registration").label("has_gst_reg"), has_active_offer_subq.label("has_active_offer"), has_inactive_offer_subq.label("has_inactive_offer"), has_bank_subq.label("has_bank"), ) .outerjoin( MerchantCategoryModel, cast(MerchantModel.category_id, Integer) == MerchantCategoryModel.id, ) .filter( MerchantModel.deleted_at.is_(None), MerchantModel.is_guest_active == 1, ) .order_by(MerchantModel.merchant_name) .all() ) return [ MerchantListItem( id=row.id, merchant_name=row.merchant_name, category_name=row.category_name, slug=row.slug, status=row.status, pre_status=row.pre_status, receive_application=row.receive_application, code=row.code, contact_number=row.contact_number, is_guest_active=row.is_guest_active, has_module=bool(row.has_module), is_agreement_signed=bool(row.is_agreement_signed), has_pan_card=bool(row.has_pan_card), has_equifax=bool(row.has_equifax), has_gst_reg=bool(row.has_gst_reg), has_active_offer=bool(row.has_active_offer), has_inactive_offer=bool(row.has_inactive_offer), has_bank=bool(row.has_bank), ) for row in rows ] def get_all(self) -> List[Merchant]: models = self.db.query(MerchantModel).filter(MerchantModel.deleted_at.is_(None)).all() return [_to_entity(m) for m in models] def get_by_id(self, merchant_id: int) -> Optional[Merchant]: model = ( self.db.query(MerchantModel) .filter(MerchantModel.id == merchant_id, MerchantModel.deleted_at.is_(None)) .first() ) if not model: return None return _to_entity(model) def _to_location_entity(self, model: MerchantLocationModel) -> MerchantLocation: return MerchantLocation( id=model.id, merchant_id=model.merchant_id, name=model.name, slug=model.slug, email=model.email, phone=model.phone, address=model.location, city=None, # city column does not exist in merchant_location table state=model.state, pincode=None, # pincode column does not exist in merchant_location table gst_number=model.gst_number, landmark=model.landmark, total_revenue_per_month=model.total_revenue_per_month, percentage_of_sales_in_emi=model.percentage_of_sales_in_emi, number_of_lenders=model.number_of_lenders, number_of_visits=model.number_of_visits, created_at=model.created_at, updated_at=model.updated_at, ) def list_merchants(self, limit: int = 5, skip: int = 0) -> List[Merchant]: location_count = ( self.db.query( MerchantLocationModel.merchant_id, func.count(MerchantLocationModel.id).label("loc_count") ) .filter(MerchantLocationModel.deleted_at == None) .group_by(MerchantLocationModel.merchant_id) .subquery() ) models = ( self.db.query(MerchantModel) .outerjoin(location_count, MerchantModel.id == location_count.c.merchant_id) .filter(MerchantModel.deleted_at == None) .order_by(desc(func.coalesce(location_count.c.loc_count, 0))) .offset(skip) .limit(limit) .all() ) return [_to_entity(m) for m in models] def count_merchants(self) -> int: return self.db.query(MerchantModel).filter(MerchantModel.deleted_at == None).count() def get_merchant_by_id(self, merchant_id: int) -> Optional[Merchant]: model = self.db.query(MerchantModel).filter(MerchantModel.id == merchant_id).first() if not model: return None return _to_entity(model) def get_locations_by_merchant(self, merchant_id: int) -> List[MerchantLocation]: models = self.db.query(MerchantLocationModel).filter( MerchantLocationModel.merchant_id == merchant_id, MerchantLocationModel.deleted_at == None, ).all() return [self._to_location_entity(m) for m in models] def get_location_by_id(self, location_id: int) -> Optional[MerchantLocation]: model = self.db.query(MerchantLocationModel).filter( MerchantLocationModel.id == location_id, MerchantLocationModel.deleted_at == None, ).first() if not model: return None return self._to_location_entity(model) def create_location(self, location: MerchantLocation) -> MerchantLocation: model = MerchantLocationModel( merchant_id=location.merchant_id, name=location.name, slug=location.slug, email=location.email, phone=location.phone, location=location.address, gst_number=location.gst_number, state=location.state, landmark=location.landmark, total_revenue_per_month=location.total_revenue_per_month, percentage_of_sales_in_emi=location.percentage_of_sales_in_emi, number_of_lenders=location.number_of_lenders, number_of_visits=location.number_of_visits, created_at=datetime.now(), updated_at=datetime.now(), ) self.db.add(model) self.db.commit() self.db.refresh(model) return self._to_location_entity(model) def update_location(self, location_id: int, location_data: dict) -> Optional[MerchantLocation]: model = self.db.query(MerchantLocationModel).filter(MerchantLocationModel.id == location_id).first() if not model: return None for key, value in location_data.items(): if hasattr(model, key): setattr(model, key, value) model.updated_at = datetime.now() self.db.commit() self.db.refresh(model) return self._to_location_entity(model)