import json import sys import mysql.connector #import venv import barcode from barcode.writer import ImageWriter sys.path.insert(0, "./include") from fastapi import FastAPI from fastapi.encoders import jsonable_encoder from fastapi.responses import JSONResponse from fastapi import FastAPI, HTTPException, Security from fastapi.security import APIKeyHeader from config import connect from starlette.responses import StreamingResponse from fastapi.openapi.docs import get_swagger_ui_html from fastapi.openapi.utils import get_openapi import secrets import io import qrcode from PIL import Image,ImageDraw from qrcode.image.styles.moduledrawers import RoundedModuleDrawer,CircleModuleDrawer from qrcode.image.styledpil import StyledPilImage from qrcode.image.styles.colormasks import SolidFillColorMask from fastapi import Depends, HTTPException, status from fastapi.security import HTTPBasic, HTTPBasicCredentials app = FastAPI(title = "Flag API",docs_url=None,redoc_url=None) security = HTTPBasic() api_key_header = APIKeyHeader(name="FLAG-API-Key") def get_api_key(api_key_header: str = Security(api_key_header)) -> str: conn = connect() cursor = conn.cursor() query = "SELECT fastapikey from flag.company;" cursor.execute(query) api_keys = [row[0].decode("utf-8") for row in cursor.fetchall()] conn.close() if api_key_header in api_keys: return api_key_header raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or missing API Key", ) def get_current_username(credentials: HTTPBasicCredentials = Depends(security)): correct_username = secrets.compare_digest(credentials.username, "user") correct_password = secrets.compare_digest(credentials.password, "password") if not (correct_username and correct_password): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect email or password", headers={"WWW-Authenticate": "Basic"}, ) return credentials.username @app.get("/docs") async def get_documentation(username: str = Depends(get_current_username)): return get_swagger_ui_html(openapi_url="/openapi.json", title="docs") class create_dict(dict): def __init__(self): self = dict() def add(self, key, value): self[key] = value @app.get("/list_box") async def read_item(api_key: str = Security(get_api_key)): conn = connect() conn = connect() cursor = conn.cursor() query = "SELECT id,model,mat,coalesce(box,'-') as box,coalesce(DATE_FORMAT(date_prod, '%d-%m-%Y'),'-') as date_prod,coalesce(var,'-'),coalesce(active,'-') as active,coalesce(state,'-') as state,coalesce(slots,'-') as slots,coalesce(geopos,'-') as geopos FROM flag.flag;" cursor.execute(query) columns = [column[0] for column in cursor.description] data = [dict(zip(columns, row)) for row in cursor.fetchall()] conn.close() return JSONResponse(content=data, media_type="application/json") @app.get("/list_company") async def read_item(api_key: str = Security(get_api_key)): conn = connect() cursor = conn.cursor() query = "SELECT id,company_name FROM flag.company;" cursor.execute(query) columns = [column[0] for column in cursor.description] data = [dict(zip(columns, row)) for row in cursor.fetchall()] conn.close() return JSONResponse(content=data, media_type="application/json",) @app.get("/lock_box_space") async def read_item(matricola: str, company_name: str,date: str, timeband: str,nrlock:str,api_key: str=Security(get_api_key)): conn = connect() cursor = conn.cursor() query = "SELECT flag.lock_box_space("+matricola+","+company_name+","+date+","+timeband+","+nrlock+");" cursor.execute(query) row = cursor.fetchone()[0].split("-") data=json.loads('{"Free Space":'+row[0]+","+'"Locked ID":'+ row[1]+"}") conn.commit() conn.close() return data @app.get("/book_box_space") async def read_item(lock_id: str,api_key: str=Security(get_api_key)): conn = connect() cursor = conn.cursor() query = "SELECT flag.book_box_space("+lock_id+");" cursor.execute(query) columns = ('QR Client', 'QR Delivery') row = tuple(cursor.fetchone()[0].split(',')) data = dict(zip(columns, row)) conn.commit() conn.close() return JSONResponse(content=data, media_type="application/json") @app.get("/unlock_box_space") async def read_item(id: str,api_key: str=Security(get_api_key)): conn = connect() cursor = conn.cursor() query = "SELECT flag.unlock_box_space("+id+");" cursor.execute(query) columns = ('Result','--') row = tuple(cursor.fetchone()[0].split(',')) data = dict(zip(columns, row)) conn.commit() conn.close() return JSONResponse(content=data, media_type="application/json") @app.get("/sql_mqtt") async def read_item(mqtt_topic: str,mqtt_value: str,api_key: str=Security(get_api_key)): conn = connect() cursor = conn.cursor() query = "SELECT mqtt.sql_mqtt ("+mqtt_topic+","+mqtt_value+")as response;" cursor.execute(query) columns = ('Result',) row = (cursor.fetchone()[0],) data = dict(zip(columns, row)) conn.commit() conn.close() return JSONResponse(content=data, media_type="application/json") @app.get("/code128/{message}") def code128(message: str): buf = io.BytesIO() #EAN13(str(100000902922), writer=ImageWriter()).write(rv) EAN = barcode.get_barcode_class('code128') my_ean = EAN(message, writer=ImageWriter()) my_ean.get_fullcode() options = {'font_size': 5,'text_distance': 2.0,'module_height':10.0,'foreground':'darkblue'} my_ean.write(buf,options) #img.save(buf,'jpeg') buf.seek(0) # important here! return StreamingResponse(buf, media_type="image/png") @app.get("/generate/{message}") def generate(message: str): Logo_link = './logo.png' basewidth = 100 img = qrcode.make(message) logo = Image.open(Logo_link) # adjust image size wpercent = (basewidth/float(logo.size[0])) hsize = int((float(logo.size[1])*float(wpercent))) logo = logo.resize((basewidth, hsize), Image.ANTIALIAS) QRcode = qrcode.QRCode( error_correction=qrcode.constants.ERROR_CORRECT_H ) QRcode.add_data(message) QRcode.make() # taking color name from user QRcolor = 'Blue' # adding color to QR code QRimg = QRcode.make_image(image_factory=StyledPilImage, fill_color=QRcolor, #module_drawer=CircleModuleDrawer(radius_ratio=1.2), eye_drawer=RoundedModuleDrawer(radius_ratio=1.8), color_mask=SolidFillColorMask(front_color=(59, 89, 152))) #embeded_image_path="./logo.png") # set size of QR code # pos = ((QRimg.size[0] - logo.size[0]) // 2, # (QRimg.size[1] - logo.size[1]) // 2) ## QRimg.paste(logo, pos) # save the QR code generated # QRimg.save('./gfg_QR.jpg') # logo=Image.open("./logo.png") imgsize=QRimg.size imgsize1=logo.size new_size = (imgsize[0], imgsize[1] + imgsize1[1]-35) img = Image.new("RGB", new_size, color="white") img.paste(QRimg, (0, 0)) img.paste(logo,(int((float(imgsize[0])-float(imgsize1[0]))/2),imgsize[0]-35)) img.save('./gfg_QR1.jpg') # logo=Image.open("./logo.png") buf = io.BytesIO() img.save(buf,'jpeg') buf.seek(0) # important here! return StreamingResponse(buf, media_type="image/jpeg") #img = qrcode.make(message) #buf = io.BytesIO() #img.save(buf) #buf.seek(0) # important here! #return StreamingResponse(buf, media_type="image/jpeg")beded