Merge branch 'bnewbold-dataset-ingest-fixes'

3 files changed, 250 insertions, 35 deletions

diff --git a/python/fatcat_tools/importers/__init__.py b/python/fatcat_tools/importers/__init__.py
index 654be2e9..e13ab552 100644
--- a/python/fatcat_tools/importers/__init__.py
+++ b/python/fatcat_tools/importers/__init__.py
@@ -37,6 +37,7 @@ from .fileset_generic import FilesetImporter
 from .grobid_metadata import GrobidMetadataImporter
 from .ingest import (
     IngestFileResultImporter,
+    IngestFilesetFileResultImporter,
     IngestFilesetResultImporter,
     IngestWebResultImporter,
     SavePaperNowFileImporter,
diff --git a/python/fatcat_tools/importers/common.py b/python/fatcat_tools/importers/common.py
index cd51a24c..2136d1da 100644
--- a/python/fatcat_tools/importers/common.py
+++ b/python/fatcat_tools/importers/common.py
@@ -916,3 +916,14 @@ def make_kafka_consumer(
     )
     print("Consuming from kafka topic {}, group {}".format(topic_name, group))
     return consumer
+
+
+def filesets_very_similar(a: FilesetEntity, b: FilesetEntity) -> bool:
+    """
+    This helper method checks if two Fileset entities are effectively equivalent: same set of files with comparable hashes.
+
+    Uses a set() of SHA1 hashes to test for equivalence.
+    """
+    a_hashes = set([f.sha1 for f in a.manifest])
+    b_hashes = set([f.sha1 for f in b.manifest])
+    return a_hashes == b_hashes
diff --git a/python/fatcat_tools/importers/ingest.py b/python/fatcat_tools/importers/ingest.py
index 4f1cc3c4..c1fed31f 100644
--- a/python/fatcat_tools/importers/ingest.py
+++ b/python/fatcat_tools/importers/ingest.py
@@ -11,7 +11,7 @@ from fatcat_openapi_client import (
     WebcaptureEntity,
 )
 
-from .common import EntityImporter, make_rel_url
+from .common import EntityImporter, filesets_very_similar, make_rel_url
 
 
 class IngestFileResultImporter(EntityImporter):
@@ -260,6 +260,16 @@ class IngestFileResultImporter(EntityImporter):
                 edit_extra["grobid_status_code"] = row["grobid"]["status_code"]
                 edit_extra["grobid_version"] = row["grobid"].get("grobid_version")
 
+        # fileset/platform metadata
+        if row.get("ingest_strategy"):
+            edit_extra["ingest_strategy"] = row["ingest_strategy"]
+        if row.get("platform_domain"):
+            edit_extra["platform_domain"] = row["platform_domain"]
+        if row.get("platform_name"):
+            edit_extra["platform_name"] = row["platform_name"]
+        if row.get("platform_id"):
+            edit_extra["platform_id"] = row["platform_id"]
+
         return edit_extra
 
     def parse_record(self, row: Dict[str, Any]) -> FileEntity:
@@ -518,7 +528,6 @@ class IngestWebResultImporter(IngestFileResultImporter):
         )
 
         edit_extra = self.parse_edit_extra(row)
-
         if edit_extra:
             wc.edit_extra = edit_extra
         return wc
@@ -675,9 +684,9 @@ class IngestFilesetResultImporter(IngestFileResultImporter):
         return True
 
     def parse_fileset_urls(self, row: Dict[str, Any]) -> List[FilesetUrl]:
-        if not row.get("strategy"):
+        if not row.get("ingest_strategy"):
             return []
-        strategy = row["strategy"]
+        strategy = row["ingest_strategy"]
         urls = []
         if strategy == "archiveorg-fileset" and row.get("archiveorg_item_name"):
             urls.append(
@@ -686,17 +695,14 @@ class IngestFilesetResultImporter(IngestFileResultImporter):
                     rel="archive-base",
                 )
             )
-        if row["strategy"].startswith("web-") and row.get("platform_base_url"):
+        if strategy.startswith("web-") and row.get("platform_base_url"):
             urls.append(
                 fatcat_openapi_client.FilesetUrl(
                     url=f"https://web.archive.org/web/{row['web_base_url_dt']}/{row['web_base_url']}",
                     rel="webarchive-base",
                 )
             )
