Skip to content
This repository was archived by the owner on Sep 28, 2021. It is now read-only.

Commit

Permalink
fix: check path to config file; feat: custom config path with parameter
Browse files Browse the repository at this point in the history
  • Loading branch information
sibalzer committed May 27, 2021
1 parent 7167596 commit ffea71d
Show file tree
Hide file tree
Showing 10 changed files with 244 additions and 138 deletions.
2 changes: 1 addition & 1 deletion linux_start.sh
Original file line number Diff line number Diff line change
@@ -1,3 +1,3 @@
#!/bin/bash
pip3 install -q -r requirements.txt
python3 src/impfbot.py
python3 src/impfbot.py --config config.ini
2 changes: 1 addition & 1 deletion linux_validate.sh
Original file line number Diff line number Diff line change
@@ -1,3 +1,3 @@
#!/bin/bash
pip3 install -q -r requirements.txt
python3 src/validate_config.py
python3 src/validate_config.py --config config.ini
2 changes: 1 addition & 1 deletion mac_start.command
Original file line number Diff line number Diff line change
@@ -1,3 +1,3 @@
#!/bin/bash
pip3 install -q -r requirements.txt
python3 src/impfbot.py
python3 src/impfbot.py --config config.ini
2 changes: 1 addition & 1 deletion mac_validate.command
Original file line number Diff line number Diff line change
@@ -1,3 +1,3 @@
#!/bin/bash
pip3 install -q -r requirements.txt
python3 src/validate_config.py
python3 src/validate_config.py --config config.ini
13 changes: 6 additions & 7 deletions src/api_wrapper.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,19 +2,18 @@

from requests.sessions import Session
from common import sleep
import settings

log = logging.getLogger(__name__)


headers = {
'Accept': 'application/json',
'User-Agent': settings.USER_AGENT
}
def fetch_api(plz: int, birthdate_timestamp: int = None, max_retries: int = 10, sleep_after_error: int = 30, sleep_after_shadowban: int = 300, user_agent: str = 'python') -> any:
url = f"https://www.impfportal-niedersachsen.de/portal/rest/appointments/findVaccinationCenterListFree/{plz}"

headers = {
'Accept': 'application/json',
'User-Agent': user_agent
}

def fetch_api(plz: int, birthdate_timestamp: int = None, max_retries: int = 10, sleep_after_error: int = 30, sleep_after_shadowban: int = 300) -> any:
url = f"https://www.impfportal-niedersachsen.de/portal/rest/appointments/findVaccinationCenterListFree/{plz}"
if birthdate_timestamp is not None:
url += f"?stiko=&count=1&birthdate={int(birthdate_timestamp)*1000}"
fail_counter = 0
Expand Down
27 changes: 24 additions & 3 deletions src/impfbot.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,12 @@
from datetime import datetime
import sys
import argparse
from log import log
import settings
import api_wrapper
import alerts

import api_wrapper

from common import sleep, sleep_until, is_night


Expand All @@ -16,16 +19,18 @@ def check_for_slot() -> None:
(datetime.now() - settings.BIRTHDATE).total_seconds()),
max_retries=10,
sleep_after_error=settings.SLEEP_BETWEEN_FAILED_REQUESTS_IN_S,
sleep_after_shadowban=settings.SLEEP_AFTER_DETECTED_SHADOWBAN_IN_MIN
sleep_after_shadowban=settings.SLEEP_AFTER_DETECTED_SHADOWBAN_IN_MIN,
user_agent=settings.USER_AGENT
)
if not result:
log.error("Result is emtpy. (Invalid ZIP Code (PLZ))")
log.error("Result is emtpy. (Invalid ZIP Code (PLZ)?)")
for elem in result:
if not elem['outOfStock']:
log.info(
f"Free slot! ({elem['freeSlotSizeOnline']}) {elem['vaccineName']}/{elem['vaccineType']}")

msg = f"Freier Impfslot ({elem['freeSlotSizeOnline']})! {elem['vaccineName']}/{elem['vaccineType']}"

alerts.alert(msg)

