diff --git a/library/azure_updater.py b/library/azure_updater.py new file mode 100644 index 0000000..a4c0d8e --- /dev/null +++ b/library/azure_updater.py @@ -0,0 +1,94 @@ +#!/usr/bin/env python3 +# This file is a part of speedtest +# Created at 01/17/2025 + +import os +import sys + +core = getattr(sys.modules["__main__"], "__file__", None) +if core: + core = os.path.abspath(core) + root = os.path.dirname(core) + if root: + os.chdir(root) + + +import requests + +URLS_FILE = "urls.txt" + +SEARCH_STRINGS = ["azurespeed.com", "windows.net", "azure"] + +BASE_URL = "https://www.azurespeed.com/api/sas?regionName={location}&blobName=100MB.bin&operation=download" + +LOCATIONS = ("australiacentral", + "australiaeast", + "australiasoutheast", + "centralindia", + "eastasia", + "japaneast", + "japanwest", + "koreacentral", + "koreasouth", + "newzealandnorth", + "southindia", + "southeastasia", + "westindia") + + +DIRECTIONS = ("southeast", "west", "south", "central", "east", "north") + + +def convert_location(location: str) -> str: + for direction in DIRECTIONS: + if direction in location: + location = location.replace(direction, "") + location = direction.capitalize() + " " + location.capitalize() + break + return location + + +def get_urls() -> list[tuple[str]]: + new_urls = [] + for location in LOCATIONS: + url = BASE_URL.format(location=location) + try: + json = requests.get(url).json() + new_url = json["url"] + except Exception: + continue + new_urls.append((location, new_url)) + return new_urls + + +def rewrite_urls(new_urls: list[tuple[str]]): + contents = [] + with open(URLS_FILE, "r") as file: + for line in file.read().splitlines(): + line = line.strip() + for str_ in SEARCH_STRINGS: + if str_.lower() in line.lower(): + break + else: + contents.append(line) + + with open(URLS_FILE, "w") as file: + for (location, new_url) in new_urls: + location = convert_location(location) + file.write("{location}|Azure|{new_url}\n".format(location=location, new_url=new_url)) + for item in contents: + file.write(item + "\n") + + + +def update_azure_lists(): + urls = get_urls() + rewrite_urls(urls) + + +def main(): + update_azure_lists() + + +if __name__ == "__main__": + main() diff --git a/speedtest.ini b/speedtest.ini index 5c3ba7b..958ced2 100644 --- a/speedtest.ini +++ b/speedtest.ini @@ -8,6 +8,8 @@ chunk_size=1024 sample_every=5 # connection timeout. This is the timeout for the server to send the first response, NOT read timeout connection_timeout=60 +# whether to update azure lists +azure_update=True [mail] smtp_host=smtp.mzjtechnology.com diff --git a/speedtest.py b/speedtest.py index 80c7cb5..80beeb8 100644 --- a/speedtest.py +++ b/speedtest.py @@ -29,6 +29,7 @@ import traceback from configparser import ConfigParser from inputimeout import inputimeout, TimeoutOccurred +from azure_updater import update_azure_lists from commons import * @@ -37,6 +38,8 @@ formatter = logging.Formatter( "%(asctime)s %(levelname)s: %(name)s: %(message)s", "%d.%m.%Y %H:%M:%S") +USER_AGENT = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36" + CONFIG_FILENAME = "speedtest.ini" URLS_FILENAME = "urls.txt" @@ -46,6 +49,8 @@ MAIL_SECTION = "mail" CHUNK_SIZE = "chunk_size" SAMPLE_EVERY = "sample_every" CONNECTION_TIMEOUT = "connection_timeout" +AZURE_UPDATE = "azure_update" + SMTP_HOST = "smtp_host" SMTP_STARTTLS = "smtp_starttls" @@ -94,11 +99,9 @@ def send_data(config: ConfigParser, email_content: str, messages: list): content_, args = text.content, text.args email_content += content_.format(*text.args) email_content += "\n" - email_content += content.footer - email_content += "\n" - email_content += "_" * 25 email_content += "\n" + from_addr = "Speedtest <%s>" % (config.get(SMTP_LOGIN)) msg = email.message.Message() msg.add_header("Content-Type", "text") @@ -106,14 +109,14 @@ def send_data(config: ConfigParser, email_content: str, messages: list): msg["Subject"] = "Speedtest report" msg["From"] = from_addr msg["To"] = config.get(SMTP_SENDTO) - client.sendmail(config.get(SMTP_LOGIN), - config.get(SMTP_SENDTO), msg.as_string()) + client.sendmail(config.get(SMTP_LOGIN), + config.get(SMTP_SENDTO), msg.as_string()) def measure_speed(url: str, chunk_size: int = 1024, sample_every: int = 5, timeout: int = 10) -> tuple[int, int]: with io.BytesIO() as f: start = time.perf_counter() - r = requests.get(url, stream=True, timeout=timeout) + r = requests.get(url, stream=True, timeout=timeout, headers={"User-Agent": USER_AGENT}) total_length = r.headers.get("content-length") dl = 0 min_ = sys.maxsize @@ -130,46 +133,51 @@ def measure_speed(url: str, chunk_size: int = 1024, sample_every: int = 5, timeo speed = dl // (time.perf_counter() - start) / 1000000 if (counter % sample_every) == 0: samples.append(speed) + + done = int(30 * dl / int(total_length)) + sys.stdout.write("\r[%s%s] %s MB/s" % ("=" * done, " " * (30-done), speed)) max_ = max(max_, speed) min_ = min(min_, speed) counter += 1 avg = sum(samples) / len(samples) + print("\n") return (avg, max_) -def test_url(url: str, chunk_size: int, sample_every: int, timeout: int) -> Message: + +def test_url(server: str, chunk_size: int, sample_every: int, timeout: int) -> tuple[int, int]: avg = 0 max_ = 0 + location, hosting, url = server.split("|") try: avg, max_ = measure_speed(url, chunk_size, sample_every, timeout) except requests.exceptions.Timeout: logger.error("Timeout to reach url: %s" % url, exc_info=True) + except requests.exceptions.ConnectionError as e: + logger.error("Unable to test url: %s (requests error: %s)" % (url, str(e))) except Exception: logger.error("Unable to test url: %s" % url, exc_info=True) else: - text = [] - text.append(Text("Average speed: {:.2f} MB/s", [avg])) - text.append(Text("Maximum speed: {:.2f} MB/s", [max_])) - content = Content("Speed test results", LineFormat(text)) - message = Message([content]) - message.server = Connection(url) - message.server.name = url - return message - - -def get_test_results(urls: list, chunk_size: int, sample_every: int, timeout: int): - messages = [] - for url in urls: - logger.debug("Going to perform speedtest on %s" % url) - message = test_url(url, chunk_size, sample_every, timeout) - if message is not None: - messages.append(message) + return avg, max_ + + +def get_test_results(servers: list[str], chunk_size: int, sample_every: int, timeout: int) -> list[dict]: + result = [] + for server in servers: + location, hosting, url = server.split("|") + logger.debug("Going to perform speedtest on %s" % server) + test_result = test_url(server, chunk_size, sample_every, timeout) + if test_result is not None: + avg, max_ = test_result + data = {"avg": avg, "max": max_, "server": server} + result.append(data) else: - logger.warning("The url %s was not tested" % url) - return messages + logger.warning("The server %s was not tested" % server) + continue + return result -def _get_urls() -> list: +def _get_test_servers() -> list: with open(URLS_FILENAME) as file: return file.read().splitlines() @@ -207,8 +215,26 @@ def get_name_and_isp() -> tuple[str, str]: return name, isp +def combine_results(results: dict) -> list[Message]: + messages = [] + for item in sorted(results, key=lambda cur: -cur["avg"]): + text = [] + avg = item["avg"] + max_ = item["max"] + location, hosting, url = item["server"].split("|") + text.append(Text("Average speed: {:.2f} MB/s", [avg])) + text.append(Text("Maximum speed: {:.2f} MB/s", [max_])) + content = Content("Speed test results", LineFormat(text)) + content.footer = "_" * 25 + message = Message([content]) + message.server = Connection(url) + message.server.name = location + " " + hosting + " " + url + messages.append(message) + return messages + + def main(): - sys.excepthook = excepthook + # sys.excepthook = excepthook parser = ConfigParser() parser.read(CONFIG_FILENAME) main_section = parser[MAIN_SECTION] @@ -218,11 +244,16 @@ def main(): sample_every = int(main_section[SAMPLE_EVERY]) chunk_size = int(main_section[CHUNK_SIZE]) timeout = int(main_section[CONNECTION_TIMEOUT]) + azure_update = main_section[AZURE_UPDATE] + if azure_update == "True": + print("Updating azure lists") + update_azure_lists() - urls = _get_urls() - logger.debug("Found %d urls" % len(urls)) + servers = _get_test_servers() + logger.debug("Found %d servers" % len(servers)) name, isp = get_name_and_isp() - messages = get_test_results(urls, chunk_size, sample_every, timeout) + results = get_test_results(servers, chunk_size, sample_every, timeout) + messages = combine_results(results) if messages is None or len(messages) == 0: print("There was an error, no tests were performed!") raise RuntimeError("No tests performed!")