-        # TODO: repository-base
-        # TODO: web-base
-
-        if row["strategy"] == "archiveorg-fileset-bundle" and row.get("archiveorg_item_name"):
+        if strategy == "archiveorg-fileset-bundle" and row.get("archiveorg_item_name"):
             urls.append(
                 fatcat_openapi_client.FilesetUrl(
                     url=f"https://archive.org/download/{row['archiveorg_item_name']}/{row['archiveorg_bundle_path']}",
@@ -704,7 +710,7 @@ class IngestFilesetResultImporter(IngestFileResultImporter):
                 )
             )
 
-        if row["strategy"] == "web-fileset-bundle" and row.get("platform_bundle_url"):
+        if strategy == "web-fileset-bundle" and row.get("platform_bundle_url"):
             urls.append(
                 fatcat_openapi_client.FilesetUrl(
                     url=f"https://web.archive.org/web/{row['web_bundle_url_dt']}/{row['web_bundle_url']}",
@@ -727,6 +733,15 @@ class IngestFilesetResultImporter(IngestFileResultImporter):
                     rel="repository-base",
                 )
             )
+        elif row.get("terminal"):
+            # fallback generic web URL
+            urls.append(
+                fatcat_openapi_client.FilesetUrl(
+                    url=row["terminal"]["terminal_url"],
+                    rel="web",
+                )
+            )
+
         return urls
 
     def parse_record(self, row: Dict[str, Any]) -> FilesetEntity:
@@ -748,12 +763,6 @@ class IngestFilesetResultImporter(IngestFileResultImporter):
             return None
 
         entity_extra: Dict[str, Any] = dict()
-        edit_extra = self.parse_edit_extra(row)
-        edit_extra["ingest_strategy"] = row["ingest_strategy"]
-        if row.get("platform"):
-            edit_extra["platform"] = row["platform"]
-        if row.get("platform_id"):
-            edit_extra["platform_id"] = row["platform_id"]
 
         entity_urls = self.parse_fileset_urls(row)
         if not entity_urls:
@@ -770,33 +779,33 @@ class IngestFilesetResultImporter(IngestFileResultImporter):
             fsf = fatcat_openapi_client.FilesetFile(
                 path=ingest_file["path"],
                 size=ingest_file["size"],
-                md5=ingest_file["md5"],
-                sha1=ingest_file["sha1"],
+                md5=ingest_file.get("md5"),
+                sha1=ingest_file.get("sha1"),
                 sha256=ingest_file.get("sha256"),
-                extra=dict(
-                    mimetype=ingest_file["mimetype"],
-                ),
+                mimetype=ingest_file.get("mimetype"),
+                extra=dict(),
             )
-            if not (fsf.md5 and fsf.sha1 and fsf.path and fsf.size):
+            if not (fsf.md5 and fsf.sha1 and fsf.path and fsf.size and fsf.mimetype):
                 self.counts["skip-partial-file-info"] += 1
                 return None
             if ingest_file.get("platform_url"):
-                # XXX: should we include this?
                 fsf.extra["original_url"] = ingest_file["platform_url"]
             if ingest_file.get("terminal_url") and ingest_file.get("terminal_dt"):
                 fsf.extra[
                     "wayback_url"
                 ] = f"https://web.archive.org/web/{ingest_file['terminal_dt']}/{ingest_file['terminal_url']}"
+            if not fsf.extra:
+                fsf.extra = None
             manifest.append(fsf)
 
         fe = fatcat_openapi_client.FilesetEntity(
             manifest=manifest,
             urls=entity_urls,
             release_ids=[release_ident],
+            extra=entity_extra or None,
         )
 
-        if entity_extra:
-            fe.extra = entity_extra
+        edit_extra = self.parse_edit_extra(row)
         if edit_extra:
             fe.edit_extra = edit_extra
         return fe
@@ -805,26 +814,29 @@ class IngestFilesetResultImporter(IngestFileResultImporter):
 
         # check for existing edits-in-progress with same URL
         for other in self._entity_queue:
