diff --git a/README.md b/README.md index 98e344d..4aeb4e1 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,26 @@ # Upwatch An Upwork webscraper that will notify you of newly published job posts in your field of work + +# PyQt5 + +# MyPy +python3 -m pip install mypy + +mypy *.py + + +# PyQt5 Stub files for MyPy +python3 -m pip install PyQt5-stubs + + +# black +python3 -m pip install black + +black *.py + + + +#LOGIC +#BeautifulSoup +#Requests +#LXML diff --git a/upwatch.py b/upwatch.py index 4cc6967..499e61c 100644 --- a/upwatch.py +++ b/upwatch.py @@ -1,19 +1,41 @@ -import requests from bs4 import BeautifulSoup # type: ignore +from typing import TypedDict +import requests import json import time import pathlib -from typing import Any, List - -# TODO: use TypedDict for these -JsonContent = Any -JobPost = Any +from typing import Optional, List, Tuple # !import re # For looking for eventual word counts in job posts & controlling the validity of url input. +JobPost = TypedDict( + "JobPost", + { + "Job Title": str, + "Payment Type": str, + "Budget": str, + "Job Description": str, + "Job Post URL": str, + }, +) + +JsonContent = TypedDict( + "JsonContent", + { + "Requests URL": str, + "Run on startup": bool, + "Scrape interval": int, + "DBMR": bool, + "Fixed Lowest Rate": int, + "Hourly Lowest Rate": int, + "Ignore no budget": bool, + "Job Posts": Optional[List[JobPost]], + }, +) + # TODO: Add to json: user agent -def read_from_json(json_path: pathlib.Path) -> JsonContent: +def read_from_json(json_path: pathlib.Path) -> Tuple[JsonContent, bool]: """ Reads all the job posts from job_posts.json """ try: with open(json_path / "job_posts.json", "r") as job_posts_json: @@ -124,14 +146,14 @@ def job_post_scraper(json_content: JsonContent) -> List[JobPost]: ) # TODO: Figure out how to fetch User Agent on current system. response.raise_for_status() break - # except requests.exceptions.HTTPError as errh: # TODO Error messages need to be communicated to user in a different way. - # print("HTTP Error:", errh) - # print("Please try a different URL") - # return - # except requests.exceptions.ConnectionError: - # print("Error Connecting") - # print("Please check you internet connection and try again.") - # return + # except requests.exceptions.HTTPError as errh: # TODO Error messages need to be communicated to user in a different way. + # print("HTTP Error:", errh) + # print("Please try a different URL") + # return + # except requests.exceptions.ConnectionError: + # print("Error Connecting") + # print("Please check you internet connection and try again.") + # return except requests.exceptions.Timeout: print("Your request timed out.") if connection_attempts == 3: @@ -163,7 +185,7 @@ def job_post_scraper(json_content: JsonContent) -> List[JobPost]: job_post_url = job_post.find("a", class_="job-title-link").attrs["href"] - job_post_dict = { + job_post_dict: JobPost = { "Job Title": job_title, "Payment Type": job_payment_type, "Budget": job_budget, diff --git a/upwatch_gui.py b/upwatch_gui.py index be1c83c..dd07b13 100644 --- a/upwatch_gui.py +++ b/upwatch_gui.py @@ -1,3 +1,4 @@ +from __future__ import annotations from PyQt5 import QtGui from PyQt5 import QtWidgets from PyQt5 import QtCore @@ -11,7 +12,7 @@ # TODO: Make "run on startup" system independent -def manage_startup_plist_file(json_content): +def manage_startup_plist_file(json_content: upwatch.JsonContent) -> None: """ Creates a plist file and saves it as a Launch Agent to run Upwatch on system startup """ plist_path = pathlib.Path("~/Library/LaunchAgents").expanduser() @@ -35,14 +36,23 @@ def manage_startup_plist_file(json_content): """ - if not (plist_path / "upwatch_startup.plist").exists() and json_content["Run on startup"]: + if ( + not (plist_path / "upwatch_startup.plist").exists() + and json_content["Run on startup"] + ): with open(plist_path / "upwatch_startup.plist", "w") as startup_plist: startup_plist.write(plist_content) - elif (plist_path / "upwatch_startup.plist").exists() and not json_content["Run on startup"]: + elif (plist_path / "upwatch_startup.plist").exists() and not json_content[ + "Run on startup" + ]: (plist_path / "upwatch_startup.plist").unlink() -def set_url(json_content, window, close_window=False): +def set_url( + json_content: upwatch.JsonContent, + window: QtWidgets.QWidget, + close_window: bool = False, +) -> None: """ Accepts user input URL and stores it in json_content """ # TODO: VALIDITY CHECK - CHECK QT DESIGNER WIDGET if window.text(): @@ -53,7 +63,9 @@ def set_url(json_content, window, close_window=False): appcore.url_dialog.window.close() -def print_url_qline(json_content, qline): # TODO: Consider if this method can be merged with set_url() +def print_url_qline( + json_content: upwatch.JsonContent, qline: QtWidgets.QWidget +) -> None: # TODO: Consider if this method can be merged with set_url() """ Shows previously input URL in text input fields """ qline.setToolTip(json_content["Requests URL"]) qline.setText(json_content["Requests URL"]) @@ -61,7 +73,7 @@ def print_url_qline(json_content, qline): # TODO: Consider if this method can b class AppCore: - def __init__(self, json_content, json_found): + def __init__(self, json_content: upwatch.JsonContent, json_found: bool) -> None: # JSON Dict with URL, Don't Bother Me Rate, Job Posts self.json_content = json_content self.json_found = json_found @@ -92,15 +104,23 @@ def __init__(self, json_content, json_found): # Create the menu self.menu = QtWidgets.QMenu() url_action = QtWidgets.QAction("Set URL") - url_action.triggered.connect(lambda: self.show_raise_window(self.url_dialog, self.url_dialog.window, True)) + url_action.triggered.connect( + lambda: self.show_raise_window( + self.url_dialog, self.url_dialog.window, True + ) + ) self.actions.append(url_action) settings_action = QtWidgets.QAction("Settings") - settings_action.triggered.connect(lambda: self.show_raise_window(settings, settings.window, True)) + settings_action.triggered.connect( + lambda: self.show_raise_window(settings, settings.window, True) + ) self.actions.append(settings_action) about_action = QtWidgets.QAction("About") - about_action.triggered.connect(lambda: self.show_raise_window(about, about.window)) + about_action.triggered.connect( + lambda: self.show_raise_window(about, about.window) + ) self.actions.append(about_action) # Add a Quit option to the menu. @@ -117,7 +137,9 @@ def __init__(self, json_content, json_found): # Launches settings window on program start if no Requests URL is defined. if not self.json_content["Requests URL"]: - self.show_raise_window(settings, settings.window) # instantiate settings class Here + self.show_raise_window( + settings, settings.window + ) # instantiate settings class Here self.worker_thread = WorkerThread(self.json_content) self.worker_thread.job_done.connect(self.on_job_done) @@ -126,34 +148,38 @@ def __init__(self, json_content, json_found): self.tray.messageClicked.connect(self.message_clicked) - def show_raise_window(self, instance, window, _print=False): + def show_raise_window( + self, instance: AppCore, window: QtWidgets.QWidget, _print: bool = False + ) -> None: # Shows the currently set URL if window is settins or url dialog if _print: print_url_qline(self.json_content, instance.url_input) window.show() window.raise_() - def test_func(self, url, event): # TODO: Move this outside class or to other class + def test_func( + self, url: str, event: QtGui.QMouseEvent + ) -> None: # TODO: Move this outside class or to other class webbrowser.open_new_tab(url) - def start_logic_thread(self): + def start_logic_thread(self) -> None: """ Calls the web scraping loop in a separate thread (to not freeze GUI) """ threading.Thread( target=upwatch.scrape_loop, args=[json_content], daemon=True ).start() - def close_program(self): + def close_program(self) -> None: """ Closes Upwatch """ upwatch.write_to_json(self.json_content, json_path) self.app.quit() - def enter_box(self, partialed, event): - partialed.setStyleSheet('text-decoration: underline;') + def enter_box(self, partialed: QtWidgets.QLabel, event: QtGui.QEnterEvent) -> None: + partialed.setStyleSheet("text-decoration: underline;") - def exit_box(self, partialed, event): - partialed.setStyleSheet('text-decoration: none;') + def exit_box(self, partialed: QtWidgets.QLabel, event: QtGui.QEvent) -> None: + partialed.setStyleSheet("text-decoration: none;") - def job_post_dialog(self): + def job_post_dialog(self) -> None: self.scroll_area = QtWidgets.QScrollArea(widgetResizable=True) self.widget = QtWidgets.QWidget() self.scroll_area.setWidget(self.widget) @@ -211,7 +237,7 @@ def job_post_dialog(self): self.scroll_area.move(800, 0) self.scroll_area.show() - def on_job_done(self, result): + def on_job_done(self, result: upwatch.List[upwatch.JobPost]) -> None: fixed_dbmr_rate = self.json_content["Fixed Lowest Rate"] hourly_dbmr_rate = self.json_content["Hourly Lowest Rate"] @@ -290,7 +316,7 @@ def on_job_done(self, result): 10000, ) - def message_clicked(self): + def message_clicked(self) -> None: if self.selected_job_posts_number == 1: webbrowser.open_new_tab(self.current_job_post["Job Post URL"]) else: @@ -299,10 +325,13 @@ def message_clicked(self): class UrlDialog: """ Creates the 'Set URL' dialog for user to specify URL to scrape from """ - def __init__(self, json_content): + + def __init__(self, json_content: upwatch.JsonContent) -> None: self.json_content = json_content self.window = QtWidgets.QDialog() - self.window.setGeometry(750, 0, 200, 30) # TODO: Make sure this dialog opens under icon + self.window.setGeometry( + 750, 0, 200, 30 + ) # TODO: Make sure this dialog opens under icon self.window.setWindowFlags( QtCore.Qt.FramelessWindowHint | QtCore.Qt.WindowStaysOnTopHint ) @@ -310,14 +339,17 @@ def __init__(self, json_content): self.url_input.setClearButtonEnabled(True) self.url_input.setPlaceholderText("Paste Valid Upwork URL here") self.url_input.resize(200, 30) # Makes QLineEdit fill size of dialog window - self.url_input.returnPressed.connect(lambda: set_url(self.json_content, self.url_input, True)) + self.url_input.returnPressed.connect( + lambda: set_url(self.json_content, self.url_input, True) + ) # TODO: Add "QRegexpValidator − Checks input against a Regex expression" class SettingsWindow: """ Creates Program's Settings Window """ - def __init__(self, json_content): + + def __init__(self, json_content: upwatch.JsonContent) -> None: self.json_content = json_content grid = QtWidgets.QGridLayout() self.window = QtWidgets.QWidget() @@ -334,7 +366,8 @@ def __init__(self, json_content): self.url_input = QtWidgets.QLineEdit() self.url_input.setPlaceholderText("https://www.upwork.com/...") self.url_input.textChanged.connect( - lambda: set_url(self.json_content, self.url_input)) + lambda: set_url(self.json_content, self.url_input) + ) # Separator lines separator = QtWidgets.QFrame() @@ -355,7 +388,9 @@ def __init__(self, json_content): # TODO: Add "Are you sure"-dialog if unchecked # Set scraping interval - self.scrape_interval_label = QtWidgets.QLabel("How often should Upwatch check \nfor new job posts? (minutes)") + self.scrape_interval_label = QtWidgets.QLabel( + "How often should Upwatch check \nfor new job posts? (minutes)" + ) self.scrape_interval_label.adjustSize() self.scrape_interval = QtWidgets.QComboBox() self.scrape_interval.addItems(["5", "10", "20", "30", "45", "60"]) @@ -413,17 +448,13 @@ def __init__(self, json_content): # Add widgets to grid layout grid.addWidget(self.url_label, 0, 0, alignment=QtCore.Qt.AlignLeft) - grid.addWidget( - self.url_input, 1, 0, 1, 2, alignment=QtCore.Qt.AlignTop - ) + grid.addWidget(self.url_input, 1, 0, 1, 2, alignment=QtCore.Qt.AlignTop) grid.addWidget(separator, 2, 0, 1, 2) grid.addWidget(self.run_on_startup, 3, 0, alignment=QtCore.Qt.AlignLeft) grid.addWidget(self.scrape_interval_label, 4, 0) grid.addWidget(self.scrape_interval, 4, 1, alignment=QtCore.Qt.AlignRight) grid.addWidget(separator_2, 5, 0, 1, 2) - grid.addWidget( - self.dbmr_groupbox, 6, 0, 3, 2, alignment=QtCore.Qt.AlignBottom - ) + grid.addWidget(self.dbmr_groupbox, 6, 0, 3, 2, alignment=QtCore.Qt.AlignBottom) low_rate_grid.addWidget( self.fixed_dbmr_label, 0, 0, alignment=QtCore.Qt.AlignBottom @@ -435,7 +466,7 @@ def __init__(self, json_content): low_rate_grid.addWidget(self.hourly_dbmr_input, 1, 1) low_rate_grid.addWidget(self.ignore_no_budget, 2, 0, 1, 2) - def set_startup_state(self): + def set_startup_state(self) -> None: """ Enables / Disables 'Run on startup' in json """ if self.json_content["Run on startup"]: # TODO: Create "are you sure"-window @@ -445,11 +476,11 @@ def set_startup_state(self): manage_startup_plist_file(self.json_content) - def set_scrape_interval(self): + def set_scrape_interval(self) -> None: """ Sets the 'Scrape interval' state in json """ self.json_content["Scrape interval"] = self.scrape_interval.currentText() - def set_dbmr_state(self): + def set_dbmr_state(self) -> None: """ Enables / Disables 'Don't bother me rate' in json """ if not self.json_content["DBMR"]: self.json_content["DBMR"] = True @@ -460,28 +491,31 @@ def set_dbmr_state(self): self.json_content["Fixed Lowest Rate"] = 0 self.json_content["Hourly Lowest Rate"] = 0 - def set_dbmr_fixed(self): + def set_dbmr_fixed(self) -> None: """ Sets the value of 'Don't bother me rate' for fixed-price job posts """ if self.fixed_dbmr_input.text(): self.json_content["Fixed Lowest Rate"] = int(self.fixed_dbmr_input.text()) else: self.json_content["Fixed Lowest Rate"] = 0 - def set_dbmr_hourly(self): + def set_dbmr_hourly(self) -> None: """ Sets the value of 'Don't bother me rate' for hourly job posts """ if self.hourly_dbmr_input.text(): self.json_content["Hourly Lowest Rate"] = int(self.hourly_dbmr_input.text()) else: self.json_content["Hourly Lowest Rate"] = 0 - def set_ignore_no_budget(self): + def set_ignore_no_budget(self) -> None: """ Enables / Disables if to ignore job posts without a specified budget in json """ - self.json_content["Ignore no budget"] = not self.json_content["Ignore no budget"] + self.json_content["Ignore no budget"] = not self.json_content[ + "Ignore no budget" + ] class AboutWindow: """ Creates program's About window """ - def __init__(self): + + def __init__(self) -> None: self.window = QtWidgets.QWidget() self.window.setWindowTitle("About") self.window.setGeometry(300, 300, 300, 300) @@ -499,11 +533,11 @@ class WorkerThread(QtCore.QThread): job_done = QtCore.pyqtSignal(object) - def __init__(self, json_content): + def __init__(self, json_content: upwatch.JsonContent) -> None: super().__init__() self.json_content = json_content - def run(self): + def run(self) -> None: """Calls the web scraping function on a scheduled interval, and sleeps in between for the time specified in json""" while not self.json_content["Requests URL"]: