mc_378_internet_archive

ENV.md

@@ -22,6 +22,7 @@ Please ensure these are properly defined in a `.env` file in the root directory.
 | `DISCORD_WEBHOOK_URL`           | The URL for the Discord webhook used for notifications                                                                                                                        | `abc123`                                                                                     |
 | `HUGGINGFACE_INFERENCE_API_KEY` | The API key required for accessing the Hugging Face Inference API.                                                                                                            | `abc123`                                                                                     |
 | `HUGGINGFACE_HUB_TOKEN`         | The API key required for uploading to the PDAP HuggingFace account via Hugging Face Hub API.                                                                                                                                                                     | `abc123`  |
+| `INTERNET_ARCHIVE_S3_KEYS`      | Keys used for saving a URL to the Internet Archives.   | 'abc123:gpb0dk` |
 
 
 
@@ -32,25 +33,27 @@ Task flags are used to enable/disable certain tasks. They are set to `1` to enab
 
 The following flags are available:
 
-| Flag                                  | Description                                            |
-|---------------------------------------|--------------------------------------------------------|
-| `SCHEDULED_TASKS_FLAG`                | All scheduled tasks.                                   |
-| `URL_HTML_TASK_FLAG`                  | URL HTML scraping task.                                |
-| `URL_RECORD_TYPE_TASK_FLAG`           | Automatically assigns Record Types to URLs.            |
+| Flag                                | Description                                            |
+|-------------------------------------|--------------------------------------------------------|
+| `SCHEDULED_TASKS_FLAG`              | All scheduled tasks.                                   |
+| `URL_HTML_TASK_FLAG`                | URL HTML scraping task.                                |
+| `URL_RECORD_TYPE_TASK_FLAG`         | Automatically assigns Record Types to URLs.            |
 | `URL_AGENCY_IDENTIFICATION_TASK_FLAG` | Automatically assigns and suggests Agencies for URLs.  |
-| `URL_SUBMIT_APPROVED_TASK_FLAG`       | Submits approved URLs to the Data Sources App.         |
-| `URL_MISC_METADATA_TASK_FLAG`         | Adds misc metadata to URLs.                            |
-| `URL_404_PROBE_TASK_FLAG`             | Probes URLs for 404 errors.                            |
-| `URL_AUTO_RELEVANCE_TASK_FLAG`        | Automatically assigns Relevances to URLs.              |
-| `URL_PROBE_TASK_FLAG`                 | Probes URLs for web metadata.                          |
-| `URL_ROOT_URL_TASK_FLAG`              | Extracts and links Root URLs to URLs.                  |
-| `SYNC_AGENCIES_TASK_FLAG`             | Synchonize agencies from Data Sources App.             |
-| `SYNC_DATA_SOURCES_TASK_FLAG`         | Synchonize data sources from Data Sources App.         |
-| `PUSH_TO_HUGGING_FACE_TASK_FLAG`      | Pushes data to HuggingFace.                            |
+| `URL_SUBMIT_APPROVED_TASK_FLAG`     | Submits approved URLs to the Data Sources App.         |
+| `URL_MISC_METADATA_TASK_FLAG`       | Adds misc metadata to URLs.                            |
+| `URL_404_PROBE_TASK_FLAG`           | Probes URLs for 404 errors.                            |
+| `URL_AUTO_RELEVANCE_TASK_FLAG`      | Automatically assigns Relevances to URLs.              |
+| `URL_PROBE_TASK_FLAG`               | Probes URLs for web metadata.                          |
+| `URL_ROOT_URL_TASK_FLAG`            | Extracts and links Root URLs to URLs.                  |
+| `SYNC_AGENCIES_TASK_FLAG`           | Synchonize agencies from Data Sources App.             |
+| `SYNC_DATA_SOURCES_TASK_FLAG`       | Synchonize data sources from Data Sources App.         |
+| `PUSH_TO_HUGGING_FACE_TASK_FLAG`    | Pushes data to HuggingFace.                            |
 | `POPULATE_BACKLOG_SNAPSHOT_TASK_FLAG` | Populates the backlog snapshot.                        |
-| `DELETE_OLD_LOGS_TASK_FLAG`           | Deletes old logs.                                      |
-| `RUN_URL_TASKS_TASK_FLAG`             | Runs URL tasks.                                        |
-| `IA_PROBE_TASK_FLAG`              | Extracts and links Internet Archives metadata to URLs. |
+| `DELETE_OLD_LOGS_TASK_FLAG`         | Deletes old logs.                                      |
+| `RUN_URL_TASKS_TASK_FLAG`           | Runs URL tasks.                                        |
+| `IA_PROBE_TASK_FLAG`            | Extracts and links Internet Archives metadata to URLs. |
+| `IA_SAVE_TASK_FLAG`             | Saves URLs to Internet Archives.                       |
+
 
 
 ## Foreign Data Wrapper (FDW)

