hotswap/delphi-database-v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement comprehensive CSV import system for legacy database migration

Browse Source 
- Added 5 new legacy models to app/models.py (FileType, FileNots, RolexV, FVarLkup, RVarLkup)
- Created app/import_legacy.py with import functions for all legacy tables:
  * Reference tables: TRNSTYPE, TRNSLKUP, FOOTERS, FILESTAT, EMPLOYEE, GRUPLKUP, FILETYPE, FVARLKUP, RVARLKUP
  * Core tables: ROLODEX, PHONE, ROLEX_V, FILES, FILES_R, FILES_V, FILENOTS, LEDGER, DEPOSITS, PAYMENTS
  * Specialized: PLANINFO, QDROS, PENSIONS and all pension-related tables
- Created app/sync_legacy_to_modern.py with sync functions to populate modern models from legacy data
- Updated admin routes in app/main.py:
  * Extended process_csv_import to support all new import types
  * Added /admin/sync endpoint for syncing legacy to modern models
  * Updated get_import_type_from_filename to recognize all CSV file patterns
- Enhanced app/templates/admin.html with:
  * Import Order Guide showing recommended import sequence
  * Sync to Modern Models section with confirmation dialog
  * Sync results display with detailed per-table statistics
  * Updated supported file formats list
- All import functions use batch processing (500 rows), proper error handling, and structured logging
- Sync functions maintain foreign key integrity and skip orphaned records with warnings
This commit is contained in:
HotSwapp
2025-10-08 09:41:38 -05:00
parent 2efbf14940
commit 4030dbd88e
6 changed files with 2545 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

207
app/main.py
View File

