Karl/ktv-python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

latest working sample

Browse Source
This commit is contained in:
Karl 2025-05-10 13:58:58 +01:00
parent 0ef3916ae7
commit 445cdc834a
17 changed files with 953 additions and 784 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
.env.sample
View File

@ -1,8 +1,8 @@
DBHOST=
DBUSER=
DBPASS=
DATABASE=
DBPORT=
TORSSRV=
TORSPWD=
DBHOST=
DBUSER=
DBPASS=
DATABASE=
DBPORT=
TORSSRV=
TORSPWD=
ENCRYPTKEY=

4
.gitignore vendored
View File

@ -1,2 +1,2 @@
.env
*.pyc
.env
*.pyc

30
.vscode/launch.json vendored
View File

@ -1,16 +1,16 @@
{
// Use IntelliSense to learn about possible attributes.
// Hover to view descriptions of existing attributes.
// For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
"version": "0.2.0",
"configurations": [
{
"name": "Python: Current File",
"type": "python",
"request": "launch",
"program": "${file}",
"console": "integratedTerminal",
"justMyCode": false
}
]
{
// Use IntelliSense to learn about possible attributes.
// Hover to view descriptions of existing attributes.
// For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
"version": "0.2.0",
"configurations": [
{
"name": "Python: Current File",
"type": "python",
"request": "launch",
"program": "${file}",
"console": "integratedTerminal",
"justMyCode": false
}
]
}

10
.vscode/settings.json vendored
View File

@ -1,6 +1,6 @@
{
"[python]": {
"editor.defaultFormatter": "ms-python.black-formatter"
},
"python.formatting.provider": "none"
{
"[python]": {
"editor.defaultFormatter": "ms-python.black-formatter"
},
"python.formatting.provider": "none"
}

6
ktvmanager/lib/DNS_list.txt
View File

@ -1,4 +1,4 @@
http://apppanel.co.uk/panel/capo/smarters/api//home.php?action=dns
http://chopzappz.xyz/v3advert/crazy/api/dns.php
http://razzlertv.xyz/customers/scotslad/virgin/api/dns.php
http://apppanel.co.uk/panel/capo/smarters/api//home.php?action=dns
http://chopzappz.xyz/v3advert/crazy/api/dns.php
http://razzlertv.xyz/customers/scotslad/virgin/api/dns.php
http://razzlertv.xyz/customers/scotslad/skyup/api/dns.php

80
ktvmanager/lib/checks.py
View File

@ -1,41 +1,41 @@
import requests
import concurrent.futures
import time
from requests_tor import RequestsTor
import os
from dotenv import load_dotenv
from ktvmanager.lib.get_urls import generate_urls_for_user
import json
load_dotenv()
rt = RequestsTor(tor_ports=(9001, 9002, 9003, 9004, 9005), tor_cport=9051, password=os.getenv('TORSPWD'), autochange_id=3)
def get_status(url):
try:
resp = rt.get(url=url, timeout=2)
if resp.status_code == 200:
if "<html><head>" not in resp.text:
response_data = json.loads(resp.text)
if response_data['user_info']['auth'] == 1:
return response_data
except:
pass
def find_url_for_user_details(username, password):
urls = generate_urls_for_user(username, password)
tm1 = time.perf_counter()
with concurrent.futures.ThreadPoolExecutor() as executor:
futures = []
for url in urls:
futures.append(executor.submit(get_status, url=url))
for future in concurrent.futures.as_completed(futures):
if future._result != None:
executor.shutdown(wait=False)
print(f"{future._result['user_info']['username']} - {future._result['server_info']['url']}:{future._result['server_info']['port']}")
tm2 = time.perf_counter()
print(f'Total time elapsed: {tm2-tm1:0.2f} seconds')
find_url_for_user_details('Karl061122', 'gkuEDWzxHD')
find_url_for_user_details('Karl130623', 'emYZWPs')
find_url_for_user_details('Karlos2306', 'Gg58Wg8MB9')
import requests
import concurrent.futures
import time
from requests_tor import RequestsTor
import os
from dotenv import load_dotenv
from get_urls import generate_urls_for_user
import json
load_dotenv()
rt = RequestsTor(tor_host=os.getenv('TORSSRV'), tor_ports=(9001, 9002, 9003, 9004, 9005), tor_cport=9051, password=os.getenv('TORSPWD'), autochange_id=3)
def get_status(url):
try:
resp = rt.get(url=url, timeout=2)
if resp.status_code == 200:
if "<html><head>" not in resp.text:
response_data = json.loads(resp.text)
if response_data['user_info']['auth'] == 1:
return response_data
except:
pass
def find_url_for_user_details(username, password):
urls = generate_urls_for_user(username, password)
tm1 = time.perf_counter()
with concurrent.futures.ThreadPoolExecutor() as executor:
futures = []
for url in urls:
futures.append(executor.submit(get_status, url=url))
for future in concurrent.futures.as_completed(futures):
if future._result != None:
executor.shutdown(wait=False)
print(f"{future._result['user_info']['username']} - {future._result['server_info']['url']}:{future._result['server_info']['port']}")
tm2 = time.perf_counter()
print(f'Total time elapsed: {tm2-tm1:0.2f} seconds')
find_url_for_user_details('Karl061122', 'gkuEDWzxHD')
find_url_for_user_details('Karl130623', 'emYZWPs')
find_url_for_user_details('Karlos2306', 'Gg58Wg8MB9')
find_url_for_user_details('Maxine2306', 'EszFDNNcb2')

92
ktvmanager/lib/database.py
View File

@ -1,46 +1,46 @@
import os
from dotenv import load_dotenv
import mysql.connector
from mysql.connector import connection
load_dotenv()
def create_connection_to_database() -> connection:
"""_summary_
Returns:
connection: _description_
"""
username = os.getenv("DBUSER")
password = os.getenv("DBPASS")
server = os.getenv("DBHOST")
database = os.getenv("DATABASE")
port = os.getenv("DBPORT")
mydb = mysql.connector.connect(
host=server, user=username, password=password, database=database, port=port
)
return mydb
def return_data_from_database(query) -> None:
connection = create_connection_to_database()
cursor = connection.cursor()
cursor.execute(query)
data = cursor.fetchall()
return data
def getUserAccounts(user) -> None:
"""_summary_
Args:
user (_type_): _description_
"""
query = "SELECT userAccounts.username, userAccounts.stream, userAccounts.streamURL, userAccounts.expiaryDate, userAccounts.password FROM users INNER JOIN userAccounts ON users.id = userAccounts.userID WHERE users.id = '1'"
a = return_data_from_database(query)
print(a)
getUserAccounts("1")
import os
from dotenv import load_dotenv
import mysql.connector
from mysql.connector import connection
load_dotenv()
def create_connection_to_database() -> connection:
"""_summary_
Returns:
connection: _description_
"""
username = os.getenv("DBUSER")
password = os.getenv("DBPASS")
server = os.getenv("DBHOST")
database = os.getenv("DATABASE")
port = os.getenv("DBPORT")
mydb = mysql.connector.connect(
host=server, user=username, password=password, database=database, port=port
)
return mydb
def return_data_from_database(query) -> None:
connection = create_connection_to_database()
cursor = connection.cursor()
cursor.execute(query)
data = cursor.fetchall()
return data
def getUserAccounts(user) -> None:
"""_summary_
Args:
user (_type_): _description_
"""
query = "SELECT userAccounts.username, userAccounts.stream, userAccounts.streamURL, userAccounts.expiaryDate, userAccounts.password FROM users INNER JOIN userAccounts ON users.id = userAccounts.userID WHERE users.id = '1'"
a = return_data_from_database(query)
print(a)
getUserAccounts("1")

34
ktvmanager/lib/encryption.py
View File

@ -1,17 +1,17 @@
from pyeasyencrypt.pyeasyencrypt import encrypt_string, decrypt_string
import os
from dotenv import load_dotenv
load_dotenv()
def encrypt_password(clear_string):
password = os.getenv("ENCRYPTKEY")
encrypted_string = encrypt_string(clear_string, password)
return encrypted_string
def decrypt_password(encrypted_string):
password = os.getenv("ENCRYPTKEY")
decrypted_string = decrypt_string(encrypted_string, password)
return decrypted_string
from pyeasyencrypt.pyeasyencrypt import encrypt_string, decrypt_string
import os
from dotenv import load_dotenv
load_dotenv()
def encrypt_password(clear_string):
password = os.getenv("ENCRYPTKEY")
encrypted_string = encrypt_string(clear_string, password)
return encrypted_string
def decrypt_password(encrypted_string):
password = os.getenv("ENCRYPTKEY")
decrypted_string = decrypt_string(encrypted_string, password)
return decrypted_string

3
ktvmanager/lib/extra_urls.txt
View File

@ -1 +1,2 @@
http://sarahgraphite.awardvpn.xyz
http://sarahgraphite.awardvpn.xyz
http://sarahgraphite.liveme.vip

97
ktvmanager/lib/get_urls.py
View File

@ -1,47 +1,50 @@
import os
from dotenv import load_dotenv
import requests
import json
from requests_tor import RequestsTor
load_dotenv()
torpass = os.getenv('TORSPWD')
rt = RequestsTor(tor_ports=(9001, 9002, 9003, 9004, 9005), tor_cport=9051, password=torpass, autochange_id=5)
def get_latest_urls_from_dns():
with open('./ktvmanager/lib/DNS_list.txt') as f:
lines = [line.rstrip('\n') for line in f]
with open('./ktvmanager/lib/extra_urls.txt') as urls:
extra_urls = [line.rstrip('\n') for line in urls]
# print(lines)
complete_list_of_urls = []
for url in lines:
data = requests.get(url)
content = json.loads(data.text)
list_of_urls = content['su'].split(',')
for url in list_of_urls:
complete_list_of_urls.append(url)
complete_list_of_urls = list(set(complete_list_of_urls))
# print(complete_list_of_urls)
for url in extra_urls:
complete_list_of_urls.append(url)
return complete_list_of_urls
def generate_urls_for_user(username, password):
new_urls = []
for url in get_latest_urls_from_dns():
hard_url = f'/player_api.php?password={password}&username={username}&action=user&sub=info'
new_url = url + hard_url
new_urls.append(new_url)
return new_urls
def check_for_url_from_details(username, password):
new_urls = []
for url in get_latest_urls_from_dns():
hard_url = f'/player_api.php?password={password}&username={username}&action=user&sub=info'
new_url = url + hard_url
new_urls.append(new_url)
print(new_urls)
import os
from dotenv import load_dotenv
import requests
import json
from requests_tor import RequestsTor
load_dotenv()
torpass = os.getenv('TORSPWD')
rt = RequestsTor(tor_ports=(9001, 9002, 9003, 9004, 9005), tor_cport=9051, password=torpass, autochange_id=5)
def get_latest_urls_from_dns():
with open('./ktvmanager/lib/DNS_list.txt') as f:
lines = [line.rstrip('\n') for line in f]
with open('./ktvmanager/lib/extra_urls.txt') as urls:
extra_urls = [line.rstrip('\n') for line in urls]
# print(lines)
complete_list_of_urls = []
for url in lines:
data = requests.get(url)
try:
content = json.loads(data.text)
except Exception:
pass
list_of_urls = content['su'].split(',')
for url in list_of_urls:
complete_list_of_urls.append(url)
complete_list_of_urls = list(set(complete_list_of_urls))
# print(complete_list_of_urls)
for url in extra_urls:
complete_list_of_urls.append(url)
return complete_list_of_urls
def generate_urls_for_user(username, password):
new_urls = []
for url in get_latest_urls_from_dns():
hard_url = f'/player_api.php?password={password}&username={username}&action=user&sub=info'
new_url = url + hard_url
new_urls.append(new_url)
return new_urls
def check_for_url_from_details(username, password):
new_urls = []
for url in get_latest_urls_from_dns():
hard_url = f'/player_api.php?password={password}&username={username}&action=user&sub=info'
new_url = url + hard_url
new_urls.append(new_url)
print(new_urls)

35
ktvmanager/server.py
View File

@ -0,0 +1,35 @@
from flask import Flask
from routes.index import index_blueprint
from routes.get_user_accounts import get_user_accounts_blueprint
from routes.single_check import single_check_blueprint
from routes.add_account import add_account_blueprint
from routes.delete_account import delete_account_blueprint
from routes.login import login_blueprint
from auth import requires_auth
from lib.get_users import get_users
app = Flask(__name__)
users = get_users()
# Routes without auth
app.register_blueprint(index_blueprint)
# Routes with basic auth
app.register_blueprint(login_blueprint, url_prefix="/login")
app.register_blueprint(get_user_accounts_blueprint, url_prefix="/getUserAccounts")
app.register_blueprint(single_check_blueprint, url_prefix="/singleCheck")
app.register_blueprint(add_account_blueprint, url_prefix="/addAccount")
app.register_blueprint(delete_account_blueprint, url_prefix="/deleteAccount")
# 404 handler
@app.errorhandler(404)
def not_found(error):
return {"error": "Not found"}, 404
# 500 error handler
@app.errorhandler(500)
def server_error(error):
return {"error": "Server error"}, 500
if __name__ == "__main__":
app.run(debug=True)

90
ktvmanager_backend/lib/get_user.py Normal file
View File

@ -0,0 +1,90 @@
import mysql.connector
def get_connection():
return mysql.connector.connect(
host='localhost',
user='your_mysql_user',
password='your_mysql_password',
database='your_database'
)
def get_user_accounts(user_id):
conn = get_connection()
cursor = conn.cursor(dictionary=True)
cursor.execute("""
SELECT
userAccounts.username,
userAccounts.stream,
userAccounts.streamURL,
userAccounts.expiaryDate,
userAccounts.password
FROM users
INNER JOIN userAccounts ON users.id = userAccounts.userID
WHERE users.id = %s
""", (user_id,))
data = cursor.fetchall()
cursor.close()
conn.close()
return data if data else 'User Not Found'
def get_user_accounts_check(user_id):
return get_user_accounts(user_id)
def get_all_user_accounts():
conn = get_connection()
cursor = conn.cursor(dictionary=True)
cursor.execute("""
SELECT
userAccounts.username,
userAccounts.password,
userAccounts.expiaryDate,
userAccounts.stream
FROM userAccounts
WHERE userAccounts.expiaryDate != '0'
ORDER BY userAccounts.id DESC
""")
data = cursor.fetchall()
cursor.close()
conn.close()
return data if data else 'User Not Found'
def get_all_unique_accounts():
conn = get_connection()
cursor = conn.cursor(dictionary=True)
cursor.execute("""
SELECT DISTINCT userAccounts.stream, userAccounts.username,
userAccounts.password, userAccounts.stream
FROM userAccounts
WHERE userAccounts.expiaryDate != '0'
GROUP BY userAccounts.stream
""")
data = cursor.fetchall()
cursor.close()
conn.close()
return data if data else 'User Not Found'
def get_user_unique_accounts(user_id):
conn = get_connection()
cursor = conn.cursor(dictionary=True)
cursor.execute("""
SELECT DISTINCT userAccounts.streamURL, userAccounts.username,
userAccounts.password, userAccounts.stream, userAccounts.streamURL
FROM userAccounts
WHERE userAccounts.expiaryDate != '0'
AND userAccounts.userId = %s
GROUP BY userAccounts.stream
ORDER BY userAccounts.stream ASC
""", (user_id,))
data = cursor.fetchall()
cursor.close()
conn.close()
return data if data else 'User Not Found'
def get_user_id(username):
conn = get_connection()
cursor = conn.cursor(dictionary=True)
cursor.execute("SELECT id FROM users WHERE userName = %s", (username,))
result = cursor.fetchone()
cursor.close()
conn.close()
return result["id"] if result else 'User Not Found'

1154
poetry.lock generated
View File

File diff suppressed because it is too large Load Diff

4
poetry.toml
View File

@ -1,2 +1,2 @@
[virtualenvs]
in-project = true
[virtualenvs]
in-project = true

44
pyproject.toml
View File

@ -1,22 +1,22 @@
[tool.poetry]
name = "ktvmanager"
version = "0.1.0"
description = ""
authors = ["Karl Hudgell <karl.hudgell@bjss.com>"]
readme = "README.md"
[tool.poetry.dependencies]
python = "^3.8"
requests = "^2.31.0"
pyodbc = "^4.0.39"
wheel = "^0.41.2"
setuptools = "^68.2.2"
python-dotenv = "^1.0.0"
mysql-connector-python = "^8.1.0"
pyeasyencrypt = "^0.1.0"
requests-tor = "^1.4"
[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"
[tool.poetry]
name = "ktvmanager"
version = "0.1.0"
description = ""
authors = ["Karl Hudgell <karl.hudgell@bjss.com>"]
readme = "README.md"
[tool.poetry.dependencies]
python = "^3.8"
requests = "^2.31.0"
pyodbc = "^4.0.39"
wheel = "^0.41.2"
setuptools = "^68.2.2"
python-dotenv = "^1.0.0"
mysql-connector-python = "^8.1.0"
pyeasyencrypt = "^0.1.0"
requests-tor = "^1.4"
[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"

16
requirements.txt Normal file
View File

@ -0,0 +1,16 @@
Brotli==1.1.0
certifi==2025.4.26
cffi==1.17.1
charset-normalizer==3.4.2
cryptography==44.0.3
idna==3.10
mysql-connector-python==9.3.0
pycparser==2.22
pyeasyencrypt==0.1.0
pyodbc==5.2.0
PySocks==1.7.1
python-dotenv==1.1.0
requests==2.32.3
requests-tor @ git+https://github.com/karl0ss/requests_tor.git@ae1e85abb3bb2c25f431bd031c6d881986c626f6
stem==1.8.2
urllib3==2.4.0

24
routes/auth.py Normal file
View File

@ -0,0 +1,24 @@
from functools import wraps
from flask import request, Response
from lib.get_users import get_users
users = get_users()
def check_auth(username, password):
return users.get(username) == password
def authenticate():
return Response(
'Could not verify your access.\n',
401,
{'WWW-Authenticate': 'Basic realm="Login Required"'}
)
def requires_auth(f):
@wraps(f)
def decorated(*args, **kwargs):
auth = request.authorization
if not auth or not check_auth(auth.username, auth.password):
return authenticate()
return f(*args, **kwargs)
return decorated