Merge branch 'master' into dev

modules/bwt.py

@@ -1,63 +0,0 @@
-#-*- coding: utf-8 -*-
-
-import os
-import traceback
-from typing import Union
-from selenium import webdriver
-from selenium.webdriver.support.ui import WebDriverWait
-from selenium.webdriver.common.by import By
-from selenium.webdriver.support import expected_conditions as EC
-from functions import *
-from modules.app import app
-from bs4 import BeautifulSoup
-
-config = jsonLoad("config.json")
-
-class EmptyCardException(Exception):
-    pass
-
-async def getWaterLeft(driver: webdriver.Remote, wait: WebDriverWait, cardid: Union[str, int], filename: int):
-        
-    url = f"https://bwtaqua.com.ua/card-topup/?id={cardid}"
-    
-    try:
-
-        # driver.execute_script(f"window.open('about:blank', '{filename}');")
-        # driver.switch_to.window(f"{filename}")
-        driver.get(f"https://bwtaqua.com.ua/card-topup/?id={cardid}")
-
-        wait.until(EC.presence_of_element_located((By.CLASS_NAME, 'js-payment-balance')))
-
-        page_source = driver.page_source
-
-        driver.execute_script("window.stop();")
-
-        with open(f'data/pages/{filename}.html', "w", encoding="utf-8") as f:
-            f.write(page_source)
-
-        output = driver.find_element(By.CLASS_NAME, "js-payment-balance").text.replace("Твій баланс ", "").replace(" л", "")
-    
-        # soup = BeautifulSoup(page_source, 'html.parser')
-        # output = (soup.find_all("h3", class_="headline headline_center headline_pink js-payment-balance")[0].getText()).replace("Твій баланс ", "").replace(" л", "")
-        
-        appendLog(f"Parsed {output} liters of water remaining (user: {filename}, cardid: {cardid})")
-     
-    except Exception as exp:
-        
-        appendLog(f"Exception occured: {exp} (user: {filename}, cardid: {cardid})")
-        
-        if app != None:
-            await app.send_message(config["owner_id"], f"**Exception occured:**\n • User: `{filename}`\n • Card: [{cardid}]({url})\n • Exception: `{exp}`\n • Traceback: `{traceback.format_exc()}`", disable_web_page_preview=True)
-        else:
-            print(f'Exception occured and could not send to user: {exp}')
-        
-        output = "Failure"
-    
-    return output
-    
-if __name__ == "__main__":
-    
-    cardid = input("Enter card number: ")
-    userid = input("Enter Telegram ID (optional): ")
-    
-    print(f"Card has {str(getWaterLeft(cardid, userid, app=None))} l. left")

modules/bwt_scrape.py

@@ -0,0 +1,36 @@
+from typing import Union
+
+from bs4 import BeautifulSoup
+from selenium.webdriver import Chrome
+from selenium.webdriver.chrome.options import Options
+
+
+def get_balance(card_id: Union[str, int]) -> Union[str, None]:
+    chrome_options = Options()
+    chrome_options.add_argument('--no-sandbox')
+    chrome_options.add_argument('--headless')
+    chrome_options.add_argument('--disable-dev-shm-usage')
+
+    driver = Chrome(options=chrome_options)
+
+    driver.get(f"https://bwtaqua.com.ua/card-topup/?id={card_id}")
+
+    html = driver.page_source
+
+    soup = BeautifulSoup(html, "html.parser")
+
+    return (
+        (
+            soup.find_all(
+                "h3",
+                class_="headline headline_center headline_pink js-payment-balance",
+            )[0].getText()
+        )
+        .replace("Твій баланс ", "")
+        .replace(" л", "")
+    )
+
+
+if __name__ == "__main__":
+    card = input("Type your card ID: ")
+    print(get_balance(card))

modules/colors.py

@@ -1,22 +0,0 @@
-RESET = '\u001b[0m'
-
-BLACK = '\u001b[30m'
-RED = '\u001b[31m'
-GREEN = '\u001b[32m'
-YELLOW = '\u001b[33m'
-BLUE = '\u001b[34m'
-MAGENTA = '\u001b[35m'
-CYAN = '\u001b[36m'
-WHITE = '\u001b[37m'
-
-BBLACK = '\u001b[30;1m'
-BRED = '\u001b[31;1m'
-BGREEN = '\u001b[32;1m'
-BYELLOW = '\u001b[33;1m'
-BBLUE = '\u001b[34;1m'
-BMAGENTA = '\u001b[35;1m'
-BCYAN = '\u001b[36;1m'
-BWHITE = '\u001b[37;1m'
-
-ULINE = '\u001b[4m'
-REVERSE = '\u001b[7m'

modules/database.py

@@ -0,0 +1,11 @@
+"""Module that provides all database collections"""
+
+import sqlite3
+from pathlib import Path
+
+from libbot.sync import config_get
+
+db: sqlite3.Connection = sqlite3.connect(Path(config_get("database")))
+cursor: sqlite3.Cursor = db.cursor()
+
+cursor.execute("CREATE TABLE IF NOT EXISTS users (id INTEGER, card TEXT, locale TEXT)")

modules/migrator.py

@@ -0,0 +1,26 @@
+from os import rename
+from pathlib import Path
+from typing import Mapping
+
+from libbot.sync import json_read
+
+from modules.database import cursor
+
+
+def migrate_database() -> None:
+    """Apply migrations from old JSON database to SQLite"""
+    if not Path("data/database.json").exists():
+        return
+
+    db_old: Mapping[str, Mapping[str, str]] = json_read(Path("data/database.json"))
+
+    for user, keys in db_old.items():
+        user_locale = None if "locale" not in keys else keys["locale"]
+        user_card = None if "card" not in keys else keys["card"]
+
+        cursor.execute(
+            "INSERT INTO users VALUES (?, ?, ?)", (int(user), user_card, user_locale)
+        )
+
+    cursor.connection.commit()
+    rename(Path("data/database.json"), Path("data/database.migrated.json"))

modules/scheduler.py

@@ -0,0 +1,3 @@
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+
+scheduler = AsyncIOScheduler()