Add KingfisherTransformMiddleware and update affected spiders

kingfisher_scrapy/base_spider.py

@@ -6,7 +6,6 @@
 from math import ceil
 from zipfile import ZipFile
 
-import ijson
 import scrapy
 from jsonpointer import resolve_pointer
 from rarfile import RarFile
@@ -34,13 +33,15 @@ class BaseSpider(scrapy.Spider):
     ``from_date`` defaults to the ``default_from_date`` class attribute, and ``until_date`` defaults to the
     ``get_default_until_date()`` return value (which is the current time, by default).
     """
-    MAX_RELEASES_PER_PACKAGE = 100
     VALID_DATE_FORMATS = {'date': '%Y-%m-%d', 'datetime': '%Y-%m-%dT%H:%M:%S'}
 
     ocds_version = '1.1'
     date_format = 'date'
     date_required = False
     unflatten = False
+    root_path = None
+    # override this if the file is in json_line format
+    file_format = None
 
     def __init__(self, sample=None, note=None, from_date=None, until_date=None, crawl_time=None,
                  keep_collection_open=None, package_pointer=None, release_pointer=None, truncate=None, *args,
@@ -242,46 +243,6 @@ def build_file_error_from_response(self, response, **kwargs):
         item.update(kwargs)
         return item
 
-    def _get_package_metadata(self, f, skip_key):
-        """
-        Returns the package metadata from a file object.
-
-        :param f: a file object
-        :param str skip_key: the key to skip
-        :returns: the package metadata
-        :rtype: dict
-        """
-        package = {}
-        for item in util.items(ijson.parse(f), '', skip_key=skip_key):
-            package.update(item)
-        return package
-
-    def parse_json_lines(self, f, *, file_name='data.json', url=None, data_type=None, encoding='utf-8'):
-        for number, line in enumerate(f, 1):
-            if self.sample and number > self.sample:
-                break
-            if isinstance(line, bytes):
-                line = line.decode(encoding=encoding)
-            yield self.build_file_item(number=number, file_name=file_name, url=url, data=line, data_type=data_type,
-                                       encoding=encoding)
-
-    def parse_json_array(self, f_package, f_list, *, file_name='data.json', url=None, data_type=None, encoding='utf-8',
-                         array_field_name='releases'):
-        if self.sample:
-            size = self.sample
-        else:
-            size = self.MAX_RELEASES_PER_PACKAGE
-
-        package = self._get_package_metadata(f_package, array_field_name)
-
-        for number, items in enumerate(util.grouper(ijson.items(f_list, f'{array_field_name}.item'), size), 1):
-            package[array_field_name] = filter(None, items)
-            data = json.dumps(package, default=util.default)
-            yield self.build_file_item(number=number, file_name=file_name, url=url, data=data, data_type=data_type,
-                                       encoding=encoding)
-            if self.sample:
-                break
-
     @classmethod
     def get_default_until_date(cls, spider):
         """
@@ -338,11 +299,11 @@ class CompressedFileSpider(BaseSpider):
 
        ``json_lines``
          Yields each line of each compressed file.
-         The archive file is saved to disk. The compressed files are *not* saved to disk.
+         Each compressed file is saved to disk. The archive file is *not* saved to disk.
        ``release_package``
          Re-packages the releases in the compressed files in groups of
          :const:`~kingfisher_scrapy.base_spider.BaseSpider.MAX_RELEASES_PER_PACKAGE`, and yields the packages.
-         The archive file is saved to disk. The compressed files are *not* saved to disk.
+         Each compressed file is saved to disk. The archive file is *not* saved to disk.
        ``None``
          Yields each compressed file.
          Each compressed file is saved to disk. The archive file is *not* saved to disk.
