Karl
4352004ed3
This commit refactors the entire backend application into a more structured and maintainable Flask project. It introduces an application factory pattern, consolidates routes into a blueprint, and implements a robust authentication and database layer. - Introduces a Flask application factory (`create_app` in `main.py`) for better organization and testability. - Consolidates all API routes into a single blueprint (`routes/api.py`) for modularity. - Implements a new basic authentication system using a decorator (`@requires_basic_auth`) to secure all endpoints. - Refactors the database access layer with standardized query execution and connection handling. - Adds new modules for core logic, including an account checker (`checker.py`) and user retrieval (`get_users.py`). - Updates the VSCode launch configuration to support the new Flask application structure. BREAKING CHANGE: The application has been completely restructured. The old `server.py` entry point is removed. The application should now be run via the app factory in `main.py`. All API endpoints now require basic authentication.
78 lines
2.7 KiB
Python
78 lines
2.7 KiB
Python
import os
|
|
import mysql.connector
|
|
from dotenv import load_dotenv
|
|
from flask import jsonify, request
|
|
from ktvmanager.lib.checker import single_account_check
|
|
from ktvmanager.lib.encryption import decrypt_password
|
|
|
|
load_dotenv()
|
|
|
|
def _create_connection():
|
|
return mysql.connector.connect(
|
|
host=os.getenv("DBHOST"),
|
|
user=os.getenv("DBUSER"),
|
|
password=os.getenv("DBPASS"),
|
|
database=os.getenv("DATABASE"),
|
|
port=os.getenv("DBPORT")
|
|
)
|
|
|
|
def _execute_query(query, params=None):
|
|
conn = _create_connection()
|
|
cursor = conn.cursor(dictionary=True)
|
|
try:
|
|
cursor.execute(query, params)
|
|
if query.strip().upper().startswith("SELECT"):
|
|
result = cursor.fetchall()
|
|
else:
|
|
conn.commit()
|
|
result = {"affected_rows": cursor.rowcount}
|
|
return result
|
|
finally:
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
def get_user_id_from_username(username):
|
|
query = "SELECT id FROM users WHERE username = %s"
|
|
result = _execute_query(query, (username,))
|
|
if result:
|
|
return result[0]['id']
|
|
return None
|
|
|
|
def get_user_accounts(user_id):
|
|
query = "SELECT * FROM userAccounts WHERE userID = %s"
|
|
accounts = _execute_query(query, (user_id,))
|
|
for account in accounts:
|
|
account['password'] = decrypt_password(account['password'])
|
|
return jsonify(accounts)
|
|
|
|
def get_stream_names():
|
|
query = "SELECT streamName FROM streams"
|
|
results = _execute_query(query)
|
|
stream_names = [row['streamName'] for row in results]
|
|
return jsonify(stream_names)
|
|
|
|
def single_check():
|
|
data = request.get_json()
|
|
# This is a placeholder for getting stream URLs. In a real application,
|
|
# this would likely come from a database query or a configuration file.
|
|
stream_urls = ["http://example.com", "http://example.org"]
|
|
result = single_account_check(data, stream_urls)
|
|
if result:
|
|
# Here you would typically update the database with the new information
|
|
return jsonify(result)
|
|
return jsonify({"message": "All checks failed"}), 400
|
|
|
|
def add_account():
|
|
data = request.get_json()
|
|
query = "INSERT INTO userAccounts (username, stream, streamURL, expiaryDate, password, userID) VALUES (%s, %s, %s, %s, %s, %s)"
|
|
params = (data['username'], data['stream'], data['streamURL'], data['expiaryDate'], data['password'], data['userID'])
|
|
result = _execute_query(query, params)
|
|
return jsonify(result)
|
|
|
|
def delete_account():
|
|
data = request.get_json()
|
|
query = "DELETE FROM userAccounts WHERE id = %s"
|
|
params = (data['id'],)
|
|
result = _execute_query(query, params)
|
|
return jsonify(result)
|