refactor for thread identification

src/core/app.py

@@ -2,13 +2,13 @@
 import re
 import time
 from configparser import ConfigParser
-from multiprocessing import Lock, Condition
+from multiprocessing import Condition, Lock
 from multiprocessing.pool import ThreadPool
-from threading import current_thread
+from threading import get_ident
 
 import requests
-from requests.exceptions import ConnectionError, Timeout
 from alive_progress import alive_bar
+from requests.exceptions import ConnectionError, Timeout
 
 from ..generator import PasswdGenerator, TupleGenerator
 from ..onion import TorProxy
@@ -17,11 +17,6 @@
 __all__ = ["App"]
 
 
-def get_thread_id(name: str) -> int:
-    """Return the thread id from the thread name"""
-    return int(name.split("-")[1].split(" ")[0])
-
-
 class App:
     def __init__(self, args: Args) -> None:
         self.consecutive_fails = 0
@@ -57,22 +52,19 @@ def post_search(self, pport: int, username_field: str, password_field: str) -> b
         last_exception: str = None
         try_no = 0
 
-        session = self.sessions.get(tid := get_thread_id(name := current_thread().name))
+        session = self.sessions.get(tid := get_ident(), None)
         if not session:
             session = self.build_session(pport)
             self.sessions.update({tid: session})
-            self.logger.debug("new session for %s", name)
+            self.logger.debug("new session for Thread-%d (worker)", tid)
 
         # 3 tries to connect to the target
         # connected if some response and status code is 200
         while try_no < self.max_tries and (not response or response.status_code != 200):
             try:
                 response = session.post(
                     self.post_url,
-                    data={
-                        self.username_field: username_field,
-                        self.password_field: password_field,
-                    },
+                    data={self.username_field: username_field, self.password_field: password_field},
                     allow_redirects=True,  # maybe new page when successfull login
                     timeout=self.timeout,  # 10 secondes by default
                 )
@@ -104,11 +96,7 @@ def post_search(self, pport: int, username_field: str, password_field: str) -> b
                     self.consecutive_fails += 1
 
         if response and self.target.search(response.text):
-            self.logger.info(
-                "Login successful using [%s]:[%s] 🎉",
-                username_field,
-                password_field,
-            )
+            self.logger.info("Login successful using [%s]:[%s] 🎉", username_field, password_field)
 
     def build_session(self, pport: int) -> requests.Session:
         """Build a new session with the new port"""