MERMS/MermsCoreBackendFlask
1
0
Fork 0
Code Issues Pull Requests Releases Activity
Files
e89271a6546e28eef1d98d9727092bc1b41b77ec
MermsCoreBackendFlask/services/web/project/__init__.py
T
CHIEFSOFT\ameye e89271a654 import json
2024-12-30 14:27:38 -05:00

550 lines
20 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
import os
import psycopg2
from dotenv import load_dotenv
from functools import wraps
import datetime
import jwt
import random
import json
import psycopg2.extras
import pandas as pd
import project.validate.validate as validate
#import project.models.members as Members
from project.models.members import Members
from flask_cors import CORS
from flasgger import Swagger, swag_from
from flask_mail import Mail, Message
from flask import (
Flask,
jsonify,
send_from_directory,
request,
)
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import create_engine
# import socket
#import SQLAlchemy
#from werkzeug.utils import secure_filename
load_dotenv()
app = Flask(__name__)
CORS(app)
app.config.from_object("project.config.Config")
db = SQLAlchemy(app)
#jwt_secret = os.getenv("JWT_SECRET")
app.config['SECRET_KEY'] = os.getenv("JWT_SECRET")
panel_url = os.getenv("PANEL_URL")
mail = Mail(app) # instantiate the mail class
# configuration of mail
app.config['MAIL_SERVER']='smtp.gmail.com'
app.config['MAIL_PORT'] = 465
#app.config['MAIL_PORT'] = 587
app.config['MAIL_USERNAME'] = 'message@chiefsoft.com'
app.config['MAIL_PASSWORD'] = 'may12002!'
app.config['MAIL_USE_TLS'] = False
app.config['MAIL_USE_SSL'] = True
mail = Mail(app)
template = {
"swagger": "2.0",
"info": {
"title": "MERMS Core API",
"description": "This API was developed using Python Flask, which provides an interface for core MERMS endpoints.",
"version": "1.0"
}
}
app.config['SWAGGER'] = {
'title': 'MERMS API',
'uiversion': 2,
'template': './resources/flasgger/swagger_ui.html'
}
Swagger(app, template=template)
@swag_from('../../docs/consume.yml')
def token_required(f):
@wraps(f)
def decorated(*args, **kwargs):
#token = request.args.get('token')
token = request.headers["Authorization"].split(" ")[1]
# print(token)
if not token:
return jsonify({'message': 'Error - missing token'}), 403
try:
data= jwt.decode(token, app.config['SECRET_KEY'], algorithms=["HS256"])
except:
return jsonify({'message': 'Token is invalid'}),403
return f(data, *args, **kwargs)
return decorated
class User(db.Model):
__tablename__ = "users"
id = db.Column(db.Integer, primary_key=True)
email = db.Column(db.String(128), unique=True, nullable=False)
active = db.Column(db.Boolean(), default=True, nullable=False)
def __init__(self, email):
self.email = email
dataUrl = os.getenv("DATABASE_URL")
connection = psycopg2.connect(dataUrl)
#engine = SQLAlchemy.create_engine(dataUrl)
engine = create_engine(dataUrl)
@app.route("/")
def hello_world():
# mail.send(
# subject="Template example",
# receivers=["ameye@chiefsoft.com"],
# html_template="email/verify.html"
# )
send_register_mail('ameye@chiefsoft.com')
# msg = Message(
# 'Hello',
# sender ='message@chiefsoft.com',
# recipients = ['ses66181@gmail.com']
# )
# msg.body = 'Hello Flask message sent from Flask-Mail'
# mail.send(msg)
return jsonify(action_data="sent")
# GLOBAL_AVG = """SELECT * FROM members WHERE id > 0;"""
#
# result = pd.read_sql(GLOBAL_AVG, engine)
# print(result)
#
# cols = result.columns.difference(['Col1'])
# d = (result.groupby('Col1')[cols]
# .apply(lambda x: x.to_dict('r'))
# .reset_index(name='Other_details')
# .to_json(orient='records'))
#
# #json_data = [json.loads(row[0]) for row in result]
#
# with connection:
# with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor:
# cursor.execute(GLOBAL_AVG)
# account = cursor.fetchall()
# print(account[0]["uid"])
# for row in account.rows:
# print(row['id'], row['uid'])
# print(account)
# # Convert the list of tuples to a list of JSON objects
# json_data = [json.loads(row[0]) for row in account]
# #print(account)
# json_data = json.dumps(account)
# print(json_data)
# connection.close()
# return jsonify(result=account, action_data=action_data, account=account)
# GLOBAL_AVG = """SELECT * FROM members WHERE id = 1;"""
# with connection:
# with connection.cursor() as cursor:
# cursor.execute(GLOBAL_AVG)
# account = cursor.fetchone()
# return jsonify(hello="ameye world")
# # return {"account": account}
@app.route("/panel/auth/login", methods=["POST"])
def start_login():
try:
data = request.json
if not data:
return {
"message": "Please provide user details",
"data": None,
"error": "Bad request"
}, 400
# validate input
is_validated = validate.validate_username_and_password(data.get('username'), data.get('password'))
if is_validated is not True:
return dict(message='Invalid data', data=None, error=is_validated), 400
member = Members().login(
data["username"],
data["password"]
)
if member:
try:
user = {}
user_data = {}
user_data["id"] = member[0]
user_data["uid"] = member[1]
# token should expire after 24 hrs
user["token"] = jwt.encode(
{"user": user_data, 'exp' : datetime.datetime.utcnow() + datetime.timedelta(minutes=3330)},
app.config["SECRET_KEY"],
algorithm="HS256"
)
user["room"] = member[1]
return {
"message": "Successfully fetched auth token",
"data": user
}
except Exception as e:
return {
"error": "Something went wrong",
"message": str(e)
}, 500
return {
"message": "Error fetching auth token!, invalid email or password",
"data": None,
"error": "Unauthorized"
}, 404
except Exception as e:
return {
"message": "Something went wrong!",
"error": str(e),
"data": None
}, 500
@app.route("/panel/auth/register", methods=["POST"])
def start_register():
try:
data = request.json
if not data:
return {
"message": "Please provide signup details",
"data": None,
"error": "Bad request"
}, 400
# validate input
print(data)
is_validated = validate.validate_signup_data(data.get('firstname'), data.get('lastname'), data.get('email'))
if is_validated is not True:
return dict(message='Invalid data', data=None, error=is_validated), 400
firstname= data.get('firstname')
lastname= data.get('lastname')
email= data.get('email')
# mycursor = mydb.cursor()
#
# sql = "INSERT INTO customers (name, address) VALUES (%s, %s)"
# val = ("John", "Highway 21")
# mycursor.execute(sql, val)
#
# mydb.commit()
# query = "INSERT INTO users (username, password) VALUES (?, ?)"
# values = ("john", "password123")
# cursor.execute(query, values)
# conn.commit()
INSERT_MEMBERS = "INSERT INTO members_pending(email,firstname,lastname) VALUES('"+firstname+"','"+lastname+"','"+email+"')"
with connection:
with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor:
cursor.execute(INSERT_MEMBERS)
connection.commit()
last_row_id=cursor.lastrowid
print('After Insert ::: ')
print(last_row_id)
send_register_mail(email)
return jsonify(hello="ameye signup path world", last_row_id=last_row_id)
except Exception as e:
return {
"message": "Something went wrong!",
"error": str(e),
"data": None
}, 500
def send_register_mail(signup_email):
#panel_url
signup_data = {}
signup_data["id"] = '100'
signup_data["uid"] = 'a4b75649-e3c5-424f-bcdb-5481e625d24b'
jwt_part = jwt.encode(
{"user": signup_data, 'exp' : datetime.datetime.utcnow() + datetime.timedelta(minutes=3330)},
app.config["SECRET_KEY"],
algorithm="HS256"
)
link_url= str(panel_url) + '/csignup/' + jwt_part
print(link_url)
msg = Message(
'verify your MERMS Account',
sender ='message@chiefsoft.com',
recipients = [signup_email,'ameye+merscopy@chiefsoft.com']
)
msg.body = """Hello Firstname,
You received this message for account verification
Follow the link:" {% link_url %}
https://dev-panel.mermsemr.com/csignup/eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyIjp7ImlkIjoiOTgyODkyODQyODI4NTI4OSIsInVpZCI6Ijk4Mjg5Mjg0MjgyODUyODkifSwiZXhwIjoxNzM1NTA0MDE4fQ.pDQvYUr_PGZMeMO2gr-B3DRQ7AM7IjVM5vSERNTviG4
For any Support
Reach Out
"""
mail.send(msg)
@app.route("/panel/auth/register/verify", methods=["POST"])
def verify_register():
#data = request.json
vrL ='eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyIjp7ImlkIjoiOTgyODkyODQyODI4NTI4OSIsInVpZCI6Ijk4Mjg5Mjg0MjgyODUyODkifSwiZXhwIjoxNzM1NTA0MDE4fQ.pDQvYUr_PGZMeMO2gr-B3DRQ7AM7IjVM5vSERNTviG4'
try:
# data= jwt.decode(data.verify_link, app.config['SECRET_KEY'], algorithms=["HS256"])
data= jwt.decode(vrL, app.config['SECRET_KEY'], algorithms=["HS256"])
except:
return jsonify({'status': 'INVALID', 'message': 'Link is invalid'}),403
print(data)
user_uid = 'a4b75649-e3c5-424f-bcdb-5481e625d24b'
FIND_USER_DETAIL= "SELECT firstname,lastname,email FROM members_pending WHERE uid::text = '"+user_uid+"'"
with connection:
with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor:
cursor.execute(FIND_USER_DETAIL)
account = cursor.fetchall()
print(account[0])
return jsonify(status="VALID", user=account[0], pending_uid=user_uid,country=[])
@app.route("/panel/auth/reset", methods=["POST"])
def start_resetpass():
try:
data = request.json
if not data:
return {
"message": "Please provide username ",
"data": None,
"error": "Bad request"
}, 400
# validate input
print(data)
is_validated = validate.validate_username(data.get('username'))
if is_validated is not True:
return dict(message='Invalid data', data=None, error=is_validated), 400
username= data.get('username')
sql = "INSERT INTO password_reset (username) VALUES (%s)"
val = (username)
with connection:
with connection.cursor() as cursor:
cursor.execute(sql, val)
connection.commit()
# last_row_id=cursor.lastrowid
# return jsonify(hello="ameye reset path world", last_row_id=last_row_id)
return jsonify(hello="ameye reset path world")
except Exception as e:
return {
"message": "Something went wrong!",
"error": str(e),
"data": None
}, 500
def send_resetpass_mail(signup_email):
msg = Message(
'Reset your MERMS Account',
sender ='message@chiefsoft.com',
recipients = [signup_email,'ameye+merscopy@chiefsoft.com']
)
msg.body = 'Hello MERMS message sent for account verification http://localhost:8090/accreset/JWT-djhgdhjgdhdggggd'
mail.send(msg)
@app.route("/panel/account")
@token_required
def panel_account(current_user):
# print(current_user["user"]["uid"])
user_uid = current_user["user"]["uid"]
# print(user_uid)
member_dash = Members().get_member_by_uid(user_uid)
print(member_dash[0])
print(member_dash[0][0])
return jsonify(hello=current_user)
@app.route("/panel/account/dash")
@token_required
def dashboard(current_user):
print( current_user["user"]["uid"])
user_uid = current_user["user"]["uid"]
FIND_USER_DETAIL= "SELECT id,uid,username,updated,email,account_name, firstname, lastname FROM members WHERE uid::text = '"+user_uid+"'"
with connection:
with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor:
cursor.execute(FIND_USER_DETAIL)
account = cursor.fetchall()
print(account[0])
dash_data = {
"username": account[0]["username"],
"account_name": account[0]["account_name"],
"firstname": account[0]["firstname"],
"lastname" : account[0]["lastname"],
"email": account[0]["email"],
"updated": account[0]["updated"],
}
return jsonify(dash_data=dash_data)
@app.route("/panel/account/products")
@token_required
def panel_products(current_user):
products_data = {
"last_update": datetime.datetime.utcnow(),
"products": [
{"uid":"A000001","icon": "icon_product", "description": "Professional Website" , "status": 'Activate now'},
{"uid":"A000002","icon": "icon_product", "description": "Professional Blog" , "status": 'Activate now'},
{"uid":"A000003","icon": "icon_product", "description": "Business Website" , "status": 'Activate now'},
{"uid":"A000004","icon": "icon_product", "description": "Business Blog Site" , "status": 'Activate now'},
{"uid":"A000005","icon": "icon_product", "description": "OpenEmr" , "status": 'Activate now'},
{"uid":"A000005","icon": "icon_product", "description": "Dummy Dummy" , "status": 'Activate now'},
]
}
PRODUCT_LIST ='SELECT id,uid,product_id,name,description,status,banner FROM products ORDER BY id ASC'
with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor:
cursor.execute(PRODUCT_LIST)
products_list = cursor.fetchall()
print(products_list)
productJS = json.dumps( [dict(ix) for ix in products_list] )
print(productJS)
return jsonify(products_data=products_data, products_list=products_list)
@app.route("/panel/account/bar")
@token_required
def recent_bar(current_user):
bar_data = {
"last_update": datetime.datetime.utcnow(),
"initial": random.randint(0, 10),
"processing": random.randint(0, 10),
"verifying" : random.randint(0, 10),
"completed" : random.randint(0, 10),
"top_bar": [
{"id": "1", "description": "Contacts" , "last_update": "10-10-2010 11:00 AM", "value": random.randint(0, 10) , "data_span":'Last 2 months'},
{"id": "2", "description": "Site Traffic" , "last_update": "10-10-2010 11:30 AM", "value": random.randint(0, 10), "data_span":'Past 12 hours'},
{"id": "3", "description": "Appointments" , "last_update": "10-12-2010 11:30 AM", "value": random.randint(0, 10), "data_span":'Last 14 days'},
{"id": "4", "description": "Purchases" , "last_update": "10-12-2010 11:30 AM", "value": random.randint(0, 10), "data_span":'Last 3 months'},
]
}
return jsonify(bar_data=bar_data)
@app.route("/panel/account/actions")
@token_required
def recent_actions(current_user):
print( current_user["user"]["uid"])
user_id = current_user["user"]["id"]
# FIND_USER_DETAIL= "SELECT id,uid,username,updated,email,account_name, firstname, lastname FROM members WHERE uid::text = '"+user_uid+"'"
# RECENT_ACTIONS = "SELECT * FROM members_actions WHERE member_id = " + user_id + " ORDER by id DESC LIMIT 4"
# RECENT_ACTIONS = "SELECT * FROM members_actions WHERE member_id::text = %s ORDER by id DESC LIMIT 4"
# print(user_id)
RECENT_ACTIONS = "SELECT * FROM members_actions WHERE member_id = 1 ORDER by id DESC LIMIT 4"
actionVal = (user_id)
with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor:
cursor.execute(RECENT_ACTIONS)
recent_actions = cursor.fetchall()
print(recent_actions)
action_data = {
"recent_actions" : recent_actions,
"last_update": datetime.datetime.utcnow(),
"initial": random.randint(0, 10),
"processing": random.randint(0, 10),
"verifying" : random.randint(0, 10),
"completed" : random.randint(0, 10),
"actions": [
{"no": "1", "description": "Welcome to MERMS" , "date": "10-10-2010 11:00 AM", "status": 'completed'},
{"no": "2", "description": "Personal Blog Setup" , "date": "10-10-2010 11:30 AM", "status": 'processing'},
{"no": "3", "description": "Web Traffic Analysis" , "date": "10-12-2010 11:30 AM", "status": 'verifying'},
]
}
return jsonify(action_data=action_data)
@app.route("/panel/account/products/url")
@token_required
def product_urls(current_user):
url_data = {
"last_update": datetime.datetime.utcnow(),
"url": [
{"no": "1", "description": "Main Website", "url":'https://starter.0ach30996.mermsemr.com', "date": "10-10-2010 11:00 AM", "status": 'Active'},
{"no": "1", "description": "Personal Blog", "url":'https://starterblo.0ach30996.mermsemr.com', "date": "10-10-2010 11:00 AM", "status": 'Updating'},
{"no": "1", "description": "Business Web", "url":'https://starterbus.0ach30996.mermsemr.com', "date": "10-10-2010 11:00 AM", "status": 'Active'},
{"no": "1", "description": "Business Blog", "url":'https://starterbblog.0ach30996.mermsemr.com', "date": "10-10-2010 11:00 AM", "status": 'Active'},
]
}
return jsonify(url_data=url_data)
@app.route("/panel/account/payments")
@token_required
def account_payments(current_user):
payments = {
"last_update": datetime.datetime.utcnow(),
"url": [
{"no": "1", "description": "Welcome to MERMS" , "date": "10-10-2010 11:00 AM", "status": 'completed'},
{"no": "2", "description": "Personal Blog Setup" , "date": "10-10-2010 11:30 AM", "status": 'processing'},
{"no": "3", "description": "Web Traffic Analysis" , "date": "10-12-2010 11:30 AM", "status": 'verifying'},
]
}
return jsonify(payments_data=payments_data)
##. Description Date Status "10-10-2021 10 AM"
@app.route("/panel/myproduct/dash")
@token_required
def myproduct(current_user):
product_id = request.args.get('product_id')
print(product_id)
if not product_id:
return {
"message": "Please provide product_id",
"data": None,
"error": "Bad request"
}, 400
print(product_id)
FIND_PRODUCT= "SELECT uid,product_id,name,description,banner,status FROM products WHERE product_id='"+product_id+"'"
with connection:
with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor:
cursor.execute(FIND_PRODUCT)
product = cursor.fetchall()
print(product[0]["uid"])
myproduct_data = {
"myproduct_uid":"",
"status": product[0]["status"],
"product_uid": product[0]["uid"],
"banner": product[0]["banner"],
"product_name": product[0]["name"],
"description": "Product Description - Commitment is something that comes from understanding that everything has its price and then having the willingness to pay that price. This is important because nobody wants to put significant effort into something, only to find out after the fact that the price was too high.The price is something not necessarily defined as financial. It could be time, effort, sacrifice, money or perhaps, something else.",
"title": product[0]["description"],
"subscription_text" : "Start with your goals in mind and then work possible.ith yand Goals. If the plan doesn’t support the vision then change it!",
"promotion_text": "Start Free Today !",
"price_text" : "90 days free and 3.95/Month",
}
return jsonify(myproduct_data=myproduct_data)
@app.route("/panel/myproduct/subscription")
@token_required
def subscription_start(current_user):
product_id = request.args.get('product_id')
print(product_id)
if not product_id:
return {
"message": "Please provide product_id",
"data": None,
"error": "Bad request"
}, 400
print(product_id)
return jsonify(myproduct_data=myproduct_data)
Reference in New Issue View Git Blame Copy Permalink