sleep(settings.COOLDOWN_AFTER_FOUND_IN_MIN, 0)
Expand All @@ -37,6 +42,22 @@ def check_for_slot() -> None:

if __name__ == "__main__":
try:
parser = argparse.ArgumentParser(
description='Notification bot for the lower saxony vaccination portal ')
parser.add_argument('-f', '-c', '--config', dest='configfile',
help='Path to config.ini file', required=False, default='config.ini')
arg = vars(parser.parse_args())

try:
settings.load(arg['configfile'])
except (settings.ParseExeption, FileNotFoundError) as e:
log.error(e)
sys.exit(1)
except settings.ParseRuntimeExeption as e:
log.warning(e)
except Exception as e:
log.warning(e)

while True:
if is_night() and settings.SLEEP_AT_NIGHT:
log.info("It's night. Sleeping until 7am")
Expand Down
280 changes: 163 additions & 117 deletions src/settings.py
Original file line number Diff line number Diff line change
@@ -1,123 +1,169 @@
import configparser
import datetime
import logging
import sys
from log import log

log = logging.getLogger(__name__)

config = configparser.ConfigParser()
config.read("config.ini")
class ParseExeption(BaseException):
pass

try:
ZIP = config["COMMON"]["postleitzahl"]
if len(ZIP) != 5:
raise Exception('Non 5 digit ZIP-Code')
if ZIP[0:2] not in ['19', '21', '26', '27', '28', '29', '30', '31', '34', '37', '38', '48', '49']:

def load(path: str):
config = configparser.ConfigParser()
dataset = config.read("config.ini")
if not dataset:
raise FileNotFoundError(f"Could not find config file. Exit...")

try:
global ZIP
ZIP = config["COMMON"]["postleitzahl"]
if len(ZIP) != 5:
raise Exception('Non 5 digit ZIP-Code')
if ZIP[0:2] not in ['19', '21', '26', '27', '28', '29', '30', '31', '34', '37', '38', '48', '49']:
ParseExeption(
"[EMAIL] Are you sure that you are living in lower saxony? Because your ZIP-Code seem suspicious...")
except KeyError as e:
raise ParseExeption(
f"[COMMON] '{e}' is missing in config. Cant run without Zip-Code.")
except Exception as e:
raise ParseExeption(f"[COMMON] Invalid ZIP-Code: {e}")

try:
global BIRTHDATE
BIRTHDATE = datetime.datetime.strptime(
config["COMMON"]["geburtstag"], r"%d.%m.%Y")
except KeyError as e:
raise ParseExeption(
f"[COMMON] '{e}' is missing in config. Cant run without birthdate. Exit.")
except Exception as e:
raise ParseExeption(f"[COMMON] Invalid birthdate: {e}")

try:
global SEND_EMAIL
SEND_EMAIL = True if config["EMAIL"]["enable"].lower(
) == "true" else False
except KeyError:
log.warn(
"[EMAIL] 'enable' is missing in config. Set to False")
SEND_EMAIL = False

try:
global SENDER, SERVER, PASSWORD, EMAIL_RECEIVERS, PORT
if SEND_EMAIL:
SENDER = config["EMAIL"]["sender"]
SERVER = config["EMAIL"]["server"]
PASSWORD = config["EMAIL"]["password"]
EMAIL_RECEIVERS = config["EMAIL"]["empfaenger"].split(',')
PORT = config["EMAIL"]["port"]
except KeyError as e:
log.warning(
f"[EMAIL] '{e}' is missing in config. Set Send Email to False")
SEND_EMAIL = False

try:
global SEND_TELEGRAM_MSG
SEND_TELEGRAM_MSG = True if config["TELEGRAM"]["enable_telegram"].lower(
) == "true" else False
except KeyError:
log.warning(
"[TELEGRAM] 'enable_telegram' is missing in config. Set to False")
SEND_TELEGRAM_MSG = False

