from flask import Flask, request, make_response, jsonify, send_from_directory, abort from flask_httpauth import HTTPBasicAuth from flask_mysqldb import MySQL import logging import datetime from datetime import timedelta, datetime import json import re import numpy as np import pandas as pd import os import requests import math import random from configparser import ConfigParser from dateutil import parser logging.basicConfig(filename='/var/www/html/logs/p_bulk_sms.log', filemode='a', format='%(asctime)s %(name)s %(levelname)s %(message)s', level=logging.INFO) #Read config.ini file config_object = ConfigParser() config_object.read("config.ini") #Get info live_hellio_info = config_object["live_hellio"] live_hellio_wr = config_object["live_hellio_wr"] sandbox_wr_info = config_object["sandbox_write"] authinfo = config_object["auth"] smsapiinfo = config_object["smsapi"] scp_info = config_object["server_4"] app = Flask(__name__) app.config['UPLOAD_EXTENSIONS'] = ['.txt', '.csv', '.xlsx'] mysql = MySQL(app) auth = HTTPBasicAuth() USER_DATA = { str(authinfo["username"]): str(authinfo["password"]) } @auth.verify_password def verify(username, password): if not (username and password): return False return USER_DATA.get(username) == password @app.route('/senderid', methods=['POST', 'GET']) @auth.login_required def senderId(): app.config['MYSQL_HOST'] = str(live_hellio_info["host"]) app.config['MYSQL_USER'] = str(live_hellio_info["user"]) app.config['MYSQL_PASSWORD'] = str(live_hellio_info["password"]) app.config['MYSQL_PORT'] = int(live_hellio_info["port"]) app.config['MYSQL_DB'] = str(live_hellio_info["db"]) if request.method == 'POST': req = request.get_json() user_id = req['user_id'] user_id = int(user_id) cur = mysql.connection try: sender_data = pd.read_sql("SELECT sender_id FROM user_sender_ids WHERE user_id={} AND status='1'".format(user_id), cur) results = sender_data['sender_id'].to_list() logging.info(f"sender_ids = {results}") except Exception as e: logging.info(e) return make_response(jsonify(code=404, error='sender id does not exist'), 200) if not results: logging.info(f"No sender ids for {user_id}") return make_response(jsonify(code=404, error='sender id does not exist'), 200) return make_response(jsonify(code=200, sender_ids=results),200) @app.route('/balance-checks', methods=['POST', 'GET']) @auth.login_required def balance(): def replace_text(row, sqd_columns_name, dict_columns_name): for i in sqd_columns_name: try: row['message'] = row['message'].replace(i, row[dict_columns_name[i]]) except: row['message'] = row['message'].replace(i, '') return row app.config['MYSQL_HOST'] = str(live_hellio_info["host"]) app.config['MYSQL_USER'] = str(live_hellio_info["user"]) app.config['MYSQL_PASSWORD'] = str(live_hellio_info["password"]) app.config['MYSQL_PORT'] = int(live_hellio_info["port"]) app.config['MYSQL_DB'] = str(live_hellio_info["db"]) if request.method == 'POST': req = request.get_json() user_id = req['user_id'] user_id = int(user_id) username = req['username'].strip() message = req['message'] remove_duplicate = req['remove_duplicate'] file_path = req['file_path'] file_ext = os.path.splitext(file_path)[1] cur = mysql.connection.cursor() if not os.path.exists(str(file_path)): return make_response(jsonify(code=404, error="File not found"), 200) file_ext = os.path.splitext(file_path)[1] if file_ext not in app.config['UPLOAD_EXTENSIONS']: return make_response(jsonify(code=500, error="File extension not supported "), 200) else: app.config['UPLOADS'] = file_path try: if file_ext == '.xlsx': data = pd.read_excel(app.config['UPLOADS']) traffic = data.shape[0] elif file_ext == '.csv': data = pd.read_csv(app.config['UPLOADS'],low_memory=False) traffic = data.shape[0] elif file_ext == '.txt': data = pd.read_csv(app.config['UPLOADS'], delim_whitespace=True, error_bad_lines=False) traffic = data.shape[0] else: return make_response(jsonify(code=500, error="File extension not supported"), 200) except FileNotFoundError: abort(404) if remove_duplicate == 1: data.drop_duplicates(subset=data.columns[0], keep='first', inplace=True) # reset index data = data.reset_index(drop=True) else: data try: cur.execute("SELECT price, new_bal FROM credits WHERE user_id='{}' AND username='{}' ORDER BY created_at desc".format(user_id, username)) credits_data = cur.fetchone() cost_price = credits_data[0] logging.info(type(cost_price)) balance = credits_data[1] logging.info(type(balance)) data['message'] = message data_copy = data.copy() dict_columns_name = {str('[' + x + ']'): x for x in list(data_copy.columns)[1:]} sqd_columns_name = [str('[' + x + ']') for x in list(data_copy.columns)[1:]] data = data.apply(replace_text, args=(sqd_columns_name, dict_columns_name,), axis=1) logging.info(f"p-bulk data: {data.head()}") data['sms_count'] = np.where(data['message'].str.len() <= 160, 1, np.where((data['message'].str.len().between(161, 306, inclusive=True)),2, np.where((data['message'].str.len().between(307, 459, inclusive=True)), 3, np.where((data['message'].str.len().between(460,621, inclusive=True)),4, np.where((data['message'].str.len().between( 622, 766, inclusive=True)), 5, np.where((data['message'].str.len().between(767, 919,inclusive=True)), 6, np.where((data['message'].str.len().between( 920,1072,inclusive=True)),7,8))))))) data['sms_cost'] = data['sms_count'] * cost_price msg_cost = data['sms_cost'].sum() sms_count = int(data['sms_count'].max()) logging.info(msg_cost) logging.info(type(msg_cost)) cur.close() if msg_cost > balance: return make_response(jsonify(code=400, error=f"Insufficient Balance, the cost of sms is Ghs{msg_cost}.'\n'You have Ghs{balance} left, you need Ghs{msg_cost-balance} to send sms"), 200) except Exception as e: logging.info(e) cur.close() return make_response(jsonify(code=500, error="user credits info does not exist"), 200) return make_response(jsonify(code=200, sms_count=sms_count, total_traffic=traffic, data="Enough Balance"),200) @app.route('/sendsms', methods=['POST', 'GET']) @auth.login_required def sms(): app.config['MYSQL_HOST'] = str(sandbox_wr_info["host"]) app.config['MYSQL_USER'] = str(sandbox_wr_info["user"]) app.config['MYSQL_PASSWORD'] = str(sandbox_wr_info["password"]) app.config['MYSQL_PORT'] = int(sandbox_wr_info["port"]) app.config['MYSQL_DB'] = str(sandbox_wr_info["db"]) def generateOTP(): digits = "0123456789" OTP = "" for i in range(4) : OTP += digits[math.floor(random.random() * 10)] created_at = datetime.utcnow() expires = created_at + timedelta(minutes=5) created_at = created_at.strftime('%Y-%m-%d %H:%M:%S') expires = expires.strftime('%Y-%m-%d %H:%M:%S') return OTP, created_at, expires if request.method == 'POST': cur = mysql.connection.cursor() req = request.get_json() user_id = req['user_id'] user_id = int(user_id) username = req['username'] destination = req['destination'] url = req['url'] otp_code, created_at, expires = generateOTP() try: json_payload = { "username" : str(smsapiinfo['username']), "password" : str(smsapiinfo['password']), "source" : str(smsapiinfo['source']), "destination" : destination, "message" : f"{username} with user_id {user_id} has queued messsages waiting for approval, visit this link {url}" } otp_payload = { "username" : str(smsapiinfo['username']), "password" : str(smsapiinfo['password']), "source" : str(smsapiinfo['source']), "destination" : destination, "message" : f"Account verification OTP is {otp_code}" } res = requests.post('https://deywuro.com/api/sms', json=json_payload) res2 = requests.post('https://deywuro.com/api/sms', json=otp_payload) resp = res.json() resp2 = res2.json() if resp['code']!= 0: return make_response(jsonify(code=resp['code'], error="Message not delivered"), 200) if resp2['code']!= 0: return make_response(jsonify(code=resp2['code'], error="OTP not delivered"), 200) logging.info(f"sms response {res}") logging.info(f"Otp response {res2}") try: cur.execute("INSERT INTO otp_codes(code, created_at, expires) VALUES(%s, %s, %s)", (otp_code, created_at, expires)) mysql.connection.commit() cur.close() except Exception as e: logging.info(e) cur.close() return make_response(jsonify(code=500, error="Query failed"), 200) except Exception as e: logging.info(e) return make_response(jsonify(code=400, error="Message not delivered"), 200) return make_response(jsonify(code=200, success="Link and OTP-code sent"), 200) @app.route('/resend/otp', methods=['POST', 'GET']) @auth.login_required def otp_code(): app.config['MYSQL_HOST'] = str(sandbox_wr_info["host"]) app.config['MYSQL_USER'] = str(sandbox_wr_info["user"]) app.config['MYSQL_PORT'] = int(sandbox_wr_info["port"]) app.config['MYSQL_PASSWORD'] = str(sandbox_wr_info["password"]) app.config['MYSQL_DB'] = str(sandbox_wr_info["db"]) def generateOTP(): digits = "0123456789" OTP = "" for i in range(4) : OTP += digits[math.floor(random.random() * 10)] created_at = datetime.utcnow() expires = created_at + timedelta(minutes=5) created_at = created_at.strftime('%Y-%m-%d %H:%M:%S') expires = expires.strftime('%Y-%m-%d %H:%M:%S') return OTP, created_at, expires if request.method == 'POST': cur = mysql.connection.cursor() req = request.get_json() destination = req['destination'] otp_code, created_at, expires = generateOTP() try: otp_payload = { "username" : str(smsapiinfo['username']), "password" : str(smsapiinfo['password']), "source" : str(smsapiinfo['source']), "destination" : destination, "message" : f"Account verification OTP is {otp_code}" } res = requests.post('https://deywuro.com/api/sms', json=otp_payload) resp = res.json() if resp['code']!= 0: return make_response(jsonify(code=resp['code'], error="OTP not delivered"), 200) logging.info(f"Otp response {res}") try: cur.execute("INSERT INTO otp_codes(code, created_at, expires) VALUES(%s, %s, %s)", (otp_code, created_at, expires)) mysql.connection.commit() cur.close() except Exception as e: logging.info(e) cur.close() return make_response(jsonify(code=500, error="Query failed"), 200) except Exception as e: logging.info(e) return make_response(jsonify(code=400, error="OTP-code not delivered"), 200) return make_response(jsonify(code=200, success="OTP-code sent"), 200) @app.route('/validate/otp', methods=['POST', 'GET']) @auth.login_required def validate_code(): app.config['MYSQL_HOST'] = str(sandbox_wr_info["host"]) app.config['MYSQL_USER'] = str(sandbox_wr_info["user"]) app.config['MYSQL_PORT'] = int(sandbox_wr_info["port"]) app.config['MYSQL_PASSWORD'] = str(sandbox_wr_info["password"]) app.config['MYSQL_DB'] = str(sandbox_wr_info["db"]) if request.method == 'POST': cur = mysql.connection.cursor() req = request.get_json() otp_code = req['otp_code'] try: sql = "DELETE FROM otp_codes WHERE expires < NOW()" cur.execute(sql) mysql.connection.commit() except Exception as e: logging.info(e) cur.close() try: sql = "SELECT code FROM otp_codes WHERE code={}".format(otp_code) cur.execute(sql) otp = cur.fetchone() if not otp: return make_response(jsonify(code=500, error="Invalid code"), 200) cur.close() except Exception as e: logging.info(e) cur.close() return make_response(jsonify(code=500, error="Invalid code"), 200) return make_response(jsonify(code=200, success="Valid code"), 200) @app.route('/bulk/info', methods=['POST', 'GET']) @auth.login_required def bulk_info(): app.config['MYSQL_HOST'] = str(live_hellio_wr["host"]) app.config['MYSQL_USER'] = str(live_hellio_wr["user"]) app.config['MYSQL_PORT'] = int(live_hellio_wr["port"]) app.config['MYSQL_PASSWORD'] = str(live_hellio_wr["password"]) app.config['MYSQL_DB'] = str(live_hellio_wr["db"]) if request.method == 'POST': req = request.get_json() username= req['username'] total_traffic = req['total_traffic'] message = req['message'] sms_count = req['sms_count'] file_path = req['file_path'] user_id = req['user_id'] user_id = int(user_id) name = req['campaign_name'] sender = req['sender_id'] sch_datetime = parser.parse(req['sched_datetime']).isoformat() sch_datetime = sch_datetime.replace('T', " ") approval_status = req['approval_status'] created_by = req['created_by'] created_at = req['created_at'] message_type = req['message_type'] b_type = req['type'] logging.info(f'bulk type : {b_type}') cur = mysql.connection.cursor() # copy file to server 4 cmd = os.system(f"/usr/bin/sudo /usr/bin/sshpass -p {scp_info['password']} /usr/bin/scp -P {scp_info['port']} {file_path} {scp_info['username']}@{scp_info['host']}:{scp_info['destination']}") if not cmd: file_path = str(scp_info["destination"]) + os.path.basename(file_path) logging.info(f'file transfer successful:{file_path}') else: logging.info('file transfer failed') return make_response(jsonify(code=500, error='file transfer failed')) if approval_status.lower() == "approved": try: sch_datetime = datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S') format_datetime = datetime.strptime(sch_datetime, '%Y-%m-%d %H:%M:%S') if format_datetime > datetime.utcnow() + timedelta(minutes=2): sched_datetime = format_datetime.strftime('%Y-%m-%d %H:%M:%S') sql = "INSERT INTO user_jobs(name, user_id, username, sender, message, sms_count, total_traffic, file_path, sched_datetime, status, message_type, created_at, created_by, type) VALUES(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)" val = (name, user_id, username, sender, message, sms_count, total_traffic, file_path, sched_datetime, "pending", message_type, created_at, created_by, b_type) cur.execute(sql, val) mysql.connection.commit() else: format_datetime = datetime.utcnow() + timedelta(minutes=2) sched_datetime = format_datetime.strftime('%Y-%m-%d %H:%M:%S') sql = "INSERT INTO user_jobs(name, user_id, username, sender, message, sms_count, total_traffic, file_path, sched_datetime, status, message_type, created_at, created_by, type) VALUES(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)" val = (name, user_id, username, sender, message, sms_count, total_traffic, file_path, sched_datetime, "pending", message_type, created_at, created_by, b_type) cur.execute(sql, val) mysql.connection.commit() except Exception as e: logging.info(e) cur.close() return make_response(jsonify(code=404, error='user job does not exist'), 200) elif approval_status.lower() == "rejected": try: sch_datetime = datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S') format_datetime = datetime.strptime(sch_datetime, '%Y-%m-%d %H:%M:%S') sched_datetime = format_datetime.strftime('%Y-%m-%d %H:%M:%S') sql = "INSERT INTO user_jobs(name, user_id, username, sender, message, sms_count, total_traffic, file_path, sched_datetime, status, message_type, created_at, created_by, type) VALUES(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)" val = (name, user_id, username, sender, message, sms_count, total_traffic, file_path, sched_datetime, "pending", message_type, created_at, created_by, b_type) cur.execute(sql, val) mysql.connection.commit() except Exception as e: logging.info(e) cur.close() return make_response(jsonify(code=404, error='user job does not exist'), 200) return make_response(jsonify(code=403, error="user job has been rejected!"),200) else: return make_response(jsonify(code=400, error="user job has not been approved!"),200) return make_response(jsonify(code=200, message="Message queued"),200) if __name__ =='__main__': app.run(host='0.0.0.0')