-            # XXX: how to duplicate check?
-            if other.original_url == fse.original_url:
+            if filesets_very_similar(other, fse):
                 self.counts["skip-in-queue"] += 1
+                self.counts["skip"] += 1
                 return False
 
         # lookup sha1, or create new entity (TODO: API doesn't support this yet)
         # existing = None
 
         # NOTE: in lieu of existing checks (by lookup), only allow one fileset per release
-        release = self.api.get_release(fse.release_ids[0], expand="filesets")
-        if release.filesets:
-            # XXX: how to duplicate check filesets?
+        if not self.bezerk_mode:
+            release = self.api.get_release(fse.release_ids[0], expand="filesets")
+
             # check if this is an existing match, or just a similar hit
-            for other in release.filesets:
-                if fse.original_url == other.original_url:
-                    # TODO: compare very similar timestamps of same time (different formats)
+            for other in release.filesets or []:
+                if filesets_very_similar(other, fse):
                     self.counts["exists"] += 1
                     return False
-            self.counts["skip-release-has-fileset"] += 1
-            return False
+
+            # for now, being conservative and just skipping if release has any other fileset
+            if release.filesets:
+                self.counts["skip-release-has-fileset"] += 1
+                self.counts["skip"] += 1
+                return False
 
         return True
 
@@ -849,6 +861,197 @@ class IngestFilesetResultImporter(IngestFileResultImporter):
             )
 
 
