enables api_db cache queries if configured with new option (#113)

Co-authored-by: msramalho <19508417+msramalho@users.noreply.github.com>

Pipfile

@@ -35,6 +35,7 @@ vk-url-scraper = "*"
 requests = {extras = ["socks"], version = "*"}
 numpy = "*"
 warcio = "*"
+jsonlines = "*"
 
 [dev-packages]
 autopep8 = "*"

src/auto_archiver/archivers/tiktok_archiver.py

@@ -1,9 +1,11 @@
-import json, os, traceback, uuid
+import json, os, traceback
 import tiktok_downloader
 from loguru import logger
 
+
 from . import Archiver
 from ..core import Metadata, Media, ArchivingContext
+from ..utils.misc import random_str
 
 
 class TiktokArchiver(Archiver):
@@ -37,7 +39,7 @@ class TiktokArchiver(Archiver):
             logger.warning(f'Other Tiktok error {error}')
 
         try:
-            filename = os.path.join(ArchivingContext.get_tmp_dir(), f'{str(uuid.uuid4())[0:8]}.mp4')
+            filename = os.path.join(ArchivingContext.get_tmp_dir(), f'{random_str(8)}.mp4')
             tiktok_media = tiktok_downloader.snaptik(url).get_media()
 
             if len(tiktok_media) <= 0:

src/auto_archiver/core/metadata.py

@@ -105,7 +105,8 @@ class Metadata:
 
     def get_timestamp(self, utc=True, iso=True) -> datetime.datetime:
         ts = self.get("timestamp")
-        if not ts: return ts
+        if not ts: return 
+        if type(ts) == float: ts = datetime.datetime.fromtimestamp(ts)
         if utc: ts = ts.replace(tzinfo=datetime.timezone.utc)
         if iso: return ts.isoformat()
         return ts

src/auto_archiver/core/orchestrator.py

@@ -77,7 +77,7 @@ class ArchivingOrchestrator:
         if cached_result:
             logger.debug("Found previously archived entry")
             for d in self.databases:
-                d.done(cached_result)
+                d.done(cached_result, cached=True)
             return cached_result
 
         # 3 - call archivers until one succeeds

src/auto_archiver/databases/api_db.py

@@ -1,3 +1,4 @@
+from typing import Union
 import requests, os
 from loguru import logger
 
@@ -14,6 +15,7 @@ class AAApiDb(Database):
     def __init__(self, config: dict) -> None:
         # without this STEP.__init__ is not called
         super().__init__(config)
+        self.allow_rearchive = bool(self.allow_rearchive)
         self.assert_valid_string("api_endpoint")
         self.assert_valid_string("api_secret")
 
@@ -21,16 +23,37 @@ class AAApiDb(Database):
     def configs() -> dict:
         return {
             "api_endpoint": {"default": None, "help": "API endpoint where calls are made to"},
-            "api_secret": {"default": None, "help": "API authentication secret"},
+            "api_secret": {"default": None, "help": "API Basic authentication secret [deprecating soon]"},
+            "api_token": {"default": None, "help": "API Bearer token, to be preferred over secret (Basic auth) going forward"},
             "public": {"default": False, "help": "whether the URL should be publicly available via the API"},
             "author_id": {"default": None, "help": "which email to assign as author"},
             "group_id": {"default": None, "help": "which group of users have access to the archive in case public=false as author"},
+            "allow_rearchive": {"default": True, "help": "if False then the API database will be queried prior to any archiving operations and stop if the link has already been archived"},
             "tags": {"default": [], "help": "what tags to add to the archived URL", "cli_set": lambda cli_val, cur_val: set(cli_val.split(","))},
         }
+    def fetch(self, item: Metadata) -> Union[Metadata, bool]:
+        """ query the database for the existence of this item"""
+        if not self.allow_rearchive: return
+        
+        params = {"url": item.get_url(), "limit": 1}
+        headers = {"Authorization": f"Bearer {self.api_token}", "accept": "application/json"}
+        response = requests.get(os.path.join(self.api_endpoint, "tasks/search-url"), params=params, headers=headers)
 
-    def done(self, item: Metadata) -> None:
+        if response.status_code == 200:
+            logger.success(f"API returned a previously archived instance: {response.json()}")
+            # TODO: can we do better than just returning the first result?
+            return Metadata.from_dict(response.json()[0]["result"])
+        
+        logger.error(f"AA API FAIL ({response.status_code}): {response.json()}")
+        return False
+
+
+    def done(self, item: Metadata, cached: bool=False) -> None:
         """archival result ready - should be saved to DB"""
-        logger.info(f"saving archive of {item.get_url()} to the AA API.")
+        if cached: 
+            logger.debug(f"skipping saving archive of {item.get_url()} to the AA API because it was cached")
+            return
+        logger.debug(f"saving archive of {item.get_url()} to the AA API.")
 
         payload = {'result': item.to_json(), 'public': self.public, 'author_id': self.author_id, 'group_id': self.group_id, 'tags': list(self.tags)}
         response = requests.post(os.path.join(self.api_endpoint, "submit-archive"), json=payload, auth=("abc", self.api_secret))
@@ -39,3 +62,5 @@ class AAApiDb(Database):
             logger.success(f"AA API: {response.json()}")
         else:
             logger.error(f"AA API FAIL ({response.status_code}): {response.json()}")
+
+    

src/auto_archiver/databases/console_db.py

@@ -27,6 +27,6 @@ class ConsoleDb(Database):
     def aborted(self, item: Metadata) -> None:
         logger.warning(f"ABORTED {item}")
 
-    def done(self, item: Metadata) -> None:
+    def done(self, item: Metadata, cached: bool=False) -> None:
         """archival result ready - should be saved to DB"""
         logger.success(f"DONE {item}")

src/auto_archiver/databases/csv_db.py

@@ -24,7 +24,7 @@ class CSVDb(Database):
             "csv_file": {"default": "db.csv", "help": "CSV file name"}
         }
 
