psy2 / app.py
DmitrMakeev's picture
Update app.py
568706e verified
raw
history blame
19.2 kB
from flask import Flask, request, render_template_string, render_template, jsonify
import sqlite3
import os
import random
import requests
import time
import re
import json
import base64
from unidecode import unidecode
api_key_sys = os.getenv('api_key_sys')
gc_url = os.getenv('gc_url')
gc_api = os.getenv('gc_api')
wa_url = os.getenv('wa_url')
wa_api_key = os.getenv('wa_api_key')
app = Flask(__name__, template_folder="./")
app.config['DEBUG'] = True
UPLOAD_FOLDER = 'static'
if not os.path.exists(UPLOAD_FOLDER):
os.makedirs(UPLOAD_FOLDER)
DATABASES = ['data_gc.db', 'data1.db', 'data2.db', 'data3.db', 'data4.db', 'data5.db']
def init_db(db_name):
conn = sqlite3.connect(db_name)
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS contacts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
phone TEXT NOT NULL,
email TEXT NOT NULL,
vk_id TEXT NOT NULL,
chat_id TEXT NOT NULL,
ws_statys TEXT NOT NULL,
ws_stop TEXT NOT NULL,
web_statys INTEGER,
fin_progress INTEGER,
shop_statys_full TEXT NOT NULL,
curator TEXT NOT NULL,
shop_statys_cur TEXT NOT NULL,
partner TEXT NOT NULL,
pr1 TEXT NOT NULL,
pr2 TEXT NOT NULL,
pr3 TEXT NOT NULL,
pr4 TEXT NOT NULL,
pr5 TEXT NOT NULL,
ad_url TEXT NOT NULL
)
''')
conn.commit()
conn.close()
for db in DATABASES:
init_db(db)
def randomize_message(template):
def replace_placeholder(match):
options = match.group(1).split('|')
return random.choice(options)
return re.sub(r'\{([^}]+)\}', replace_placeholder, template)
def clean_phone_number(phone):
if phone.startswith('+'):
return phone[1:]
return phone
def send_message(chat_id, message):
#base_url = os.getenv('wa_url')
#api_key = os.getenv('wa_api_key')
full_url = f"{wa_url}{wa_api_key}"
payload = {
"chatId": chat_id,
"message": message
}
headers = {
'Content-Type': 'application/json'
}
response = requests.request("POST", full_url, headers=headers, json=payload)
try:
response_json = response.json()
except ValueError:
response_json = {"error": "Invalid JSON response"}
return response_json
def check_and_send_mailings(mesage_db1, clean_db):
try:
results = []
for database in DATABASES:
conn = sqlite3.connect(database)
cursor = conn.cursor()
cursor.execute('SELECT name, phone FROM contacts')
contacts = cursor.fetchall()
conn.close()
for contact in contacts:
name = contact[0]
chat_id = f"{clean_phone_number(contact[1])}@c.us"
message = randomize_message(mesage_db1)
message = message.replace('[[nemes]]', name) # Подстановка имени
send_result = send_message(chat_id, message)
results.append({
"chat_id": chat_id,
"message": message,
"result": send_result
})
if clean_db == '1':
conn = sqlite3.connect(database)
cursor = conn.cursor()
cursor.execute('DELETE FROM contacts')
conn.commit()
conn.close()
return jsonify({"status": "success", "results": results}), 200
except Exception as e:
print(f"Error sending mailings: {e}")
return jsonify({"status": "error", "message": str(e)}), 500
#С проверкой sys
@app.route('/start_db', methods=['GET'])
def start_mailings():
mesage_db1 = request.args.get('mesage')
clean_db = request.args.get('clean_db')
api_sys_control = request.args.get('api_sys')
if not mesage_db1:
return "Parameter 'mesage' is required.", 400
if api_sys_control != api_key_sys:
return "EUR 22", 200
return check_and_send_mailings(mesage_db1, clean_db)
@app.route('/add_data_gc', methods=['GET'])
def add_data_gc():
try:
name = request.args.get('name')
phone = request.args.get('phone')
email = request.args.get('email')
api_sys_control = request.args.get('api_sys')
if not name or not phone or not email:
return "Parameters 'name', 'phone', and 'email' are required.", 400
if api_sys_control != api_key_sys:
return "EUR 22", 200
# Проверка и удаление плюса в начале номера телефона, если он есть
if phone.startswith('+'):
phone = phone[1:]
conn = sqlite3.connect('data_gc.db')
cursor = conn.cursor()
cursor.execute('SELECT * FROM contacts WHERE phone = ? OR email = ?', (phone, email))
existing_contact = cursor.fetchone()
if existing_contact:
return f"Contact with phone {phone} or email {email} already exists."
cursor.execute('''
INSERT INTO contacts (name, phone, email, ws_statys, ws_stop, web_statys, fin_progress, shop_statys, curator, partner, pr1, pr2, pr3)
VALUES (?, ?, ?, 0, 0, 0, 0, '', '', '', '', '', '')
''', (name, phone, email))
conn.commit()
conn.close()
return f"Contact added: {name} - {phone} - {email}", 200
except Exception as e:
print(f"Error adding contact: {e}")
return "Internal Server Error", 500
@app.route('/add_data_ras', methods=['GET'])
def add_contact():
try:
name = request.args.get('name')
phone = request.args.get('phone')
email = request.args.get('email')
api_sys_control = request.args.get('api_sys')
if not name or not phone or not email:
return "Parameters 'name', 'phone', and 'email' are required.", 400
if api_sys_control != api_key_sys:
return "EUR 22", 200
# Проверка и удаление плюса в начале номера телефона, если он есть
if phone.startswith('+'):
phone = phone[1:]
conn = sqlite3.connect('data1.db')
cursor = conn.cursor()
cursor.execute('INSERT INTO contacts (name, phone, email) VALUES (?, ?, ?)', (name, phone, email))
conn.commit()
conn.close()
return f"Contact added: {name} - {phone} - {email}", 200
except Exception as e:
print(f"Error adding contact: {e}")
return "Internal Server Error", 500
@app.route('/data_gc')
def show_data_gc():
try:
api_sys_control = request.args.get('api_sys')
if api_sys_control != api_key_sys:
return "EUR 22", 200
conn = sqlite3.connect('data_gc.db')
cursor = conn.cursor()
cursor.execute('SELECT name, phone, email FROM contacts')
contacts = cursor.fetchall()
cursor.execute('SELECT COUNT(*) FROM contacts')
total_users = cursor.fetchone()[0]
conn.close()
return render_template('data_gc.html', contacts=contacts, total_users=total_users)
except Exception as e:
print(f"Error showing contacts: {e}")
return "Internal Server Error", 500
@app.route('/data_gc_tab', methods=['GET'])
def data_gc_tab():
api_sys_control = request.args.get('api_sys')
if api_sys_control != api_key_sys:
return "EUR 22", 200
return render_template('data_gc_tab.html')
@app.route('/data_gc_tab_out', methods=['GET'])
def data_gc_tab_out():
try:
api_sys_control = request.args.get('api_sys')
if api_sys_control != api_key_sys:
return "EUR 22", 200
conn = sqlite3.connect('data_gc.db')
cursor = conn.cursor()
cursor.execute('SELECT name, phone, email FROM contacts')
contacts = cursor.fetchall()
conn.close()
contacts_json = [{'name': contact[0], 'phone': contact[1], 'email': contact[2]} for contact in contacts]
return jsonify(contacts_json), 200
except Exception as e:
print(f"Error getting data from data_gc: {e}")
return "Internal Server Error", 500
@app.route('/data_ras')
def show_data_ras():
try:
api_sys_control = request.args.get('api_sys')
if api_sys_control != api_key_sys:
return "EUR 22", 200
conn = sqlite3.connect('data1.db')
cursor = conn.cursor()
cursor.execute('SELECT name, phone, email FROM contacts')
contacts = cursor.fetchall()
cursor.execute('SELECT COUNT(*) FROM contacts')
total_users = cursor.fetchone()[0]
conn.close()
return render_template('data_ras.html', contacts=contacts, total_users=total_users)
except Exception as e:
print(f"Error showing contacts: {e}")
return "Internal Server Error", 500
@app.route('/ver', methods=['GET'])
def veref():
api_sys_control = request.args.get('api_sys')
if api_sys_control != api_key_sys:
return "EUR 22", 200
return render_template('ver.html')
@app.route('/se_mes', methods=['GET'])
def se_mes():
api_sys_control = request.args.get('api_sys')
if api_sys_control != api_key_sys:
return "EUR 22", 200
return render_template('se_mes.html')
@app.route('/se_mes_im', methods=['GET'])
def se_mes_im():
api_sys_control = request.args.get('api_sys')
if api_sys_control != api_key_sys:
return "EUR 22", 200
return render_template('se_mes_im.html')
@app.route('/se_mes_ran', methods=['GET'])
def se_mes_ran():
api_sys_control = request.args.get('api_sys')
if api_sys_control != api_key_sys:
return "EUR 22", 200
return render_template('se_mes_ran.html')
@app.route('/se_mes_im_ran', methods=['GET'])
def se_mes_im_ran():
api_sys_control = request.args.get('api_sys')
if api_sys_control != api_key_sys:
return "EUR 22", 200
return render_template('se_mes_im_ran.html')
@app.route('/se_mes_im2', methods=['GET'])
def se_mes_im2():
api_sys_control = request.args.get('api_sys')
if api_sys_control != api_key_sys:
return "EUR 22", 200
return render_template('se_mes_im2.html')
@app.route('/se_mes_f', methods=['GET'])
def se_mes_f():
api_sys_control = request.args.get('api_sys')
if api_sys_control != api_key_sys:
return "EUR 22", 200
return render_template('se_mes_f.html')
@app.route('/up_gr', methods=['GET'])
def up_gr():
api_sys_control = request.args.get('api_sys')
if api_sys_control != api_key_sys:
return "EUR 22", 200
return render_template('up_gr.html')
@app.route('/up_user_gp', methods=['GET'])
def up_user_gp():
api_sys_control = request.args.get('api_sys')
if api_sys_control != api_key_sys:
return "EUR 22", 200
return render_template('up_user_gp.html')
@app.route('/del_user_gp', methods=['GET'])
def del_user_gp():
api_sys_control = request.args.get('api_sys')
if api_sys_control != api_key_sys:
return "EUR 22", 200
return render_template('del_user_gp.html')
@app.route('/up_ad', methods=['GET'])
def up_ad():
api_sys_control = request.args.get('api_sys')
if api_sys_control != api_key_sys:
return "EUR 22", 200
return render_template('up_ad.html')
@app.route('/del_ad', methods=['GET'])
def del_ad():
api_sys_control = request.args.get('api_sys')
if api_sys_control != api_key_sys:
return "EUR 22", 200
return render_template('del_ad.html')
@app.route('/se_opr', methods=['GET'])
def se_opr():
api_sys_control = request.args.get('api_sys')
if api_sys_control != api_key_sys:
return "EUR 22", 200
return render_template('se_opr.html')
@app.route('/online', methods=['GET'])
def online():
api_sys_control = request.args.get('api_sys')
if api_sys_control != api_key_sys:
return "EUR 22", 200
return render_template('online.html')
@app.route('/se_mes_f_gc', methods=['GET'])
def se_mes_f_gc():
api_sys_control = request.args.get('api_sys')
if api_sys_control != api_key_sys:
return "EUR 22", 200
return render_template('se_mes_f_gc.html')
@app.route('/total_users', methods=['GET'])
def total_users():
try:
api_sys_control = request.args.get('api_sys')
if api_sys_control != api_key_sys:
return "EUR 22", 200
total_users_gc = 0
total_users_ras = 0
conn_gc = sqlite3.connect('data_gc.db')
cursor_gc = conn_gc.cursor()
cursor_gc.execute('SELECT COUNT(*) FROM contacts')
total_users_gc = cursor_gc.fetchone()[0]
conn_gc.close()
conn_ras = sqlite3.connect('data1.db')
cursor_ras = conn_ras.cursor()
cursor_ras.execute('SELECT COUNT(*) FROM contacts')
total_users_ras = cursor_ras.fetchone()[0]
conn_ras.close()
return jsonify({
"total_users_gc": total_users_gc,
"total_users_ras": total_users_ras
}), 200
except Exception as e:
print(f"Error getting total users: {e}")
return "Internal Server Error", 500
@app.route('/all_users_gc', methods=['GET'])
def all_users_gc():
try:
api_sys_control = request.args.get('api_sys')
if api_sys_control != api_key_sys:
return "EUR 22", 200
conn = sqlite3.connect('data_gc.db')
cursor = conn.cursor()
cursor.execute('SELECT name, phone, email FROM contacts')
contacts = cursor.fetchall()
conn.close()
return jsonify(contacts), 200
except Exception as e:
print(f"Error getting all users from data_gc: {e}")
return "Internal Server Error", 500
@app.route('/all_users_ras', methods=['GET'])
def all_users_ras():
try:
api_sys_control = request.args.get('api_sys')
if api_sys_control != api_key_sys:
return "EUR 22", 200
conn = sqlite3.connect('data1.db')
cursor = conn.cursor()
cursor.execute('SELECT name, phone, email FROM contacts')
contacts = cursor.fetchall()
conn.close()
return jsonify(contacts), 200
except Exception as e:
print(f"Error getting all users from data_ras: {e}")
return "Internal Server Error", 500
@app.route('/gc_db_no_email', methods=['GET'])
def gc_db_no_email():
try:
api_sys_control = request.args.get('api_sys')
if api_sys_control != api_key_sys:
return "EUR 22", 200
# Чтение параметров из GET-запроса
name_d = request.args.get('name', '')
phone_d = request.args.get('phone', '')
pr1_d = request.args.get('pr1', '')
pr2_d = request.args.get('pr2', '')
pr3_d = request.args.get('pr3', '')
# Проверка базы данных на наличие email по номеру телефона
conn = sqlite3.connect('data_gc.db')
cursor = conn.cursor()
cursor.execute('SELECT email FROM contacts WHERE phone = ?', (phone_d,))
result = cursor.fetchone()
email_d = result[0] if result else ''
conn.close()
# Формирование JSON
json_data = {
"user": {
"email": email_d,
"phone": phone_d,
"first_name": name_d,
"addfields": {
"pr1": pr1_d,
"pr2": pr2_d,
"pr3": pr3_d
}
},
"system": {
"refresh_if_exists": 1
},
"session": {
"utm_source": "",
"utm_medium": "",
"utm_content": "",
"utm_campaign": "",
"utm_group": "",
"gcpc": "",
"gcao": "",
"referer": ""
}
}
# Конвертация JSON в Base64
json_str = json.dumps(json_data)
params_d = base64.b64encode(json_str.encode('utf-8')).decode('utf-8')
# Данные для отправки в теле запроса
data = {
'key': gc_api,
'action': 'add',
'params': params_d
}
# Отправка POST-запроса с данными в формате "form-data"
response = requests.post(gc_url, data=data)
# Возвращаем ответ от тестового адреса
return {
'status_code': response.status_code,
'response_body': response.text
}
except Exception as e:
print(f"Error in gc_db_no_email: {e}")
return "Internal Server Error", 500
@app.route('/gc_db_email', methods=['GET'])
def gc_db_email():
try:
api_sys_control = request.args.get('api_sys')
if api_sys_control != api_key_sys:
return "EUR 22", 200
# Чтение параметров из GET-запроса
name_d = request.args.get('name', '')
email_d = request.args.get('email', '')
phone_d = request.args.get('phone', '')
pr1_d = request.args.get('pr1', '')
pr2_d = request.args.get('pr2', '')
pr3_d = request.args.get('pr3', '')
# Формирование JSON
json_data = {
"user": {
"email": email_d,
"phone": phone_d,
"first_name": name_d,
"addfields": {
"pr1": pr1_d,
"pr2": pr2_d,
"pr3": pr3_d
}
},
"system": {
"refresh_if_exists": 1
},
"session": {
"utm_source": "",
"utm_medium": "",
"utm_content": "",
"utm_campaign": "",
"utm_group": "",
"gcpc": "",
"gcao": "",
"referer": ""
}
}
# Конвертация JSON в Base64
json_str = json.dumps(json_data)
params_d = base64.b64encode(json_str.encode('utf-8')).decode('utf-8')
# Данные для отправки в теле запроса
data = {
'key': gc_api,
'action': action_d,
'params': params_d
}
# Отправка POST-запроса с данными в формате "form-data"
response = requests.post(gc_url, data=data)
# Возвращаем ответ от тестового адреса
return {
'status_code': response.status_code,
'response_body': response.text
}
except Exception as e:
print(f"Error in gc_db_email: {e}")
return "Internal Server Error", 500
if __name__ == '__main__':
app.run(host='0.0.0.0', port=int(os.environ.get('PORT', 7860)))