Merge pull request #276 from Der-Henning/dev

Merge dev to main

requirements-build.txt

@@ -1,4 +1,4 @@
 -r requirements.txt
 altgraph==0.17.3
-pyinstaller==5.7.0
+pyinstaller==5.8.0
 pyinstaller-hooks-contrib==2022.15

requirements.txt

@@ -4,6 +4,7 @@ certifi==2022.12.7
 charset-normalizer==3.0.1
 colorlog==6.7.0
 cron-descriptor==1.2.35
+humanize==4.6.0
 idna==3.4
 packaging==23.0
 prometheus-client==0.16.0

sample.env

@@ -6,6 +6,7 @@ ITEM_IDS=
 METRICS=false
 METRICS_PORT=8000
 DISABLE_TESTS=false
+LOCALE=en_US
 
 TGTG_USERNAME=
 TGTG_ACCESS_TOKEN=

scanner.spec

@@ -3,6 +3,7 @@ from PyInstaller.utils.hooks import collect_data_files
 
 datas = []
 datas += collect_data_files('cron_descriptor')
+datas += collect_data_files('humanize')
 
 
 block_cipher = None

src/config.sample.ini

@@ -19,6 +19,7 @@ MetricsPort = 8000
 DisableTests = false
 ## Disable all console outputs. only displays errors or Console notifier messages
 quiet = false
+locale = en_US
 
 [TGTG]
 ## TGTG Username / Login EMail

src/main.py

@@ -2,9 +2,11 @@
 import http.client as http_client
 import json
 import logging
+import platform
+import signal
 import sys
 from os import path
-from typing import NoReturn
+from typing import Any, NoReturn
 
 import colorlog
 import requests
@@ -28,9 +30,13 @@
 # set to 1 to debug http headers
 http_client.HTTPConnection.debuglevel = 0
 
+SYS_PLATFORM = platform.system()
+IS_WINDOWS = SYS_PLATFORM.lower() in ('windows', 'cygwin')
+
 
 def main() -> NoReturn:
     """Wrapper for Scanner and Helper functions."""
