Merge pull request #27 from brave-intl/fix/formatting

Fix formatting

Author: Sneagan

.gitignore

@@ -2,6 +2,7 @@
 __pycache__/
 *.py[cod]
 *$py.class
+poetry.lock
 
 # C extensions
 *.so

generate-address-list.py

@@ -1,64 +1,108 @@
 #!/usr/bin/env python3
 
-import xml.etree.ElementTree as ET
 import argparse
-import pathlib
-import json
 import hashlib
+import json
+import pathlib
+import xml.etree.ElementTree as ET
+
 from ofac_scraper import OfacWebsiteScraper
 
 FEATURE_TYPE_TEXT = "Digital Currency Address - "
-NAMESPACE = {'sdn': 'https://sanctionslistservice.ofac.treas.gov/api/PublicationPreview/exports/ADVANCED_XML'}
+NAMESPACE = {
+    "sdn": "https://sanctionslistservice.ofac.treas.gov/api/PublicationPreview/exports/"
+    "ADVANCED_XML"
+}
 
 # List of implemented output formats
 OUTPUT_FORMATS = ["TXT", "JSON"]
 
 SDN_ADVANCED_FILE_PATH = "sdn_advanced.xml"
 
+
 def parse_arguments():
     parser = argparse.ArgumentParser(
-        description='Tool to extract sanctioned digital currency addresses from the OFAC special designated nationals XML file (sdn_advanced.xml)')
-    parser.add_argument('assets', nargs='*',
-                        default=[], help='the asset for which the sanctioned addresses should be extracted (default: XBT (Bitcoin))')
-    parser.add_argument('-sdn', '--special-designated-nationals-list', dest='sdn', type=argparse.FileType('rb'),
-                        help='the path to the sdn_advanced.xml file (can be downloaded from https://www.treasury.gov/ofac/downloads/sanctions/1.0/sdn_advanced.xml)', default=SDN_ADVANCED_FILE_PATH)
-    parser.add_argument('-f', '--output-format',  dest='format', nargs='*', choices=OUTPUT_FORMATS,
-                        default=OUTPUT_FORMATS[0], help='the output file format of the address list (default: TXT)')
-    parser.add_argument('-path', '--output-path', dest='outpath',  type=pathlib.Path, default=pathlib.Path(
-        "./"), help='the path where the lists should be written to (default: current working directory ("./")')
+        description="Tool to extract sanctioned digital currency addresses from the "
+        "OFAC special designated nationals XML file (sdn_advanced.xml)"
+    )
+    parser.add_argument(
+        "assets",
+        nargs="*",
+        default=[],
+        help="the asset for which the sanctioned addresses should be extracted "
+        "(default: XBT (Bitcoin))",
+    )
+    parser.add_argument(
+        "-sdn",
+        "--special-designated-nationals-list",
+        dest="sdn",
+        type=argparse.FileType("rb"),
+        help="the path to the sdn_advanced.xml file (can be downloaded from "
+        "https://www.treasury.gov/ofac/downloads/sanctions/1.0/sdn_advanced.xml)",
+        default=SDN_ADVANCED_FILE_PATH,
+    )
+    parser.add_argument(
+        "-f",
+        "--output-format",
+        dest="format",
+        nargs="*",
+        choices=OUTPUT_FORMATS,
+        default=OUTPUT_FORMATS[0],
+        help="the output file format of the address list (default: TXT)",
+    )
+    parser.add_argument(
+        "-path",
+        "--output-path",
+        dest="outpath",
+        type=pathlib.Path,
+        default=pathlib.Path("./"),
+        help="the path where the lists should be written to (default: current working "
+        'directory ("./")',
+    )
     return parser.parse_args()
 
+
 def feature_type_text(asset):
     """returns text we expect in a <FeatureType></FeatureType> tag for a given asset"""
     return "Digital Currency Address - " + asset
 
+
 def get_possible_assets(root):
     """
     Returns a list of possible digital currency assets from the parsed XML.
     """
     assets = []
-    feature_types = root.findall('sdn:ReferenceValueSets/sdn:FeatureTypeValues/sdn:FeatureType', NAMESPACE)
+    feature_types = root.findall(
+        "sdn:ReferenceValueSets/sdn:FeatureTypeValues/sdn:FeatureType", NAMESPACE
+    )
     for feature_type in feature_types:
-        if feature_type.text.startswith('Digital Currency Address - '):
-            asset = feature_type.text.replace('Digital Currency Address - ', '')
+        if feature_type.text.startswith("Digital Currency Address - "):
+            asset = feature_type.text.replace("Digital Currency Address - ", "")
             assets.append(asset)
     return assets
 
+
 def get_address_id(root, asset):
     """returns the feature id of the given asset"""
     feature_type = root.find(
-        "sdn:ReferenceValueSets/sdn:FeatureTypeValues/*[.='{}']".format(feature_type_text(asset)), NAMESPACE)
-    if feature_type == None:
-        raise LookupError("No FeatureType with the name {} found".format(
-            feature_type_text(asset)))
+        f"sdn:ReferenceValueSets/sdn:FeatureTypeValues"
+        f"/*[.='{feature_type_text(asset)}']",
+        NAMESPACE,
+    )
+    if feature_type is None:
+        raise LookupError(
+            f"No FeatureType with the name {feature_type_text(asset)} found"
+        )
     address_id = feature_type.attrib["ID"]
     return address_id
 
 
 def get_sanctioned_addresses(root, address_id):
     """returns a list of sanctioned addresses for the given address_id"""
-    addresses = list()
-    for feature in root.findall("sdn:DistinctParties//*[@FeatureTypeID='{}']".format(address_id), NAMESPACE):
+    addresses = []
+    for feature in root.findall(
+        f"sdn:DistinctParties//*[@FeatureTypeID='{address_id}']", NAMESPACE
+    ):
         for version_detail in feature.findall(".//sdn:VersionDetail", NAMESPACE):
             addresses.append(version_detail.text)
     return addresses
@@ -72,14 +116,15 @@ def write_addresses(addresses, asset, output_formats, outpath):
 
 
 def write_addresses_txt(addresses, asset, outpath):
-    with open("{}/sanctioned_addresses_{}.txt".format(outpath, asset), 'w') as out:
+    with open(f"{outpath}/sanctioned_addresses_{asset}.txt", "w") as out:
         for address in addresses:
-            out.write(address+"\n")
+            out.write(address + "\n")
 
 
 def write_addresses_json(addresses, asset, outpath):
-    with open("{}/sanctioned_addresses_{}.json".format(outpath, asset), 'w') as out:
-        out.write(json.dumps(addresses, indent=2)+"\n")
+    with open(f"{outpath}/sanctioned_addresses_{asset}.json", "w") as out:
+        out.write(json.dumps(addresses, indent=2) + "\n")
+
 
 def compute_sha256(file_path):
     sha256_hash = hashlib.sha256()
@@ -88,10 +133,12 @@ def compute_sha256(file_path):
             sha256_hash.update(chunk)
     return sha256_hash.hexdigest()
 
+
 def write_checksum_file(sha256, checksum_file_path):
     with open(checksum_file_path, "w") as checksum_file:
         checksum_file.write(f"SHA256({SDN_ADVANCED_FILE_PATH}) = {sha256}\n")
 
+
 def main():
     args = parse_arguments()
 
@@ -109,17 +156,17 @@ def main():
     tree = ET.parse(args.sdn)
     root = tree.getroot()
 
-    assets = list()
-    if type(args.assets) == str:
+    assets = []
+    if isinstance(args.format, str):
         assets.append(args.assets)
     else:
         assets = args.assets
 
     if len(assets) == 0:
         assets = get_possible_assets(root)
 
-    output_formats = list()
-    if type(args.format) == str:
+    output_formats = []
+    if isinstance(args.format, str):
         output_formats.append(args.format)
     else:
         output_formats = args.format
@@ -150,5 +197,6 @@ def main():
 
     write_checksum_file(sha256_checksum_from_site, "data/sdn_advanced_checksum.txt")
 
+
 if __name__ == "__main__":
     main()

ofac_scraper.py

@@ -1,14 +1,13 @@
-import logging
 import time
+
 from selenium import webdriver
 from selenium.common.exceptions import TimeoutException
-from selenium.webdriver.chrome.service import Service
 from selenium.webdriver.chrome.options import Options
+from selenium.webdriver.chrome.service import Service
 from selenium.webdriver.common.action_chains import ActionChains
 from selenium.webdriver.common.by import By
+from selenium.webdriver.support import expected_conditions as ec
 from selenium.webdriver.support.ui import WebDriverWait
-from selenium.webdriver.support import expected_conditions as EC
-from webdriver_manager.chrome import ChromeDriverManager
 
 
 class OfacWebsiteScraper:
@@ -29,19 +28,20 @@ def open_website(self, url):
 
     def wait_for_element(self, by, value, timeout=30):
         return WebDriverWait(self.driver, timeout).until(
-            EC.presence_of_element_located((by, value))
+            ec.presence_of_element_located((by, value))
         )
 
     def get_element_text(self, by, value):
         return self.driver.find_element(by, value).text
 
     def get_sha256_checksum(self):
-        MAX_RETRIES = 10
-        RETRY_DELAY = 10  # seconds to wait before retrying
+        max_retries = 10
+        retry_delay = 10
 
-        for attempt in range(MAX_RETRIES):
+        for attempt in range(max_retries):
             print(
-                f"Attempting to get SHA-256 checksum (attempt {attempt + 1}/{MAX_RETRIES})..."
+                f"Attempting to get SHA-256 checksum (attempt"
+                f" {attempt + 1}/{max_retries})..."
             )
             try:
                 self.open_website("https://sanctionslist.ofac.treas.gov/Home/SdnList")
@@ -53,7 +53,9 @@ def get_sha256_checksum(self):
 
                 # Scroll to (waiting for animation) and Click the 'File
                 # Signatures' button with the known ID
-                header_element = self.driver.find_element(By.ID, "accordion__heading-:r1:")
+                header_element = self.driver.find_element(
+                    By.ID, "accordion__heading-:r1:"
+                )
                 ActionChains(self.driver).move_to_element(header_element).perform()
                 time.sleep(1)
                 header_element.click()
@@ -76,15 +78,16 @@ def get_sha256_checksum(self):
                 sha256_checksum = checksums_content.split("SHA-256: ")[1].split("\n")[0]
                 return sha256_checksum
 
-            except TimeoutException:
+            except TimeoutException as e:
                 print(
-                    f"Timeout occurred on attempt {attempt + 1}/{MAX_RETRIES}. Retrying in {RETRY_DELAY} seconds..."
+                    f"Timeout occurred on attempt {attempt + 1}/{max_retries}. "
+                    f"Retrying in {retry_delay} seconds..."
                 )
-                time.sleep(RETRY_DELAY)
-                if attempt == MAX_RETRIES - 1:
+                time.sleep(retry_delay)
+                if attempt == max_retries - 1:
                     raise TimeoutException(
                         "Max retries reached. The website is not responding."
-                    )
+                    ) from e
 
     def close(self):
         self.driver.quit()

pyproject.toml

@@ -0,0 +1,64 @@
+[tool.poetry]
+name = "ofac-sanctions-list"
+version = "0.1.0"
+description = "Extract and manage OFAC sanctioned digital currency addresses"
+readme = "README.md"
+packages = [
+    {include = "ofac_address_extractor.py"},
+    {include = "update_s3_objects.py"},
+    {include = "ofac_scraper.py"}
+]
+
+[tool.poetry.scripts]
+ofac-extract = "ofac_address_extractor:main"
+ofac-s3-sync = "update_s3_objects:main"
+
+[tool.poetry.dependencies]
+python = "^3.11"
+boto3 = "^1.40.73"
+botocore = "^1.35.73"
+
+[tool.poetry.group.dev.dependencies]
+pytest = "8.4.2"
+pytest-cov = "4.1.0"
+mypy = "1.18.2"
+ruff = "0.8.6"
+bandit = {extras = ["toml"], version = "1.8.6"}
+pre-commit = "4.3.0"
+vulture = "2.14"
+types-boto3 = "^1.0.2"
+boto3-stubs = {extras = ["s3"], version = "^1.40.73"}
+
+[tool.ruff]
+line-length = 88
+target-version = "py311"
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "N", "UP", "B", "A", "C4"]
+ignore = []
+
+[tool.mypy]
+python_version = "3.11"
+warn_return_any = true
+warn_unused_configs = true
+disallow_untyped_defs = true
+
+[[tool.mypy.overrides]]
+module = ["boto3.*", "botocore.*"]
+ignore_missing_imports = true
+
+[tool.bandit]
+exclude_dirs = ["tests", "tmp", "data"]
+skips = ["B404", "B603", "B607"]
+
+[tool.pytest.ini_options]
+testpaths = ["."]
+python_files = ["test_*.py"]
+
+[tool.vulture]
+exclude = ["tests/", "tmp/", "data/"]
+min_confidence = 80
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"