"""Backfill purchase-flow vouchers and statuses for existing records.""" import argparse import asyncio import sys from decimal import Decimal from pathlib import Path from sqlalchemy import select sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from app.application.services.purchase_service import PurchaseService from app.application.services.voucher_service import VoucherService from app.core.database import close_database_connection, connect_to_database, get_session_factory from app.domain.entities.user_registration import UserRegistration from app.domain.enums import PurchasePaymentStatus, VendorBillStatus, VoucherStatus, VoucherType from app.infrastructure.db.models import CompanyModel, VoucherModel from app.infrastructure.repositories.chart_of_account_mysql_repository import ( MySQLChartOfAccountRepository, ) from app.infrastructure.repositories.company_mysql_repository import MySQLCompanyRepository from app.infrastructure.repositories.inventory_mysql_repository import MySQLInventoryRepository from app.infrastructure.repositories.purchase_mysql_repository import MySQLPurchaseRepository from app.infrastructure.repositories.voucher_mysql_repository import MySQLVoucherRepository def _system_user(user_id: str) -> UserRegistration: return UserRegistration( id=user_id, username="system", email="system@example.com", full_name="System Backfill", password_hash="", is_active=True, ) async def _voucher_exists_for_reference(session, company_id: str, reference: str) -> bool: result = await session.execute( select(VoucherModel.id).where( VoucherModel.company_id == company_id, VoucherModel.voucher_type == VoucherType.PURCHASE.value, VoucherModel.reference == reference, VoucherModel.status == VoucherStatus.POSTED.value, ) ) return result.scalar_one_or_none() is not None async def run_backfill(dry_run: bool = False) -> None: await connect_to_database() factory = get_session_factory() totals = { "companies": 0, "grn_vouchers_created": 0, "bill_vouchers_created": 0, "payment_vouchers_created": 0, "po_status_refreshed": 0, } async with factory() as session: company_rows = ( await session.execute( select(CompanyModel.company_id, CompanyModel.user_id).order_by(CompanyModel.created_at) ) ).all() totals["companies"] = len(company_rows) company_repo = MySQLCompanyRepository(session) account_repo = MySQLChartOfAccountRepository(session) voucher_repo = MySQLVoucherRepository(session) purchase_repo = MySQLPurchaseRepository(session) inventory_repo = MySQLInventoryRepository(session) voucher_service = VoucherService(voucher_repo, account_repo, company_repo) purchase_service = PurchaseService( purchase_repo, inventory_repo, account_repo, company_repo, voucher_service, ) for company_id, user_id in company_rows: user = _system_user(user_id) touched_po_ids: set[str] = set() print(f"\nCompany: {company_id}") # 1) Backfill GRN vouchers for already-confirmed receipts. receipts = await purchase_repo.list_goods_receipts( company_id=company_id, status=None, purchase_order_id=None, skip=0, limit=100000, ) for receipt in receipts: if receipt.status.value != "confirmed": continue if await _voucher_exists_for_reference(session, company_id, receipt.grn_number): continue if dry_run: print(f" [dry-run] GRN voucher -> {receipt.grn_number}") totals["grn_vouchers_created"] += 1 else: await purchase_service._post_grn_accounting_entry(user, company_id, receipt) totals["grn_vouchers_created"] += 1 print(f" GRN voucher created -> {receipt.grn_number}") # 2) Backfill vouchers on existing posted/paid bills where voucher_id is missing. bills = await purchase_repo.list_vendor_bills( company_id=company_id, status=None, vendor_id=None, skip=0, limit=100000, ) for bill in bills: if bill.voucher_id: continue if bill.status not in ( VendorBillStatus.POSTED, VendorBillStatus.PARTIALLY_PAID, VendorBillStatus.PAID, ): continue if bill.total_amount <= Decimal("0.00"): continue ap_account_id = bill.ap_account_id if not ap_account_id: vendor = await purchase_repo.get_vendor(bill.vendor_id, company_id) ap_account_id = vendor.account_id if vendor else None if not ap_account_id: print(f" Skipped bill {bill.bill_number}: missing AP account") continue debit_entries: list[dict] = [] grni_account = await account_repo.get_by_code(company_id, "2120") if bill.goods_receipt_id and grni_account and not grni_account.is_group: debit_entries.append( { "account_id": grni_account.id or "", "description": f"GRN accrual clear for bill {bill.bill_number}", "debit_amount": bill.total_amount, "credit_amount": Decimal("0.00"), } ) else: for line in bill.lines: account_id = line.account_id if not account_id and line.item_id: item = await inventory_repo.get_item(line.item_id, company_id) if item: account_id = item.inventory_account_id or item.expense_account_id if not account_id: print( f" Skipped bill {bill.bill_number}: no account for line {line.line_number}" ) debit_entries = [] break debit_entries.append( { "account_id": account_id, "description": line.description or f"Bill {bill.bill_number}", "debit_amount": line.line_total, "credit_amount": Decimal("0.00"), } ) if not debit_entries: continue aggregated: dict[str, dict] = {} for entry in debit_entries: key = entry["account_id"] if key not in aggregated: aggregated[key] = { "account_id": key, "description": entry["description"], "debit_amount": Decimal("0.00"), "credit_amount": Decimal("0.00"), } aggregated[key]["debit_amount"] += Decimal(str(entry["debit_amount"])) entries = list(aggregated.values()) entries.append( { "account_id": ap_account_id, "description": f"AP for bill {bill.bill_number}", "debit_amount": Decimal("0.00"), "credit_amount": bill.total_amount, } ) if dry_run: print(f" [dry-run] Bill voucher -> {bill.bill_number}") totals["bill_vouchers_created"] += 1 else: voucher = await voucher_service.create_voucher( user, company_id, { "voucher_type": VoucherType.PURCHASE.value, "voucher_date": bill.bill_date, "reference": bill.bill_number, "narration": bill.notes or f"Vendor bill {bill.bill_number}", "entries": entries, }, ) voucher = await voucher_service.post_voucher(user, company_id, voucher.id or "") bill.voucher_id = voucher.id bill.ap_account_id = ap_account_id bill.updated_at = bill.updated_at await purchase_repo.update_vendor_bill(bill.id or "", bill) totals["bill_vouchers_created"] += 1 print(f" Bill voucher created -> {bill.bill_number}") if bill.purchase_order_id: touched_po_ids.add(bill.purchase_order_id) # 3) Backfill vouchers on already-posted payments where voucher_id is missing. payments = await purchase_repo.list_payments( company_id=company_id, status=None, vendor_id=None, skip=0, limit=100000, ) for payment in payments: if payment.voucher_id: continue if payment.status != PurchasePaymentStatus.POSTED: continue if not payment.bank_account_id or not payment.ap_account_id: print(f" Skipped payment {payment.payment_number}: missing bank/AP account") continue if payment.total_amount <= Decimal("0.00"): continue if dry_run: print(f" [dry-run] Payment voucher -> {payment.payment_number}") totals["payment_vouchers_created"] += 1 else: voucher = await voucher_service.create_voucher( user, company_id, { "voucher_type": VoucherType.PAYMENT.value, "voucher_date": payment.payment_date, "reference": payment.payment_number, "narration": payment.notes or f"Purchase payment {payment.payment_number}", "entries": [ { "account_id": payment.ap_account_id, "description": f"AP payment {payment.payment_number}", "debit_amount": payment.total_amount, "credit_amount": Decimal("0.00"), }, { "account_id": payment.bank_account_id, "description": f"Bank/Cash for {payment.payment_number}", "debit_amount": Decimal("0.00"), "credit_amount": payment.total_amount, }, ], }, ) voucher = await voucher_service.post_voucher(user, company_id, voucher.id or "") payment.voucher_id = voucher.id await purchase_repo.update_payment(payment.id or "", payment) totals["payment_vouchers_created"] += 1 print(f" Payment voucher created -> {payment.payment_number}") for alloc in payment.allocations: bill = await purchase_repo.get_vendor_bill(alloc.vendor_bill_id, company_id) if bill and bill.purchase_order_id: touched_po_ids.add(bill.purchase_order_id) # 4) Refresh PO workflow status after backfill. if dry_run: totals["po_status_refreshed"] += len(touched_po_ids) else: for po_id in touched_po_ids: await purchase_service._refresh_purchase_order_workflow_status(company_id, po_id) totals["po_status_refreshed"] += 1 if touched_po_ids: print(f" Refreshed PO statuses: {len(touched_po_ids)}") await close_database_connection() print("\nBackfill complete") print(f"Companies scanned: {totals['companies']}") print(f"GRN vouchers created: {totals['grn_vouchers_created']}") print(f"Bill vouchers created: {totals['bill_vouchers_created']}") print(f"Payment vouchers created: {totals['payment_vouchers_created']}") print(f"PO statuses refreshed: {totals['po_status_refreshed']}") if __name__ == "__main__": parser = argparse.ArgumentParser(description="Backfill purchase vouchers/status for existing records") parser.add_argument("--dry-run", action="store_true", help="Show what would change without writing") args = parser.parse_args() asyncio.run(run_backfill(dry_run=args.dry_run))