Karl
1c3918354c
All checks were successful
Build and Publish Docker Image / build-and-push (push) Successful in 1m13s
84 lines
3.1 KiB
Python
84 lines
3.1 KiB
Python
import mysql.connector.pooling
|
|
from flask import jsonify, request, current_app
|
|
from ktvmanager.lib.checker import single_account_check
|
|
from ktvmanager.lib.encryption import encrypt_password, decrypt_password
|
|
from ktvmanager.lib.get_urls import get_latest_urls_from_dns
|
|
|
|
db_pool = None
|
|
|
|
def initialize_db_pool():
|
|
global db_pool
|
|
db_pool = mysql.connector.pooling.MySQLConnectionPool(
|
|
pool_name="ktv_pool",
|
|
pool_size=5,
|
|
host=current_app.config["DBHOST"],
|
|
user=current_app.config["DBUSER"],
|
|
password=current_app.config["DBPASS"],
|
|
database=current_app.config["DATABASE"],
|
|
port=current_app.config["DBPORT"]
|
|
)
|
|
|
|
def _execute_query(query, params=None):
|
|
conn = db_pool.get_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:
|
|
try:
|
|
account['password'] = decrypt_password(account['password'])
|
|
except Exception as e:
|
|
# Log the error to the console for debugging
|
|
print(f"Password decryption failed for account ID {account.get('id', 'N/A')}: {e}")
|
|
account['password'] = "DECRYPTION_FAILED"
|
|
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()
|
|
stream_urls = current_app.config["STREAM_URLS"]
|
|
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(user_id):
|
|
data = request.form
|
|
res = single_account_check(data, get_latest_urls_from_dns())
|
|
encrypted_password = encrypt_password(data['password'])
|
|
query = "INSERT INTO userAccounts (username, stream, streamURL, expiaryDate, password, userID) VALUES (%s, %s, %s, %s, %s, %s)"
|
|
params = (data['username'], data['stream'], res['url'], res['data']['user_info']['exp_date'], encrypted_password, user_id)
|
|
result = _execute_query(query, params)
|
|
return jsonify(result)
|
|
|
|
def delete_account(user_id):
|
|
data = request.form
|
|
query = "DELETE FROM userAccounts WHERE username = %s AND stream = %s AND userId = %s"
|
|
params = (data['user'],data['stream'],user_id)
|
|
result = _execute_query(query, params)
|
|
return jsonify(result)
|