Revamped external manual data, split sitemap

akariv · akariv · commit 732fe4d0a8e7 · 2024-04-02T12:03:51.000+03:00
diff --git a/operators/deploy/__init__.py b/operators/deploy/__init__.py
@@ -61,7 +61,7 @@ def operator(*_):
         source = DF.Flow(
             load_from_airtable(settings.AIRTABLE_ALTERNATE_BASE, spec.table, settings.AIRTABLE_VIEW, settings.AIRTABLE_API_KEY),
             DF.select_fields([spec.id_field] + select_fields),
-        ).results()[0][0]
+        ).results(onerror=None)[0][0]
         source = dict((row[spec.id_field], row) for row in source)
 
         DF.Flow(
diff --git a/operators/derive/autocomplete.py b/operators/derive/autocomplete.py
@@ -50,7 +50,7 @@ def prepare_locations():
         tmpfile.close()
         all_places = DF.Flow(
             DF.load(tmpfile.name, format='datapackage'),
-        ).results()[0][0]
+        ).results(onerror=None)[0][0]
         keys = [n for rec in all_places for n in rec['name']]
         mapping = dict((n ,rec['bounds']) for rec in all_places for n in rec['name'])
         return keys, mapping
diff --git a/operators/derive/manual_fixes.py b/operators/derive/manual_fixes.py
@@ -25,7 +25,7 @@ def fetch_aux_table(self, var, table):
             var = DF.Flow(
                 load_from_airtable(settings.AIRTABLE_BASE, table, settings.AIRTABLE_VIEW, settings.AIRTABLE_API_KEY),
                 DF.select_fields([AIRTABLE_ID_FIELD, 'id']),
-            ).results()[0][0]
+            ).results(onerror=None)[0][0]
             logger.info(f'Got {len(var)} {table} records')
             var = dict((r['id'], r) for r in var)
         return var
diff --git a/operators/derive/to_dp.py b/operators/derive/to_dp.py
@@ -564,7 +564,7 @@ def card_data_flow():
             resources=['situations'],
         ),
         DF.select_fields(['key', 'id', 'name', 'synonyms']),
-    ).results()[0][0]
+    ).results(onerror=None)[0][0]
     situations = dict(
         (s.pop('key'), s) for s in situations
     ) | dict(
@@ -576,7 +576,7 @@ def card_data_flow():
             resources=['responses'],
         ),
         DF.select_fields(['key', 'id', 'name', 'synonyms']),
-    ).results()[0][0]
+    ).results(onerror=None)[0][0]
     responses = dict(
         (r.pop('key'), r) for r in responses
     ) | dict(
diff --git a/operators/derive/to_sitemap.py b/operators/derive/to_sitemap.py
@@ -21,30 +21,42 @@ def data_api_sitemap_flow():
         DF.concatenate(dict(path=[]), target=dict(name='sitemap', path='sitemap.csv')),
         DF.set_type('path', transform=lambda v: v.replace("'", '&apos;').replace('"', '&quot;')),
         DF.printer()
-    ).results()[0][0]
+    ).results(onerror=None)[0][0]
     today = datetime.date.today().isoformat()
