WSGI Python Flask apache2 на ubuntu возвращает ошибку
У меня есть VPS сервер, на котором нужно запустить API для обработки регистраций. При запуске apache сервера (по этой инструкции - https://www.8host.com/blog/razvyortyvanie-prilozheniya-flask-na-servere-ubuntu/) вылезает ошибка SyntaxError: invalid syntax.
[Wed Aug 16 13:35:07.173522 2023] [wsgi:error] [pid 63783:tid 139809708308224] Traceback (most recent call last):
[Wed Aug 16 13:35:07.173537 2023] [wsgi:error] [pid 63783:tid 139809708308224] File "/var/www/API/API.wsgi", line 6, in <module>
[Wed Aug 16 13:35:07.173575 2023] [wsgi:error] [pid 63783:tid 139809708308224] from API import app as application
[Wed Aug 16 13:35:07.173602 2023] [wsgi:error] [pid 63783:tid 139809708308224] File "/var/www/API/API/__init__.py", line 54
[Wed Aug 16 13:35:07.173606 2023] [wsgi:error] [pid 63783:tid 139809708308224] async def start(message: types.Message):
[Wed Aug 16 13:35:07.173609 2023] [wsgi:error] [pid 63783:tid 139809708308224] ^
[Wed Aug 16 13:35:07.173612 2023] [wsgi:error] [pid 63783:tid 139809708308224] SyntaxError: invalid syntax
Хотя при запуске "в лоб" через SSH соединение в консоли Ubuntu всё работет, ошибок не выдаёт.
Вот сам код, некоторые функции закоментированны или пока не выполняют то, что я хочу, но это не первостепенно.
# -*- coding: utf-8 -*-
import asyncio
import contextvars
import functools
import random
import sqlite3
from datetime import datetime
from flask import Flask, jsonify, request
from flask_cors import CORS
from aiogram import Bot, types
from aiogram.dispatcher import Dispatcher
from aiogram.utils import executor
import functions as func
# try:
# conn = sqlite3.connect("db/database.db")
# cursor = conn.cursor()
#
# except sqlite3.Error as error:
# print("Возникла ошибка в Д - ", error)
bot = Bot(token="MY-TOKEN")
dp = Dispatcher(bot)
# @dp.message_handler(commands=["start"])
# async def start(message: types.Message):
#
# try:
# args = str(message.get_args())
#
# local_database_id = cursor.execute("SELECT id FROM users WHERE local_id = (?)", (args,)).fetchone()
#
# if local_database_id:
# await bot.send_message(message.from_user.id, "Регистрация завершена, ваш ID - " + args)
# except:
# await bot.send_message(message.from_user.id, "Привет. Это бот от Sample Code. \nВот наш сайт: https://")
import re
import sqlite3
import random
# from telebot import TeleBot, types
try:
conn = sqlite3.connect("db/database.db")
cursor = conn.cursor()
except sqlite3.Error as error:
print("Возникла ошибка в БД - ", error)
# bot = TeleBot(token="мой-токен")
@dp.message_handler(commands=["start"])
async def start(message: types.Message):
try:
args = extract_start_argument(message.text)
local_database_id = cursor.execute("SELECT id, password, phone_verification, phone FROM users WHERE local_id = ?", (args,)).fetchone()
print(1)
if local_database_id:
user_id, password, phone_verification, phone = local_database_id
print(2)
if not password:
password = generate_password()
cursor.execute("UPDATE users SET password = ? WHERE id = ?", (password, user_id))
cursor.execute("UPDATE users SET telegram_id = ? WHERE id = ?", (message.from_user.id, user_id))
conn.commit()
print(3)
if not phone_verification:
await bot.send_message(user_id,
f"Привет! Пожалуйста, подтверди свой номер телефона.", reply_markup=create_phone_verification_keyboard())
else:
await bot.send_message(user_id, "Регистрация уже завершена.")
else:
await bot.send_message(message.from_user.id, "Неверный local_id.")
except:
await bot.send_message(message.from_user.id, "Привет. Это бот от Sample Code. \nВот наш сайт: https://")
@dp.message_handler(content_types=["contact"])
async def handle_contact(message: types.Message):
try:
user_id = message.from_user.id
local_database_id = cursor.execute("SELECT id, phone, phone_verification FROM users WHERE telegram_id = ?",
(user_id,)).fetchone()
if local_database_id:
_, db_phone, phone_verification = local_database_id
provided_phone = message.contact.phone_number
if db_phone == provided_phone:
if not phone_verification:
cursor.execute("UPDATE users SET phone_verification = ? WHERE id = ?", ("verified", _))
conn.commit()
password = cursor.execute("SELECT password FORM users WHERE id = ?", (_,)).fetchone()
formatted_password = f"`{password}`"
await bot.send_message(user_id, f"Номер телефона успешно подтверждён. Твой пароль - {formatted_password}", parse_mode="MarkdownV2")
else:
await bot.send_message(user_id, "Номер телефона уже был подтверждён ранее.")
else:
cursor.execute("UPDATE users SET phone = ? WHERE id = ?", (provided_phone, user_id))
conn.commit()
await bot.send_message(user_id, "Номер телефона обновлён. Пожалуйста, подтвердите его.")
else:
await bot.send_message(user_id, "Ошибка: пользователь не найден.")
except:
await bot.send_message(user_id, "Произошла ошибка при обработке контакта.")
def generate_password(length=12):
characters = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789!@#$%^&*()_+-=[]{}|;:,.<>?~"
password = "".join(random.choice(characters) for _ in range(length))
return password
def create_phone_verification_keyboard():
keyboard = types.ReplyKeyboardMarkup(one_time_keyboard=True, resize_keyboard=True)
button = types.KeyboardButton("Подтвердить номер телефона", request_contact=True)
keyboard.add(button)
return keyboard
def extract_start_argument(text):
match = re.search(r"/start\s+(\S+)", text)
if match:
return match.group(1)
return None
# if __name__ == "__main__":
# bot.polling()
async def to_thread(func, /, *args, **kwargs):
loop = asyncio.get_running_loop()
ctx = contextvars.copy_context()
func_call = functools.partial(ctx.run(), func, *args, **kwargs)
return await loop.run_in_executor(None, func_call)
# app = Flask(__name__, subdomain_matching=True)
# app.config['SERVER_NAME'] = "api.сайт.ru:5555"
# app.config['SERVER_NAME'] = "api.сайт.ru"
app = Flask(__name__)
CORS(app)
@app.route("/api")
async def api():
return "<h1><br>To get access to the API, please contact us - почта@gmail.com</br></h1>"
@app.route("/api/register", methods=['GET'])
async def register():
try:
conn = sqlite3.connect("db/database.db")
cursor = conn.cursor()
except sqlite3.Error as error:
print("Возникла ошибка в БД - ", error)
def phone_unically(phone):
phone_db = cursor.execute("SELECT phone FROM users WHERE phone = (?)", (phone,)).fetchone()
if phone_db:
return False
else:
return True
def email_unically(email):
email_db = cursor.execute("SELECT email FROM users WHERE email = (?)", (email,)).fetchone()
if email_db:
return False
else:
return True
try:
full_name = request.args.get('full_name')
except:
return jsonify({"error": "ФИО не указано", "status":"bad"})
try:
birth_date = request.args.get('birth_date')
current_date = datetime.now()
# Преобразуем строку с датой рождения в объект datetime
birth_date_1 = datetime.strptime(birth_date, "%Y-%m-%d")
# Вычисляем разницу между текущей датой и датой рождения
age = current_date.year - birth_date_1.year
# Проверяем, достиг ли возраст 18 лет или более
if current_date.month < birth_date_1.month or (
current_date.month == birth_date_1.month and current_date.day < birth_date_1.day):
age -= 1
if age >= 14:
pass
else:
return jsonify({"error": "Вам меньше 14 лет", "status":"bad"})
except:
return jsonify({"error": "Дата рождения не указана", "status":"bad"})
try:
email = request.args.get('email')
if email_unically(email):
pass
else:
return jsonify({"error": "Указанный EMAIL уже был использован для регистрации", "status":"bad"})
except:
return jsonify({"error": "EMAIL не указан", "status":"bad"})
try:
phone = request.args.get('phone').replace(" ", "")
if phone_unically(phone):
pass
else:
return jsonify({"error": "Указанный номер телефона уже был использован для регистрации", "status":"bad"})
except:
return jsonify({"error": "Номер телефона не указан", "status":"bad"})
local_id = str(random.randint(0, 999999))
print(local_id)
cursor.execute("INSERT INTO users (full_name, birth_date, email, phone, local_id) VALUES (?,?,?,?,?)", (full_name, birth_date, email, phone, local_id,))
conn.commit()
# user_id = await func.get_chat_id(phone)
# print(user_id)
# await bot.send_message(user_id, "ТЕСТ")
await bot.send_message(1060614786, f"<b>Новая регистрация на курс! \n\nДанные:</b> \nФИО: {full_name}\nТелефон: {phone}\nEmail: {email}\nLocal id: {local_id}", parse_mode="HTML")
return jsonify({"error": "", "status":"ok", "local_id": local_id})
@app.route("/api/no_tg", methods=["GET"])
async def no_tg():
try:
local_id = request.args.get('local_id')
except:
return jsonify({"error": "ФИО не указано", "status":"bad"})
await bot.send_message(1060614786, f"Пользователь *{local_id}* не может подтвердить регистрацию в телеграмме, отпрвьте инструкцию на почту")
return jsonify({"status":"ok"})
async def start(host, port):
# asyncio.create_task(await to_thread(app.run(host=host, port=port)))
asyncio.create_task(await to_thread(app.run()))
if __name__ == "__main__":
async def start_func():
asyncio.create_task(await to_thread(executor.start_polling(dispatcher=dp, skip_updates=False, on_startup=await start(host="0.0.0.0", port="5555"))))
asyncio.run(start_func())
Источник: Stack Overflow на русском