+class IngestFilesetFileResultImporter(IngestFileResultImporter):
+    """
+    Variant of IngestFileResultImporter for processing dataset (Fileset) ingest
+    results, which resulted in a single file, into File entities.
+    """
+
+    def __init__(self, api: ApiClient, **kwargs) -> None:
+
+        eg_desc = (
+            kwargs.pop("editgroup_description", None)
+            or "Single files crawled from web using sandcrawler ingest tool, in dataset mode"
+        )
+        eg_extra = kwargs.pop("editgroup_extra", dict())
+        eg_extra["agent"] = eg_extra.get(
+            "agent", "fatcat_tools.IngestFilesetFileResultImporter"
+        )
+        kwargs["do_updates"] = False
+        super().__init__(api, editgroup_description=eg_desc, editgroup_extra=eg_extra, **kwargs)
+        self.max_file_count = 300
+
+    def want_fileset(self, row: Dict[str, Any]) -> bool:
+
+        manifest: Optional[List[Any]] = row.get("manifest")
+        if not manifest or len(manifest) == 0:
+            self.counts["skip-empty-manifest"] += 1
+            return False
+
+        if len(manifest) > 1:
+            self.counts["skip-multiple-files"] += 1
+            return False
+
+        assert len(manifest) == 1
+        return True
+
+    def want(self, row: Dict[str, Any]) -> bool:
+
+        if not self.want_ingest(row):
+            return False
+
+        if row.get("status") != "success-file":
+            self.counts["skip-status"] += 1
+            return False
+
+        # fileset-specific filters
+        if row["request"].get("ingest_type") not in [
+            "dataset",
+        ]:
+            self.counts["skip-ingest-type"] += 1
+            return False
+
+        if not self.want_fileset(row):
+            return False
+
+        return True
+
+    def parse_fileset_urls(self, row: Dict[str, Any]) -> List[FilesetUrl]:
+        if not row.get("ingest_strategy"):
+            return []
+        strategy = row["ingest_strategy"]
+        urls = []
+        # XXX
+        if strategy == "archiveorg-fileset" and row.get("archiveorg_item_name"):
+            urls.append(
+                fatcat_openapi_client.FilesetUrl(
+                    url=f"https://archive.org/download/{row['archiveorg_item_name']}/",
+                    rel="archive-base",
+                )
+            )
+        if strategy.startswith("web-") and row.get("platform_base_url"):
+            urls.append(
+                fatcat_openapi_client.FilesetUrl(
+                    url=f"https://web.archive.org/web/{row['web_base_url_dt']}/{row['web_base_url']}",
+                    rel="webarchive-base",
+                )
+            )
+        if strategy == "archiveorg-fileset-bundle" and row.get("archiveorg_item_name"):
+            urls.append(
+                fatcat_openapi_client.FilesetUrl(
+                    url=f"https://archive.org/download/{row['archiveorg_item_name']}/{row['archiveorg_bundle_path']}",
+                    rel="archive-bundle",
+                )
+            )
+
+        if strategy == "web-fileset-bundle" and row.get("platform_bundle_url"):
+            urls.append(
+                fatcat_openapi_client.FilesetUrl(
+                    url=f"https://web.archive.org/web/{row['web_bundle_url_dt']}/{row['web_bundle_url']}",
+                    rel="webarchive-bundle",
+                )
+            )
+
+        # add any additional / platform URLs here
+        if row.get("platform_bundle_url"):
+            urls.append(
+                fatcat_openapi_client.FilesetUrl(
+                    url=row["platform_bundle_url"],
+                    rel="repository-bundle",
+                )
+            )
+        if row.get("platform_base_url"):
+            urls.append(
+                fatcat_openapi_client.FilesetUrl(
+                    url=row["platform_bundle_url"],
+                    rel="repository-base",
+                )
+            )
+        elif row.get("terminal"):
+            # fallback generic web URL
+            urls.append(
+                fatcat_openapi_client.FilesetUrl(
+                    url=row["terminal"]["terminal_url"],
+                    rel="web",
+                )
+            )
+
+        return urls
+
+    def parse_record(self, row: Dict[str, Any]) -> FileEntity:
+
+        request = row["request"]
+
+        # double check that want() filtered request correctly
+        if request.get("ingest_type") not in [
+            "dataset",
+        ]:
+            self.counts["skip-ingest-type"] += 1
+            return None
+
+        # identify release by fatcat ident, or extid lookup
+        release_ident = self.parse_ingest_release_ident(row)
+
+        if not release_ident:
+            self.counts["skip-release-not-found"] += 1
+            return None
+
+        assert row["file_count"] == len(row["manifest"]) == 1
+        file_meta = row["manifest"][0]
+        # print(file_meta)
+        assert file_meta["status"] == "success"
+
+        # add file-level access URLs
+        entity_urls = []
+        if file_meta.get("platform_url"):
+            entity_urls.append(FileUrl(rel="web", url=file_meta["platform_url"]))
+        if file_meta.get("terminal_url") and file_meta.get("terminal_dt"):
+            entity_urls.append(
+                FileUrl(
+                    rel="webarchive",
+                    url=f"https://web.archive.org/web/{file_meta['terminal_dt']}/{file_meta['terminal_url']}",
+                )
+            )
+        if row["ingest_strategy"] == "archiveorg-file":
+            entity_urls.append(
+                FileUrl(
+                    rel="archive",
+                    url=f"https://archive.org/download/{row['archiveorg_item_name']}/{file_meta['path']}",
+                )
+            )
+
+        if not entity_urls:
+            self.counts["skip-no-access-url"] += 1
+            return None
+
+        entity_extra: Dict[str, Any] = dict()
+        entity_extra["path"] = file_meta["path"]
+
+        # this is to work around a bug in old sandcrawler ingest code
+        if file_meta["md5"] == file_meta["sha1"]:
+            self.counts["skip-bad-hashes"] += 1
+            return None
+
+        fe = FileEntity(
+            md5=file_meta["md5"],
+            sha1=file_meta["sha1"],
+            sha256=file_meta["sha256"],
+            size=file_meta["size"],
+            mimetype=file_meta["mimetype"],
+            release_ids=[release_ident],
+            urls=entity_urls,
+            extra=entity_extra or None,
+        )
+        if not (fe.md5 and fe.sha1 and fe.sha256 and (fe.size is not None) and fe.mimetype):
+            self.counts["skip-partial-file-info"] += 1
+            return None
+
+        edit_extra = self.parse_edit_extra(row)
+        if edit_extra:
+            fe.edit_extra = edit_extra
+        return fe
+
+
 class SavePaperNowFilesetImporter(IngestFilesetResultImporter):
     """
     Like SavePaperNowFileImporter, but for fileset/dataset ingest.