WIP on release-to-sim fetching

2 files changed, 124 insertions, 12 deletions

diff --git a/fatcat_scholar/sandcrawler.py b/fatcat_scholar/sandcrawler.py
new file mode 100644
index 0000000..db6014f
--- /dev/null
+++ b/fatcat_scholar/sandcrawler.py
@@ -0,0 +1,75 @@
+
+import json
+import minio
+import requests
+from typing import Dict, Optional, Any
+
+class SandcrawlerPostgrestClient():
+
+    def __init__(self, api_url: str):
+        self.api_url = api_url
+        
+    def get_grobid(self, sha1: str) -> Optional[Dict[str, Any]]:
+        resp = requests.get(self.api_url + "/grobid", params=dict(sha1hex='eq.'+sha1))
+        resp.raise_for_status()
+        resp_json = resp.json()
+        if resp_json:
+            return resp_json[0]
+        else:
+            return None
+
+
+class SandcrawlerMinioClient(object):
+
+    def __init__(self, host_url: str, access_key: Optional[str] = None, secret_key: Optional[str] = None, default_bucket: Optional[str] = "sandcrawler"):
+        """
+        host is minio connection string (host:port)
+        access and secret key are as expected
+        default_bucket can be supplied so that it doesn't need to be repeated for each function call
+
+        Example config:
+
+            host="localhost:9000",
+            access_key=os.environ['MINIO_ACCESS_KEY'],
+            secret_key=os.environ['MINIO_SECRET_KEY'],
+        """
+        self.mc = minio.Minio(
+            host_url,
+            access_key=access_key,
+            secret_key=secret_key,
+            secure=False,
+        )
+        self.default_bucket = default_bucket
+
+    def _blob_path(self, folder, sha1hex, extension, prefix):
+        if not extension:
+            extension = ""
+        if not prefix:
+            prefix = ""
+        assert len(sha1hex) == 40
+        obj_path = "{}{}/{}/{}/{}{}".format(
+            prefix,
+            folder, 
+            sha1hex[0:2],
+            sha1hex[2:4],
+            sha1hex,
+            extension,
+        )
+        return obj_path 
+
+    def get_blob(self, folder, sha1hex, extension="", prefix="", bucket=None):
+        """
+        sha1hex is sha1 of the blob itself
+
+        Fetched blob from the given bucket/folder, using the sandcrawler SHA1 path convention
+        """
+        obj_path = self._blob_path(folder, sha1hex, extension, prefix)
+        if not bucket:
+            bucket = self.default_bucket
+        assert bucket
+        blob = self.mc.get_object(
+            bucket,
+            obj_path,
+        )
+        # TODO: optionally verify SHA-1?
+        return blob.data
diff --git a/fatcat_scholar/work_pipeline.py b/fatcat_scholar/work_pipeline.py
index 7ed0eac..161327a 100644
--- a/fatcat_scholar/work_pipeline.py
+++ b/fatcat_scholar/work_pipeline.py
@@ -1,4 +1,5 @@
 
+import os
 import io
 import sys
 import argparse
@@ -9,6 +10,7 @@ import internetarchive
 
 from fatcat_scholar.api_entities import *
 from fatcat_scholar.djvu import djvu_extract_leaf_texts
+from fatcat_scholar.sandcrawler import SandcrawlerPostgrestClient, SandcrawlerMinioClient
 from fatcat_scholar.issue_db import IssueDB, SimIssueRow
 from fatcat_scholar.es_transform import es_biblio_from_release, es_release_from_release, DocType
 
@@ -79,21 +81,41 @@ def fulltext_pref_list(releases: List[ReleaseEntity]) -> List[str]:
 
 class WorkPipeline():
 
-    def __init__(self, issue_db: IssueDB):
+    def __init__(self, issue_db: IssueDB, sandcrawler_db_client: SandcrawlerPostgrestClient, sandcrawler_s3_client: SandcrawlerMinioClient):
         self.issue_db: IssueDB = issue_db
         self.ia_client = internetarchive.get_session()
