159 lines
4.7 KiB
Python
159 lines
4.7 KiB
Python
#from git import Repo # hash ostatniego commitu
|
|
import os
|
|
import time
|
|
import toml
|
|
import lewy_db
|
|
|
|
global db, config, randomly_generated_passcode
|
|
|
|
class colors:
|
|
HEADER = '\033[95m'
|
|
OKBLUE = '\033[94m'
|
|
OKCYAN = '\033[96m'
|
|
OKGREEN = '\033[92m'
|
|
WARNING = '\033[93m'
|
|
FAIL = '\033[91m'
|
|
ENDC = '\033[0m'
|
|
BOLD = '\033[1m'
|
|
UNDERLINE = '\033[4m'
|
|
ENDL = '\n'
|
|
|
|
def safeTraverse(obj: dict, path: list, default=None):
|
|
result = obj
|
|
try:
|
|
for x in path:
|
|
result = result[x]
|
|
except (KeyError, TypeError):
|
|
result = default
|
|
print(f"error reading: {' -> '.join(path)} - returning: {default}")
|
|
finally:
|
|
return result
|
|
|
|
def getCommit() -> str | None:
|
|
try:
|
|
return Repo(search_parent_directories=True).head.object.hexsha
|
|
except Exception as e:
|
|
return None
|
|
|
|
def getCommitInFormattedHTML():
|
|
repo = "<center><p style=\"color: white;\">Brak informacji o wersji skryptu</p></center>"
|
|
commit = getCommit()
|
|
|
|
if commit is not None:
|
|
repo = f"<center><p style=\"color: white;\">Commit: <a href='https://gitea.7o7.cx/roberteam/lewangoalski/commit/{commit}'>{commit[:11]}</a></p></center>"
|
|
|
|
return repo
|
|
|
|
def getCommitWithFailsafe():
|
|
commit = getCommit()
|
|
|
|
if commit is None:
|
|
commit = "(unknown commit)"
|
|
else:
|
|
commit = "#" + commit
|
|
|
|
return commit[:12]
|
|
|
|
def ensureRandomlyGeneratedPassword():
|
|
global randomly_generated_passcode
|
|
|
|
# iff the passcode is 0, as we manually set it elsewhere!
|
|
if randomly_generated_passcode == 0:
|
|
# generate a pseudorandom one and use it in the temporary config
|
|
randomly_generated_passcode = str(int(time.time() * 1337 % 899_999 + 100_000))
|
|
|
|
print(f"{colors.WARNING}WARNING{colors.ENDC}: Default config populated with one-time, insecure pseudorandom API key: {colors.OKCYAN}{randomly_generated_passcode}{colors.ENDC}.\n"
|
|
f" The API key is not the Flask debugger PIN. You need to provide a config file for persistence!{colors.ENDL}")
|
|
|
|
def getConfig(configfile: str) -> dict:
|
|
global randomly_generated_passcode
|
|
|
|
if not os.path.exists(configfile):
|
|
dummy_config = {'general': {'db_path_url': 'sqlite:///lewangoalski.sqlite', 'is_proxied': False, 'public_facing_url': 'http://127.0.0.1:5000/', 'db_prefix': 'lewy_sqlite'}, 'api': {'api_key': 'CHANGEME'}, 'scraper': {'user-agent': ''}, 'sportsmen': {'tracked_ids': ["MVC8zHZD", "WGOY4FSt", "vgOOdZbd", "Wn6E2SED", "AiH2zDve", "dUShzrBp", "UmV9iQmE", "tpV0VX0S", "vw8ZV7HC", "Qgx2trzH", "2oMimkAU", "WfXv1DCa", "0vgcq6un", "v5HSlEAa", "4S9fNUYh"]}}
|
|
# if a passcode has not been provided by the user (config file doesn't exist, and user didn't specify it using an argument)
|
|
print(f"{colors.WARNING}WARNING{colors.ENDC}: Using default, baked in config data. {colors.ENDL}"
|
|
f" Consider copying and editing the provided example file ({colors.OKCYAN}config.example.toml{colors.ENDC}).")
|
|
|
|
ensureRandomlyGeneratedPassword()
|
|
|
|
dummy_config['api']['api_key'] = str(randomly_generated_passcode)
|
|
return dummy_config
|
|
|
|
else:
|
|
return toml.load(configfile)
|
|
|
|
def setupDb(app, config) -> lewy_db.baza:
|
|
global db
|
|
db = lewy_db.baza(app, config)
|
|
return db
|
|
|
|
def getDb() -> lewy_db.baza:
|
|
"""
|
|
Akcesor dla wrappera bazy danych wspólnego dla całego projektu
|
|
(klasy baza z lewy_db)
|
|
"""
|
|
return db
|
|
|
|
def setConfig(configfile):
|
|
"""
|
|
Zapewnia, że konfiguracja nie jest pusta,
|
|
nawet, gdy sam plik jest pusty.
|
|
|
|
:param configfile: Ścieżka do pliku
|
|
:type configfile: str
|
|
"""
|
|
global config
|
|
config = getConfig(configfile)
|
|
|
|
if safeTraverse(config['api']['api_key'], []) is None or not config['api']['api_key']:
|
|
ensureRandomlyGeneratedPassword()
|
|
config['api']['api_key'] = str(randomly_generated_passcode)
|
|
|
|
|
|
def getHeaders():
|
|
"""
|
|
Zwraca hardkodowane nagłówki do scrapowania, bądź te,
|
|
z config.toml (o ile użytkownik jakieś podał).
|
|
"""
|
|
|
|
# NOTE: use ESR user-agent
|
|
# user_agent = 'Mozilla/5.0 (Windows NT 10.0; rv:130.0) Gecko/20100101 Firefox/130.0'
|
|
user_agent = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:128.0) Gecko/20100101 Firefox/128.0'
|
|
|
|
if safeTraverse(config[scraper]['user-agent'], []) is not None:
|
|
user_agent = config[scraper]['user-agent']
|
|
|
|
return user_agent
|
|
|
|
def getUptime():
|
|
"""
|
|
Zwraca informację o czasie działania serwera.
|
|
"""
|
|
return int(time.time()) - starttime
|
|
|
|
def extractIpAndPortFromPublicUrl() -> tuple:
|
|
|
|
"""
|
|
Pobiera dane z konfiguracji i zwraca
|
|
krotkę: adres IP i port.
|
|
"""
|
|
|
|
ip, port = "127.0.0.1", "5000"
|
|
|
|
try:
|
|
url = config['general']['public_facing_url'].replace(":/", "")
|
|
url_parts = url.split('/')
|
|
ip_and_port = url_parts[1]
|
|
ip, port = ip_and_port.split(':')
|
|
except:
|
|
pass
|
|
|
|
return ip, port
|
|
|
|
# Please leave at the bottom of this file.
|
|
config = {}
|
|
configfile = "config.toml"
|
|
version = getCommitWithFailsafe()
|
|
apiVersion = "1"
|
|
randomly_generated_passcode = 0
|