@@ -363,15 +324,12 @@ def start_requests(self):
     """
 
     encoding = 'utf-8'
-    skip_pluck = 'Archive files are not supported'
     compressed_file_format = None
     archive_format = 'zip'
     file_name_must_contain = ''
 
     @handle_http_error
     def parse(self, response):
-        if self.compressed_file_format:
-            yield self.build_file_from_response(response, data_type=self.archive_format, post_to_api=False)
         if self.archive_format == 'zip':
             cls = ZipFile
         else:
@@ -390,20 +348,19 @@ def parse(self, response):
                 basename += '.json'
 
             data = archive_file.open(filename)
+            package = None
+            # for compressed_file_format == 'release_package' we need to read the file twice: once to extract the
+            # package metadata and then to extract the releases themselves
+            if self.compressed_file_format == 'release_package':
+                package = archive_file.open(filename)
 
-            kwargs = {
+            yield File({
                 'file_name': basename,
-                'url': response.request.url,
+                'data': {'data': data, 'package': package},
                 'data_type': self.data_type,
-                'encoding': self.encoding,
-            }
-            if self.compressed_file_format == 'json_lines':
-                yield from self.parse_json_lines(data, **kwargs)
-            elif self.compressed_file_format == 'release_package':
-                package = archive_file.open(filename)
-                yield from self.parse_json_array(package, data, **kwargs)
-            else:
-                yield self.build_file(data=data.read(), **kwargs)
+                'url': response.request.url,
+                'encoding': self.encoding
+            })
 
 
 class LinksSpider(SimpleSpider):

kingfisher_scrapy/extensions.py

@@ -69,14 +69,17 @@ def item_scraped(self, item, spider):
 
         Returns a dict with the metadata.
         """
-        if not isinstance(item, File):
+        if not isinstance(item, (File, FileItem)):
             return
 
         # The crawl's relative directory, in the format `<spider_name>[_sample]/<YYMMDD_HHMMSS>`.
         name = spider.name
         if spider.sample:
             name += '_sample'
-        path = os.path.join(name, spider.get_start_time('%Y%m%d_%H%M%S'), item['file_name'])
+        file_name = item['file_name']
+        if isinstance(item, FileItem):
+            file_name = f"{item['number']}-{item['file_name']}"
+        path = os.path.join(name, spider.get_start_time('%Y%m%d_%H%M%S'), file_name)
 
         self._write_file(path, item['data'])
 

kingfisher_scrapy/item_schema/item.json

