from textwrap import wrap from fastapi import APIRouter, HTTPException, File, UploadFile, Form from fastapi.responses import FileResponse, JSONResponse import requests from pydantic import BaseModel from minio import Minio import Minio_Storage from reportlab.pdfgen import canvas from reportlab.lib.pagesizes import A4 from reportlab.lib.utils import ImageReader import io from datetime import datetime #SuperToken Auth from front end from supertokens_python.recipe.session.framework.fastapi import verify_session from supertokens_python.recipe.session import SessionContainer from fastapi import Depends import Minio_Storage.minioConnection router = APIRouter() class minioPullRequest(BaseModel): file_path: str class minioDeleteRequest(BaseModel): file_path: str class medCertUploud(BaseModel): app_id: str fullName: str id_no: str docfname: str busName: str busAddr: str busNo: str busEmail: str startDate: str endDate: str returnDate: str logo_path: str sig_path: str @router.get("/minio/pull/file/{app_id}/{folder}/{file_name}/{env}", tags=["Minio"]) async def pull_File_from_user(app_id: str, folder: str, file_name: str, env: str, session: SessionContainer = Depends(verify_session())): #, session: SessionContainer = Depends(verify_session()) path = app_id + "/" + folder + "/" + file_name try: # uploudFile(app_id, file.filename, extension[1], content) client = Minio_Storage.minioConnection.minioConnect(env) miniourl = client.presigned_get_object("mih", path) # temp = minioResponse.data#.encode('utf-8').strip() # print(temp) # print("=======================================================================") # temp = temp.decode('utf-8') #print(miniourl) except Exception as error: raise HTTPException(status_code=404, detail=miniourl) # return {"message": error} return { "minioURL": miniourl, } @router.post("/minio/upload/file/", tags=["Minio"]) async def upload_File_to_user(file: UploadFile = File(...), app_id: str= Form(...), folder: str= Form(...), session: SessionContainer = Depends(verify_session())): extension = file.filename.split(".") content = file.file try: uploudFile(app_id, folder, file.filename, extension[1], content) except Exception as error: raise HTTPException(status_code=404, detail="Failed to Uploud Record") return {"message": "Successfully Delete File"} # return { # "app_id": app_id, # "file name": file.filename, # "extension": extension[0], # "file contents": file.file.read(), # } @router.delete("/minio/delete/file/", tags=["Minio"]) async def delete_File_of_user(requestItem: minioDeleteRequest, session: SessionContainer = Depends(verify_session())): #, session: SessionContainer = Depends(verify_session()) path = requestItem.file_path try: # uploudFile(app_id, file.filename, extension[1], content) client = Minio_Storage.minioConnection.minioConnect("dev") minioError = client.remove_object("mih", path) except Exception as error: raise HTTPException(status_code=404, detail=minioError) # return {"message": error} return {"message": "Successfully deleted File"} # return { # "app_id": app_id, # "file name": file.filename, # "extension": extension[0], # "file contents": file.file.read(), # } # Get List of all files by patient @router.post("/minio/generate/med-cert/", tags=["Minio"]) async def upload_File_to_user(requestItem: medCertUploud, session: SessionContainer = Depends(verify_session())): # uploudMedCert(requestItem.app_id, # requestItem.fullName, # requestItem.docfname, # requestItem.startDate, # requestItem.endDate, # requestItem.returnDate, # requestItem.logo_path, # requestItem.sig_path) uploudMedCert(requestItem) return {"message": "Successfully Generated File"} def uploudFile(app_id, folder, fileName, extension, content): client = Minio_Storage.minioConnection.minioConnect("dev") found = client.bucket_exists("mih") if not found: client.make_bucket("mih") else: print("Bucket already exists") fname = app_id + "/" + folder + "/" + fileName client.put_object("mih", fname, content, length=-1, part_size=10*1024*1024, content_type=f"application/{extension}") #"minio""localhost:9000" # def uploudMedCert(app_id, fullName, docfname, startDate, endDate, returnDate, logo_path, sig_path): def uploudMedCert(requestItem: medCertUploud): client = Minio_Storage.minioConnection.minioConnect("dev") # generateMedCertPDF(fullName, docfname, startDate, endDate, returnDate, logo_path, sig_path) generateMedCertPDF(requestItem) today = datetime.today().strftime('%Y-%m-%d') found = client.bucket_exists("mih") if not found: client.make_bucket("mih") else: print("Bucket already exists") fileName = f"{requestItem.app_id}/patient_files/Med-Cert-{requestItem.fullName}-{today}.pdf" client.fput_object("mih", fileName, "temp.pdf") # def generateMedCertPDF(fullName, docfname, startDate, endDate, returnDate, logo_path, sig_path): def generateMedCertPDF(requestItem: medCertUploud): client = Minio_Storage.minioConnection.minioConnect("dev") new_logo_path = requestItem.logo_path.replace(" ","-") new_sig_path = requestItem.sig_path.replace(" ","-") minioLogo = client.get_object("mih", new_logo_path).read() imageLogo = ImageReader(io.BytesIO(minioLogo)) minioSig = client.get_object("mih", new_sig_path).read() imageSig = ImageReader(io.BytesIO(minioSig)) w,h = A4 today = datetime.today().strftime('%d-%m-%Y') myCanvas = canvas.Canvas("temp.pdf", pagesize=A4) #Business Logo myCanvas.drawImage(imageLogo, 50, h - 125,100,100, mask='auto') #Business Details myCanvas.setFont('Helvetica-Bold', 10) myCanvas.drawRightString(w - 50,h - 40, f"Name: {requestItem.busName}") myCanvas.drawRightString(w - 50,h - 55, f"Address: {requestItem.busAddr}") myCanvas.drawRightString(w - 50,h - 70, f"Contact No.: {requestItem.busNo}") myCanvas.drawRightString(w - 50,h - 85, f"Email: {requestItem.busEmail}") myCanvas.line(50,h-150,w-50,h-150) #Todays Date myCanvas.setFont('Helvetica', 12) issueDate = str(today) myCanvas.drawRightString(w - 50,h - 180,issueDate) #Title myCanvas.setFont('Helvetica-Bold', 20) myCanvas.drawString(w-375, h - 200, "Medical Certificate") #Body myCanvas.setFont('Helvetica', 12) body = "" body += "This is to certify that " + requestItem.fullName.upper() + " (" + requestItem.id_no+ ") was seen by " + requestItem.docfname.upper() + " on " + requestItem.startDate + "." body += "\nHe/She is unfit to attend work/school from " + requestItem.startDate + " up to and including " + requestItem.endDate + "." body += "\nHe/She will return on " + requestItem.returnDate + "." y = 250 for line in wrap(body, 90): myCanvas.drawString(50, h-y, line) y += 30 # myCanvas.drawString(50, h-250,line1) # myCanvas.drawString(50, h-280,line2) # myCanvas.drawString(50, h-310,line3) #Signature myCanvas.drawImage(imageSig, 50, h - 690,100,100) myCanvas.line(50,h-700,200,h-700) myCanvas.drawString(50, h-720, requestItem.docfname.upper()) #QR Verification qrText = requestItem.fullName.upper() + " booked off from " + requestItem.startDate + " to " + requestItem.endDate + " by " + requestItem.docfname.upper() + ".\nPowered by Mzansi Innovation Hub." qrText = qrText.replace(" ","+") url = f"https://api.qrserver.com/v1/create-qr-code/?data={qrText}&size=100x100" response = requests.get(url) image = ImageReader(io.BytesIO(response.content)) myCanvas.drawImage(image,w-150, h-700,100,100) myCanvas.setFont('Helvetica-Bold', 15) myCanvas.drawString(w-150,h-720,"Scan to verify") myCanvas.save() #Def get #uploudFile("Yasien Meth","Dr D Oct","18-06-2024","20-06-2024","21-06-2024") def byteToMb(size): sizekb = size /1000 sizemb = sizekb / 1000 return sizemb # print(byteToMb(229574))