166 lines
5.5 KiB
Python
166 lines
5.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Main file processor for end-to-end ACH file processing.
|
|
Orchestrates download, parsing, mapping, and database insertion.
|
|
"""
|
|
|
|
import os
|
|
import tempfile
|
|
import gzip
|
|
import shutil
|
|
from pathlib import Path
|
|
from logging_config import get_logger
|
|
from upi_parser import UPIParser
|
|
from db.repository import Repository
|
|
from db.models import ProcessedFile
|
|
from sftp.sftp_client import SFTPClient
|
|
from .data_mapper import UPIDataMapper
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class FileProcessor:
|
|
|
|
|
|
def __init__(self, repository: Repository = None, sftp_client: SFTPClient = None):
|
|
self.repository = repository or Repository()
|
|
self.sftp_client = sftp_client or SFTPClient()
|
|
self.temp_dir = tempfile.gettempdir()
|
|
|
|
def process_file(
|
|
self,
|
|
filename: str,
|
|
bankcode: str,
|
|
remote_path: str
|
|
) -> bool:
|
|
|
|
local_path = os.path.join(self.temp_dir, filename)
|
|
|
|
try:
|
|
logger.info(f"Starting processing: {filename} (bank: {bankcode})")
|
|
|
|
# Step 1: Check if already processed
|
|
if self.repository.is_file_processed(filename, bankcode):
|
|
logger.info(f"File already processed for {bankcode}: {filename}")
|
|
return True
|
|
|
|
# Step 2: Download file
|
|
if not self.sftp_client.download_file(remote_path, local_path):
|
|
raise Exception(f"Failed to download file: {remote_path}")
|
|
|
|
# NEW: Handle .gz file
|
|
file_to_parse = local_path
|
|
|
|
if filename.lower().endswith(".gz"):
|
|
logger.info(f"Extracting GZ file: {filename}")
|
|
|
|
extracted_path = local_path[:-3] # remove .gz
|
|
|
|
with gzip.open(local_path, 'rb') as f_in:
|
|
with open(extracted_path, 'wb') as f_out:
|
|
shutil.copyfileobj(f_in, f_out)
|
|
|
|
file_to_parse = extracted_path
|
|
|
|
# Step 3: Parse file
|
|
parser = UPIParser(file_to_parse)
|
|
transactions, metadata, summary = parser.parse()
|
|
|
|
if not transactions:
|
|
logger.warning(f"No transactions found in {filename}")
|
|
|
|
processed_file = ProcessedFile(
|
|
filename=filename, # ORIGINAL .gz NAME
|
|
bankcode=bankcode,
|
|
file_path=remote_path,
|
|
transaction_count=0,
|
|
status='SUCCESS'
|
|
)
|
|
self.repository.mark_file_processed(processed_file)
|
|
return True
|
|
|
|
# Step 4: Map transactions
|
|
mapped_records = UPIDataMapper.map_transactions(transactions, bankcode)
|
|
|
|
# Step 5: Insert to database
|
|
inserted_count, skipped_count = self.repository.bulk_insert_transactions(mapped_records)
|
|
|
|
# Step 6: Mark processed
|
|
processed_file = ProcessedFile(
|
|
filename=filename, # ORIGINAL .gz NAME
|
|
bankcode=bankcode,
|
|
file_path=remote_path,
|
|
transaction_count=inserted_count,
|
|
status='SUCCESS'
|
|
)
|
|
self.repository.mark_file_processed(processed_file)
|
|
|
|
logger.info(
|
|
f"Successfully processed {filename}: "
|
|
f"{inserted_count} inserted, {skipped_count} skipped (non-ipks accounts)"
|
|
)
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error processing {filename}: {e}", exc_info=True)
|
|
|
|
try:
|
|
processed_file = ProcessedFile(
|
|
filename=filename,
|
|
bankcode=bankcode,
|
|
file_path=remote_path,
|
|
transaction_count=0,
|
|
status='FAILED',
|
|
error_message=str(e)[:2000]
|
|
)
|
|
self.repository.mark_file_processed(processed_file)
|
|
except Exception as mark_error:
|
|
logger.error(f"Failed to mark file as failed: {mark_error}")
|
|
|
|
return False
|
|
|
|
finally:
|
|
# UPDATED CLEANUP (handles both .gz + extracted file)
|
|
try:
|
|
if os.path.exists(local_path):
|
|
os.remove(local_path)
|
|
|
|
if 'file_to_parse' in locals() and file_to_parse != local_path:
|
|
if os.path.exists(file_to_parse):
|
|
os.remove(file_to_parse)
|
|
|
|
logger.debug(f"Cleanup done for {filename}")
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Error cleaning up local file {local_path}: {e}")
|
|
|
|
def process_files(self, files_to_process: list) -> dict:
|
|
|
|
stats = {
|
|
'total': len(files_to_process),
|
|
'successful': 0,
|
|
'failed': 0,
|
|
'files': []
|
|
}
|
|
|
|
for filename, bankcode, remote_path in files_to_process:
|
|
success = self.process_file(filename, bankcode, remote_path)
|
|
stats['successful'] += 1 if success else 0
|
|
stats['failed'] += 0 if success else 1
|
|
stats['files'].append({
|
|
'filename': filename,
|
|
'bankcode': bankcode,
|
|
'success': success
|
|
})
|
|
|
|
logger.info(f"Processing complete: {stats['successful']}/{stats['total']} successful")
|
|
return stats
|
|
|
|
def __enter__(self):
|
|
if self.sftp_client and not self.sftp_client.sftp:
|
|
self.sftp_client.connect()
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
if self.sftp_client:
|
|
self.sftp_client.disconnect() |