+    _register_signals()
     parser = argparse.ArgumentParser(
         description="TooGoodToGo scanner and notifier.",
         prog="scanner"
@@ -245,6 +251,24 @@ def _print_welcome_message() -> None:
     log.info("")
 
 
+def _register_signals() -> None:
+    # TODO: Define SIGUSR1, SIGUSR2
+    signal.signal(signal.SIGINT, _handle_exit_signal)
+    signal.signal(signal.SIGTERM, _handle_exit_signal)
+    if hasattr(signal, "SIGBREAK"):
+        signal.signal(getattr(signal, "SIGBREAK"), _handle_exit_signal)
+    if not IS_WINDOWS:
+        signal.signal(signal.SIGHUP, _handle_exit_signal)
+        # TODO: SIGQUIT is ideally meant to terminate with core dumps
+        signal.signal(signal.SIGQUIT, _handle_exit_signal)
+
+
+def _handle_exit_signal(signum: int, _frame: Any) -> None:
+    log = logging.getLogger("tgtg")
+    log.debug('Received signal %d' % signum)
+    raise KeyboardInterrupt
+
+
 def query_yes_no(question, default="yes") -> bool:
     """Ask a yes/no question via raw_input() and return their answer.
 

src/models/config.py

@@ -7,6 +7,8 @@
 from pathlib import Path
 from typing import Any
 
+import humanize
+
 from models.cron import Cron
 from models.errors import ConfigurationError
 
@@ -18,6 +20,7 @@
     'sleep_time': 60,
     'schedule_cron': Cron('* * * * *'),
     'debug': False,
+    'locale': "en_US",
     'metrics': False,
     'metrics_port': 8000,
     'disable_tests': False,
@@ -75,7 +78,7 @@
         'url': '',
         'method': 'POST',
         'body': '',
-        'type': '',
+        'type': 'text/plain',
         'headers': {},
         'timeout': 60,
         'cron': Cron('* * * * *')
@@ -105,6 +108,7 @@ class Config():
     sleep_time: int
     schedule_cron: str
     debug: bool
+    locale: str
     metrics: bool
     metrics_port: int
     disable_tests: bool
@@ -135,6 +139,8 @@ def __init__(self, file: str = None):
 
         self.token_path = environ.get("TGTG_TOKEN_PATH", None)
         self._load_tokens()
+        if (self.locale and not self.locale.startswith('en')):
+            humanize.i18n.activate(self.locale)
 
     def _open(self, file: str, mode: str) -> TextIOWrapper:
         return open(Path(self.token_path, file), mode, encoding='utf-8')
@@ -238,6 +244,7 @@ def _read_ini(self) -> None:
             self._ini_get_boolean(config, "MAIN", "DisableTests",
                                   "disable_tests")
             self._ini_get_boolean(config, "MAIN", "quiet", "quiet")
+            self._ini_get(config, "MAIN", "locale", "locale")
 
             self._ini_get(config, "TGTG", "Username", "tgtg.username")
             self._ini_get(config, "TGTG", "AccessToken", "tgtg.access_token")
@@ -350,6 +357,7 @@ def _read_env(self) -> None:
             self._env_get_int("METRICS_PORT", "metrics_port")
             self._env_get_boolean("DISABLE_TESTS", "disable_tests")
             self._env_get_boolean("QUIET", "quiet")
+            self._env_get("LOCALE", "locale")
 
             self._env_get("TGTG_USERNAME", "tgtg.username")
             self._env_get("TGTG_ACCESS_TOKEN", "tgtg.access_token")
@@ -419,7 +427,7 @@ def set(self, section: str, option: str, value: Any) -> bool:
             try:
                 config = configparser.ConfigParser()
                 config.optionxform = str
-                config.read(self.file)
+                config.read(self.file, encoding='utf-8')
                 if section not in config.sections():
                     config.add_section(section)
                 config.set(section, option, str(value))
@@ -440,7 +448,7 @@ def save_tokens(self, access_token: str, refresh_token: str,
             try:
                 config = configparser.ConfigParser()
                 config.optionxform = str
-                config.read(self.file)
+                config.read(self.file, encoding='utf-8')
                 if "TGTG" not in config.sections():
                     config.add_section("TGTG")
                 config.set("TGTG", "AccessToken", access_token)

src/models/item.py

@@ -1,6 +1,8 @@
 import datetime
 import re
 
+import humanize
+
 from models.errors import MaskConfigurationError
 
 ATTRS = ["item_id", "items_available", "display_name", "description",
@@ -17,6 +19,7 @@ class Item():
     """
 
     def __init__(self, data: dict):
+
         self.items_available = data.get("items_available", 0)
         self.display_name = data.get("display_name", "-")
         self.favorite = "Yes" if data.get("favorite", False) else "No"
@@ -96,9 +99,10 @@ def pickupdate(self) -> str:
             pto = self._datetimeparse(self.pickup_interval_end)
             prange = (f"{pfr.hour:02d}:{pfr.minute:02d} - "
                       f"{pto.hour:02d}:{pto.minute:02d}")
+            tommorow = now + datetime.timedelta(days=1)
             if now.date() == pfr.date():
-                return f"Today, {prange}"
+                return f"{humanize.naturalday(now)}, {prange}"
             if (pfr.date() - now.date()).days == 1:
-                return f"Tomorrow, {prange}"
+                return f"{humanize.naturalday(tommorow)}, {prange}"
             return f"{pfr.day}/{pfr.month}, {prange}"
         return "-"

src/scanner.py

@@ -64,7 +64,8 @@ def _job(self) -> None:
         for item_id in self.item_ids:
             try:
                 if item_id != "":
-                    items.append(Item(self.tgtg_client.get_item(item_id)))
+                    items.append(
+                        Item(self.tgtg_client.get_item(item_id)))
             except TgtgAPIError as err:
                 log.error(err)
         items += self._get_favorites()

src/tgtg/tgtg_client.py

@@ -173,34 +173,31 @@ def _post(self, path, **kwargs) -> requests.Response:
         if response.status_code in (HTTPStatus.OK, HTTPStatus.ACCEPTED):
             self.captcha_error_count = 0
             return response
-        try:
-            response.json()
-        except ValueError:
-            # Status Code == 403 and no json contend
-            # --> Blocked due to rate limit / wrong user_agent.
-            # 1. Try: Get latest APK Version from google
-            # 2. Try: Reset session
-            # 3. Try: Delete datadome cookie and reset session
-            # 10.Try: Sleep 10 minutes, and reset session
-            if response.status_code == 403:
-                log.debug("Captcha Error 403!")
-                self.captcha_error_count += 1
-                if self.captcha_error_count == 1:
-                    self.user_agent = self._get_user_agent()
-                elif self.captcha_error_count == 2:
-                    self.session = self._create_session()
-                elif self.captcha_error_count == 4:
-                    self.datadome_cookie = None
-                    self.session = self._create_session()
-                elif self.captcha_error_count >= 10:
-                    log.warning(
-                        "Too many captcha Errors! Sleeping for 10 minutes...")
-                    time.sleep(10 * 60)
-                    log.info("Retrying ...")
-                    self.captcha_error_count = 0
-                    self.session = self._create_session()
-                time.sleep(1)
-                return self._post(path, **kwargs)
+        # Status Code == 403
+        # --> Blocked due to rate limit / wrong user_agent.
+        # 1. Try: Get latest APK Version from google
+        # 2. Try: Reset session
+        # 3. Try: Delete datadome cookie and reset session
+        # 10.Try: Sleep 10 minutes, and reset session
+        if response.status_code == 403:
+            log.debug("Captcha Error 403!")
+            self.captcha_error_count += 1
+            if self.captcha_error_count == 1:
+                self.user_agent = self._get_user_agent()
+            elif self.captcha_error_count == 2:
+                self.session = self._create_session()
+            elif self.captcha_error_count == 4:
+                self.datadome_cookie = None
+                self.session = self._create_session()
+            elif self.captcha_error_count >= 10:
+                log.warning(
+                    "Too many captcha Errors! Sleeping for 10 minutes...")
+                time.sleep(10 * 60)
+                log.info("Retrying ...")
+                self.captcha_error_count = 0
+                self.session = self._create_session()
+            time.sleep(1)
+            return self._post(path, **kwargs)
         raise TgtgAPIError(response.status_code, response.content)
 
     def _get_user_agent(self) -> str:

tests/conftest.py

@@ -3,7 +3,6 @@
 
 import pytest
 
-from models.config import Config
 from models.item import Item
 
 
@@ -42,11 +41,6 @@ def temp_path():
     shutil.rmtree(temp_path)
 
 
-@pytest.fixture
-def default_config():
-    return Config("")
-
-
 @pytest.fixture
 def test_item(tgtg_item: dict):
     return Item(tgtg_item)

tests/test_config.py

@@ -1,30 +1,34 @@
 import configparser
+from importlib import reload
 from pathlib import Path
 
 import pytest
 
-from models import Config, Cron
-from models.config import DEFAULT_CONFIG
+import models.config
+from models.cron import Cron
 
 
 def test_default_ini_config():
-    config = Config("")
-    for key in DEFAULT_CONFIG:
+    reload(models.config)
+    config = models.config.Config("")
+    for key in models.config.DEFAULT_CONFIG:
         assert hasattr(config, key)
-        assert getattr(config, key) == DEFAULT_CONFIG.get(key)
+        assert getattr(config, key) == models.config.DEFAULT_CONFIG.get(key)
 
 
 def test_default_env_config():
-    config = Config()
-    for key in DEFAULT_CONFIG:
+    reload(models.config)
+    config = models.config.Config()
+    for key in models.config.DEFAULT_CONFIG:
         assert hasattr(config, key)
-        assert getattr(config, key) == DEFAULT_CONFIG.get(key)
+        assert getattr(config, key) == models.config.DEFAULT_CONFIG.get(key)
 
 
 def test_config_set(temp_path: Path):
+    reload(models.config)
     config_path = Path(temp_path, "config.ini")
     config_path.touch(exist_ok=True)
-    config = Config(config_path.absolute())
+    config = models.config.Config(config_path.absolute())
 
     assert config.set("MAIN", "debug", True)
 
@@ -35,9 +39,10 @@ def test_config_set(temp_path: Path):
 
 
 def test_save_tokens_to_ini(temp_path: Path):
+    reload(models.config)
     config_path = Path(temp_path, "config.ini")
     config_path.touch(exist_ok=True)
-    config = Config(config_path.absolute())
+    config = models.config.Config(config_path.absolute())
     config.save_tokens("test_access_token", "test_refresh_token",
                        "test_user_id", "test_cookie")
 
@@ -51,9 +56,10 @@ def test_save_tokens_to_ini(temp_path: Path):
 
 
 def test_token_path(temp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    reload(models.config)
     monkeypatch.setenv("TGTG_TOKEN_PATH", str(temp_path.absolute()))
 
-    config = Config()
+    config = models.config.Config()
     config.save_tokens("test_access_token", "test_refresh_token",
                        "test_user_id", "test_cookie")
     config._load_tokens()
@@ -65,6 +71,7 @@ def test_token_path(temp_path: Path, monkeypatch: pytest.MonkeyPatch):
 
 
 def test_ini_get(temp_path: Path):
+    reload(models.config)
     config_path = Path(temp_path, "config.ini")
 
     with open(config_path, 'w', encoding='utf-8') as file:
@@ -80,7 +87,7 @@ def test_ini_get(temp_path: Path):
             '${{price}} € \\nÀ récupérer"}'
         ])
 
-    config = Config(config_path.absolute())
+    config = models.config.Config(config_path.absolute())
 
     assert config.debug is True
     assert config.item_ids == ["23423", "32432", "234532"]
@@ -93,6 +100,7 @@ def test_ini_get(temp_path: Path):
 
 
 def test_env_get(monkeypatch: pytest.MonkeyPatch):
+    reload(models.config)
     monkeypatch.setenv("DEBUG", "true")
     monkeypatch.setenv("ITEM_IDS", "23423, 32432, 234532")
     monkeypatch.setenv("WEBHOOK_TIMEOUT", "42")
@@ -101,7 +109,7 @@ def test_env_get(monkeypatch: pytest.MonkeyPatch):
     monkeypatch.setenv("WEBHOOK_BODY", '{"content": "${{items_available}} '
                        'panier(s) à ${{price}} € \\nÀ récupérer"}')
 
-    config = Config()
+    config = models.config.Config()
 
     assert config.debug is True
     assert config.item_ids == ["23423", "32432", "234532"]