make Entity a NamedTuple

Signed-off-by: Niv Sardi <[email protected]>

python/common/defaults.py

@@ -0,0 +1 @@
+DATA_PATH='./data'

python/entity.py

@@ -1,16 +1,23 @@
 #!/usr/bin/env python3
-
-class Entity():
-    _DATA_PATH = './data'
-    def __init__(self, name, bco, url=None, logo=None):
-        self.name = name
-        self.bco = bco
-        self.url = url
-        self.logo = logo
+import csv
+from typing import NamedTuple
+
+def read_entities(fn):
+    with open('./data/entidades.csv', newline='') as csvfile:
+        reader = csv.DictReader(csvfile)
+        bcos = { d['bco']:update(d, {'id': i}) for i, d in enumerate(reader)}
+    return bcos
+
+class Entity(NamedTuple):
+    name: str
+    id: int = 0
+    bco: str = "debug"
+    url: str = None
+    logo: str = None
 
     def __repr__(self):
         return f"""
-Entity:
+Entity {self.id}:
         name: {self.name}
         bco:  {self.bco}
         url:  {self.url}
@@ -22,26 +29,22 @@ Entity:
         self = apply(cls, l)
         return self
 
+    # this now looks horrible…
     @classmethod
     def from_dict(cls, d):
-        self = cls(None, None)
-        
-        for f in d.keys():
-            setattr(self, f, d[f])
+        o = {'name': None, 'id': 0, 'bco': None, 'url': None, 'logo': None}
+        o.update(d)
+        self = cls(o['name'], o['id'], o['bco'], o['url'], o['logo'])
         return self
 
     @classmethod
     def row_names(cls):
-        return ['name', 'bco', 'url', 'logo']
-
-    @property
-    def DATA_PATH(self):
-        return self._DATA_PATH
+        return ['id', 'name', 'bco', 'url', 'logo']
 
     def to_row(self):
-        return [self.name, self.bco, self.url, self.logo]
+        return [self.id, self.name, self.bco, self.url, self.logo]
 
 if __name__ == '__main__':
-    e = Entity.from_dict({'url': 'blah'})
+    e = Entity.from_dict({'name': 'test', 'url': 'blah'})
     assert(e.url == 'blah')
     print(e)

python/main.py

@@ -8,34 +8,32 @@ from progress.bar import ChargingBar
 
 from entity import Entity
 from common import selectors
+from common import defaults
 
-pathlib.Path(f"{Entity._DATA_PATH}/logos").mkdir(parents=True, exist_ok=True)
+pathlib.Path(f'{defaults.DATA_PATH}/logos').mkdir(parents=True, exist_ok=True)
 
 DATA_FILE = './data/entidades.csv'
-URL = "http://www.bcra.gob.ar/SistemasFinancierosYdePagos/Entidades_financieras.asp"
+URL = 'http://www.bcra.gob.ar/SistemasFinancierosYdePagos/Entidades_financieras.asp'
 page = requests.get(URL)
-soup = BeautifulSoup(page.content, "html.parser")
+soup = BeautifulSoup(page.content, 'html.parser')
 
-options = soup.find(class_="form-control").find_all('option')
-with open(f"{DATA_FILE}.tmp", 'w', newline='') as csvfile:
+options = soup.find(class_='form-control').find_all('option')
+with open(f'{DATA_FILE}.tmp', 'w', newline='') as csvfile:
     writer = csv.writer(csvfile)
     writer.writerow(Entity.row_names())
 
+    i = 0
     bar = ChargingBar('Processing', max=len(options))
     for o in options[1:]:
-        e = Entity(
-            name = o.text,
-            bco = o.attrs['value']
-        )
-        page = requests.post(URL, data={'bco': e.bco})
-        soup = BeautifulSoup(page.content, "html.parser")
+        (name, bco)= (o.text, o.attrs['value'])
+        page = requests.post(URL, data={'bco': bco})
+        soup = BeautifulSoup(page.content, 'html.parser')
         try:
             img = soup.select_one(selectors.logosbancos).attrs['src']
-            img = img.replace("../", "https://www.bcra.gob.ar/")
+            img = img.replace('../', 'https://www.bcra.gob.ar/')
         except AttributeError as err:
-            print('img', e.name, err)
+            print('img', name, err)
             img = None
-        e.logo = img
 
         a = soup.select_one(selectors.entity_http)
         try:
@@ -48,10 +46,11 @@ with open(f"{DATA_FILE}.tmp", 'w', newline='') as csvfile:
             except TypeError:
                 print('ERROR', a)
 
-        e.url = a
+        e = Entity(name, id=i, bco=bco, logo=img, url=a)
         writer.writerow(e.to_row())
+        i+=1
         bar.next()
     bar.finish()
 
-shutil.move(f"{DATA_FILE}.tmp", DATA_FILE)
-print("scrape finished")
+shutil.move(f'{DATA_FILE}.tmp', DATA_FILE)
+print('scrape finished')

python/screenshot.py

@@ -9,6 +9,7 @@ from selenium.webdriver.common.by import By
 
 from common import selectors
 from entity import Entity
+from common import defaults
 
 options = webdriver.FirefoxOptions()
 options.add_argument("--headless")
@@ -24,13 +25,13 @@ def sc_entity(e: Entity):
     print(e)
     driver.implicitly_wait(10)
     driver.get(e.url)
-    driver.save_screenshot(f"{e.DATA_PATH}/{e.bco}.png")
-    driver.save_full_page_screenshot(f"{e.DATA_PATH}/{e.bco}.full.png")
+    driver.save_screenshot(f"{defaults.DATA_PATH}/{e.bco}.png")
+    driver.save_full_page_screenshot(f"{defaults.DATA_PATH}/{e.bco}.full.png")
 
     logos = driver.find_elements(By.CSS_SELECTOR, selectors.img_logo) or []
     logos.extend(driver.find_elements(By.CSS_SELECTOR, selectors.id_logo) or [])
     logos.extend(driver.find_elements(By.CSS_SELECTOR, selectors.cls_logo) or [])
-    with open(f"{e.DATA_PATH}/{e.bco}.full.txt", 'w') as f:
+    with open(f"{defaults.DATA_PATH}/{e.bco}.full.txt", 'w') as f:
         for i in logos:
             f.write(f"{e.bco} {coord_to_point(i.rect)}\n")
 

python/vendor.py

@@ -1,7 +1,5 @@
 #!/usr/bin/env python3
 import pathlib
-
-import shutil
 import csv
 import concurrent.futures
 import requests
@@ -9,6 +7,7 @@ import requests
 from progress.bar import ChargingBar
 
 from entity import Entity
+from common import defaults
 import screenshot
 import web
 
@@ -31,6 +30,7 @@ def from_csv(fn):
                     (cert, logos) = f.result()
                 except Exception as exc:
                     print('%r generated an exception: %s' % (url, exc))
+                    raise
                 else:
                     print(cert, logos)
                 bar.next()
@@ -40,6 +40,6 @@ def from_csv(fn):
 #exit()
 
 if __name__ == '__main__':
-    #pathlib.Path(e.DATA_PATH).mkdir(parents=True, exist_ok=True)
-    pathlib.Path(f"{Entity._DATA_PATH}/logos").mkdir(parents=True, exist_ok=True)
-    from_csv(f"{Entity._DATA_PATH}/entidades.csv")
+    #pathlib.Path(defaults.DATA_PATH).mkdir(parents=True, exist_ok=True)
+    pathlib.Path(f"{defaults.DATA_PATH}/logos").mkdir(parents=True, exist_ok=True)
+    from_csv(f"{defaults.DATA_PATH}/entidades.csv")

python/web.py

@@ -1,22 +1,25 @@
 #!/usr/bin/env python3
 import ssl
+import shutil
+import requests
 from bs4 import BeautifulSoup
 
 from entity import Entity
-from common import selectors
+from common import selectors, defaults
+
 def get_page(e: Entity):
     try:
         page = requests.get(e.url)
     except Exception:
-        e.url = e.url.replace('http', 'https')
-        page = requests.get(e.url)
+        url = e.url.replace('http', 'https')
+        page = requests.get(url)
     return page
 
 def get_cert(e: Entity):
     ssl_url = e.url.split("/")[2]
     try:
         cert = ssl.get_server_certificate((ssl_url, 443), ca_certs=None)
-        fn = f"{e.DATA_PATH}/{e.bco}.cert"
+        fn = f"{defaults.DATA_PATH}/{e.bco}.cert"
         with open(fn, 'w') as f:
             f.write(cert)
     except Exception as err:
@@ -24,23 +27,27 @@ def get_cert(e: Entity):
             f.write(str(err))
     return fn
 
+def get_img_logo(src: string):
+        ext = src.split('.')[-1].split('/')[-1]
+        res = requests.get(src, stream=True)
+
+        fn = f"{defaults.DATA_PATH}/logos/{e.bco}.{i}.{ext}"
+        with open(fn, "wb") as f:
+            shutil.copyfileobj(res.raw, f)
+        return fn
+
 def get_logos(e: Entity, page):
     soup = BeautifulSoup(page.content, "html.parser")
-    logos = soup.select(selectors.logo)
+    logos = soup.select(selectors.img_logo)
+    logos.extend(soup.select(selectors.id_logo))
+    logos.extend(soup.select(selectors.cls_logo))
 
     i = 0
     lfn = []
     for l in logos:
-        src = l.attrs['src']
-        ext = src.split('.')[-1].split('/')[-1]
-        try:
-            res = requests.get(src, stream=True)
-        except Exception:
-            res = requests.get(f"{e.url}/{src}")
-
-        fn = f"{e.DATA_PATH}/logos/{e.bco}.{i}.{ext}"
-        with open(fn, "wb") as f:
-            shutil.copyfileobj(res.raw, f)
-        lfn.append(fn)
+        if 'src' in l.attrs:
+            src = l.attrs['src']
+            if not src.startswith('http'): src = e.url + src
+            lfn.append(get_img_logo(src))
         i+=1
     return lfn