mirror of
https://github.com/bellingcat/auto-archiver-api.git
synced 2026-06-08 03:28:35 +03:00
Update misc.py
This commit is contained in:
@@ -1,6 +1,11 @@
|
||||
import base64
|
||||
from typing import List
|
||||
|
||||
from auto_archiver.core import Media, Metadata
|
||||
from fastapi.encoders import jsonable_encoder
|
||||
from loguru import logger
|
||||
|
||||
from app.shared.db import models
|
||||
|
||||
|
||||
def custom_jsonable_encoder(obj):
|
||||
@@ -14,3 +19,44 @@ def convert_priority_to_queue_dict(priority: str) -> dict:
|
||||
"priority": 0 if priority == "high" else 10,
|
||||
"queue": f"{priority}_priority",
|
||||
}
|
||||
|
||||
|
||||
def convert_if_media(media):
|
||||
if isinstance(media, Media):
|
||||
return media
|
||||
elif isinstance(media, dict):
|
||||
try:
|
||||
return Media.from_dict(media)
|
||||
except Exception as e:
|
||||
logger.debug(f"error parsing {media} : {e}")
|
||||
return False
|
||||
|
||||
|
||||
def get_all_urls(result: Metadata) -> List[models.ArchiveUrl]:
|
||||
db_urls = []
|
||||
for m in result.media:
|
||||
for i, url in enumerate(m.urls):
|
||||
db_urls.append(
|
||||
models.ArchiveUrl(url=url, key=m.get("id", f"media_{i}"))
|
||||
)
|
||||
for k, prop in m.properties.items():
|
||||
if prop_converted := convert_if_media(prop):
|
||||
for i, url in enumerate(prop_converted.urls):
|
||||
db_urls.append(
|
||||
models.ArchiveUrl(
|
||||
url=url, key=prop_converted.get("id", f"{k}_{i}")
|
||||
)
|
||||
)
|
||||
if isinstance(prop, list):
|
||||
for i, prop_media in enumerate(prop):
|
||||
if prop_media := convert_if_media(prop_media):
|
||||
for j, url in enumerate(prop_media.urls):
|
||||
db_urls.append(
|
||||
models.ArchiveUrl(
|
||||
url=url,
|
||||
key=prop_media.get(
|
||||
"id", f"{k}{prop_media.key}_{i}.{j}"
|
||||
),
|
||||
)
|
||||
)
|
||||
return db_urls
|
||||
|
||||
Reference in New Issue
Block a user