@@ -32,14 +32,8 @@
           "enum": [
             "record",
             "release",
-            "release_list",
             "record_package",
             "release_package",
-            "record_package_list",
-            "release_package_list",
-            "record_package_list_in_results",
-            "release_package_list_in_results",
-            "release_in_Release",
             "zip",
             "rar",
             "tar.gz"

kingfisher_scrapy/items.py

@@ -22,11 +22,8 @@ class File(KingfisherItem):
     files_store = scrapy.Field()
 
 
-class FileItem(KingfisherItem):
+class FileItem(File):
     number = scrapy.Field()
-    data = scrapy.Field()
-    data_type = scrapy.Field()
-    encoding = scrapy.Field()
 
 
 class FileError(KingfisherItem):

kingfisher_scrapy/middlewares.py

@@ -1,9 +1,14 @@
 # https://docs.scrapy.org/en/latest/topics/spider-middleware.html
-
+import json
 from datetime import datetime
 
+import ijson
 import scrapy
 
+from kingfisher_scrapy import util
+from kingfisher_scrapy.base_spider import CompressedFileSpider
+from kingfisher_scrapy.items import File
+
 
 class ParaguayAuthMiddleware:
     """
@@ -74,3 +79,82 @@ def process_request(request, spider):
         if 'token_request' in request.meta and request.meta['token_request']:
             return
         request.headers['Authorization'] = spider.access_token
+
+
+class KingfisherTransformMiddleware:
+    """
+    Middleware that checks for File items that comes from CompressedFileSpider or returns non-well-packaged OCDS
+    File item and transform them into a release or record package
+    """
+    MAX_RELEASES_PER_PACKAGE = 100
+
+    def process_spider_output(self, response, result, spider):
+        for item in result:
+
+            if isinstance(item, File) and (spider.root_path is not None or isinstance(spider, CompressedFileSpider)):
+                kwargs = {
+                    'file_name': item['file_name'],
+                    'url': response.request.url,
+                    'data_type': spider.data_type,
+                    'encoding': item['encoding'],
+                }
+                data = item['data']
+                package = item['data']
+                compressed_file = False
+                if isinstance(spider, CompressedFileSpider):
+                    data = item['data']['data']
+                    package = item['data']['package']
+                    compressed_file = True
+                # if it is a compressed file and the file dont need any transformations
+                if compressed_file and spider.compressed_file_format is None:
+                    yield spider.build_file(data=data.read(), **kwargs)
+                # if it is a compressed file or regular file but as json_lines
+                elif spider.file_format or (compressed_file and spider.compressed_file_format == 'json_lines'):
+                    yield from self._parse_json_lines(spider, data, **kwargs)
+                # otherwise is must be a release or record package or a list of them
+                else:
+                    yield from self._parse_json_array(spider, package, data, **kwargs)
+            else:
+                yield item
+
+    def _parse_json_array(self, spider, package_data, list_data, *, file_name='data.json', url=None, data_type=None,
+                          encoding='utf-8'):
+
+        list_type = 'releases'
+        if 'record' in data_type:
+            list_type = 'records'
+
+        package = self._get_package_metadata(package_data, list_type, data_type, spider.root_path)
+        # we change the data_type into a valid one:release_package or record_package
+        data_type = data_type if 'package' in data_type else f'{data_type}_package'
+
+        # we yield a release o record package with a maximum of self.MAX_RELEASES_PER_PACKAGE releases or records
+        for number, items in enumerate(util.grouper(ijson.items(list_data, spider.root_path),
+                                                    self.MAX_RELEASES_PER_PACKAGE), 1):
+            package[list_type] = filter(None, items)
+            data = json.dumps(package, default=util.default)
+            yield spider.build_file_item(number=number, file_name=file_name, url=url, data=data,
+                                         data_type=data_type, encoding=encoding)
+
+    def _parse_json_lines(self, spider, data, *, file_name='data.json', url=None, data_type=None, encoding='utf-8'):
+        for number, line in enumerate(data, 1):
+            if isinstance(line, bytes):
+                line = line.decode(encoding=encoding)
+            yield from self._parse_json_array(spider, line, line, file_name=file_name, url=url, data_type=data_type,
+                                              encoding=encoding)
+            return
+
+    def _get_package_metadata(self, data, skip_key, data_type, root_path):
+        """
+        Returns the package metadata from a file object.
+
+        :param data: a data object
+        :param str skip_key: the key to skip
+        :returns: the package metadata
+        :rtype: dict
+        """
+        package = {}
+        if 'package' in data_type:
+            for item in util.items(ijson.parse(data), root_path, skip_key=skip_key):
+                package.update(item)
+        return package

kingfisher_scrapy/settings.py

@@ -57,6 +57,9 @@
 #SPIDER_MIDDLEWARES = {
 #    'kingfisher_scrapy.middlewares.MyCustomSpiderMiddleware': 543,
 #}
+SPIDER_MIDDLEWARES = {
+    'kingfisher_scrapy.middlewares.KingfisherTransformMiddleware': 543
+}
 
 # Enable or disable downloader middlewares
 # See https://docs.scrapy.org/en/latest/topics/downloader-middleware.html

kingfisher_scrapy/spiders/afghanistan_records.py

@@ -16,6 +16,7 @@ class AfghanistanRecords(SimpleSpider):
     name = 'afghanistan_records'
     data_type = 'record'
     skip_pluck = 'Already covered (see code for details)'  # afghanistan_releases
+    root_path = ''
 
     download_delay = 1
 

kingfisher_scrapy/spiders/afghanistan_releases.py

@@ -15,6 +15,7 @@ class AfghanistanReleases(SimpleSpider):
     """
     name = 'afghanistan_releases'
     data_type = 'release'
+    root_path = ''
 
     download_delay = 1.5
 

kingfisher_scrapy/spiders/argentina_buenos_aires.py

@@ -19,6 +19,7 @@ class ArgentinaBuenosAires(CompressedFileSpider):
     name = 'argentina_buenos_aires'
     data_type = 'release_package'
     compressed_file_format = 'release_package'
+    root_path = ''
 
     # the data list service takes too long to be downloaded, so we increase the download timeout
     download_timeout = 1000

kingfisher_scrapy/spiders/argentina_vialidad.py

@@ -11,7 +11,8 @@ class ArgentinaVialidad(SimpleSpider):
       https://datosabiertos.vialidad.gob.ar/ui/index.html#!/datos_abiertos
     """
     name = 'argentina_vialidad'
-    data_type = 'release_package_list'
+    data_type = 'release_package'
+    root_path = 'item'
 
     def start_requests(self):
         url = 'https://datosabiertos.vialidad.gob.ar/api/ocds/package/all'

kingfisher_scrapy/spiders/colombia_bulk.py

@@ -13,9 +13,10 @@ class ColombiaBulk(CompressedFileSpider):
     """
 
     name = 'colombia_bulk'
-    data_type = 'release_in_Release'
+    data_type = 'release'
     encoding = 'iso-8859-1'
     compressed_file_format = 'json_lines'
+    root_path = 'Release'
 
     download_timeout = 99999
     custom_settings = {

kingfisher_scrapy/spiders/digiwhist_base.py

@@ -14,8 +14,9 @@ class DigiwhistBase(BaseSpider):
     Bulk download documentation
       https://opentender.eu/download
     """
-    skip_pluck = 'JSON Lines is not supported'
     data_type = 'release_package'
+    root_path = ''
+    file_format = 'json_lines'
 
     def start_requests(self):
         # See scrapy.spiders.Spider.start_requests
@@ -24,9 +25,8 @@ def start_requests(self):
 
     @handle_http_error
     def parse(self, response):
-        yield self.build_file_from_response(response, data_type='tar.gz', post_to_api=False)
 
         # Load a line at the time, pass it to API
         with tarfile.open(fileobj=BytesIO(response.body), mode="r:gz") as tar:
             with tar.extractfile(tar.getnames()[0]) as readfp:
-                yield from self.parse_json_lines(readfp, url=self.start_urls[0], data_type=self.data_type)
+                yield self.build_file_from_response(data=readfp, response=response, file_name='data.json')

kingfisher_scrapy/spiders/dominican_republic.py

@@ -15,6 +15,7 @@ class DominicanRepublic(CompressedFileSpider):
     data_type = 'release_package'
     compressed_file_format = 'release_package'
     archive_format = 'rar'
+    root_path = ''
 
     def start_requests(self):
         yield scrapy.Request(

kingfisher_scrapy/spiders/indonesia_bandung.py

@@ -31,6 +31,7 @@ class IndonesiaBandung(BaseSpider):
     """
     name = 'indonesia_bandung'
     data_type = 'release'
+    root_path = ''
 
     def start_requests(self):
         pattern = 'https://birms.bandung.go.id/api/packages/year/{}'

kingfisher_scrapy/spiders/kenya_makueni.py

@@ -14,7 +14,8 @@ class KenyaMakueni(IndexSpider):
       https://opencontracting.makueni.go.ke/swagger-ui.html#/ocds-controller
     """
     name = 'kenya_makueni'
-    data_type = 'release_package_list'
+    data_type = 'release_package'
+    root_path = 'item'
     limit = 10
     additional_params = {'pageSize': limit}
     yield_list_results = False

kingfisher_scrapy/spiders/kosovo.py

@@ -17,9 +17,10 @@ class Kosovo(SimpleSpider):
       https://ocdskrpp-test.rks-gov.net/Help
     """
     name = 'kosovo'
-    data_type = 'release_list'
+    data_type = 'release'
     date_format = 'datetime'
     default_from_date = '2000-01-01T00:00:00'
+    root_path = 'item'
 
     def start_requests(self):
         stages = ['Award', 'Tender', 'Bid']