refactor folder strucure

1 files changed, 89 insertions, 0 deletions

diff --git a/sec_certs/dataset/fips_algorithm.py b/sec_certs/dataset/fips_algorithm.py
new file mode 100644
index 00000000..a89637c1
--- /dev/null
+++ b/sec_certs/dataset/fips_algorithm.py
@@ -0,0 +1,89 @@
+import json
+import logging
+from pathlib import Path
+from typing import Dict, Union
+
+from bs4 import BeautifulSoup
+
+from sec_certs import helpers as helpers, constants as constants, cert_processing as cert_processing, files as files
+from sec_certs.dataset.dataset import Dataset
+from sec_certs.serialization import ComplexSerializableType, CustomJSONEncoder, CustomJSONDecoder
+from sec_certs.certificate.fips import FIPSCertificate
+
+class FIPSAlgorithmDataset(Dataset, ComplexSerializableType):
+
+    def get_certs_from_web(self):
+        self.root_dir.mkdir(exist_ok=True)
+        algs_paths, algs_urls = [], []
+
+        # get first page to find out how many pages there are
+        helpers.download_file(
+            constants.FIPS_ALG_URL + '1',
+            self.root_dir / "page1.html")
+
+        with open(self.root_dir / "page1.html", "r") as alg_file:
+            soup = BeautifulSoup(alg_file.read(), 'html.parser')
+            num_pages = soup.select('span[data-total-pages]')[0].attrs
+
+        for i in range(1, int(num_pages['data-total-pages'])):
+            if not (self.root_dir / f'page{i}.html').exists():
+                algs_urls.append(
+                    constants.FIPS_ALG_URL + str(i))
+                algs_paths.append(self.root_dir / f"page{i}.html")
+
+        logging.info(f"downloading {len(algs_urls)} algs html files")
+        cert_processing.process_parallel(FIPSCertificate.download_html_page, list(zip(algs_urls, algs_paths)),
+                                         constants.N_THREADS)
+
+        self.parse_html()
+
+    def parse_html(self):
+        def split_alg(alg_string):
+            cert_type = alg_string.rstrip('0123456789')
+            cert_id = alg_string[len(cert_type):]
+            return cert_type.strip(), cert_id.strip()
+
+        for f in files.search_files(self.root_dir):
+            with open(f, 'r', encoding='utf-8') as handle:
+                html_soup = BeautifulSoup(handle.read(), 'html.parser')
+
+            table = html_soup.find('table', class_='table table-condensed publications-table table-bordered')
+            spans = table.find_all('span')
+            for span in spans:
+                elements = span.find_all('td')
+                vendor, implementation = elements[0].text, elements[1].text
+                elements_sliced = elements[2:]
+                for i in range(0, len(elements_sliced), 2):
+                    alg_type, alg_id = split_alg(elements_sliced[i].text.strip())
+                    validation_date = elements_sliced[i + 1].text.strip()
+                    fips_alg = FIPSCertificate.Algorithm(alg_id, vendor, implementation, alg_type, validation_date)
+                    if alg_id not in self.certs:
+                        self.certs[alg_id] = []
+                    self.certs[alg_id].append(fips_alg)
+
+    def convert_all_pdfs(self):
+        raise NotImplementedError('Not meant to be implemented')
+
+    def download_all_pdfs(self):
+        raise NotImplementedError('Not meant to be implemented')
+
+    def to_dict(self):
+        return {"certs": self.certs}
+
+    @classmethod
+    def from_dict(cls, dct: Dict):
+        certs = dct['certs']
+        dset = cls(certs, Path('../'), 'algorithms', 'algorithms used in dataset')
+        return dset
+
+    def to_json(self, output_path: Union[str, Path]):
+        with Path(output_path).open('w') as handle:
+            json.dump(self, handle, indent=4, cls=CustomJSONEncoder)
+
+    @classmethod
+    def from_json(cls, input_path: Union[str, Path]):
+        input_path = Path(input_path)
+        with input_path.open('r') as handle:
+            dset = json.load(handle, cls=CustomJSONDecoder)
+        dset.root_dir = input_path.parent.absolute()
+        return dset
\ No newline at end of file