-    with tempfile.NamedTemporaryFile(mode='w') as buff:
-        buff.write('<?xml version="1.0" encoding="UTF-8"?><urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">')
-        for row in urls:
-            buff.write('<url><loc>https://www.kolsherut.org.il{}</loc><lastmod>{}</lastmod></url>\n'.format(row['path'], today))
-        buff.write('</urlset>')
-        dumper = dump_to_ckan(settings.CKAN_HOST, settings.CKAN_API_KEY, settings.CKAN_OWNER_ORG, force_format=False)
-        datapackage = dict(
-            name='sitemap',
-            resources=[dict(
-                name='sitemap',
-                path='sitemap.xml',
+    with tempfile.TemporaryDirectory() as tmpdir:
+        idx = 0
+        resources = []
+        while len(urls) > 0:
+            res_name = f'sitemap_{idx}' if idx > 0 else 'sitemap'
+            base_filename = f'{res_name}.xml'
+            filename = f'{tmpdir}/{base_filename}'
+            with open(filename, 'w') as buff:
+                buff.write('<?xml version="1.0" encoding="UTF-8"?><urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">')
+                _urls = urls[:50000]
+                for row in _urls:
+                    buff.write('<url><loc>https://www.kolsherut.org.il{}</loc><lastmod>{}</lastmod></url>\n'.format(row['path'], today))
+                buff.write('</urlset>')
+                urls = urls[50000:]
+            resources.append(dict(
+                name=res_name,
+                path=base_filename,
                 format='xml',
                 schema=dict(
                     fields=[dict(name='path', type='string')]
                 )
-            )],
+            ))
+            idx += 1
+
+        dumper = dump_to_ckan(settings.CKAN_HOST, settings.CKAN_API_KEY, settings.CKAN_OWNER_ORG, force_format=False)
+        datapackage = dict(
+            name='sitemap',
+            resources=resources,
         )
         dumper.datapackage = Package(datapackage)
         dumper.write_ckan_dataset(dumper.datapackage)
-        buff.flush()
         print(dumper.datapackage.resources[0].descriptor)
-        dumper.write_file_to_output(buff.name, 'sitemap.xml')
+        for resource in resources:
+            dumper.write_file_to_output(f'{tmpdir}/{resource.path}', resource.path)
 
 
 def operator(*_):
diff --git a/operators/manual_data_entry/external.py b/operators/manual_data_entry/external.py
@@ -55,7 +55,27 @@ def func(rows):
                         emit['Org Email'] = row.get('Org Email')
                         emit['Org Website'] = row.get('Org Website')
                     emit['Data Source'] = row['Source Name']
-                    emit['taxonomies'] = [service['קטגוריה'], service['אוכלוסיית יעד'], service['שפה'], service.get('שפה-2'), service.get('שפה-3'), service.get('שפה-4'), service.get('שפה-5')]
+                    try:
+                        emit['taxonomies'] = [service['קטגוריה'], service['אוכלוסיית יעד'], service['שפה'], service.get('שפה-2'), service.get('שפה-3'), service.get('שפה-4'), service.get('שפה-5')]
+                    except KeyError:
+                        pass
+                    try:
+                        emit['target_audiences'] = service['אוכלוסיות יעד']
+                    except KeyError:
+                        emit['target_audiences'] = None
+                    emergency_service_msg = 'יש לתייג כשירות חירום'
+                    emergency_service = False
+                    try:
+                        emergency_service = service['שירות למצב החירום'] is True
+                    except KeyError:
+                        pass
+                    try:
+                        notes = service['הערות חופשיות'] or ''
+                        notes = (notes + '\n\n' + emergency_service_msg) if emergency_service else notes
+                        notes = notes.strip()
+                        emit['notes'] = notes
+                    except KeyError:
+                        emit['notes'] = None if not emergency_service else emergency_service_msg
                     yield emit
 
     return DF.Flow(
@@ -69,27 +89,30 @@ def func(rows):
         DF.add_field('Service Website', 'string'),
         DF.add_field('Service Email', 'string'),        
         DF.add_field('Data Source', 'string'),
-        DF.add_field('taxonomies', 'array', []),
+        DF.add_field('taxonomies', 'array'),
+        DF.add_field('target_audiences', 'string'),
+        DF.add_field('notes', 'string'),
         func
     )
 
 def handle_taxonomies(taxonomies):
     def func(row):
         responses = set()
         situations = set()
-        for t in row['taxonomies']:
-            if not t:
-                continue
-            t = t.strip()
-            if t in taxonomies:
-                responses.update(taxonomies[t]['response_ids'] or [])
-                situations.update(taxonomies[t]['situation_ids'] or [])
-        row['responses_ids'] = list(responses)
-        row['situations_ids'] = list(situations)
+        if row.get('taxonomies'):
+            for t in row['taxonomies']:
+                if not t:
+                    continue
+                t = t.strip()
+                if t in taxonomies:
+                    responses.update(taxonomies[t]['response_ids'] or [])
+                    situations.update(taxonomies[t]['situation_ids'] or [])
+            row['responses_ids'] = list(responses)
+            row['situations_ids'] = list(situations)
 
     return DF.Flow(
-        DF.add_field('responses_ids', 'array', []),
-        DF.add_field('situations_ids', 'array', []),
+        DF.add_field('responses_ids', 'array'),
+        DF.add_field('situations_ids', 'array'),
         func,
         DF.delete_fields(['taxonomies'])
     )
diff --git a/operators/manual_data_entry/mde_utils.py b/operators/manual_data_entry/mde_utils.py
@@ -34,6 +34,15 @@ def func(row):
     return func
 
 
+def handle_no_taxonomies():
+    def func(row):
+        data = row['data']
+        if not data.get('responses'):
+            data.pop('responses', None)
+        if not data.get('situations'):
+            data.pop('situations', None)
+    return func
+
 # ORGS
 def org_updater():
     def func(row):
@@ -206,7 +215,7 @@ def mde_service_flow(data_sources, source_id):
         DF.update_resource(-1, name='services'),
         DF.select_fields(['Org Id', 'Branch Address', 'Branch Geocode', 'Data Source',
                           'Service Name', 'Service Description', 'Service Conditions', 'Service Phone Number', 'Service Email', 'Service Website',
-                          'responses_ids', 'situations_ids']),
+                          'responses_ids', 'situations_ids', 'target_audiences', 'notes']),
         DF.rename_fields({
             'Org Id': 'organization',
             'Data Source': 'data_source',
@@ -231,14 +240,17 @@ def mde_service_flow(data_sources, source_id):
             phone_numbers=r.get('phone_numbers'),
             data_source=r['data_source'],
             email_address=r.get('email_address'),
+            target_audiences=r.get('target_audiences'),
+            notes=r.get('notes'),
         )),
