Files
delphi-database-v2/app/models.py
HotSwapp 4030dbd88e Implement comprehensive CSV import system for legacy database migration
- 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
2025-10-08 09:41:38 -05:00

751 lines
23 KiB
Python

"""
SQLAlchemy models for the Delphi database.
All models inherit from Base which is configured in the database module.
"""
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, Float, Text, Boolean, Date, Numeric, Index, UniqueConstraint, ForeignKeyConstraint
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql import func
# Create Base for SQLAlchemy 1.x compatibility
Base = declarative_base()
class User(Base):
"""
User model for authentication.
Stores user credentials and basic information for login functionality.
"""
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String(50), unique=True, index=True, nullable=False)
password_hash = Column(String(255), nullable=False)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
def __repr__(self):
return f"<User(id={self.id}, username='{self.username}')>"
class Client(Base):
"""
Client model representing individuals or entities.
Core client information imported from ROLODEX data.
"""
__tablename__ = "clients"
id = Column(Integer, primary_key=True, index=True)
rolodex_id = Column(String(20), unique=True, index=True)
last_name = Column(String(50))
first_name = Column(String(50))
middle_initial = Column(String(10))
company = Column(String(100))
address = Column(String(255))
city = Column(String(50))
state = Column(String(2))
zip_code = Column(String(10))
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
# Relationships
phones = relationship("Phone", back_populates="client")
cases = relationship("Case", back_populates="client")
def __repr__(self):
return f"<Client(id={self.id}, name='{self.first_name} {self.last_name}')>"
class Phone(Base):
"""
Phone number model linked to clients.
Stores phone number information for clients.
"""
__tablename__ = "phones"
id = Column(Integer, primary_key=True, index=True)
client_id = Column(Integer, ForeignKey("clients.id"), nullable=False)
phone_type = Column(String(20)) # home, work, mobile, fax, etc.
phone_number = Column(String(20))
extension = Column(String(10))
created_at = Column(DateTime(timezone=True), server_default=func.now())
# Relationships
client = relationship("Client", back_populates="phones")
def __repr__(self):
return f"<Phone(id={self.id}, number='{self.phone_number}')>"
class Case(Base):
"""
Case model representing legal cases or files.
Main case information imported from FILES data.
"""
__tablename__ = "cases"
id = Column(Integer, primary_key=True, index=True)
file_no = Column(String(20), unique=True, index=True, nullable=False)
client_id = Column(Integer, ForeignKey("clients.id"), nullable=False)
status = Column(String(20), default="active")
case_type = Column(String(50))
description = Column(Text)
open_date = Column(DateTime(timezone=True))
close_date = Column(DateTime(timezone=True))
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
# Relationships
client = relationship("Client", back_populates="cases")
transactions = relationship("Transaction", back_populates="case")
documents = relationship("Document", back_populates="case")
payments = relationship("Payment", back_populates="case")
def __repr__(self):
return f"<Case(id={self.id}, file_no='{self.file_no}')>"
class Transaction(Base):
"""
Transaction model for financial transactions.
Records financial activities related to cases.
"""
__tablename__ = "transactions"
id = Column(Integer, primary_key=True, index=True)
case_id = Column(Integer, ForeignKey("cases.id"), nullable=False)
transaction_date = Column(DateTime(timezone=True))
# Legacy/basic fields
transaction_type = Column(String(20)) # Maps to legacy T_Type
amount = Column(Float)
description = Column(Text) # Maps to legacy Note
reference = Column(String(50)) # Previously used for Item_No
# Ledger-specific fields (added for File Cabinet MVP)
item_no = Column(Integer)
employee_number = Column(String(20)) # Empl_Num
t_code = Column(String(10)) # T_Code
t_type_l = Column(String(1)) # T_Type_L (Credit/Debit marker)
quantity = Column(Float)
rate = Column(Float)
billed = Column(String(1)) # 'Y' or 'N'
created_at = Column(DateTime(timezone=True), server_default=func.now())
# Relationships
case = relationship("Case", back_populates="transactions")
def __repr__(self):
return f"<Transaction(id={self.id}, amount={self.amount})>"
class Document(Base):
"""
Document model for case-related documents.
Stores information about documents associated with cases.
"""
__tablename__ = "documents"
id = Column(Integer, primary_key=True, index=True)
case_id = Column(Integer, ForeignKey("cases.id"), nullable=False)
document_type = Column(String(50))
file_name = Column(String(255))
file_path = Column(String(500))
description = Column(Text)
uploaded_date = Column(DateTime(timezone=True))
created_at = Column(DateTime(timezone=True), server_default=func.now())
# Relationships
case = relationship("Case", back_populates="documents")
def __repr__(self):
return f"<Document(id={self.id}, file_name='{self.file_name}')>"
class Payment(Base):
"""
Payment model for payment records.
Records payments made or received for cases.
"""
__tablename__ = "payments"
id = Column(Integer, primary_key=True, index=True)
case_id = Column(Integer, ForeignKey("cases.id"), nullable=False)
payment_date = Column(DateTime(timezone=True))
payment_type = Column(String(20))
amount = Column(Float)
description = Column(Text)
check_number = Column(String(20))
created_at = Column(DateTime(timezone=True), server_default=func.now())
# Relationships
case = relationship("Case", back_populates="payments")
def __repr__(self):
return f"<Payment(id={self.id}, amount={self.amount})>"
class ImportLog(Base):
"""
ImportLog model for tracking CSV import operations.
Records the history and results of bulk data imports from legacy CSV files.
"""
__tablename__ = "import_logs"
id = Column(Integer, primary_key=True, index=True)
import_type = Column(String(50), nullable=False) # client, phone, case, transaction, document, payment
file_name = Column(String(255), nullable=False)
file_path = Column(String(500), nullable=False)
status = Column(String(20), default="pending") # pending, running, completed, failed
total_rows = Column(Integer, default=0)
processed_rows = Column(Integer, default=0)
success_count = Column(Integer, default=0)
error_count = Column(Integer, default=0)
error_details = Column(Text) # JSON string of error details
started_at = Column(DateTime(timezone=True), server_default=func.now())
completed_at = Column(DateTime(timezone=True))
created_at = Column(DateTime(timezone=True), server_default=func.now())
def __repr__(self):
return f"<ImportLog(id={self.id}, type='{self.import_type}', status='{self.status}')>"
# -----------------------------
# Legacy schema models (read-only/migration support)
# Derived from docs/legacy-schema.md
# -----------------------------
class Rolodex(Base):
"""
Legacy ROLODEX master table.
Primary key is the human-readable `Id`.
"""
__tablename__ = "rolodex"
id = Column(String, primary_key=True, index=True) # Id (TEXT)
prefix = Column(String)
first = Column(String)
middle = Column(String)
last = Column(String, index=True)
suffix = Column(String)
title = Column(String)
a1 = Column(String)
a2 = Column(String)
a3 = Column(String)
city = Column(String)
abrev = Column(String(2)) # state abbreviation
st = Column(String) # state name
zip = Column(String(10))
email = Column(String)
dob = Column(Date)
ss = Column(String) # SS#
legal_status = Column(String)
group = Column(String)
memo = Column(Text)
__table_args__ = (
Index("ix_rolodex_last_first", "last", "first"),
Index("ix_rolodex_group", "group"),
Index("ix_rolodex_email", "email"),
)
class LegacyPhone(Base):
"""
Legacy PHONE table (phones by Rolodex Id).
Composite PK on (id, phone) to prevent duplicates per person.
"""
__tablename__ = "phone"
id = Column(String, ForeignKey("rolodex.id", ondelete="CASCADE"), primary_key=True, index=True)
phone = Column(String, primary_key=True)
location = Column(String)
__table_args__ = (
Index("ix_phone_id", "id"),
)
class TrnsType(Base):
"""TRNSTYPE (transaction groups)."""
__tablename__ = "trnstype"
t_type = Column(String, primary_key=True)
t_type_l = Column(String)
header = Column(String)
footer = Column(String)
class TrnsLkup(Base):
"""TRNSLKUP (transaction codes)."""
__tablename__ = "trnslkup"
t_code = Column(String, primary_key=True)
t_type = Column(String, ForeignKey("trnstype.t_type"), index=True)
t_type_l = Column(String)
amount = Column(Numeric(12, 2))
description = Column(Text)
class Footers(Base):
"""FOOTERS table for footer codes and labels."""
__tablename__ = "footers"
f_code = Column(String, primary_key=True)
f_footer = Column(Text)
class FileStat(Base):
"""FILESTAT table for file statuses."""
__tablename__ = "filestat"
status = Column(String, primary_key=True)
definition = Column(Text)
send = Column(String)
footer_code = Column(String, ForeignKey("footers.f_code"), index=True)
class Employee(Base):
"""EMPLOYEE table linking employees to `Rolodex` and rate."""
__tablename__ = "employee"
empl_num = Column(String, primary_key=True)
empl_id = Column(String, ForeignKey("rolodex.id"), index=True)
rate_per_hour = Column(Numeric(12, 2))
class States(Base):
"""STATES reference table."""
__tablename__ = "states"
abrev = Column(String(2), primary_key=True)
st = Column(String)
class GroupLkup(Base):
"""GRUPLKUP reference table."""
__tablename__ = "gruplkup"
code = Column(String, primary_key=True)
description = Column(Text)
title = Column(String)
class Printers(Base):
"""PRINTERS configuration table."""
__tablename__ = "printers"
number = Column(Integer, primary_key=True)
name = Column(String)
port = Column(String)
page_break = Column(String)
setup_st = Column(String)
phone_book = Column(String)
rolodex_info = Column(String)
envelope = Column(String)
file_cabinet = Column(String)
accounts = Column(String)
statements = Column(String)
calendar = Column(String)
reset_st = Column(String)
b_underline = Column(String)
e_underline = Column(String)
b_bold = Column(String)
e_bold = Column(String)
class Setup(Base):
"""
SETUP application configuration. Not strictly keyed in legacy; introduce surrogate PK.
"""
__tablename__ = "setup"
id = Column(Integer, primary_key=True, autoincrement=True)
appl_title = Column(String)
l_head1 = Column(String)
l_head2 = Column(String)
l_head3 = Column(String)
l_head4 = Column(String)
l_head5 = Column(String)
l_head6 = Column(String)
l_head7 = Column(String)
l_head8 = Column(String)
l_head9 = Column(String)
l_head10 = Column(String)
default_printer = Column(Integer, ForeignKey("printers.number"), index=True)
class LegacyFile(Base):
"""
FILES (file cabinet) primary table.
"""
__tablename__ = "files"
file_no = Column(String, primary_key=True, index=True)
id = Column(String, ForeignKey("rolodex.id"), index=True)
file_type = Column(String, index=True)
regarding = Column(Text)
opened = Column(Date)
closed = Column(Date)
empl_num = Column(String, ForeignKey("employee.empl_num"), index=True)
rate_per_hour = Column(Numeric(12, 2))
status = Column(String, ForeignKey("filestat.status"), index=True)
footer_code = Column(String, ForeignKey("footers.f_code"), index=True)
opposing = Column(String, ForeignKey("rolodex.id"), index=True)
hours = Column(Numeric(12, 2))
hours_p = Column(Numeric(12, 2))
trust_bal = Column(Numeric(12, 2))
trust_bal_p = Column(Numeric(12, 2))
hourly_fees = Column(Numeric(12, 2))
hourly_fees_p = Column(Numeric(12, 2))
flat_fees = Column(Numeric(12, 2))
flat_fees_p = Column(Numeric(12, 2))
disbursements = Column(Numeric(12, 2))
disbursements_p = Column(Numeric(12, 2))
credit_bal = Column(Numeric(12, 2))
credit_bal_p = Column(Numeric(12, 2))
total_charges = Column(Numeric(12, 2))
total_charges_p = Column(Numeric(12, 2))
amount_owing = Column(Numeric(12, 2))
amount_owing_p = Column(Numeric(12, 2))
transferable = Column(Numeric(12, 2))
memo = Column(Text)
__table_args__ = (
Index("ix_files_id", "id"),
Index("ix_files_opposing", "opposing"),
Index("ix_files_status", "status"),
Index("ix_files_type", "file_type"),
)
class FilesR(Base):
"""FILES_R relationships per file."""
__tablename__ = "files_r"
file_no = Column(String, ForeignKey("files.file_no", ondelete="CASCADE"), primary_key=True)
relationship = Column(String, primary_key=True)
rolodex_id = Column(String, ForeignKey("rolodex.id", ondelete="CASCADE"), primary_key=True)
__table_args__ = (
Index("ix_files_r_rolodex_id", "rolodex_id"),
)
class FilesV(Base):
"""FILES_V variables per file."""
__tablename__ = "files_v"
file_no = Column(String, ForeignKey("files.file_no", ondelete="CASCADE"), primary_key=True)
identifier = Column(String, primary_key=True)
response = Column(Text)
class Ledger(Base):
"""LEDGER entries for time/charges per file."""
__tablename__ = "ledger"
file_no = Column(String, ForeignKey("files.file_no", ondelete="CASCADE"), primary_key=True)
date = Column(Date, index=True)
item_no = Column(Integer, primary_key=True)
empl_num = Column(String, ForeignKey("employee.empl_num"), index=True)
t_code = Column(String, ForeignKey("trnslkup.t_code"), index=True)
t_type = Column(String, ForeignKey("trnstype.t_type"), index=True)
t_type_l = Column(String)
quantity = Column(Numeric(12, 2))
rate = Column(Numeric(12, 2))
amount = Column(Numeric(12, 2))
billed = Column(String(1)) # 'Y' or 'N'
note = Column(Text)
__table_args__ = (
Index("ix_ledger_file_date", "file_no", "date"),
)
class Deposits(Base):
"""DEPOSITS daily totals."""
__tablename__ = "deposits"
deposit_date = Column(Date, primary_key=True)
total = Column(Numeric(12, 2))
class LegacyPayment(Base):
"""PAYMENTS legacy payments (separate from modern `payments`)."""
__tablename__ = "payments_legacy"
id = Column(Integer, primary_key=True, autoincrement=True)
deposit_date = Column(Date, ForeignKey("deposits.deposit_date"), index=True)
file_no = Column(String, ForeignKey("files.file_no"), index=True)
rolodex_id = Column(String, ForeignKey("rolodex.id"), index=True)
regarding = Column(Text)
amount = Column(Numeric(12, 2))
note = Column(Text)
class PlanInfo(Base):
"""PLANINFO reference table."""
__tablename__ = "planinfo"
plan_id = Column(String, primary_key=True)
plan_name = Column(String)
plan_type = Column(String)
empl_id_no = Column(String)
plan_no = Column(String)
nra = Column(String)
era = Column(String)
errf = Column(String)
colas = Column(String)
divided_by = Column(String)
drafted = Column(String)
benefit_c = Column(String)
qdro_c = Column(String)
rev = Column(String) # ^REV
pa = Column(String) # ^PA
form_name = Column(String)
drafted_on = Column(Date)
memo = Column(Text)
class Qdros(Base):
"""QDROS table for QDRO case data."""
__tablename__ = "qdros"
file_no = Column(String, ForeignKey("files.file_no"), primary_key=True)
version = Column(String, primary_key=True)
plan_id = Column(String, ForeignKey("planinfo.plan_id"), index=True)
_1 = Column(String) # ^1
_2 = Column(String) # ^2
part = Column(String) # ^Part
altp = Column(String) # ^AltP
pet = Column(String) # ^Pet
res = Column(String) # ^Res
case_type = Column(String)
case_code = Column(String)
section = Column(String)
case_number = Column(String)
judgment_date = Column(Date)
valuation_date = Column(Date)
married_on = Column(Date)
percent_awarded = Column(Numeric(12, 2))
ven_city = Column(String)
ven_cnty = Column(String)
ven_st = Column(String(2))
draft_out = Column(Date)
draft_apr = Column(Date)
final_out = Column(Date)
judge = Column(String)
form_name = Column(String)
class Pensions(Base):
"""PENSIONS primary table; composite key (file_no, version)."""
__tablename__ = "pensions"
file_no = Column(String, ForeignKey("files.file_no"), primary_key=True)
version = Column(String, primary_key=True)
plan_id = Column(String, ForeignKey("planinfo.plan_id"), index=True)
plan_name = Column(String)
title = Column(String)
first = Column(String)
last = Column(String)
birth = Column(Date)
race = Column(String)
sex = Column(String)
info = Column(Date)
valu = Column(Date)
accrued = Column(Numeric(12, 2))
vested_per = Column(Numeric(12, 2))
start_age = Column(Numeric(12, 2))
cola = Column(Numeric(12, 2))
max_cola = Column(Numeric(12, 2))
withdrawal = Column(Numeric(12, 2))
pre_dr = Column(Numeric(12, 2))
post_dr = Column(Numeric(12, 2))
tax_rate = Column(Numeric(12, 2))
class PensionResults(Base):
"""RESULTS derived values per pension (by file_no, version)."""
__tablename__ = "pension_results"
file_no = Column(String, primary_key=True)
version = Column(String, primary_key=True)
accrued = Column(Numeric(12, 2))
start_age = Column(Numeric(12, 2))
cola = Column(Numeric(12, 2))
withdrawal = Column(Numeric(12, 2))
pre_dr = Column(Numeric(12, 2))
post_dr = Column(Numeric(12, 2))
tax_rate = Column(Numeric(12, 2))
age = Column(Numeric(12, 2))
years_from = Column(Numeric(12, 2))
life_exp = Column(Numeric(12, 2))
ev_monthly = Column(Numeric(12, 2))
payments = Column(Numeric(12, 2))
pay_out = Column(Numeric(12, 2))
fund_value = Column(Numeric(12, 2))
pv = Column(Numeric(12, 2))
mortality = Column(Numeric(12, 2))
pv_am = Column(Numeric(12, 2))
pv_amt = Column(Numeric(12, 2))
pv_pre_db = Column(Numeric(12, 2))
pv_annuity = Column(Numeric(12, 2))
wv_at = Column(Numeric(12, 2))
pv_plan = Column(Numeric(12, 2))
years_married = Column(Numeric(12, 2))
years_service = Column(Numeric(12, 2))
marr_per = Column(Numeric(12, 2))
marr_amt = Column(Numeric(12, 2))
__table_args__ = (
ForeignKeyConstraint(["file_no", "version"], ["pensions.file_no", "pensions.version"], ondelete="CASCADE"),
Index("ix_pension_results_file_version", "file_no", "version"),
)
class PensionMarriage(Base):
"""MARRIAGE periods related to pensions."""
__tablename__ = "pension_marriage"
id = Column(Integer, primary_key=True, autoincrement=True)
file_no = Column(String, nullable=False)
version = Column(String, nullable=False)
married_from = Column(Date)
married_to = Column(Date)
married_years = Column(Numeric(12, 2))
service_from = Column(Date)
service_to = Column(Date)
service_years = Column(Numeric(12, 2))
marital_pct = Column(Numeric(12, 2))
__table_args__ = (
ForeignKeyConstraint(["file_no", "version"], ["pensions.file_no", "pensions.version"], ondelete="CASCADE"),
Index("ix_pension_marriage_file_version", "file_no", "version"),
)
class PensionDeath(Base):
"""DEATH related amounts for pensions."""
__tablename__ = "pension_death"
file_no = Column(String, primary_key=True)
version = Column(String, primary_key=True)
lump1 = Column(Numeric(12, 2))
lump2 = Column(Numeric(12, 2))
growth1 = Column(Numeric(12, 2))
growth2 = Column(Numeric(12, 2))
disc1 = Column(Numeric(12, 2))
disc2 = Column(Numeric(12, 2))
__table_args__ = (
ForeignKeyConstraint(["file_no", "version"], ["pensions.file_no", "pensions.version"], ondelete="CASCADE"),
)
class PensionSchedule(Base):
"""SCHEDULE vesting schedule for pensions."""
__tablename__ = "pension_schedule"
file_no = Column(String, primary_key=True)
version = Column(String, primary_key=True)
vests_on = Column(Date)
vests_at = Column(Numeric(12, 2))
__table_args__ = (
ForeignKeyConstraint(["file_no", "version"], ["pensions.file_no", "pensions.version"], ondelete="CASCADE"),
)
class PensionSeparate(Base):
"""SEPARATE calculations for pensions."""
__tablename__ = "pension_separate"
file_no = Column(String, primary_key=True)
version = Column(String, primary_key=True)
separation_rate = Column(Numeric(12, 2))
__table_args__ = (
ForeignKeyConstraint(["file_no", "version"], ["pensions.file_no", "pensions.version"], ondelete="CASCADE"),
)
class FileType(Base):
"""FILETYPE reference table for file/case types."""
__tablename__ = "filetype"
file_type = Column(String, primary_key=True)
def __repr__(self):
return f"<FileType(file_type='{self.file_type}')>"
class FileNots(Base):
"""FILENOTS table for file memos/notes."""
__tablename__ = "filenots"
file_no = Column(String, ForeignKey("files.file_no", ondelete="CASCADE"), primary_key=True)
memo_date = Column(Date, primary_key=True)
memo_note = Column(Text)
__table_args__ = (
Index("ix_filenots_file_no", "file_no"),
)
def __repr__(self):
return f"<FileNots(file_no='{self.file_no}', date='{self.memo_date}')>"
class RolexV(Base):
"""ROLEX_V variables per rolodex entry."""
__tablename__ = "rolex_v"
id = Column(String, ForeignKey("rolodex.id", ondelete="CASCADE"), primary_key=True)
identifier = Column(String, primary_key=True)
response = Column(Text)
__table_args__ = (
Index("ix_rolex_v_id", "id"),
)
def __repr__(self):
return f"<RolexV(id='{self.id}', identifier='{self.identifier}')>"
class FVarLkup(Base):
"""FVARLKUP file variable lookup table."""
__tablename__ = "fvarlkup"
identifier = Column(String, primary_key=True)
query = Column(Text)
response = Column(Text)
def __repr__(self):
return f"<FVarLkup(identifier='{self.identifier}')>"
class RVarLkup(Base):
"""RVARLKUP rolodex variable lookup table."""
__tablename__ = "rvarlkup"
identifier = Column(String, primary_key=True)
query = Column(Text)
def __repr__(self):
return f"<RVarLkup(identifier='{self.identifier}')>"