-    def done(self, item: Metadata) -> None:
+    def done(self, item: Metadata, cached: bool=False) -> None:
         """archival result ready - should be saved to DB"""
         logger.success(f"DONE {item}")
         is_empty = not os.path.isfile(self.csv_file) or os.path.getsize(self.csv_file) == 0

src/auto_archiver/databases/database.py

@@ -36,6 +36,6 @@ class Database(Step, ABC):
         return False
 
     @abstractmethod
-    def done(self, item: Metadata) -> None:
+    def done(self, item: Metadata, cached: bool=False) -> None:
         """archival result ready - should be saved to DB"""
         pass

src/auto_archiver/databases/gsheet_db.py

@@ -41,7 +41,7 @@ class GsheetsDb(Database):
         """check if the given item has been archived already"""
         return False
 
-    def done(self, item: Metadata) -> None:
+    def done(self, item: Metadata, cached: bool=False) -> None:
         """archival result ready - should be saved to DB"""
         logger.success(f"DONE {item.get_url()}")
         gw, row = self._retrieve_gsheet(item)
@@ -57,8 +57,10 @@ class GsheetsDb(Database):
                     cell_updates.append((row, col, final_value))
             except Exception as e:
                 logger.error(f"Unable to batch {col}={final_value} due to {e}")
-
+        status_message = item.status
-        cell_updates.append((row, 'status', item.status))
+        if cached:
+            status_message = f"[cached] {status_message}"
+        cell_updates.append((row, 'status', status_message))
 
         media: Media = item.get_final_media()
         if hasattr(media, "urls"):

src/auto_archiver/enrichers/screenshot_enricher.py

@@ -1,9 +1,10 @@
 from loguru import logger
-import time, uuid, os
+import time, os
 from selenium.common.exceptions import TimeoutException
 
+
 from . import Enricher
-from ..utils import Webdriver, UrlUtil
+from ..utils import Webdriver, UrlUtil, random_str  
 from ..core import Media, Metadata, ArchivingContext
 
 class ScreenshotEnricher(Enricher):
@@ -29,7 +30,7 @@ class ScreenshotEnricher(Enricher):
             try:
                 driver.get(url)
                 time.sleep(int(self.sleep_before_screenshot))
-                screenshot_file = os.path.join(ArchivingContext.get_tmp_dir(), f"screenshot_{str(uuid.uuid4())[0:8]}.png")
+                screenshot_file = os.path.join(ArchivingContext.get_tmp_dir(), f"screenshot_{random_str(8)}.png")
                 driver.save_screenshot(screenshot_file)
                 to_enrich.add_media(Media(filename=screenshot_file), id="screenshot")
             except TimeoutException:

src/auto_archiver/enrichers/thumbnail_enricher.py

@@ -1,8 +1,9 @@
-import ffmpeg, os, uuid
+import ffmpeg, os
 from loguru import logger
 
 from . import Enricher
 from ..core import Media, Metadata, ArchivingContext
+from ..utils.misc import random_str
 
 
 class ThumbnailEnricher(Enricher):
@@ -23,7 +24,7 @@ class ThumbnailEnricher(Enricher):
         logger.debug(f"generating thumbnails")
         for i, m in enumerate(to_enrich.media[::]):
             if m.is_video():
-                folder = os.path.join(ArchivingContext.get_tmp_dir(), str(uuid.uuid4()))
+                folder = os.path.join(ArchivingContext.get_tmp_dir(), random_str(24))
                 os.makedirs(folder, exist_ok=True)
                 logger.debug(f"generating thumbnails for {m.filename}")
                 fps, duration = 0.5, m.get("duration")

src/auto_archiver/enrichers/wacz_enricher.py

@@ -1,5 +1,6 @@
+import jsonlines
 import mimetypes
-import os, shutil, subprocess, uuid
+import os, shutil, subprocess
 from zipfile import ZipFile
 from loguru import logger
 from warcio.archiveiterator import ArchiveIterator
@@ -7,7 +8,7 @@ from warcio.archiveiterator import ArchiveIterator
 from ..core import Media, Metadata, ArchivingContext
 from . import Enricher
 from ..archivers import Archiver
-from ..utils import UrlUtil
+from ..utils import UrlUtil, random_str
 
 
 class WaczArchiverEnricher(Enricher, Archiver):
@@ -46,7 +47,7 @@ class WaczArchiverEnricher(Enricher, Archiver):
 
         url = to_enrich.get_url()
 
-        collection = str(uuid.uuid4())[0:8]
+        collection = random_str(8)
         browsertrix_home_host = os.environ.get('BROWSERTRIX_HOME_HOST') or os.path.abspath(ArchivingContext.get_tmp_dir())
         browsertrix_home_container = os.environ.get('BROWSERTRIX_HOME_CONTAINER') or browsertrix_home_host
 
@@ -106,6 +107,24 @@ class WaczArchiverEnricher(Enricher, Archiver):
         to_enrich.add_media(Media(wacz_fn), "browsertrix")
         if self.extract_media:
             self.extract_media_from_wacz(to_enrich, wacz_fn)
+
+        if use_docker:
+            jsonl_fn = os.path.join(browsertrix_home_container, "collections", collection, "pages", "pages.jsonl")
+        else:
+            jsonl_fn = os.path.join("collections", collection, "pages", "pages.jsonl")
+
+        if not os.path.exists(jsonl_fn):
+            logger.warning(f"Unable to locate and pages.jsonl  {jsonl_fn=}")
+        else:
+            logger.info(f"Parsing pages.jsonl  {jsonl_fn=}")
+            with jsonlines.open(jsonl_fn) as reader:
+                for obj in reader:
+                    if 'title' in obj:
+                        to_enrich.set_title(obj['title'])
+                    if 'text' in obj:
+                        to_enrich.set_content(obj['text'])
+
+
         return True
 
     def extract_media_from_wacz(self, to_enrich: Metadata, wacz_filename: str) -> None:

src/auto_archiver/formatters/html_formatter.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 from dataclasses import dataclass
-import mimetypes, uuid, os, pathlib
+import mimetypes, os, pathlib
 from jinja2 import Environment, FileSystemLoader
 from urllib.parse import quote
 from loguru import logger
@@ -9,6 +9,7 @@ from ..version import __version__
 from ..core import Metadata, Media, ArchivingContext
 from . import Formatter
 from ..enrichers import HashEnricher
+from ..utils.misc import random_str
 
 
 @dataclass
@@ -44,10 +45,10 @@ class HtmlFormatter(Formatter):
             metadata=item.metadata,
             version=__version__
         )
