cleanup and add more utf8 tld's

Author: mboot-github

.pylintrc

@@ -9,14 +9,18 @@ export SHELL
 MIN_PYTHON_VERSION := $(shell basename $$( ls /usr/bin/python3.[0-9][0-9] | awk '{print $0; exit}' ) )
 export MIN_PYTHON_VERSION
 
+run: pypiTest
+
 # ====================================================
 # uploading to pypi an pypiTestUpload
 # build a test-mypi and download the image in a venv ane run a test
 pypiTest: pypiTestUpload testTestPypi
 
 # this is only the upload now for pypi builders
-pypiTestUpload:
-	./bin/upload_to_pypiTest.sh
+pypiTestUpload: # untested ./bin/upload_to_pypiTest.sh
+	VERSION=$(shell cat ./work/version );
+	ls -l ./dist/*$(VERSION)*
+	twine upload -r testpypi dist/*$(VERSION)*
 
 testTestPypi:
 	./bin/testTestPyPiUpload.sh 2>tmp/$@-2 | tee tmp/$@-1
@@ -25,5 +29,7 @@ testTestPypi:
 # needs to be called manually, can be transferred to github action
 release: pypi
 
-pypi:
-	./bin/upload_to_pypi.sh
+pypi: # ./bin/upload_to_pypi.sh
+	VERSION=$(shell cat ./work/version )
+	ls -l ./dist/*$(VERSION)*
+	twine upload --verbose dist/*$(VERSION)*

Makefile.pypi

@@ -68,7 +68,7 @@ format:
 	ruff format *.py
 
 check:
-	ls
+	ruff check --fix *.py
 
 test:
 	$(COMMON_VENV) \

analizer/Makefile

@@ -1,8 +1,6 @@
 #! /usr/bin/env python3
 
-"""
-Analyze all tld's currently in the iana root db
-"""
+# Analyze all tld's currently in the iana root db
 
 import io
 import re
@@ -11,6 +9,7 @@
     LRUCache,
     Resolver,
 )
+
 from ianaCrawler import IanaCrawler
 from ianaDatabase import IanaDatabase
 from pslGrabber import PslGrabber
@@ -29,7 +28,7 @@ def prepDb(
 
 def prepResolver() -> Resolver:
     resolver: Resolver = Resolver()
-    resolver.cache = LRUCache()  # type: ignore
+    resolver.cache = LRUCache()
 
 
 def updateAllIanaTldData(

analizer/analizeIanaTld.py

@@ -7,9 +7,6 @@
 
 import requests
 from tld import get_tld
-from Typing import (
-    List,
-)
 
 FILE_NAME: str = "registrar-ids-1.csv"
 FILE_URL: str = f"https://www.iana.org/assignments/registrar-ids/{FILE_NAME}"
@@ -19,16 +16,15 @@ def getFileFromUrl(fileName: str, url: str) -> None:
     r = requests.get(
         url,
         allow_redirects=True,
+        timeout=300,
     )
-    pathlib.Path(fileName).open("wb").write(
-        r.content,
-    )
+    pathlib.Path(fileName).write_bytes(r.content)
     return fileName
 
 
 def readCsvFile(fileName: str):
-    result: List[List[str]] = []
-    with pathlib.Path(fileName).open() as csv_file:
+    result: list[list[str]] = []
+    with pathlib.Path(fileName).open("r", encoding="utf8") as csv_file:
         csv_reader = csv.reader(csv_file, delimiter=",")
 
         line_count = 0
@@ -46,15 +42,15 @@ def readCsvFile(fileName: str):
 
 def xMain() -> None:
     fileName: str = getFileFromUrl(FILE_NAME, FILE_URL)
-    result: List[List[str]] = readCsvFile(fileName)
+    result: list[list[str]] = readCsvFile(fileName)
 
-    rdapList: List[str] = []
-    hostList: List[str] = []
-    tldList: List[str] = []
-    fldList: List[str] = []
+    rdapList: list[str] = []
+    hostList: list[str] = []
+    tldList: list[str] = []
+    fldList: list[str] = []
 
     for row in result:
-        if row[3].strip() == "":
+        if not row[3].strip():
             continue
 
         rdap = row[3]

analizer/get_registrars.py

@@ -60,8 +60,7 @@ def _getPageFromUrlIntoSoupWithRetry(
                 print(e, file=sys.stderr)
                 time.sleep(15)
 
-        soup = BeautifulSoup(response.text, "html.parser")
-        return soup
+        return BeautifulSoup(response.text, "html.parser")
 
     def _getAdditionalItem(
         self,
@@ -73,8 +72,8 @@ def _getAdditionalItem(
                 z: str = f"{what}:"
                 if z in data[i]:
                     return data[i].replace(z, "").strip()
-            except Exception as _:
-                _ = _
+            except Exception as e:
+                _ = e
                 return None
         return None
 
@@ -105,7 +104,7 @@ def _resolveWhois(
                 except Exception as e:
                     print(whois, e, n, file=sys.stderr)
                     time.sleep(5)
-                    n = n - 1
+                    n -= 1
 
             for a in answer:
                 s = str(a)
@@ -122,11 +121,14 @@ def extractInfoFromPageSoup(
         soup: BeautifulSoup,
         tldItem: list[Any],
     ) -> None:
+        # URL for registration services: http://www.aaa.com
+        # WHOIS Server: whois.nic.aaa
+        # RDAP Server: https://rdap.nic.aaa/
         zz = {
             "Whois": "WHOIS Server",
+            "Rdap": "RDAP Server",
             "RegistrationUrl": "URL for registration services",
         }
-
         for key, val in zz.items():
             regDataW: str | None = self._getTldParagraphWithString(soup, val)
             if not regDataW:
@@ -136,7 +138,8 @@ def extractInfoFromPageSoup(
             regDataW = regDataW.replace(val, key)
             regDataA = regDataW.split("\n")
             for s in [key]:
-                tldItem.append(self._getAdditionalItem(s, regDataA))
+                s = self._getAdditionalItem(s, regDataA)
+                tldItem.append(s)
 
     def doWhoisServerResolve_DoesItExist(
         self,
@@ -174,10 +177,12 @@ def _processOneTableData(self, trs: list[str]) -> list[str]:
             try:
                 link = each.find("a")["href"]
                 aa = link.split("/")
-                record.append(aa[-1].replace(".html", ""))
-                record.append(each.text.strip())
-            except Exception as _:
-                _ = _
+                s = aa[-1].replace(".html", "")
+                record.append(s)
+                s = each.text.strip()
+                record.append(s)
+            except Exception as e:
+                _ = f"{e}"
                 record.append(each.text)
         return record
 
@@ -194,10 +199,7 @@ def _processOneTableRow(self, tr: str) -> None:
         self.records.append(record)
 
     def getTldInfoAllFromIanaUrl(self) -> None:
-        """
-        Extract all current defined tld names from the main iana root db page
-
-        """
+        """Extract all current defined tld names from the main iana root db page"""
         soup = self._getPageFromUrlIntoSoupWithRetry(self._getUrl())
         table: Any = soup.find("table")  # the first table has the tld data
 
@@ -217,6 +219,7 @@ def addInfoToAllTld(self) -> None:
         self.columns.insert(4, "Whois")  # is there a whois server defined
         self.columns.insert(5, "RegistrationUrl")  # is there a registration url defined
         self.columns.insert(6, "DnsResolve-A")  # if we have a whois server does it actually resolve to sometething real
+        self.columns.insert(7, "Rdap")  # is there a whois server defined
 
         for tldItem in self.records:  # tldItem is a list
             rr = self._addInfoToOneTld(tldItem)

analizer/ianaCrawler.py

@@ -27,7 +27,8 @@ def connectDb(
 
     def testValidConnection(self) -> None:
         if self.conn is None:
-            raise Exception("No valid connection to the database exist")
+            msg = "No valid connection to the database exist"
+            raise Exception(msg)
 
     def selectSql(
         self,
@@ -38,14 +39,11 @@ def selectSql(
         cur: Any = self.conn.cursor()
 
         try:
-            if data:
-                result = cur.execute(sql, data)
-            else:
-                result = cur.execute(sql)
+            result = cur.execute(sql, data) if data else cur.execute(sql)
 
         except Exception as e:
             print(sql, data, e, file=sys.stderr)
-            exit(101)
+            sys.exit(101)
         return result, cur
 
     def doSql(
@@ -58,17 +56,14 @@ def doSql(
         cur: Any = self.conn.cursor()
 
         try:
-            if data:
-                result = cur.execute(sql, data)
-            else:
-                result = cur.execute(sql)
+            result = cur.execute(sql, data) if data else cur.execute(sql)
 
             if withCommit:
                 self.conn.commit()
 
         except Exception as e:
             print(sql, e, file=sys.stderr)
-            exit(101)
+            sys.exit(101)
         return result
 
     def createTableTld(self) -> None:
@@ -80,11 +75,11 @@ def createTableTld(self) -> None:
     TLD_Manager     TEXT,
     Whois           TEXT,
     'DnsResolve-A'  TEXT,
-    RegistrationUrl TEXT
+    RegistrationUrl TEXT,
+    Rdap            TEXT
 );
 """
-        rr = self.doSql(sql)
-        return rr
+        return self.doSql(sql)
 
     def createTablePsl(self) -> None:
         sql = """
@@ -97,8 +92,7 @@ def createTablePsl(self) -> None:
     PRIMARY KEY (Tld, Psl)
 );
 """
-        rr = self.doSql(sql)
-        return rr
+        return self.doSql(sql)
 
     def prepData(
         self,

analizer/ianaDatabase.py

@@ -9,6 +9,7 @@
 )
 
 import idna as idna2
+
 from ianaDatabase import IanaDatabase
 
 # the next 2 belong together
@@ -70,7 +71,7 @@ def __init__(
 
     def _normalizeRow(self) -> None:
         self.tld = self.row[0].replace("'", "")
-        self.tld2 = "".join(map(lambda s: s and re.sub(r"[^\w\s]", "", s), self.row[1]))
+        self.tld2 = "".join(map(lambda s: s and re.sub(r"[^\w\s]", "", s), self.row[1]))  # noqa: C417
         self.tld3 = self.row[1].replace(".", "").replace("'", "").replace("\u200f", "").replace("\u200e", "")
         self.tld4 = self.tld3
 
@@ -131,10 +132,7 @@ def _skipSpecialResolve(self):
         if "whois.centralnicregistry.com." in self.resolve and self._doCentralNic(serverHint, self.thisTld):
             return True
 
-        if "whois.donuts.co" in self.resolve and self._doCentralNic(serverHint, self.thisTld):
-            return True
-
-        return False
+        return "whois.donuts.co" in self.resolve and self._doCentralNic(serverHint, self.thisTld)
 
     def _doUtf8Preparations(self):
         try:
@@ -144,10 +142,6 @@ def _doUtf8Preparations(self):
             return None
 
         self.tld4 = self.tld4.encode("idna").decode()
-        if self.tld != self.tld2:
-            if 0 and self.tld2 not in self.ss:
-                print("# idna", self.tld, self.tld2, self.tld3, self.tld4, self.tld.encode("idna"))
-
         if self.tld != self.tld3:
             print(f"#SKIP {self.tld} {self.tld2} {self.tld3}")
             return True
@@ -158,10 +152,7 @@ def _skipKnowTld(self):
         if self.tld2 == self.tld and self.tld in self.allKnownTldDict:
             return True
 
-        if self.tld2 in self.allKnownTldDict and self.tld in self.allKnownTldDict:
-            return True
-
-        return False
+        return self.tld2 in self.allKnownTldDict and self.tld in self.allKnownTldDict
 
     def _doNoManagerTld(self):
         if self.manager == "NULL":
@@ -194,14 +185,11 @@ def _doNoWhois(self):
             # unclear,
             # we have existing ns records indicating some self.tld's actually exist
             # but have no whois, lets skip for now
-            # TODO add ns records
+            # TO_DO add ns records
             if self.tld not in self.allKnownTldDict:
                 print(f'# ZZ["{self.tld}"] = ' + '{"_privateRegistry": True} # noWhois ')
 
-        if self.w == "NULL":
-            return True
-
-        return False
+        return self.w == "NULL"
 
     def _doCleanuphois(self):
         def xx(zz):
@@ -263,9 +251,9 @@ def processRow(
 def extractServers(aDict: dict[str, Any]) -> dict[str, Any]:
     servers: dict[str, Any] = {}
     k = "_server"
-    for key in aDict:
-        if k in aDict[key]:
-            server = aDict[key][k]
+    for key, value in aDict.items():
+        if k in value:
+            server = value[k]
             if server not in servers:
                 servers[server] = []
             servers[server].append(key)
@@ -323,7 +311,7 @@ def xMain() -> None:
 
     iad = IanaDatabase(verbose=verbose)
     iad.connectDb(dbFileName)
-    rr, cur = getAllDataTld(iad)
+    _, cur = getAllDataTld(iad)
     for row in cur:
         ot = OneTld(tld_regexpr.ZZ, verbose=verbose)
         ot.processRow(row, allTld, ss)

analizer/investigateTld.py

@@ -29,8 +29,7 @@ def getData(
         self,
         url: str,
     ) -> Any:
-        response = self.Session.get(url)
-        return response
+        return self.Session.get(url)
 
     def ColumnsPsl(self) -> list[str]:
         return [