-        # TODO: postgrest client
-        # TODO: minio client
+        self.sandcrawler_db_client = sandcrawler_db_client
+        self.sandcrawler_s3_client = sandcrawler_s3_client
 
-    def fetch_file_grobid(self, fe: FileEntity) -> Optional[Any]:
+    def fetch_file_grobid(self, fe: FileEntity, release_ident: str) -> Optional[Any]:
         """
         tei_xml: str
         release_ident: Optional[str]
         file_ident: Optional[str]
         """
-        return None
+        if not fe.sha1:
+            return None
+        if not fe.urls:
+            return None
+        grobid_meta = self.sandcrawler_db_client.get_grobid(fe.sha1)
+        if grobid_meta['status'] != 'success':
+            return None
+        #print(grobid_meta)
+        grobid_xml = self.sandcrawler_s3_client.get_blob(
+            folder="grobid",
+            sha1hex=fe.sha1,
+            extension=".tei.xml",
+            prefix="",
+            bucket="sandcrawler",
+        )
+        #print(grobid_xml)
+        return dict(
+            tei_xml=grobid_xml,
+            release_ident=release_ident,
+            file_ident=fe.ident,
+        )
 
-    def fetch_file_pdftotext(self, fe: FileEntity) -> Optional[Any]:
+    def fetch_file_pdftotext(self, fe: FileEntity, release_ident: str) -> Optional[Any]:
         """
         raw_text: str
         release_ident: Optional[str]
@@ -190,8 +212,8 @@ class WorkPipeline():
         #print(f"pref_idents={pref_idents}", file=sys.stderr)
 
         # find best accessible fatcat file
-        grobid_fulltext: Optional[Any]
-        pdftotext_fulltext: Optional[Any]
+        grobid_fulltext: Optional[Any] = None
+        pdftotext_fulltext: Optional[Any] = None
         for ident in pref_idents:
             release = release_dict[ident]
             if not release.files:
@@ -199,8 +221,8 @@ class WorkPipeline():
             for fe in release.files:
                 if not fe.sha1 or fe.mimetype not in (None, "application/pdf"):
                     continue
-                grobid_fulltext = self.fetch_file_grobid(fe)
-                pdftotext_fulltext = self.fetch_file_pdftotext(fe)
+                grobid_fulltext = self.fetch_file_grobid(fe, ident)
+                pdftotext_fulltext = self.fetch_file_pdftotext(fe, ident)
                 if grobid_fulltext or pdftotext_fulltext:
                     break
             if grobid_fulltext or pdftotext_fulltext:
@@ -279,6 +301,14 @@ def main():
         help="sqlite3 database file to open",
         default='data/issue_db.sqlite',
         type=str)
+    parser.add_argument("--sandcrawler-db-api",
+        help="Sandcrawler Postgrest API endpoint",
+        default='http://aitio.us.archive.org:3030',
+        type=str)
+    parser.add_argument("--sandcrawler-s3-api",
+        help="Sandcrawler S3 (minio/seaweedfs) API endpoint",
+        default='aitio.us.archive.org:9000',
+        type=str)
 
     sub = subparsers.add_parser('run_releases',
         help="takes expanded release entity JSON, sorted by work_ident")
@@ -292,8 +322,15 @@ def main():
         print("tell me what to do! (try --help)")
         sys.exit(-1)
 
-    idb = IssueDB(args.issue_db_file)
-    wp = WorkPipeline(issue_db=idb)
+    wp = WorkPipeline(
+        issue_db=IssueDB(args.issue_db_file),
+        sandcrawler_db_client=SandcrawlerPostgrestClient(api_url=args.sandcrawler_db_api),
+        sandcrawler_s3_client=SandcrawlerMinioClient(
+            host_url=args.sandcrawler_s3_api,
+            access_key=os.environ.get('MINIO_ACCESS_KEY'),
+            secret_key=os.environ.get('MINIO_SECRET_KEY'),
+        ),
+    )
 
     if args.func == 'run_releases':
         wp.run_releases(args.json_file)