-        html_path = os.path.join(ArchivingContext.get_tmp_dir(), f"formatted{str(uuid.uuid4())}.html")
+        html_path = os.path.join(ArchivingContext.get_tmp_dir(), f"formatted{random_str(24)}.html")
         with open(html_path, mode="w", encoding="utf-8") as outf:
             outf.write(content)
-        final_media = Media(filename=html_path)
+        final_media = Media(filename=html_path, _mimetype="text/html")
 
         he = HashEnricher({"hash_enricher": {"algorithm": ArchivingContext.get("hash_enricher.algorithm"), "chunksize": 1.6e7}})
         if len(hd := he.calculate_hash(final_media.filename)):

src/auto_archiver/storages/s3.py

@@ -1,14 +1,14 @@
 
-from typing import IO, Any
+from typing import IO
-import boto3, uuid, os, mimetypes
+import boto3, os
-from botocore.errorfactory import ClientError
+
-from ..core import Metadata
+from ..utils.misc import random_str
 from ..core import Media
 from ..storages import Storage
+from ..enrichers import HashEnricher
 from loguru import logger
-from slugify import slugify
-
 
+NO_DUPLICATES_FOLDER = "no-dups/"
 class S3Storage(Storage):
     name = "s3_storage"
 
@@ -21,6 +21,9 @@ class S3Storage(Storage):
             aws_access_key_id=self.key,
             aws_secret_access_key=self.secret
         )
+        self.random_no_duplicate = bool(self.random_no_duplicate)
+        if self.random_no_duplicate:
+            logger.warning("random_no_duplicate is set to True, this will override `path_generator`, `filename_generator` and `folder`.")
 
     @staticmethod
     def configs() -> dict:
@@ -31,7 +34,7 @@ class S3Storage(Storage):
                 "region": {"default": None, "help": "S3 region name"},
                 "key": {"default": None, "help": "S3 API key"},
                 "secret": {"default": None, "help": "S3 API secret"},
