studenthub_notifier.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ziele:\n",
    "- Mit Tor Netzwerk verbinden\n",
    "- fhnw studenthub einloggen\n",
    "- ects laden\n",
    "- mail benachrichtigung Änderung\n",
    "- (modul heruasfinden)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from selenium import webdriver\n",
    "from selenium.webdriver.firefox.service import Service\n",
    "from selenium.webdriver.firefox.options import Options\n",
    "from selenium.webdriver.firefox.firefox_binary import FirefoxBinary\n",
    "from selenium.webdriver.support.ui import WebDriverWait\n",
    "from selenium.webdriver.support import expected_conditions as EC\n",
    "from selenium.webdriver.common.by import By\n",
    "from selenium.webdriver.common.keys import Keys\n",
    "import subprocess\n",
    "\n",
    "#import slack_notifications as slack\n",
    "import smtplib\n",
    "from email.mime.multipart import MIMEMultipart\n",
    "from email.mime.text import MIMEText\n",
    "import random\n",
    "\n",
    "import os\n",
    "import time\n",
    "import json"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# credential file must be present and correct\n",
    "credentials = json.loads(open(\"fhnw.json\").read())\n",
    "fhnw_username = credentials['username']\n",
    "fhnw_password = credentials['password']\n",
    "email_sender = credentials['email_sender']\n",
    "email_receiver = credentials['email_receiver']\n",
    "email_password = credentials['email_password']\n",
    "tor_path = credentials['tor_path']\n",
    "repeat = credentials['repeat']\n",
    "# slack_token = credentials['slack_access']\n",
    "# slack_channel = credentials['slack_channel']\n",
    "# slack.ACCESS_TOKEN = slack_token\n",
    "# slack_username = \"studenthub_bot\"\n",
    "\n",
    "try:\n",
    "    prev_data = json.loads(open(\"studenthub.json\").read())\n",
    "    prev_credits = prev_data['credits']\n",
    "except:\n",
    "    prev_data = {}\n",
    "    prev_credits = 0\n",
    "    prev_modules = {}\n",
    "\n",
    "# dict to save newly loaded studenthub data\n",
    "data = {}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "def send_mail(subject, message):\n",
    "    server=smtplib.SMTP_SSL(\"smtp.zoho.com\", 465)\n",
    "    # smtp_server.ehlo()\n",
    "    #smtp_server.starttls()\n",
    "    server.login(email_sender, email_password)\n",
    "    #writing the message in HTML\n",
    "\n",
    "    msg = MIMEMultipart()\n",
    "    msg['From'] = email_sender\n",
    "    msg['To'] = email_receiver\n",
    "    msg['Subject'] = subject\n",
    "    body = message\n",
    "    msg.attach(MIMEText(body, 'plain'))\n",
    "\n",
    "    email_text = msg.as_string()\n",
    "    server.sendmail(email_sender, email_receiver, email_text)\n",
    "    # quit mail server\n",
    "    server.quit()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "buffer_timeout = 10\n",
    "page_load_timeout = 30\n",
    "max_retry = 3\n",
    "retry = 0\n",
    "last_error = \"\"\n",
    "\n",
    "while(True):\n",
    "    try:\n",
    "        profile_path = f'{tor_path}\\\\Browser\\\\TorBrowser\\\\Data\\\\Browser\\\\profile.default'\n",
    "        # needed to be done to work\n",
    "        os.startfile(f'{tor_path}\\\\Browser\\\\TorBrowser\\\\Tor\\\\tor.exe')\n",
    "        time.sleep(buffer_timeout)\n",
    "\n",
    "        options = Options()\n",
    "        options.set_preference('profile', profile_path)\n",
    "        options.set_preference('network.proxy.type', 1)\n",
    "        options.set_preference('network.proxy.socks', '127.0.0.1')\n",
    "        options.set_preference('network.proxy.socks_port', 9050)\n",
    "        options.set_preference(\"network.proxy.socks_remote_dns\", False)\n",
    "        #binary = FirefoxBinary(rf'{tor_path}\\Browser\\firefox.exe')\n",
    "        options.binary_location = f'{tor_path}\\\\Browser\\\\firefox.exe'\n",
    "\n",
    "        service = Service(r'./geckodriver.exe')\n",
    "\n",
    "        driver = webdriver.Firefox(service=service, options=options)\n",
    "\n",
    "        time.sleep(buffer_timeout)\n",
    "        driver.get(\"https://studenthub.technik.fhnw.ch/\")\n",
    "\n",
    "        #wait for browser to start and page to load\n",
    "        #fhnw_login = WebDriverWait(driver, page_load_timeout * 2).until(EC.presence_of_element_located((By.ID, 'fhnwLoginLink')))\n",
    "        # fix: Message: Element <a id=\"fhnwLoginLink\" class=\"btn-sts-login\" href=\"{{provider.href}}\"> could not be scrolled into view\n",
    "        time.sleep(buffer_timeout)\n",
    "        #login\n",
    "        #driver.find_element(By.ID, 'fhnwLoginLink').click()\n",
    "        username = WebDriverWait(driver, page_load_timeout * 2).until(EC.presence_of_element_located((By.ID, 'username')))\n",
    "        time.sleep(buffer_timeout)\n",
    "        driver.find_element(By.ID, 'username').send_keys(fhnw_username)\n",
    "        driver.find_element(By.ID, 'password').send_keys(fhnw_password)\n",
    "        driver.find_element(By.NAME, \"_eventId_proceed\").click()\n",
    "\n",
    "        # get credits\n",
    "        key = \"studiengang-angerechnet-fehlgeschlagen\"\n",
    "        credits = WebDriverWait(driver, page_load_timeout * 2).until(EC.presence_of_element_located((By.CLASS_NAME, key)))\n",
    "        time.sleep(buffer_timeout)\n",
    "        credits_div = driver.find_element(By.CLASS_NAME, key)\n",
    "        credits_text = credits_div.get_attribute(\"innerText\")\n",
    "        curr_credits = credits_text.split(\"/\")[0]\n",
    "        data['credits'] = curr_credits\n",
    "\n",
    "        # do not round grades\n",
    "        driver.find_element(By.ID, \"filter\").click()\n",
    "        driver.find_element(By.XPATH, \"//span[.='Noten runden']\").click()\n",
    "        driver.find_element(By.ID, \"filter\").click()\n",
    "\n",
    "        passed_modules = driver.find_elements(By.CLASS_NAME, 'grade-module')\n",
    "        delta_text = \"\"\n",
    "        modules = {}\n",
    "\n",
    "        for passed_module in passed_modules:\n",
    "            grade = passed_module.get_attribute(\"innerText\")\n",
    "            # returns <Name>\\n<Grade>\n",
    "            name = passed_module.find_element(By.XPATH, \"./../..\").get_attribute(\"innerText\").split(\"\\n\")[0].strip()\n",
    "            modules[name] = grade\n",
    "            if((not name in prev_data) or (modules[name] != prev_data[name])):\n",
    "                if(grade != \"\"):\n",
    "                    delta_text += f\"Modul <{name}> wurde neu mit der Note {grade} bewertet. \\r\\n\"\n",
    "\n",
    "        # append credits to data dict\n",
    "        data.update(modules)\n",
    "\n",
    "        # send email notification\n",
    "        if(delta_text != \"\"): # and prev_data != {}\n",
    "            send_mail(f\"Studenthub credits changed! {prev_credits} -> {curr_credits}\", delta_text + \"\\r\\n\\r\\nLiebe Grüsse\\r\\nStudenthub Notifier, Pascal Berger\")\n",
    "            prev_credits = curr_credits\n",
    "        else:\n",
    "            print(f\"No credits change <{credits_text}>\")\n",
    "\n",
    "        # rewrite new data to dump\n",
    "        open(\"studenthub.json\", \"w\").write(json.dumps(data, indent=2))\n",
    "\n",
    "        # close browser session\n",
    "        driver.quit()\n",
    "        if(repeat != 1):\n",
    "            break\n",
    "        prev_data = data\n",
    "\n",
    "        # configure intervall, 30 min - 1 h \n",
    "        delay_sec = 1 # 60*60\n",
    "        delay_sec += random.randint(1800, 3600)\n",
    "        print(f\"waiting {round(delay_sec / 60)} minutes\")\n",
    "        # do sleep one second at a time that the notebook is interruptable\n",
    "        for i in range(delay_sec):\n",
    "            time.sleep(1)\n",
    "        retry = 0\n",
    "    except Exception as e:\n",
    "        # quit browser if possible\n",
    "        try:\n",
    "            driver.quit()\n",
    "        except:\n",
    "            pass\n",
    "\n",
    "        # error printing, handling\n",
    "        print(e)\n",
    "        last_error = str(e)\n",
    "        if(retry >= max_retry - 1):\n",
    "            print(\"max retry reached, aborting\")\n",
    "            break\n",
    "\n",
    "        print(\"retrying\")\n",
    "        retry += 1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# send abort mail\n",
    "send_mail(f\"Studenthub notifier aborted!\", f\"Error occured:\\r\\n<{last_error}>\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "Spielwiese"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# send slack notification (not working right now)\n",
    "# if(delta_text != \"\"):\n",
    "#     slack.send_notify(slack_channel, username=slack_username, text=f'@studenthubnotificatio {delta_text}')"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "studenthub_notifier-v9uZMKUh",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.5"
  },
  "orig_nbformat": 4,
  "vscode": {
   "interpreter": {
    "hash": "3eb6f79f057af9bbad0829210926a9dd96404007ff219818b402afda27c12313"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}