try:
if SEND_TELEGRAM_MSG:
global TOKEN, CHAT_IDS
TOKEN = config["TELEGRAM"]["token"]
CHAT_IDS = config["TELEGRAM"]["chat_id"].split(',')
except KeyError as e:
log.warning(
f"[TELEGRAM] '{e}' is missing in config. Set Telegram to False")
SEND_EMAIL = False

try:
global OPEN_BROWSER
OPEN_BROWSER = True if config["WEBBROWSER"]["open_browser"].lower(
) == "true" else False
except KeyError:
log.warning(
"'open_browser' is missing in config. Set to False")
OPEN_BROWSER = False

try:
global SLEEP_BETWEEN_REQUESTS_IN_S
SLEEP_BETWEEN_REQUESTS_IN_S = int(
config["ADVANCED"]["sleep_between_requests_in_s"])
except KeyError:
log.warning(
"'sleep_between_requests_in_s' is missing in config. Set to 5min")
SLEEP_BETWEEN_REQUESTS_IN_S = 300
except ValueError:
log.warning(
"'sleep_between_requests_in_s' is not a number. Set to 5min")
SLEEP_BETWEEN_REQUESTS_IN_S = 300

try:
global SLEEP_BETWEEN_FAILED_REQUESTS_IN_S
SLEEP_BETWEEN_FAILED_REQUESTS_IN_S = int(
config["ADVANCED"]["sleep_between_failed_requests_in_s"])
except KeyError:
log.warning(
"'sleep_between_failed_requests_in_s' is missing in config. Set to 30s")
SLEEP_BETWEEN_FAILED_REQUESTS_IN_S = 30
except ValueError:
log.warning(
"'sleep_between_failed_requests_in_s' is not a number. Set to 30s")
SLEEP_BETWEEN_REQUESTS_IN_S = 300

try:
global SLEEP_AFTER_DETECTED_SHADOWBAN_IN_MIN
SLEEP_AFTER_DETECTED_SHADOWBAN_IN_MIN = int(
config["ADVANCED"]["sleep_after_ipban_in_min"])*60
except KeyError:
log.warning(
"'sleep_after_ipban_in_min' is missing in config. Set to 3h")
SLEEP_AFTER_DETECTED_SHADOWBAN_IN_MIN = 60*180
except ValueError:
log.warning(
"'sleep_after_ipban_in_min' is not a number. Set to 3h")
SLEEP_BETWEEN_REQUESTS_IN_S = 300

try:
global COOLDOWN_AFTER_FOUND_IN_MIN
COOLDOWN_AFTER_FOUND_IN_MIN = int(
config["ADVANCED"]["cooldown_after_found_in_min"])*60
except KeyError:
log.warning(
"'cooldown_after_found_in_min' is missing in config. Set to 15min")
COOLDOWN_AFTER_FOUND_IN_MIN = 60*15
except ValueError:
log.warning(
"'cooldown_after_found_in_min' is not a number. Set to 15min")
SLEEP_BETWEEN_REQUESTS_IN_S = 300

try:
global JITTER
JITTER = int(
config["ADVANCED"]["jitter"])
except KeyError:
log.warning("'jitter' is missing in config. Set to 15s")
JITTER = 15
except ValueError:
log.warning(
"'jitter' is not a number. Set to 15s")
SLEEP_BETWEEN_REQUESTS_IN_S = 300

try:
global SLEEP_AT_NIGHT
SLEEP_AT_NIGHT = True if config["ADVANCED"]["sleep_at_night"].lower(
) == "true" else False
except KeyError:
log.warning(
"'sleep_at_night' is missing in config. Set to True")
SLEEP_AT_NIGHT = True

try:
global USER_AGENT
USER_AGENT = config["ADVANCED"]["user_agent"]
except KeyError:
log.warning(
"[EMAIL] Are you sure that you are living in lower saxony? Because your ZIP-Code seem suspicious...")
except KeyError as e:
log.warning(
f"[COMMON] '{e}' is missing in Config. Cant run without Zip-Code. Exit.")
sys.exit(1)
except Exception as e:
log.warning(f"[COMMON] Invalid ZIP-Code: {e}")