alembic/versions/2025_08_17_1830-8a70ee509a74_add_internet_archives_upload_task.py

@@ -0,0 +1,43 @@
+"""Add internet archives upload task
+
+Revision ID: 8a70ee509a74
+Revises: 2a7192657354
+Create Date: 2025-08-17 18:30:18.353605
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+from src.util.alembic_helpers import id_column, url_id_column, created_at_column
+
+# revision identifiers, used by Alembic.
+revision: str = '8a70ee509a74'
+down_revision: Union[str, None] = '2a7192657354'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+IA_PROBE_METADATA_TABLE_NAME_OLD = "urls_internet_archive_metadata"
+IA_PROBE_METADATA_TABLE_NAME_NEW = "url_internet_archives_probe_metadata"
+
+IA_UPLOAD_METADATA_TABLE_NAME = "url_internet_archives_save_metadata"
+
+def upgrade() -> None:
+    _create_internet_archive_save_metadata_table()
+    op.rename_table(IA_PROBE_METADATA_TABLE_NAME_OLD, IA_PROBE_METADATA_TABLE_NAME_NEW)
+
+
+
+def downgrade() -> None:
+    op.drop_table(IA_UPLOAD_METADATA_TABLE_NAME)
+    op.rename_table(IA_PROBE_METADATA_TABLE_NAME_NEW, IA_PROBE_METADATA_TABLE_NAME_OLD)
+
+def _create_internet_archive_save_metadata_table() -> None:
+    op.create_table(
+        IA_UPLOAD_METADATA_TABLE_NAME,
+        id_column(),
+        url_id_column(),
+        created_at_column(),
+        sa.Column('last_uploaded_at', sa.DateTime(), nullable=False, server_default=sa.text('now()')),
+    )

src/core/tasks/scheduled/impl/internet_archives/probe/convert.py

@@ -1,6 +1,5 @@
+from src.db.models.impl.url.internet_archives.probe.pydantic import URLInternetArchiveMetadataPydantic
 from src.external.internet_archives.models.ia_url_mapping import InternetArchivesURLMapping
-from src.db.models.impl.flag.checked_for_ia.pydantic import FlagURLCheckedForInternetArchivesPydantic
-from src.db.models.impl.url.ia_metadata.pydantic import URLInternetArchiveMetadataPydantic
 from src.util.url_mapper import URLMapper
 
 

src/core/tasks/scheduled/impl/internet_archives/probe/operator.py

@@ -14,7 +14,7 @@
 from src.db.enums import TaskType
 from src.db.models.impl.flag.checked_for_ia.pydantic import FlagURLCheckedForInternetArchivesPydantic
 from src.db.models.impl.url.error_info.pydantic import URLErrorPydanticInfo
-from src.db.models.impl.url.ia_metadata.pydantic import URLInternetArchiveMetadataPydantic
+from src.db.models.impl.url.internet_archives.probe.pydantic import URLInternetArchiveMetadataPydantic
 from src.external.internet_archives.client import InternetArchivesClient
 from src.external.internet_archives.models.ia_url_mapping import InternetArchivesURLMapping
 from src.util.url_mapper import URLMapper

src/core/tasks/scheduled/impl/internet_archives/save/filter.py

@@ -0,0 +1,14 @@
+from src.core.tasks.scheduled.impl.internet_archives.save.models.mapping import URLInternetArchivesSaveResponseMapping
+from src.core.tasks.scheduled.impl.internet_archives.save.models.subset import IASaveURLMappingSubsets
+
+
+def filter_save_responses(
+    resp_mappings: list[URLInternetArchivesSaveResponseMapping]
+) -> IASaveURLMappingSubsets:
+    subsets = IASaveURLMappingSubsets()
+    for resp_mapping in resp_mappings:
+        if resp_mapping.response.has_error:
+            subsets.error.append(resp_mapping.response)
+        else:
+            subsets.success.append(resp_mapping.response)
+    return subsets

src/core/tasks/scheduled/impl/internet_archives/save/mapper.py

@@ -0,0 +1,18 @@
+from src.core.tasks.scheduled.impl.internet_archives.save.models.entry import InternetArchivesSaveTaskEntry
+
+
+class URLToEntryMapper:
+
+    def __init__(self, entries: list[InternetArchivesSaveTaskEntry]):
+        self._url_to_entry: dict[str, InternetArchivesSaveTaskEntry] = {
+            entry.url: entry for entry in entries
+        }
+
+    def get_is_new(self, url: str) -> bool:
+        return self._url_to_entry[url].is_new
+
+    def get_url_id(self, url: str) -> int:
+        return self._url_to_entry[url].url_id
+
+    def get_all_urls(self) -> list[str]:
+        return list(self._url_to_entry.keys())

src/core/tasks/scheduled/impl/internet_archives/save/models/entry.py

@@ -0,0 +1,15 @@
+from pydantic import BaseModel
+
+from src.db.dtos.url.mapping import URLMapping
+
+
+class InternetArchivesSaveTaskEntry(BaseModel):
+    url: str
+    url_id: int
+    is_new: bool
+
+    def to_url_mapping(self) -> URLMapping:
+        return URLMapping(
+            url_id=self.url_id,
+            url=self.url
+        )

src/core/tasks/scheduled/impl/internet_archives/save/models/mapping.py

@@ -0,0 +1,8 @@
+from pydantic import BaseModel
+
+from src.external.internet_archives.models.save_response import InternetArchivesSaveResponseInfo
+
+
+class URLInternetArchivesSaveResponseMapping(BaseModel):
+    url: str
+    response: InternetArchivesSaveResponseInfo

src/core/tasks/scheduled/impl/internet_archives/save/models/subset.py

@@ -0,0 +1,8 @@
+from pydantic import BaseModel
+
+from src.core.tasks.scheduled.impl.internet_archives.save.models.mapping import URLInternetArchivesSaveResponseMapping
+
+
+class IASaveURLMappingSubsets(BaseModel):
+    error: list[URLInternetArchivesSaveResponseMapping] = []
+    success: list[URLInternetArchivesSaveResponseMapping] = []

src/core/tasks/scheduled/impl/internet_archives/save/operator.py

@@ -0,0 +1,134 @@
+from src.core.tasks.mixins.link_urls import LinkURLsMixin
+from src.core.tasks.mixins.prereq import HasPrerequisitesMixin
+from src.core.tasks.scheduled.impl.internet_archives.save.filter import filter_save_responses
+from src.core.tasks.scheduled.impl.internet_archives.save.mapper import URLToEntryMapper
+from src.core.tasks.scheduled.impl.internet_archives.save.models.entry import InternetArchivesSaveTaskEntry
+from src.core.tasks.scheduled.impl.internet_archives.save.models.mapping import URLInternetArchivesSaveResponseMapping
+from src.core.tasks.scheduled.impl.internet_archives.save.models.subset import IASaveURLMappingSubsets
+from src.core.tasks.scheduled.impl.internet_archives.save.queries.get import \
+    GetURLsForInternetArchivesSaveTaskQueryBuilder
+from src.core.tasks.scheduled.impl.internet_archives.save.queries.prereq import \
+    MeetsPrerequisitesForInternetArchivesSaveQueryBuilder
+from src.core.tasks.scheduled.impl.internet_archives.save.queries.update import \
+    UpdateInternetArchivesSaveMetadataQueryBuilder
+from src.core.tasks.scheduled.templates.operator import ScheduledTaskOperatorBase
+from src.db.client.async_ import AsyncDatabaseClient
+from src.db.enums import TaskType
+from src.db.models.impl.url.error_info.pydantic import URLErrorPydanticInfo
+from src.db.models.impl.url.internet_archives.save.pydantic import URLInternetArchiveSaveMetadataPydantic
+from src.external.internet_archives.client import InternetArchivesClient
+from src.external.internet_archives.models.save_response import InternetArchivesSaveResponseInfo
+
+
+class InternetArchivesSaveTaskOperator(
+    ScheduledTaskOperatorBase,
+    HasPrerequisitesMixin,
+    LinkURLsMixin
+):
+
+    def __init__(
+        self,
+        adb_client: AsyncDatabaseClient,
+        ia_client: InternetArchivesClient
+    ):
+        super().__init__(adb_client)
+        self.ia_client = ia_client
+
+    async def meets_task_prerequisites(self) -> bool:
+        return await self.adb_client.run_query_builder(
+            MeetsPrerequisitesForInternetArchivesSaveQueryBuilder()
+        )
+
+    @property
+    def task_type(self) -> TaskType:
+        return TaskType.IA_SAVE
+
+    async def inner_task_logic(self) -> None:
+        entries: list[InternetArchivesSaveTaskEntry] = await self._get_valid_urls()
+        mapper = URLToEntryMapper(entries)
+        url_ids = [entry.url_id for entry in entries]
+        await self.link_urls_to_task(url_ids=url_ids)
+
+        # Save all to internet archives and get responses
+        resp_mappings: list[URLInternetArchivesSaveResponseMapping] = await self._save_all_to_internet_archives(
+            mapper.get_all_urls()
+        )
+
+        # Separate errors from successful saves
+        subsets: IASaveURLMappingSubsets = filter_save_responses(resp_mappings)
+
+        # Save errors
+        await self._add_errors_to_db(mapper, responses=subsets.error)
+
+        # Save successful saves that are new archive entries
+        await self._save_new_saves_to_db(mapper, ia_mappings=subsets.success)
+
+        # Save successful saves that are existing archive entries
+        await self._save_existing_saves_to_db(mapper, ia_mappings=subsets.success)
+
+
+
+    async def _save_all_to_internet_archives(self, urls: list[str]) -> list[URLInternetArchivesSaveResponseMapping]:
+        resp_mappings: list[URLInternetArchivesSaveResponseMapping] = []
+        for url in urls:
+            resp: InternetArchivesSaveResponseInfo = await self.ia_client.save_to_internet_archives(url)
+            mapping = URLInternetArchivesSaveResponseMapping(
+                url=url,
+                response=resp
+            )
+            resp_mappings.append(mapping)
+        return resp_mappings
+
+    async def _get_valid_urls(self) -> list[InternetArchivesSaveTaskEntry]:
+        return await self.adb_client.run_query_builder(
+            GetURLsForInternetArchivesSaveTaskQueryBuilder()
+        )
+
+    async def _add_errors_to_db(
+        self,
+        mapper: URLToEntryMapper,
+        responses: list[InternetArchivesSaveResponseInfo]
+    ) -> None:
+        error_info_list: list[URLErrorPydanticInfo] = []
+        for response in responses:
+            url_id = mapper.get_url_id(response.url)
+            url_error_info = URLErrorPydanticInfo(
+                url_id=url_id,
+                error=response.error,
+                task_id=self.task_id
+            )
+            error_info_list.append(url_error_info)
+        await self.adb_client.bulk_insert(error_info_list)
+
+    async def _save_new_saves_to_db(
+        self,
+        mapper: URLToEntryMapper,
+        ia_mappings: list[URLInternetArchivesSaveResponseMapping]
+    ) -> None:
+        insert_objects: list[URLInternetArchiveSaveMetadataPydantic] = []
+        for ia_mapping in ia_mappings:
+            is_new = mapper.get_is_new(ia_mapping.url)
+            if not is_new:
+                continue
+            insert_object = URLInternetArchiveSaveMetadataPydantic(
+                url_id=mapper.get_url_id(ia_mapping.url),
+            )
+            insert_objects.append(insert_object)
+        await self.adb_client.bulk_insert(insert_objects)
+
+    async def _save_existing_saves_to_db(
+        self,
+        mapper: URLToEntryMapper,
+        ia_mappings: list[URLInternetArchivesSaveResponseMapping]
+    ) -> None:
+        url_ids: list[int] = []
+        for ia_mapping in ia_mappings:
+            is_new = mapper.get_is_new(ia_mapping.url)
+            if is_new:
+                continue
+            url_ids.append(mapper.get_url_id(ia_mapping.url))
+        await self.adb_client.run_query_builder(
+            UpdateInternetArchivesSaveMetadataQueryBuilder(
+                url_ids=url_ids
+            )
+        )

src/core/tasks/scheduled/impl/internet_archives/save/queries/get.py

@@ -0,0 +1,29 @@
+from typing import Sequence
+
+from sqlalchemy import RowMapping
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.core.tasks.scheduled.impl.internet_archives.save.models.entry import InternetArchivesSaveTaskEntry
+from src.core.tasks.scheduled.impl.internet_archives.save.queries.shared.get_valid_entries import \
+    IA_SAVE_VALID_ENTRIES_QUERY
+from src.db.helpers.session import session_helper as sh
+from src.db.queries.base.builder import QueryBuilderBase
+
+
+class GetURLsForInternetArchivesSaveTaskQueryBuilder(QueryBuilderBase):
+
+    async def run(self, session: AsyncSession) -> list[InternetArchivesSaveTaskEntry]:
+        query = (
+            IA_SAVE_VALID_ENTRIES_QUERY
+            # Limit to 15, which is the maximum number of URLs that can be saved at once.
+            .limit(15)
+        )
+
+        db_mappings: Sequence[RowMapping] = await sh.mappings(session, query=query)
+        return [
+            InternetArchivesSaveTaskEntry(
+                url_id=mapping["id"],
+                url=mapping["url"],
+                is_new=mapping["is_new"],
+            ) for mapping in db_mappings
+        ]