aetheropro/converter
0
1import asyncio2import json3import xml.etree.ElementTree as ET4import pandas as pd5import os6from pathlib import Path7import logging8from concurrent.futures import ProcessPoolExecutor9from typing import List, Dict, Any10import uuid11from datetime import datetime12from fastapi import FastAPI, UploadFile, File, HTTPException13from fastapi.responses import FileResponse14from fastapi.middleware.cors import CORSMiddleware15from pydantic import BaseModel16import tempfile17import shutil18 19logging.basicConfig(20 level=logging.INFO,21 format='%(asctime)s - %(levelname)s - %(message)s',22 handlers=[23 logging.FileHandler('/tmp/conversion.log'),24 logging.StreamHandler()25 ]26)27logger = logging.getLogger(__name__)28 29app = FastAPI(title="JSON/XML to Excel Converter API")30 31app.add_middleware(32 CORSMiddleware,33 allow_origins=["*"], # Add your frontend URL34 allow_credentials=True,35 allow_methods=["*"],36 allow_headers=["*"],37)38 39class ConversionRequest(BaseModel):40 input_dir: str | None = None41 42def parse_json_content(content: str, file_path: str) -> List[Dict[str, Any]]:43 """Parse JSON content and return structured data."""44 try:45 data = json.loads(content)46 claims = data.get('Submission', {}).get('Claim', [])47 if not isinstance(claims, list):48 claims = [claims]49 50 result = []51 payer_id = data.get('Submission', {}).get('Header', {}).get('PayerID', '')52 header = data.get('Submission', {}).get('Header', {})53 54 for claim in claims:55 base = {56 'FileName': os.path.basename(file_path),57 'SenderID': header.get('SenderID', ''),58 'ReceiverID': header.get('ReceiverID', ''),59 'TransactionDate': header.get('TransactionDate', ''),60 'RecordCount': header.get('RecordCount', ''),61 'DispositionFlag': header.get('DispositionFlag', ''),62 'ID': claim.get('ID', ''),63 'MemberID': claim.get('MemberID', ''),64 'PayerID': payer_id,65 'ProviderID': claim.get('ProviderID', ''),66 'Gross': claim.get('Gross', ''),67 'PatientShare': claim.get('PatientShare', ''),68 'Net': claim.get('Net', ''),69 'FacilityID': claim.get('Encounter', {}).get('FacilityID', ''),70 'Type': claim.get('Encounter', {}).get('Type', ''),71 'PatientID': claim.get('Encounter', {}).get('PatientID', ''),72 'Start': claim.get('Encounter', {}).get('Start', ''),73 'End': claim.get('Encounter', {}).get('End', ''),74 'StartType': claim.get('Encounter', {}).get('StartType', ''),75 'EndType': claim.get('Encounter', {}).get('EndType', ''),76 'TransferSource': '',77 'TransferDestination': ''78 }79 80 diagnoses = claim.get('Diagnosis', [])81 if not isinstance(diagnoses, list):82 diagnoses = [diagnoses]83 for diagnosis in diagnoses:84 diagnosis_data = {**base, 'Type2': diagnosis.get('Type', ''), 'Code': diagnosis.get('Code', '')}85 result.append(diagnosis_data)86 87 activities = claim.get('Activity', [])88 if not isinstance(activities, list):89 activities = [activities]90 for activity in activities:91 activity_data = {92 **base,93 'ID3': activity.get('ID', ''),94 'Start4': activity.get('Start', ''),95 'Type5': activity.get('Type', ''),96 'Code6': activity.get('Code', ''),97 'Quantity': activity.get('Quantity', ''),98 'Net7': activity.get('Net', ''),99 'Clinician': activity.get('Clinician', ''),100 'PriorAuthorizationID': activity.get('PriorAuthorizationID', '')101 }102 observations = activity.get('Observation', [])103 if not isinstance(observations, list):104 observations = [observations]105 for observation in observations:106 observation_data = {107 **activity_data,108 'Type8': observation.get('Type', ''),109 'Code9': observation.get('Code', ''),110 'Value': observation.get('Value', ''),111 'ValueType': observation.get('ValueType', '')112 }113 result.append(observation_data)114 115 return result116 117 except json.JSONDecodeError as e:118 logger.error(f"JSON parsing error in {file_path}: {str(e)}")119 return []120 except Exception as e:121 logger.error(f"Error processing JSON file {file_path}: {str(e)}")122 return []123 124async def process_json_file(file_path: str, executor: ProcessPoolExecutor) -> List[Dict[str, Any]]:125 """Process a single JSON file and return structured data."""126 try:127 loop = asyncio.get_event_loop()128 with open(file_path, 'r', encoding='utf-8') as f:129 content = f.read()130 return await loop.run_in_executor(executor, parse_json_content, content, file_path)131 132 except Exception as e:133 logger.error(f"Error reading JSON file {file_path}: {str(e)}")134 return []135 136async def process_xml_file(file_path: str, executor: ProcessPoolExecutor) -> List[Dict[str, Any]]:137 try:138 loop = asyncio.get_event_loop()139 tree = await loop.run_in_executor(executor, ET.parse, file_path)140 root = tree.getroot()141 142 data = []143 header = root.find('.//Header')144 if header is None:145 logger.error(f"No Header found in XML file {file_path}")146 return []147 148 base_header = {149 'FileName': os.path.basename(file_path),150 'SenderID': header.findtext('SenderID', ''),151 'ReceiverID': header.findtext('ReceiverID', ''),152 'TransactionDate': header.findtext('TransactionDate', ''),153 'RecordCount': header.findtext('RecordCount', ''),154 'DispositionFlag': header.findtext('DispositionFlag', '')155 }156 157 for claim in root.findall('.//Claim'):158 base_claim = {159 **base_header,160 'ID': claim.findtext('ID', ''),161 'MemberID': claim.findtext('MemberID', ''),162 'PayerID': claim.findtext('PayerID', ''),163 'ProviderID': claim.findtext('ProviderID', ''),164 'Gross': claim.findtext('Gross', ''),165 'PatientShare': claim.findtext('PatientShare', ''),166 'Net': claim.findtext('Net', ''),167 'FacilityID': claim.find('.//Encounter/FacilityID').text if claim.find('.//Encounter/FacilityID') is not None else '',168 'Type': claim.find('.//Encounter/Type').text if claim.find('.//Encounter/Type') is not None else '',169 'PatientID': claim.find('.//Encounter/PatientID').text if claim.find('.//Encounter/PatientID') is not None else '',170 'Start': claim.find('.//Encounter/Start').text if claim.find('.//Encounter/Start') is not None else '',171 'End': claim.find('.//Encounter/End').text if claim.find('.//Encounter/End') is not None else '',172 'StartType': claim.find('.//Encounter/StartType').text if claim.find('.//Encounter/StartType') is not None else '',173 'EndType': claim.find('.//Encounter/EndType').text if claim.find('.//Encounter/EndType') is not None else '',174 'TransferSource': '',175 'TransferDestination': ''176 }177 178 for diagnosis in claim.findall('.//Diagnosis'):179 diagnosis_data = {180 **base_claim,181 'Type2': diagnosis.findtext('Type', ''),182 'Code': diagnosis.findtext('Code', '')183 }184 data.append(diagnosis_data)185 186 for activity in claim.findall('.//Activity'):187 activity_data = {188 **base_claim,189 'ID3': activity.findtext('ID', ''),190 'Start4': activity.findtext('Start', ''),191 'Type5': activity.findtext('Type', ''),192 'Code6': activity.findtext('Code', ''),193 'Quantity': activity.findtext('Quantity', ''),194 'Net7': activity.findtext('Net', ''),195 'Clinician': activity.findtext('Clinician', ''),196 'PriorAuthorizationID': activity.findtext('PriorAuthorizationID', '')197 }198 for observation in activity.findall('.//Observation'):199 observation_data = {200 **activity_data,201 'Type8': observation.findtext('Type', ''),202 'Code9': observation.findtext('Code', ''),203 'Value': observation.findtext('Value', ''),204 'ValueType': observation.findtext('ValueType', '')205 }206 data.append(observation_data)207 208 return data209 210 except ET.ParseError as e:211 logger.error(f"XML parsing error in {file_path}: {str(e)}")212 return []213 except Exception as e:214 logger.error(f"Error processing XML file {file_path}: {str(e)}")215 return []216 217async def convert_to_excel(input_files: List[str], output_file: str, executor: ProcessPoolExecutor) -> tuple[int, List[str]]:218 try:219 data = []220 imported_files = 0221 error_files = []222 BATCH_SIZE = 100223 224 for i in range(0, len(input_files), BATCH_SIZE):225 batch = input_files[i:i + BATCH_SIZE]226 tasks = []227 for file in batch:228 if file.lower().endswith('.json'):229 tasks.append(process_json_file(file, executor))230 elif file.lower().endswith('.xml'):231 tasks.append(process_xml_file(file, executor))232 233 results = await asyncio.gather(*tasks, return_exceptions=True)234 235 for file, result in zip(batch, results):236 if isinstance(result, Exception):237 logger.error(f"Error processing {file}: {str(result)}")238 error_files.append(os.path.basename(file))239 elif result:240 data.extend(result)241 imported_files += 1242 else:243 error_files.append(os.path.basename(file))244 245 if data:246 df = pd.DataFrame(data)247 if len(df) > 1048576:248 sheets = []249 for i in range(0, len(df), 1048576):250 sheets.append(df[i:i + 1048576])251 with pd.ExcelWriter(output_file, engine='openpyxl') as writer:252 for idx, sheet in enumerate(sheets):253 sheet.to_excel(writer, sheet_name=f'Report_{idx + 1}', index=False)254 else:255 df.to_excel(output_file, sheet_name='Report', index=False)256 logger.info(f"Excel file saved: {output_file}")257 258 return imported_files, error_files259 260 except Exception as e:261 logger.error(f"Error during conversion: {str(e)}")262 return imported_files, error_files + [f"Critical error: {str(e)}"]263 264from fastapi import BackgroundTasks # Add this import265 266@app.post("/convert", response_class=FileResponse)267async def convert_files(268 input_dir: ConversionRequest | None = None,269 files: List[UploadFile] = File(None),270 background_tasks: BackgroundTasks = None # Add BackgroundTasks parameter271):272 temp_dir = None273 try:274 # Save output file to /tmp275 output_file = f"/tmp/report_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:8]}.xlsx"276 input_files = []277 278 if input_dir and input_dir.input_dir:279 input_path = Path(input_dir.input_dir)280 if not input_path.is_dir():281 raise HTTPException(status_code=400, detail="Invalid input directory")282 input_files = [str(f) for f in input_path.glob('*') if f.suffix.lower() in ('.json', '.xml')]283 284 elif files:285 temp_dir = tempfile.mkdtemp()286 for file in files:287 if not file.filename.lower().endswith(('.json', '.xml')):288 continue289 file_path = os.path.join(temp_dir, file.filename)290 with open(file_path, 'wb') as f:291 f.write(await file.read())292 input_files.append(file_path)293 294 if not input_files:295 raise HTTPException(status_code=400, detail="No valid JSON or XML files provided")296 297 with ProcessPoolExecutor() as executor:298 imported_files, error_files = await convert_to_excel(input_files, output_file, executor)299 300 if not os.path.exists(output_file):301 raise HTTPException(status_code=500, detail="Failed to generate Excel file")302 303 # Log completion304 logger.info(f"Conversion complete. Imported files: {imported_files}, Errors: {error_files if error_files else 'None'}")305 306 # Prepare the FileResponse307 response = FileResponse(308 output_file,309 filename=os.path.basename(output_file),310 media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'311 )312 313 # Schedule file deletion as a background task314 def delete_file(file_path: str):315 try:316 os.remove(file_path)317 logger.info(f"Deleted temporary Excel file: {file_path}")318 except Exception as e:319 logger.error(f"Failed to delete temporary Excel file {file_path}: {str(e)}")320 321 background_tasks.add_task(delete_file, output_file)322 323 return response324 325 except Exception as e:326 logger.error(f"API error: {str(e)}")327 raise HTTPException(status_code=500, detail=str(e))328 329 finally:330 if temp_dir and os.path.exists(temp_dir):331 shutil.rmtree(temp_dir, ignore_errors=True)332 333@app.get("/health")334async def health_check():335 """Health check endpoint."""336 return {"status": "ok", "message": "Converter API is running."}337 338@app.get("/")339async def root():340 """Root endpoint."""341 return {"message": "Welcome to the JSON/XML to Excel Converter API. Use /docs for API documentation."}