from fastapi import FastAPI, HTTPException, Request, File from starlette.datastructures import UploadFile import aiofiles from starlette.background import BackgroundTask import importlib.util from fastapi.responses import FileResponse import os, traceback, inspect import zipfile from typing import Any, Dict import shutil from tool import notebook_executor from datetime import datetime from script.bounce_penal_update import function1 as bounce_penal_update_function from script.overdue_update import function1 as overdue_update_function UPLOAD_DIR = "pre_part_payment_uploads" os.makedirs(UPLOAD_DIR, exist_ok=True) app = FastAPI() # Dynamic route where is the module and is the function @app.post("/api/{module_name}") async def dynamic_api(module_name: str, request: Request): try: if module_name == 'pre_part_payment': print('In if block of pre_part_payment.') # form = await request.form() # timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") # original_filename = form.get('file').filename # new_filename = f"{timestamp}_{original_filename}" # file_path = os.path.join(UPLOAD_DIR, new_filename) # with open(file_path, "wb") as f: # contents = await form.get('file').read() # f.write(contents) # data = dict(form) # data["uploaded_file_path"] = file_path form = await request.form() # Handle the uploaded file uploaded_file = form.get('file') print('File get called.') timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") original_filename = uploaded_file.filename new_filename = f"{timestamp}_{original_filename}" file_path = os.path.join(UPLOAD_DIR, new_filename) # Save the file contents = await uploaded_file.read() with open(file_path, "wb") as f: f.write(contents) print('File contents written in binary') # Extract only non-file form fields data = {k: v for k, v in form.items() if k != 'file'} data["uploaded_file_path"] = file_path print('Extracted non-file form fields.') else: print('In Else block') data = await request.json() # Path where your modules are stored module_directory = "./script" # Full path to the module file (assuming .py extension) module_path = os.path.join(module_directory, f"{module_name}.py") # Check if the file exists if not os.path.exists(module_path): raise HTTPException(status_code=404, detail=f"Module '{module_name}' not found") # Dynamically load the module from the given file path spec = importlib.util.spec_from_file_location(module_name, module_path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) # Check if the function exists in the module if hasattr(module, "function1"): # Get the function object dynamically func = getattr(module, "function1") # Ensure the function is async (for FastAPI async support) if not callable(func) or not func.__code__.co_flags & 0x80: raise HTTPException(status_code=400, detail=f"Function 'function1' is not async") # Call the async function dynamically, passing query params as arguments #result = await func(data) sig = inspect.signature(func) # If the function expects no parameters if len(sig.parameters) == 0: result = await func() else: result = await func(data) return result else: raise HTTPException(status_code=404, detail=f"Function 'funtion1' not found in module '{module_name}'") except Exception as e: raise HTTPException(status_code=500, detail=f"Internal Server Error: {str(traceback.format_exc())}") @app.post("/merchant-risk") async def dynamic_api(request: Request): try: empty_folder("./script/merchant_risk/input") empty_folder("./script/merchant_risk/output") form = await request.form() folder_path = "./script/merchant_risk/input" os.makedirs(folder_path, exist_ok=True) for key, value in form.items(): if isinstance(value, UploadFile): cleaned_key = key.replace("files[", "").replace("]", "") _, ext = os.path.splitext(value.filename) filename = f"{cleaned_key}{ext}" file_path = os.path.join(folder_path, filename) value.file.seek(0) # optional: reset file pointer async with aiofiles.open(file_path, "wb") as out_file: while chunk := await value.read(1024): # read in chunks await out_file.write(chunk) else: print("No file format") notebook_executor.run_notebook(form) except Exception as e: raise HTTPException(status_code=500, detail=f"Internal Server Error: {str(traceback.format_exc())}") @app.get("/download-zip") def download_zip_and_clean(): folder_to_zip = "./script/merchant_risk/output" zip_path = "output.zip" zip_folder(folder_to_zip, zip_path) return FileResponse( path=zip_path, media_type='application/zip', filename="output.zip", background=BackgroundTask(lambda: os.remove(zip_path)) ) def empty_folder(folder_path: str): for filename in os.listdir(folder_path): file_path = os.path.join(folder_path, filename) try: if os.path.isfile(file_path) or os.path.islink(file_path): os.unlink(file_path) # delete file or symlink elif os.path.isdir(file_path): shutil.rmtree(file_path) # delete folder except Exception as e: print(f"Failed to delete {file_path}. Reason: {e}") def zip_folder(folder_path: str, output_zip_path: str): with zipfile.ZipFile(output_zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf: for root, _, files in os.walk(folder_path): for file in files: file_path = os.path.join(root, file) # Archive file relative to the folder (no absolute paths) arcname = os.path.relpath(file_path, start=folder_path) zipf.write(file_path, arcname) @app.post("/api/overdue_update") async def overdue_update(request: Request): try: # Parse JSON body directly from request data = await request.json() result = await overdue_update_function(data) return result except Exception: raise HTTPException(status_code=500, detail=f"Overdue update Script Error: {traceback.format_exc()}") @app.post("/api/bounce_penal_update") async def bounce_penal_update(request: Request): try: # Parse JSON body directly from request data = await request.json() result = await bounce_penal_update_function(data) return result except Exception: raise HTTPException(status_code=500, detail=f"Bounce penal update Script Error: {traceback.format_exc()}") if __name__ == "__main__": import uvicorn uvicorn.run("app:app", host="127.0.0.1", port=5001, reload=True, log_level="debug")