+        handle_no_taxonomies(),
         DF.add_field('id', 'string', lambda r: mde_id(r['branch_id'], r['name'])),
         DF.select_fields(['id', 'data']),
     ).results()[0][0]
 
     print('COLLECTED {} relevant services'.format(len(services)))
     airtable_updater(settings.AIRTABLE_SERVICE_TABLE, source_id,
-        ['id', 'name', 'description', 'payment_details', 'phone_numbers', 'email_address', 'urls', 'situations', 'responses', 'branches', 'data_sources'],
+        ['id', 'name', 'description', 'payment_details', 'phone_numbers', 'email_address', 'urls', 'situations', 'responses', 'branches', 'data_sources', 'target_audiences', 'notes'],
         services,
         service_updater(data_sources),
         airtable_base=settings.AIRTABLE_DATA_IMPORT_BASE
diff --git a/operators/soproc/click_scraper.py b/operators/soproc/click_scraper.py
@@ -155,8 +155,8 @@ def scrape_click():
         DF.update_resource(-1, name='click'),
         decode_and_clean(),
         filter_results(),
-        DF.add_field('data_sources', 'string', lambda r: 'https://clickrevaha.molsa.gov.il/product-page/{product_id}#השירות בהרחבה ב״קליק לרווחה״'.format(**r)),
-        DF.add_field('urls', 'string', lambda r: r['data_sources']),
+        DF.add_field('data_sources', 'string', None),
+        DF.add_field('urls', 'string', None),
         DF.select_fields(list(SELECT_FIELDS.keys())),
         DF.rename_fields(SELECT_FIELDS),
         DF.set_type('details',
diff --git a/requirements.txt b/requirements.txt
@@ -1,4 +1,4 @@
-dataflows>=0.5.4
+dataflows>=0.5.5
 dataflows-airtable>=0.2.3
 dataflows-elasticsearch>=0.1.1
 dataflows-ckan>=0.3.8

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-dataflows>=0.5.4`
	`1`	`+dataflows>=0.5.5`
`2`	`2`	`dataflows-airtable>=0.2.3`
`3`	`3`	`dataflows-elasticsearch>=0.1.1`
`4`	`4`	`dataflows-ckan>=0.3.8`