bellingcat/auto-archiver
1
0
Fork 0
mirror of https://github.com/bellingcat/auto-archiver.git synced 2026-06-12 13:18:28 +03:00
Code Issues Packages Projects Releases Wiki Activity

Add documentation, pre-commit hook, more make commands and

Browse Source
This commit is contained in:
erinhmclark
2025-03-13 13:21:32 +00:00
parent 6e52a534e7
commit e76551ba22
21 changed files with 558 additions and 270 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/auto_archiver/modules/gsheet_feeder_db/__manifest__.py
View File

@@ -12,9 +12,7 @@
"default": None,
"help": "the id of the sheet to archive (alternative to 'sheet' config)",
},
"header": {"default": 1,
"help": "index of the header row (starts at 1)",
"type": "int"},
"header": {"default": 1, "help": "index of the header row (starts at 1)", "type": "int"},
"service_account": {
"default": "secrets/service_account.json",
"help": "service account JSON file path. Learn how to create one: https://gspread.readthedocs.io/en/latest/oauth2.html",

18
src/auto_archiver/modules/local_storage/local_storage.py
View File

@@ -1,4 +1,3 @@
import shutil
from typing import IO
import os
@@ -8,12 +7,13 @@ from auto_archiver.core import Media
from auto_archiver.core import Storage
from auto_archiver.core.consts import SetupError
class LocalStorage(Storage):
def setup(self) -> None:
if len(self.save_to) > 200:
raise SetupError("Your save_to path is too long, this will cause issues saving files on your computer. Please use a shorter path.")
raise SetupError(
"Your save_to path is too long, this will cause issues saving files on your computer. Please use a shorter path."
)
def get_cdn_url(self, media: Media) -> str:
dest = media.key
@@ -25,18 +25,18 @@ class LocalStorage(Storage):
def set_key(self, media, url, metadata):
# clarify we want to save the file to the save_to folder
old_folder = metadata.get('folder', '')
metadata.set_context('folder', os.path.join(self.save_to, metadata.get('folder', '')))
old_folder = metadata.get("folder", "")
metadata.set_context("folder", os.path.join(self.save_to, metadata.get("folder", "")))
super().set_key(media, url, metadata)
# don't impact other storages that might want a different 'folder' set
metadata.set_context('folder', old_folder)
metadata.set_context("folder", old_folder)
def upload(self, media: Media, **kwargs) -> bool:
# override parent so that we can use shutil.copy2 and keep metadata
dest = media.key
os.makedirs(os.path.dirname(dest), exist_ok=True)
logger.debug(f'[{self.__class__.__name__}] storing file {media.filename} with key {media.key} to {dest}')
logger.debug(f"[{self.__class__.__name__}] storing file {media.filename} with key {media.key} to {dest}")
res = shutil.copy2(media.filename, dest)
logger.info(res)
@@ -44,4 +44,4 @@ class LocalStorage(Storage):
# must be implemented even if unused
def uploadf(self, file: IO[bytes], key: str, **kwargs: dict) -> bool:
pass
pass

37
src/auto_archiver/modules/s3_storage/s3_storage.py
View File

@@ -1,4 +1,3 @@
from typing import IO
import boto3
@@ -11,18 +10,20 @@ from auto_archiver.utils.misc import calculate_file_hash, random_str
NO_DUPLICATES_FOLDER = "no-dups/"
class S3Storage(Storage):
class S3Storage(Storage):
def setup(self) -> None:
self.s3 = boto3.client(
's3',
"s3",
region_name=self.region,
endpoint_url=self.endpoint_url.format(region=self.region),
aws_access_key_id=self.key,
aws_secret_access_key=self.secret
aws_secret_access_key=self.secret,
)
if self.random_no_duplicate:
logger.warning("random_no_duplicate is set to True, this will override `path_generator`, `filename_generator` and `folder`.")
logger.warning(
"random_no_duplicate is set to True, this will override `path_generator`, `filename_generator` and `folder`."
)
def get_cdn_url(self, media: Media) -> str:
return self.cdn_url.format(bucket=self.bucket, region=self.region, key=media.key)
@@ -32,13 +33,13 @@ class S3Storage(Storage):
return True
extra_args = kwargs.get("extra_args", {})
if not self.private and 'ACL' not in extra_args:
extra_args['ACL'] = 'public-read'
if not self.private and "ACL" not in extra_args:
extra_args["ACL"] = "public-read"
if 'ContentType' not in extra_args:
if "ContentType" not in extra_args:
try:
if media.mimetype:
extra_args['ContentType'] = media.mimetype
extra_args["ContentType"] = media.mimetype
except Exception as e:
logger.warning(f"Unable to get mimetype for {media.key=}, error: {e}")
self.s3.upload_fileobj(file, Bucket=self.bucket, Key=media.key, ExtraArgs=extra_args)
@@ -50,21 +51,21 @@ class S3Storage(Storage):
hd = calculate_file_hash(media.filename)
path = os.path.join(NO_DUPLICATES_FOLDER, hd[:24])
if existing_key:=self.file_in_folder(path):
if existing_key := self.file_in_folder(path):
media._key = existing_key
media.set("previously archived", True)
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:
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
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

2
src/auto_archiver/modules/tiktok_tikwm_extractor/__init__.py
View File

@@ -1 +1 @@
from .tiktok_tikwm_extractor import TiktokTikwmExtractor
from .tiktok_tikwm_extractor import TiktokTikwmExtractor

7
src/auto_archiver/modules/tiktok_tikwm_extractor/__manifest__.py
View File

@@ -2,10 +2,7 @@
"name": "Tiktok Tikwm Extractor",
"type": ["extractor"],
"requires_setup": False,
"dependencies": {
"python": ["loguru", "requests"],
"bin": []
},
"dependencies": {"python": ["loguru", "requests"], "bin": []},
"description": """
Uses an unofficial TikTok video download platform's API to download videos: https://tikwm.com/
@@ -19,5 +16,5 @@
- If tikwm.com is down, this extractor will not work.
- If tikwm.com changes their API, this extractor may break.
- If no video is found, this extractor will consider the extraction failed.
"""
""",
}

7
src/auto_archiver/modules/tiktok_tikwm_extractor/tiktok_tikwm_extractor.py
View File

@@ -12,11 +12,12 @@ class TiktokTikwmExtractor(Extractor):
"""
Extractor for TikTok that uses an unofficial API and can capture content that requires a login, like sensitive content.
"""
TIKWM_ENDPOINT = "https://www.tikwm.com/api/?url={url}"
def download(self, item: Metadata) -> bool | Metadata:
url = item.get_url()
if not re.match(TikTokIE._VALID_URL, url):
return False
@@ -33,7 +34,7 @@ class TiktokTikwmExtractor(Extractor):
logger.error(f"failed to parse JSON response from tikwm.com for {url=}")
return False
if not json_response.get('msg') == 'success' or not (api_data := json_response.get('data', {})):
if not json_response.get("msg") == "success" or not (api_data := json_response.get("data", {})):
logger.error(f"failed to get a valid response from tikwm.com for {url=}: {json_response}")
return False
@@ -67,7 +68,7 @@ class TiktokTikwmExtractor(Extractor):
if created_at := api_data.pop("create_time", None):
result.set_timestamp(datetime.fromtimestamp(created_at, tz=timezone.utc))
if (author := api_data.pop("author", None)):
if author := api_data.pop("author", None):
result.set("author", author)
result.set("api_data", api_data)

4
src/auto_archiver/modules/wacz_extractor_enricher/__manifest__.py
View File

@@ -14,9 +14,7 @@
"help": "browsertrix-profile (for profile generation see https://github.com/webrecorder/browsertrix-crawler#creating-and-using-browser-profiles).",
},
"docker_commands": {"default": None, "help": "if a custom docker invocation is needed"},
"timeout": {"default": 120,
"help": "timeout for WACZ generation in seconds",
"type": "int"},
"timeout": {"default": 120, "help": "timeout for WACZ generation in seconds", "type": "int"},
"extract_media": {
"default": False,
"type": "bool",