try:
BIRTHDATE = datetime.datetime.strptime(
config["COMMON"]["geburtstag"], r"%d.%m.%Y")
except KeyError as e:
log.warning(
f"[COMMON] '{e}' is missing in Config. Cant run without birthdate. Exit.")
sys.exit(1)
except Exception as e:
log.warning(f"[COMMON] Invalid birthdate: {e}")

try:
SEND_EMAIL = True if config["EMAIL"]["enable"].lower() == "true" else False
except KeyError:
log.warning(
"[EMAIL] 'enable' is missing in Config. Set False")
SEND_EMAIL = False

try:
if SEND_EMAIL:
SENDER = config["EMAIL"]["sender"]
SERVER = config["EMAIL"]["server"]
PASSWORD = config["EMAIL"]["password"]
EMAIL_RECEIVERS = config["EMAIL"]["empfaenger"].split(',')
PORT = config["EMAIL"]["port"]
except KeyError as e:
log.warning(f"[EMAIL] '{e}' is missing in Config. Set Email to False")
SEND_EMAIL = False

try:
SEND_TELEGRAM_MSG = True if config["TELEGRAM"]["enable_telegram"].lower(
) == "true" else False
except KeyError:
log.warning("[TELEGRAM] 'enable_telegram' is missing in Config. Set False")
SEND_TELEGRAM_MSG = False

try:
if SEND_TELEGRAM_MSG:
TOKEN = config["TELEGRAM"]["token"]
CHAT_IDS = config["TELEGRAM"]["chat_id"].split(',')
except KeyError as e:
log.warning(
f"[TELEGRAM] '{e}' is missing in Config. Set Telegram to False")
SEND_EMAIL = False

try:
OPEN_BROWSER = True if config["WEBBROWSER"]["open_browser"].lower(
) == "true" else False
except KeyError:
log.warning("'open_browser' is missing in Config. Set False")
OPEN_BROWSER = False

try:
SLEEP_BETWEEN_REQUESTS_IN_S = int(
config["ADVANCED"]["sleep_between_requests_in_s"])
except KeyError:
log.warning(
"'sleep_between_requests_in_s' is missing in Config. Set 5min")
SLEEP_BETWEEN_REQUESTS_IN_S = 300

try:
SLEEP_BETWEEN_FAILED_REQUESTS_IN_S = int(
config["ADVANCED"]["sleep_between_failed_requests_in_s"])
except KeyError:
log.warning(
"'sleep_between_failed_requests_in_s' is missing in Config. Set 30s")
SLEEP_BETWEEN_FAILED_REQUESTS_IN_S = 30

try:
SLEEP_AFTER_DETECTED_SHADOWBAN_IN_MIN = int(
config["ADVANCED"]["sleep_after_ipban_in_min"])*60
except KeyError:
log.warning("'sleep_after_ipban_in_min' is missing in Config. Set 3h")
SLEEP_AFTER_DETECTED_SHADOWBAN_IN_MIN = 60*180
try:
COOLDOWN_AFTER_FOUND_IN_MIN = int(
config["ADVANCED"]["cooldown_after_found_in_min"])*60
except KeyError:
log.warning("'cooldown_after_found_in_min' is missing in Config. Set 15min")
COOLDOWN_AFTER_FOUND_IN_MIN = 60*15

try:
JITTER = int(
config["ADVANCED"]["jitter"])
except KeyError:
log.warning("'jitter' is missing in Config. Set 15")
JITTER = 15

try:
SLEEP_AT_NIGHT = True if config["ADVANCED"]["sleep_at_night"].lower(
) == "true" else False
except KeyError:
log.warning("'sleep_at_night' is missing in Config. Set True")
SLEEP_AT_NIGHT = True

try:
USER_AGENT = config["ADVANCED"]["user_agent"]
except KeyError:
log.warning("'user_agent' is missing in config. set impfbot")
USER_AGENT = 'impfbot'
"'user_agent' is missing in config. Set to 'impfbot'")
USER_AGENT = 'impfbot'
Loading

0 comments on commit ffea71d

Please sign in to comment.