@@ -49,6 +49,8 @@ from .schemas import (
FilesListResponse,
LedgerListResponse,
)
from . import import_legacy
from . import sync_legacy_to_modern
# Load environment variables
load_dotenv()
@@ -237,45 +239,77 @@ app.mount("/static", StaticFiles(directory="static"), name="static")
def get_import_type_from_filename(filename: str) -> str:
"""
Determine import type based on filename pattern.
Determine import type based on filename pattern for legacy CSV files.
Args:
filename: Name of the uploaded CSV file
Returns:
Import type string (client, phone, case, transaction, document, payment)
Import type string matching the import function keys
"""
filename_upper = filename.upper()
# Strip extension and normalize
base = filename_upper.rsplit('.', 1)[0]
# Support files saved with explicit type prefixes (e.g., CLIENT_<uuid>.csv)
if base.startswith('CLIENT_'):
return 'client'
if base.startswith('PHONE_'):
# Reference tables
if 'TRNSTYPE' in base:
return 'trnstype'
if 'TRNSLKUP' in base:
return 'trnslkup'
if 'FOOTER' in base:
return 'footers'
if 'FILESTAT' in base:
return 'filestat'
if 'EMPLOYEE' in base:
return 'employee'
if 'GRUPLKUP' in base or 'GROUPLKUP' in base:
return 'gruplkup'
if 'FILETYPE' in base:
return 'filetype'
if 'FVARLKUP' in base:
return 'fvarlkup'
if 'RVARLKUP' in base:
return 'rvarlkup'
# Core data tables
if 'ROLEX_V' in base or 'ROLEXV' in base:
return 'rolex_v'
if 'ROLODEX' in base or 'ROLEX' in base:
return 'rolodex'
if 'FILES_R' in base or 'FILESR' in base:
return 'files_r'
if 'FILES_V' in base or 'FILESV' in base:
return 'files_v'
if 'FILENOTS' in base or 'FILE_NOTS' in base:
return 'filenots'
if 'FILES' in base or 'FILE' in base:
return 'files'
if 'PHONE' in base:
return 'phone'
if base.startswith('CASE_'):
return 'case'
if base.startswith('TRANSACTION_'):
return 'transaction'
if base.startswith('DOCUMENT_'):
return 'document'
if base.startswith('PAYMENT_'):
return 'payment'
# Legacy/real file name patterns
if base.startswith('ROLODEX') or base.startswith('ROLEX') or 'ROLODEX' in base or 'ROLEX' in base:
return 'client'
if base.startswith('PHONE') or 'PHONE' in base:
return 'phone'
if base.startswith('FILES') or base.startswith('FILE') or 'FILES' in base:
return 'case'
if base.startswith('LEDGER') or 'LEDGER' in base or base.startswith('TRNSACTN') or 'TRNSACTN' in base:
return 'transaction'
if base.startswith('QDROS') or base.startswith('QDRO') or 'QDRO' in base:
return 'document'
if base.startswith('PAYMENTS') or base.startswith('DEPOSITS') or 'PAYMENT' in base or 'DEPOSIT' in base:
return 'payment'
if 'LEDGER' in base:
return 'ledger'
if 'DEPOSITS' in base or 'DEPOSIT' in base:
return 'deposits'
if 'PAYMENTS' in base or 'PAYMENT' in base:
return 'payments'
# Specialized tables
if 'PLANINFO' in base or 'PLAN_INFO' in base:
return 'planinfo'
if 'QDROS' in base or 'QDRO' in base:
return 'qdros'
if 'MARRIAGE' in base:
return 'pension_marriage'
if 'DEATH' in base:
return 'pension_death'
if 'SCHEDULE' in base:
return 'pension_schedule'
if 'SEPARATE' in base:
return 'pension_separate'
if 'RESULTS' in base:
return 'pension_results'
if 'PENSIONS' in base or 'PENSION' in base:
return 'pensions'
raise ValueError(f"Unknown file type for filename: {filename}")
@@ -874,23 +908,49 @@ def import_payments_data(db: Session, file_path: str) -> Dict[str, Any]:
def process_csv_import(db: Session, import_type: str, file_path: str) -> Dict[str, Any]:
"""
Process CSV import based on type.
Process CSV import based on type using legacy import functions.
Args:
db: Database session
import_type: Type of import (client, phone, case, transaction, document, payment)
import_type: Type of import
file_path: Path to CSV file
Returns:
Dict with import results
"""
import_functions = {
'client': import_rolodex_data,
'phone': import_phone_data,
'case': import_files_data,
'transaction': import_ledger_data,
'document': import_qdros_data,
'payment': import_payments_data
# Reference tables (import first)
'trnstype': import_legacy.import_trnstype,
'trnslkup': import_legacy.import_trnslkup,
'footers': import_legacy.import_footers,
'filestat': import_legacy.import_filestat,
'employee': import_legacy.import_employee,
'gruplkup': import_legacy.import_gruplkup,
'filetype': import_legacy.import_filetype,
'fvarlkup': import_legacy.import_fvarlkup,
'rvarlkup': import_legacy.import_rvarlkup,
# Core data tables
'rolodex': import_legacy.import_rolodex,
'phone': import_legacy.import_phone,
'rolex_v': import_legacy.import_rolex_v,
'files': import_legacy.import_files,
'files_r': import_legacy.import_files_r,
'files_v': import_legacy.import_files_v,
'filenots': import_legacy.import_filenots,
'ledger': import_legacy.import_ledger,
'deposits': import_legacy.import_deposits,
'payments': import_legacy.import_payments,
# Specialized tables
'planinfo': import_legacy.import_planinfo,
'qdros': import_legacy.import_qdros,
'pensions': import_legacy.import_pensions,
'pension_marriage': import_legacy.import_pension_marriage,
'pension_death': import_legacy.import_pension_death,
'pension_schedule': import_legacy.import_pension_schedule,
'pension_separate': import_legacy.import_pension_separate,
'pension_results': import_legacy.import_pension_results,
}
import_func = import_functions.get(import_type)
@@ -1566,7 +1626,17 @@ async def admin_import_data(
return RedirectResponse(url="/login", status_code=302)
# Validate data type
valid_types = ['client', 'phone', 'case', 'transaction', 'document', 'payment']
valid_types = [
# Reference tables
'trnstype', 'trnslkup', 'footers', 'filestat', 'employee',
'gruplkup', 'filetype', 'fvarlkup', 'rvarlkup',
# Core data tables
'rolodex', 'phone', 'rolex_v', 'files', 'files_r', 'files_v',
'filenots', 'ledger', 'deposits', 'payments',
# Specialized tables
'planinfo', 'qdros', 'pensions', 'pension_marriage',
'pension_death', 'pension_schedule', 'pension_separate', 'pension_results'
]
if data_type not in valid_types:
return templates.TemplateResponse("admin.html", {
"request": request,
@@ -1670,6 +1740,69 @@ async def admin_import_data(
})
@app.post("/admin/sync")
async def admin_sync_data(
request: Request,
db: Session = Depends(get_db)
):
"""
Sync legacy database models to modern application models.
This route triggers the sync process to populate the simplified
modern models (Client, Phone, Case, Transaction, Payment, Document)
from the comprehensive legacy models.
"""
# Check authentication
user = get_current_user_from_session(request.session)
if not user:
return RedirectResponse(url="/login", status_code=302)
# Get form data for confirmation
form = await request.form()
clear_existing = form.get("clear_existing") == "true"
try:
logger.info(
"admin_sync_starting",
clear_existing=clear_existing,
username=user.username
)
# Run all sync functions
results = sync_legacy_to_modern.sync_all(db, clear_existing=clear_existing)
# Calculate totals
total_synced = sum(r['success'] for r in results.values() if r)
total_skipped = sum(r['skipped'] for r in results.values() if r)
total_errors = sum(len(r['errors']) for r in results.values() if r)
logger.info(
"admin_sync_complete",
total_synced=total_synced,
total_skipped=total_skipped,
total_errors=total_errors,
username=user.username
)
return templates.TemplateResponse("admin.html", {
"request": request,
"user": user,
"sync_results": results,
"total_synced": total_synced,
"total_skipped": total_skipped,
"total_sync_errors": total_errors,
"show_sync_results": True
})
except Exception as e:
logger.error("admin_sync_failed", error=str(e), username=user.username)
return templates.TemplateResponse("admin.html", {
"request": request,
"user": user,
"error": f"Sync failed: {str(e)}"
})
@app.get("/admin")
async def admin_panel(request: Request, db: Session = Depends(get_db)):
"""