Merge pull request alephdata#423 from alephdata/release/4.0.0

4.0.0 release

Author: stchris

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 3.22.0
+current_version = 4.0.0-rc28
 tag_name = {new_version}
 commit = True
 tag = True

.github/workflows/build.yml

@@ -15,7 +15,7 @@ jobs:
       - name: Set up
         run: |
           docker --version
-          docker-compose --version
+          docker compose --version
           echo "${GITHUB_REF}"
       - name: Start services
         run: |
@@ -25,6 +25,8 @@ jobs:
           make cached-build
       - name: Install development dependencies
         run: make dev
+        env:
+          PIP_BREAK_SYSTEM_PACKAGES: 1
       - name: Check formatting
         run: make format-check
       - name: Run linter (ruff)

.github/workflows/daily.yml

@@ -13,7 +13,7 @@ jobs:
       - name: Set up
         run: |
           docker --version
-          docker-compose --version
+          docker compose --version
       - name: Build docker cache
         run: |
           make fresh-cache

.github/workflows/main.yml

@@ -0,0 +1,19 @@
+name: Auto Assign to Project
+
+on:
+  issues:
+    types: [opened, labeled]
+  pull_request_target:
+    types: [opened, labeled]
+
+jobs:
+  assign_one_project:
+    runs-on: ubuntu-latest
+    name: Assign to One Project
+    steps:
+      - uses: actions/add-to-project@v0.3.0
+        with:
+          # You can target a repository in a different organization
+          # to the issue
+          project-url: https://github.com/orgs/alephdata/projects/10
+          github-token: ${{ secrets.ALEPH_GITHUB_TOKEN }}

Dockerfile

@@ -9,6 +9,8 @@ LABEL org.opencontainers.image.source https://github.com/alephdata/ingest-file
 # RUN echo "deb http://http.us.debian.org/debian stretch non-free" >/etc/apt/sources.list.d/nonfree.list
 RUN apt-get -qq -y update \
     && apt-get -qq -y install build-essential locales ca-certificates \
+    # git
+    git \
     # python deps (mostly to install their dependencies)
     python3-pip python3-dev python3-pil \
     # tesseract
@@ -121,8 +123,6 @@ RUN groupadd -g 1000 -r app \
 RUN mkdir /models/ && \
     curl -o "/models/model_type_prediction.ftz" "https://public.data.occrp.org/develop/models/types/type-08012020-7a69d1b.ftz"
 
-# Having updated pip/setuptools seems to break the test run for some reason (12/01/2022)
-# RUN pip3 install --no-cache-dir -U pip setuptools
 COPY requirements.txt /tmp/
 RUN pip3 install --no-cache-dir --prefer-binary --upgrade pip
 RUN pip3 install --no-cache-dir --prefer-binary --upgrade setuptools wheel
@@ -155,7 +155,8 @@ ENV ARCHIVE_TYPE=file \
     ARCHIVE_PATH=/data \
     FTM_STORE_URI=postgresql://aleph:aleph@postgres/aleph \
     REDIS_URL=redis://redis:6379/0 \
-    TESSDATA_PREFIX=/usr/share/tesseract-ocr/4.00/tessdata
+    TESSDATA_PREFIX=/usr/share/tesseract-ocr/4.00/tessdata \
+    LD_PRELOAD=/usr/lib/aarch64-linux-gnu/libgomp.so.1
 
 # USER app
 CMD ingestors process

Makefile

@@ -1,5 +1,5 @@
 INGEST=ghcr.io/alephdata/ingest-file
-COMPOSE=docker-compose
+COMPOSE=docker compose
 DOCKER=$(COMPOSE) run --rm ingest-file
 
 .PHONY: build

docker-compose.yml

@@ -12,6 +12,9 @@ services:
     image: redis:alpine
     command: ["redis-server", "--save", "3600", "10"]
 
+  rabbitmq:
+    image: rabbitmq:3.9-management-alpine
+
   ingest-file:
     build:
       context: .
@@ -22,7 +25,7 @@ services:
       - /data:mode=777
     environment:
       FTM_STORE_URI: postgresql://ingest:ingest@postgres/ingest
-      LOG_FORMAT: TEXT  # TEXT or JSON
+      LOG_FORMAT: TEXT # TEXT or JSON
     volumes:
       - "./ingestors:/ingestors/ingestors"
       - "./tests:/ingestors/tests"
@@ -33,3 +36,4 @@ services:
     depends_on:
       - postgres
       - redis
+      - rabbitmq

ingestors/__init__.py

@@ -2,10 +2,11 @@
 
 import logging
 
-__version__ = "3.22.0"
+__version__ = "4.0.0-rc28"
 
 logging.getLogger("chardet").setLevel(logging.INFO)
 logging.getLogger("PIL").setLevel(logging.INFO)
 logging.getLogger("google.auth").setLevel(logging.INFO)
 logging.getLogger("urllib3").setLevel(logging.WARNING)
 logging.getLogger("msglite").setLevel(logging.WARNING)
+logging.getLogger("pika").setLevel(logging.WARNING)

ingestors/cli.py

@@ -1,23 +1,26 @@
 import sys
 import click
 import logging
+import uuid
 from pprint import pprint
+from random import randrange
+
 from ftmstore import get_dataset
-from servicelayer.cache import get_redis, get_fakeredis
+from servicelayer.cache import get_redis
 from servicelayer.logs import configure_logging