-                # TODO: how to have sth like a custom folder? has to come from the feeders
+                "random_no_duplicate": {"default": False, "help": f"if set, it will override `path_generator`, `filename_generator` and `folder`. It will check if the file already exists and if so it will not upload it again. Creates a new root folder path `{NO_DUPLICATES_FOLDER}`"},
                 "endpoint_url": {
                     "default": 'https://{region}.digitaloceanspaces.com',
                     "help": "S3 bucket endpoint, {region} are inserted at runtime"
@@ -47,6 +50,22 @@ class S3Storage(Storage):
         return self.cdn_url.format(bucket=self.bucket, region=self.region, key=media.key)
 
     def uploadf(self, file: IO[bytes], media: Media, **kwargs: dict) -> None:
+        if not self.is_upload_needed(media): return True
+
+        if self.random_no_duplicate:
+            # checks if a folder with the hash already exists, if so it skips the upload
+            he = HashEnricher({"hash_enricher": {"algorithm": "SHA-256", "chunksize": 1.6e7}})
+            hd = he.calculate_hash(media.filename)
+            path = os.path.join(NO_DUPLICATES_FOLDER, hd[:24])
+
+            if existing_key:=self.file_in_folder(path):
+                media.key = existing_key
+                logger.debug(f"skipping upload of {media.filename} because it already exists in {media.key}")
+                return True
+            
+            _, ext = os.path.splitext(media.key)
+            media.key = os.path.join(path, f"{random_str(24)}{ext}")
+
         extra_args = kwargs.get("extra_args", {})
         if not self.private and 'ACL' not in extra_args:
             extra_args['ACL'] = 'public-read'
@@ -60,14 +79,30 @@ class S3Storage(Storage):
 
         self.s3.upload_fileobj(file, Bucket=self.bucket, Key=media.key, ExtraArgs=extra_args)
         return True
+    
+    def is_upload_needed(self, media: Media) -> bool:
+        if self.random_no_duplicate:
+            # checks if a folder with the hash already exists, if so it skips the upload
+            he = HashEnricher({"hash_enricher": {"algorithm": "SHA-256", "chunksize": 1.6e7}})
+            hd = he.calculate_hash(media.filename)
+            path = os.path.join(NO_DUPLICATES_FOLDER, hd[:24])
+
+            if existing_key:=self.file_in_folder(path):
+                media.key = existing_key
+                logger.debug(f"skipping upload of {media.filename} because it already exists in {media.key}")
+                return False
+            
+            _, ext = os.path.splitext(media.key)
+            media.key = os.path.join(path, f"{random_str(24)}{ext}")
+        return True
+    
+    
+    def file_in_folder(self, path:str) -> str:
+        # checks if path exists and is not an empty folder
+        if not path.endswith('/'):
+            path = path + '/' 
+        resp = self.s3.list_objects(Bucket=self.bucket, Prefix=path, Delimiter='/', MaxKeys=1)
+        if 'Contents' in resp:
+            return resp['Contents'][0]['Key']
+        return False
 
-    # def exists(self, key: str) -> bool:
-    #     """
-    #     Tests if a given file with key=key exists in the bucket
-    #     """
-    #     try:
-    #         self.s3.head_object(Bucket=self.bucket, Key=key)
-    #         return True
-    #     except ClientError as e:
-    #         logger.warning(f"got a ClientError when checking if {key=} exists in bucket={self.bucket}: {e}")
-    #     return False

src/auto_archiver/storages/storage.py

@@ -2,11 +2,13 @@ from __future__ import annotations
 from abc import abstractmethod
 from dataclasses import dataclass
 from typing import IO
+import os
+
+from ..utils.misc import random_str
 
 from ..core import Media, Step, ArchivingContext
 from ..enrichers import HashEnricher
 from loguru import logger
-import os, uuid
 from slugify import slugify
 
 
@@ -72,10 +74,10 @@ class Storage(Step):
             filename = slugify(filename)  # in case it comes with os.sep
         elif self.path_generator == "url": path = slugify(url)
         elif self.path_generator == "random":
-            path = ArchivingContext.get("random_path", str(uuid.uuid4())[:16], True)
+            path = ArchivingContext.get("random_path", random_str(24), True)
 
         # filename_generator logic
-        if self.filename_generator == "random": filename = str(uuid.uuid4())[:16]
+        if self.filename_generator == "random": filename = random_str(24)
         elif self.filename_generator == "static":
             he = HashEnricher({"hash_enricher": {"algorithm": ArchivingContext.get("hash_enricher.algorithm"), "chunksize": 1.6e7}})
             hd = he.calculate_hash(media.filename)

src/auto_archiver/utils/misc.py

@@ -1,5 +1,6 @@
 
 import os, json, requests
+import uuid
 from datetime import datetime
 from loguru import logger
 
@@ -49,3 +50,7 @@ def update_nested_dict(dictionary, update_dict):
             update_nested_dict(dictionary[key], value)
         else:
             dictionary[key] = value
+
+def random_str(length: int = 32) -> str:
+    assert length <= 32, "length must be less than 32 as UUID4 is used"
+    return str(uuid.uuid4()).replace("-", "")[:length]

src/auto_archiver/utils/url.py

@@ -65,6 +65,9 @@ class UrlUtil:
         if "vk.com/images/" in url: return False
         if "vk.com/images/reaction/" in url: return False
 
+        # wikipedia
+        if "wikipedia.org/static" in url: return False
+
         return True
 
     @staticmethod

src/auto_archiver/utils/webdriver.py

@@ -15,7 +15,7 @@ class Webdriver:
 
     def __enter__(self) -> webdriver:
         options = webdriver.FirefoxOptions()
-        options.headless = True
+        options.add_argument("--headless")
         options.set_preference('network.protocol-handler.external.tg', False)
         try:
             self.driver = webdriver.Firefox(options=options)

src/auto_archiver/version.py

@@ -1,9 +1,9 @@
 
 _MAJOR = "0"
-_MINOR = "6"
+_MINOR = "7"
 # On main and in a nightly release the patch should be one ahead of the last
 # released build.
-_PATCH = "13"
+_PATCH = "2"
 # This is mainly for nightly builds which have the suffix ".dev$DATE". See
 # https://semver.org/#is-v123-a-semantic-version for the semantics.
 _SUFFIX = ""