save errors in local db

db/local_db_manager.py

@@ -0,0 +1,44 @@
+import sqlalchemy
+from sqlalchemy import MetaData, Column, String, Integer, DateTime, Table
+from sqlalchemy.orm import Session
+from sqlalchemy_utils import database_exists, create_database
+
+from pojo.captcha_error_contact_pojo import ContactInErrorPojo
+
+
+class LocalDbManager:
+
+    def __init__(self, path: str):
+        self.session = Session(self.init_db(path))
+
+    def init_db(self, path: str):
+        uri_for_db = "sqlite:///{}/{}.db".format(path, "appointment")
+        print(uri_for_db)
+        # 2.-Turn on database engine
+        db_engine = sqlalchemy.create_engine(uri_for_db)  # ensure this is the correct path for the sqlite file.
+        if not database_exists(uri_for_db):
+            create_database(uri_for_db)
+        connextion = db_engine.connect()
+        if not db_engine.dialect.has_table(connextion,
+                                           ContactInErrorPojo.__tablename__):  # If table don't exist, Create.
+            metadata = MetaData(db_engine)
+            # Create a table with the appropriate Columns
+            Table(ContactInErrorPojo.__tablename__, metadata,
+                  Column('mail', String, primary_key=True, nullable=False),
+                  Column('phone', String),
+                  Column('passport', String),
+                  Column('last_name', String),
+                  Column('first_name', String),
+                  Column('ccid', String),
+                  Column('position', Integer),
+                  Column('error_type', Integer),
+                  Column('update_at', DateTime),
+                  Column('create_at', DateTime))
+            # Implement the creation
+            metadata.create_all()
+
+        return db_engine
+
+    def insert_or_update(self, instance: ContactInErrorPojo):
+        self.session.merge(instance)
+        self.session.commit()

definitions.py

@@ -1,26 +1,14 @@
 import configparser
-import os
 import getpass
+import os
 from pathlib import Path
 
-import sqlalchemy as sqlalchemy
+from db.local_db_manager import LocalDbManager
-from sqlalchemy_utils import create_database
-from sqlalchemy_utils.functions import database_exists
-
-
-def init_db(path: str):
-    uri_for_db = "sqlite:///{}/{}.db".format(path, "appointment")
-    print(uri_for_db)
-    # 2.-Turn on database engine
-    dbEngine = sqlalchemy.create_engine(uri_for_db)  # ensure this is the correct path for the sqlite file.
-    if not database_exists(uri_for_db):
-        create_database(uri_for_db)
-    dbEngine.connect()
-
 
 home = str(Path.home())
 config = configparser.ConfigParser()
 print("home path: " + home)
+
 # check the config file exsistence
 config_file_path = home + "/config.ini"
 config.read(config_file_path)
@@ -29,6 +17,5 @@ FIREBASE_CONFIG_FILE = config['DEFAULT']['firebase_config_file']
 LOGS_DIR = config['DEFAULT']['LOGS_DIR']
 username = getpass.getuser()
 LOG_SOURCE = username
-
+local_db_manager = LocalDbManager(home)
-init_db(home)
 ROOT_DIR = os.path.dirname(os.path.abspath(__file__))

pojo/captcha_error_contact_pojo.py

@@ -0,0 +1,43 @@
+from sqlalchemy import Column, String, Integer, DateTime, func
+from sqlalchemy.orm import declarative_base
+
+import definitions
+from pojo.contact_pojo import ContactPojo
+
+Base = declarative_base()
+
+ERROR_TYPE_CAPTCHA = 1
+
+
+class ContactInErrorPojo(Base):
+    __tablename__ = "contacts_in_error"
+
+    mail: str = Column(String, primary_key=True)
+    phone: str = Column(String)
+    passport: str = Column(String)
+    last_name: str = Column(String)
+    first_name: str = Column(String)
+    ccid: str = Column(String)
+    position: int = Column(Integer)
+    error_type = Column(Integer)
+    update_at = Column(DateTime, onupdate=func.now())
+    create_at = Column(DateTime, default=func.now())
+
+
+def get_captcha_error_contact_from_contact(contact: ContactPojo, error_type: int) -> ContactInErrorPojo:
+    captcha_error = ContactInErrorPojo()
+    captcha_error.mail = contact.mail
+    captcha_error.ccid = contact.ccid
+    captcha_error.phone = contact.phone
+    captcha_error.passport = contact.passport
+    captcha_error.first_name = contact.first_name
+    captcha_error.last_name = contact.last_name
+    captcha_error.position = contact.position
+    captcha_error.error_type = error_type
+    return captcha_error
+
+
+if __name__ == '__main__':
+    conact = ContactPojo(mail="panleici3m@gmail.com", phone_number="649114592", ccid="", position=0,
+                         passport_number="3322111", first_name="Lei", last_name="PAAaN")
+    definitions.local_db_manager.insert_or_update(get_captcha_error_contact_from_contact(conact, ERROR_TYPE_CAPTCHA))

requirements.txt

@@ -10,3 +10,6 @@ openpyxl==3.0.9
 google-cloud-firestore==2.4.0
 PySimpleGUI==4.60.0
 SpeechRecognition==3.8.1
+SQLAlchemy~=1.4.37
+requests~=2.27.1
+Mako~=1.2.0

workers/commandor_page.py

@@ -8,10 +8,12 @@ from typing import Union
 
 from playwright.sync_api import sync_playwright
 
+import definitions
 import params
 from params import PROXY_SERVER, PROXY_PASSWORD
 from pojo.ModeEnum import ModeEnum
 from pojo.ReserveResultPojo import ReserveResultPojo, PublishType
+from pojo.captcha_error_contact_pojo import get_captcha_error_contact_from_contact, ERROR_TYPE_CAPTCHA
 from pojo.contact_pojo import ContactPojo
 from workers.SolveCaptch import SolveCaptcha
 
@@ -256,6 +258,9 @@ class CommandorPage:
             # this email has been already used
             self.is_captcha_in_error = True
             if not self.is_finished:
+                # save the error to database with contact info
+                definitions.local_db_manager.insert_or_update(
+                    get_captcha_error_contact_from_contact(self.contact, ERROR_TYPE_CAPTCHA))
                 params.oracle_log_sender.send_captcha_error(self.contact)
             self.is_finished = True
             # no need to retry captcha, if retry ,will generate DOUBLE_REQUEST_ERROR_MESSAGE