119 lines
4.3 KiB
Python
119 lines
4.3 KiB
Python
import schedule
|
|
import signal
|
|
import json
|
|
import time
|
|
import socket
|
|
import logging
|
|
from utils.ColoredFormatter import ColoredFormatter
|
|
from bs4 import BeautifulSoup as bs
|
|
from selenium import webdriver
|
|
from selenium.webdriver.chrome.options import Options
|
|
from selenium.webdriver.common.by import By
|
|
from selenium.webdriver import ActionChains
|
|
from selenium.webdriver.support.wait import WebDriverWait
|
|
from selenium.webdriver.support import expected_conditions as ec
|
|
|
|
def setupLogger():
|
|
logging_format = "[%(asctime)s] (%(levelname)s) %(module)s - %(funcName)s: %(message)s"
|
|
logging.basicConfig(level=logging.INFO, format=logging_format)
|
|
(logger := logging.getLogger(__name__)).setLevel(logging.INFO)
|
|
logger.propagate = False
|
|
(logger_handler := logging.StreamHandler()).setFormatter(
|
|
ColoredFormatter(fmt=logging_format)
|
|
)
|
|
logger.addHandler(logger_handler)
|
|
|
|
def interrupt_handler(signum, driver: webdriver.Chrome):
|
|
print(f'Handling signal {signum} ({signal.Signals(signum).name}).')
|
|
|
|
schedule.clear()
|
|
driver.quit()
|
|
raise SystemExit
|
|
|
|
def setupChromeDriver():
|
|
logging.log(logging.CRITICAL, 'Setting up Chrome driver.')
|
|
chrome_options = Options()
|
|
chrome_options.add_argument("--headless")
|
|
chrome_options.add_argument("--no-sandbox")
|
|
chrome_options.add_argument("--mute-audio")
|
|
chrome_options.add_argument("--window-size=1280,720")
|
|
chrome_options.add_argument("--disable-dev-shm-usage")
|
|
chrome_options.add_argument("--disable-features=WebRtcHideLocalIpsWithMdns")
|
|
chrome_options.add_experimental_option('prefs', {'intl.accept_languages': 'en,en_US'})
|
|
#chrome_options.add_extension('./qryn-webrtc-exporter.crx')
|
|
|
|
driver = webdriver.Chrome(options=chrome_options)
|
|
#driver = webdriver.Remote(command_executor='http://localhost:4444', options=chrome_options)
|
|
return driver
|
|
|
|
def saveStats(stats: dict):
|
|
try:
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
sock.sendto(json.dumps(stats).encode(), ('localhost', 8094))
|
|
sock.close()
|
|
print(f'Sent stats: {stats}')
|
|
except socket.error as e:
|
|
print(f'Got socket error: {e}')
|
|
|
|
def downloadStats(driver: webdriver.Chrome):
|
|
html = driver.find_element(By.CLASS_NAME ,'vjs-stats-list').get_attribute('innerHTML')
|
|
if html is not None:
|
|
htmlBS = bs(html, 'html.parser')
|
|
else:
|
|
raise ValueError("html is None")
|
|
|
|
stats = htmlBS.find_all('div', attrs={'style': 'display: block;'})
|
|
|
|
statsDict = {
|
|
stat.div.text: stat.span.text.replace('\u21d3', 'down').replace('down/', 'down /').replace('\u21d1 ', 'up').replace('\u21d1', 'up').replace('\u00b7', '-').strip()
|
|
for stat in stats
|
|
}
|
|
|
|
for stat in statsDict:
|
|
if 'Buffer State' == stat:
|
|
statsDict[stat] = statsDict[stat][1:-1].split(', ')
|
|
|
|
# statsDict = {
|
|
# 'userName': dict(
|
|
# map(
|
|
# lambda stat: (
|
|
# stat.div.text,
|
|
# stat.span.text.replace('\u21d3', 'down').replace('down/', 'down /').replace('\u21d1 ', 'up').replace('\u21d1', 'up').replace('\u00b7', '-').strip()
|
|
# ), stats
|
|
# )
|
|
# )
|
|
# }
|
|
|
|
statsDict.update({'Timestamp': time.strftime('%Y-%m-%dT%H:%M:%S%z')})
|
|
statsDict['userName'] = 'user'
|
|
|
|
saveStats(statsDict)
|
|
|
|
def setupStats(driver: webdriver.Chrome, url: str):
|
|
actions = ActionChains(driver)
|
|
wait = WebDriverWait(driver, 30, poll_frequency=0.2)
|
|
|
|
driver.get(url)
|
|
|
|
wait.until(ec.presence_of_element_located((By.CLASS_NAME, 'vjs-big-play-button')))
|
|
actions.click(driver.find_element(By.CLASS_NAME ,'video-js')).perform()
|
|
actions.context_click(driver.find_element(By.CLASS_NAME ,'video-js')).perform()
|
|
statsForNerds = driver.find_elements(By.CLASS_NAME ,'vjs-menu-item')
|
|
actions.pause(2)
|
|
actions.click(statsForNerds[-1]).perform()
|
|
wait.until(ec.text_to_be_present_in_element((By.CLASS_NAME, 'vjs-stats-list'), 'Player'))
|
|
|
|
return driver
|
|
|
|
if __name__ == '__main__':
|
|
setupLogger()
|
|
|
|
driver = setupChromeDriver()
|
|
|
|
signal.signal(signal.SIGINT, lambda signum, frame: interrupt_handler(signum, driver))
|
|
|
|
setupStats(driver, "https://tube.kobim.cloud/w/9hAbiwai4rsbw9QnPpPkCd")
|
|
|
|
schedule.every(1).seconds.do(downloadStats, driver)
|
|
while True:
|
|
schedule.run_pending() |