-from servicelayer.jobs import Job, Dataset
+from servicelayer.taskqueue import Dataset, Task
 from servicelayer import settings as sl_settings
 from servicelayer.archive.util import ensure_path
+from servicelayer import settings as sls
 from servicelayer.tags import Tags
 
 from ingestors import settings
 from ingestors.manager import Manager
 from ingestors.directory import DirectoryIngestor
 from ingestors.analysis import Analyzer
-from ingestors.worker import IngestWorker, OP_ANALYZE, OP_INGEST
+from ingestors.worker import get_worker
 
 log = logging.getLogger(__name__)
-STAGES = [OP_ANALYZE, OP_INGEST]
 
 
 @click.group()
@@ -30,7 +33,7 @@ def cli():
 def process(sync):
     """Start the queue and process tasks as they come. Blocks while waiting"""
     num_threads = None if sync else sl_settings.WORKER_THREADS
-    worker = IngestWorker(stages=STAGES, num_threads=num_threads)
+    worker = get_worker(num_threads=num_threads)
     code = worker.run()
     sys.exit(code)
 
@@ -50,11 +53,22 @@ def killthekitten():
     conn.flushall()
 
 
-def _ingest_path(db, conn, dataset, path, languages=[]):
+def _ingest_path(db, dataset, path, languages=[]):
     context = {"languages": languages}
-    job = Job.create(conn, dataset)
-    stage = job.get_stage(OP_INGEST)
-    manager = Manager(db, stage, context)
+
+    priority = priority = randrange(1, sls.RABBITMQ_MAX_PRIORITY + 1)
+
+    task = Task(
+        task_id=uuid.uuid4().hex,
+        job_id=uuid.uuid4().hex,
+        collection_id=dataset,
+        delivery_tag="",
+        operation=settings.STAGE_INGEST,
+        priority=priority,
+        context=context,
+        payload={},
+    )
+    manager = Manager(db, task)
     path = ensure_path(path)
     if path is not None:
         if path.is_file():
@@ -76,15 +90,14 @@ def _ingest_path(db, conn, dataset, path, languages=[]):
 @click.argument("path", type=click.Path(exists=True))
 def ingest(path, dataset, languages=None):
     """Queue a set of files for ingest."""
-    conn = get_redis()
-    db = get_dataset(dataset, OP_INGEST)
-    _ingest_path(db, conn, dataset, path, languages=languages)
+    db = get_dataset(dataset, settings.STAGE_INGEST)
+    _ingest_path(db, dataset, path, languages=languages)
 
 
 @cli.command()
 @click.option("--dataset", required=True, help="Name of the dataset")
 def analyze(dataset):
-    db = get_dataset(dataset, OP_ANALYZE)
+    db = get_dataset(dataset, settings.STAGE_ANALYZE)
     analyzer = None
     for entity in db.partials():
         if analyzer is None or analyzer.entity.id != entity.id:
@@ -102,13 +115,20 @@ def analyze(dataset):
 @click.argument("path", type=click.Path(exists=True))
 def debug(path, languages=None):
     """Debug the ingest for the given path."""
-    conn = get_fakeredis()
     settings.fts.DATABASE_URI = "sqlite:////tmp/debug.sqlite3"
-    db = get_dataset("debug", origin=OP_INGEST, database_uri=settings.fts.DATABASE_URI)
+
+    # collection ID that is meant for testing purposes only
+    debug_datatset_id = 100
+
+    db = get_dataset(
+        debug_datatset_id,
+        origin=settings.STAGE_INGEST,
+        database_uri=settings.fts.DATABASE_URI,
+    )
     db.delete()
-    _ingest_path(db, conn, "debug", path, languages=languages)
-    worker = IngestWorker(conn=conn, stages=STAGES)
-    worker.sync()
+    _ingest_path(db, debug_datatset_id, path, languages=languages)
+    worker = get_worker()
+    worker.process(blocking=False)
     for entity in db.iterate():
         pprint(entity.to_dict())
 

ingestors/documents/pdf.py

@@ -19,6 +19,33 @@ class PDFIngestor(Ingestor, PDFSupport):
     EXTENSIONS = ["pdf"]
     SCORE = 6
 
+    def extract_xmp_metadata(self, pdf, entity):
+        try:
+            xmp = pdf.xmp_metadata
+            if xmp is None:
+                return
+            entity.add("messageId", xmp["xmpmm"].get("documentid"))
+            entity.add("title", xmp["dc"].get("title"))
+            entity.add("generator", xmp["pdf"].get("producer"))
+            entity.add("language", xmp["dc"].get("language"))
+            entity.add("authoredAt", xmp["xmp"].get("createdate"))
+            entity.add("modifiedAt", xmp["xmp"].get("modifydate"))
+        except Exception as ex:
+            log.warning("Error reading XMP: %r", ex)
+
+    def extract_metadata(self, pdf, entity):
+        meta = pdf.metadata
+        if meta is not None:
+            entity.add("title", meta.get("title"))
+            entity.add("author", meta.get("author"))
+            entity.add("generator", meta.get("creator"))
+            entity.add("generator", meta.get("producer"))
+            entity.add("keywords", meta.get("subject"))
+            if "creationdate" in meta:
+                entity.add("authoredAt", meta.get("creationdate"))
+            if "moddate" in meta:
+                entity.add("modifiedAt", meta.get("moddate"))
+
     def ingest(self, file_path, entity):
         """Ingestor implementation."""
         try: