mc_submit_meta_urls

ENV.md

@@ -57,15 +57,17 @@ Note that some tasks/subtasks are themselves enabled by other tasks.
 
 ### Scheduled Task Flags
 
-| Flag                                | Description                                                        |
-|-------------------------------------|--------------------------------------------------------------------|
-| `SCHEDULED_TASKS_FLAG`              | All scheduled tasks. Disabling disables all other scheduled tasks. |
-| `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. |
-| `IA_SAVE_TASK_FLAG`             | Saves URLs to Internet Archives.                       |
+| Flag                                | Description                                                                   |
+|-------------------------------------|-------------------------------------------------------------------------------|
+| `SCHEDULED_TASKS_FLAG`              | All scheduled tasks. Disabling disables all other scheduled tasks.            |
+| `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.                        |
+| `IA_SAVE_TASK_FLAG`             | Saves URLs to Internet Archives.                                              |
+| `MARK_TASK_NEVER_COMPLETED_TASK_FLAG` | Marks tasks that were started but never completed (usually due to a restart). |
+| `DELETE_STALE_SCREENSHOTS_TASK_FLAG` | Deletes stale screenshots for URLs already validated.                         |
 
 ### URL Task Flags
 
@@ -87,6 +89,7 @@ URL Task Flags are collectively controlled by the `RUN_URL_TASKS_TASK_FLAG` flag
 | `URL_AUTO_VALIDATE_TASK_FLAG`       | Automatically validates URLs.                         |
 | `URL_AUTO_NAME_TASK_FLAG`           | Automatically names URLs.                             |
 | `URL_SUSPEND_TASK_FLAG`             | Suspends URLs meeting suspension criteria.            |
+| `URL_SUBMIT_META_URLS_TASK_FLAG`    | Submits meta URLs to the Data Sources App.            |
 
 ### Agency ID Subtasks
 

alembic/versions/2025_09_30_1046-84a3de626ad8_add_link_user_submitted_url_table.py

@@ -7,8 +7,8 @@
 """
 from typing import Sequence, Union
 
-from alembic import op
 import sqlalchemy as sa
+from alembic import op
 
 from src.util.alembic_helpers import url_id_column, user_id_column, created_at_column
 

alembic/versions/2025_09_30_1613-241fd3925f5d_add_logic_for_meta_url_submissions.py

@@ -0,0 +1,63 @@
+"""Add logic for meta URL submissions
+
+Revision ID: 241fd3925f5d
+Revises: 84a3de626ad8
+Create Date: 2025-09-30 16:13:03.980113
+
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+from src.util.alembic_helpers import url_id_column, created_at_column, agency_id_column
+
+# revision identifiers, used by Alembic.
+revision: str = '241fd3925f5d'
+down_revision: Union[str, None] = '84a3de626ad8'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.execute("""ALTER TYPE task_type ADD VALUE 'Submit Meta URLs'""")
+    op.create_table(
+        "url_ds_meta_url",
+        url_id_column(),
+        agency_id_column(),
+        sa.Column("ds_meta_url_id", sa.Integer(), nullable=False),
+        created_at_column(),
+        sa.PrimaryKeyConstraint(
+            "url_id",
+            "agency_id"
+        ),
+        sa.UniqueConstraint(
+            "ds_meta_url_id"
+        )
+    )
+    op.execute("""ALTER TYPE task_type ADD VALUE 'Delete Stale Screenshots'""")
+    op.execute("""ALTER TYPE task_type ADD VALUE 'Mark Task Never Completed'""")
+    op.execute("""
+        CREATE TYPE task_status_enum as ENUM(
+            'complete',
+            'in-process',
+            'error',
+            'aborted',
+            'never-completed'
+        )
+    """)
+    op.execute("""
+            ALTER TABLE tasks
+              ALTER COLUMN task_status DROP DEFAULT,
+              ALTER COLUMN task_status TYPE task_status_enum
+              USING (
+                CASE task_status::text                               -- old enum -> text
+                  WHEN 'ready to label'  THEN 'complete'::task_status_enum
+                  ELSE task_status::text::task_status_enum
+                END
+              );
+    """)
+
+
+def downgrade() -> None:
+    pass

src/api/endpoints/task/by_id/dto.py

@@ -3,6 +3,7 @@
 
 from pydantic import BaseModel
 
+from src.db.models.impl.task.enums import TaskStatus
 from src.db.models.impl.url.core.pydantic.info import URLInfo
 from src.db.models.impl.url.error_info.pydantic import URLErrorInfoPydantic
 from src.db.enums import TaskType
@@ -11,7 +12,7 @@
 
 class TaskInfo(BaseModel):
     task_type: TaskType
-    task_status: BatchStatus
+    task_status: TaskStatus
     updated_at: datetime.datetime
     error_info: str | None = None
     urls: list[URLInfo]

src/api/endpoints/task/by_id/query.py

@@ -5,6 +5,7 @@
 from src.api.endpoints.task.by_id.dto import TaskInfo
 from src.collectors.enums import URLStatus
 from src.core.enums import BatchStatus
+from src.db.models.impl.task.enums import TaskStatus
 from src.db.models.impl.url.core.pydantic.info import URLInfo
 from src.db.models.impl.url.error_info.pydantic import URLErrorInfoPydantic
 from src.db.enums import TaskType
@@ -59,7 +60,7 @@ async def run(self, session: AsyncSession) -> TaskInfo:
             errored_urls.append(url_error_info)
         return TaskInfo(
             task_type=TaskType(task.task_type),
-            task_status=BatchStatus(task.task_status),
+            task_status=TaskStatus(task.task_status),
             error_info=error,
             updated_at=task.updated_at,
             urls=url_infos,

src/core/tasks/base/operator.py

@@ -6,6 +6,7 @@
 from src.core.tasks.url.enums import TaskOperatorOutcome
 from src.db.client.async_ import AsyncDatabaseClient
 from src.db.enums import TaskType
+from src.db.models.impl.task.enums import TaskStatus
 
 
 class TaskOperatorBase(ABC):
@@ -60,7 +61,7 @@ async def inner_task_logic(self) -> None:
         raise NotImplementedError
 
     async def handle_task_error(self, e):
-        await self.adb_client.update_task_status(task_id=self.task_id, status=BatchStatus.ERROR)
+        await self.adb_client.update_task_status(task_id=self.task_id, status=TaskStatus.ERROR)
         await self.adb_client.add_task_error(
             task_id=self.task_id,
             error=str(e)

src/core/tasks/handler.py

@@ -7,6 +7,7 @@
 from src.core.tasks.url.enums import TaskOperatorOutcome
 from src.db.client.async_ import AsyncDatabaseClient
 from src.db.enums import TaskType
+from src.db.models.impl.task.enums import TaskStatus
 
 
 class TaskHandler:
@@ -42,13 +43,14 @@ async def handle_outcome(self, run_info: TaskOperatorRunInfo):  #
             case TaskOperatorOutcome.SUCCESS:
                 await self.adb_client.update_task_status(
                     task_id=run_info.task_id,
-                    status=BatchStatus.READY_TO_LABEL
+                    status=TaskStatus.COMPLETE
                 )
 
     async def handle_task_error(self, run_info: TaskOperatorRunInfo):  #
         await self.adb_client.update_task_status(
             task_id=run_info.task_id,
-            status=BatchStatus.ERROR)
+            status=TaskStatus.ERROR
+        )
         await self.adb_client.add_task_error(
             task_id=run_info.task_id,
             error=run_info.message

src/core/tasks/scheduled/impl/delete_logs/operator.py

@@ -1,16 +1,21 @@
+import datetime
+
+from sqlalchemy import delete
+
 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.log.sqlalchemy import Log
 
 
 class DeleteOldLogsTaskOperator(ScheduledTaskOperatorBase):
 
-    def __init__(self, adb_client: AsyncDatabaseClient):
-        super().__init__(adb_client)
-
     @property
     def task_type(self) -> TaskType:
         return TaskType.DELETE_OLD_LOGS
 
     async def inner_task_logic(self) -> None:
-        await self.adb_client.delete_old_logs()
+        statement = delete(Log).where(
+            Log.created_at < datetime.datetime.now() - datetime.timedelta(days=7)
+        )
+        await self.adb_client.execute(statement)

src/core/tasks/scheduled/impl/delete_stale_screenshots/operator.py

@@ -0,0 +1,15 @@
+from src.core.tasks.scheduled.impl.delete_stale_screenshots.query import DeleteStaleScreenshotsQueryBuilder
+from src.core.tasks.scheduled.templates.operator import ScheduledTaskOperatorBase
+from src.db.enums import TaskType
+
+
+class DeleteStaleScreenshotsTaskOperator(ScheduledTaskOperatorBase):
+
+    @property
+    def task_type(self) -> TaskType:
+        return TaskType.DELETE_STALE_SCREENSHOTS
+
+    async def inner_task_logic(self) -> None:
+        await self.adb_client.run_query_builder(
+            DeleteStaleScreenshotsQueryBuilder()
+        )

src/core/tasks/scheduled/impl/delete_stale_screenshots/query.py

@@ -0,0 +1,29 @@
+from typing import Any
+
+from sqlalchemy import delete, exists, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.db.models.impl.flag.url_validated.sqlalchemy import FlagURLValidated
+from src.db.models.impl.url.screenshot.sqlalchemy import URLScreenshot
+from src.db.queries.base.builder import QueryBuilderBase
+
+
+class DeleteStaleScreenshotsQueryBuilder(QueryBuilderBase):
+
+    async def run(self, session: AsyncSession) -> Any:
+
+        statement = (
+            delete(
+                URLScreenshot
+            )
+            .where(
+                exists(
+                    select(
+                        FlagURLValidated,
+                        FlagURLValidated.url_id == URLScreenshot.url_id,
+                    )
+                )
+            )
+        )
+
+        await session.execute(statement)

src/core/tasks/scheduled/impl/mark_never_completed/operator.py

@@ -0,0 +1,15 @@
+from src.core.tasks.scheduled.impl.mark_never_completed.query import MarkTaskNeverCompletedQueryBuilder
+from src.core.tasks.scheduled.templates.operator import ScheduledTaskOperatorBase
+from src.db.enums import TaskType
+
+
+class MarkTaskNeverCompletedOperator(ScheduledTaskOperatorBase):
+
+    @property
+    def task_type(self) -> TaskType:
+        return TaskType.MARK_TASK_NEVER_COMPLETED
+
+    async def inner_task_logic(self) -> None:
+        await self.adb_client.run_query_builder(
+            MarkTaskNeverCompletedQueryBuilder()
+        )

src/core/tasks/scheduled/impl/mark_never_completed/query.py

@@ -0,0 +1,27 @@
+from datetime import timedelta, datetime
+from typing import Any
+
+from sqlalchemy import update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.core.enums import BatchStatus
+from src.db.enums import TaskType
+from src.db.models.impl.task.core import Task
+from src.db.queries.base.builder import QueryBuilderBase
+
+
+class MarkTaskNeverCompletedQueryBuilder(QueryBuilderBase):
+
+    async def run(self, session: AsyncSession) -> Any:
+        statement = (
+            update(
+                Task
+            ).values(
+                task_status=BatchStatus.ABORTED.value
+            ).
+            where(
+                Task.task_status == BatchStatus.IN_PROCESS,
+                Task.updated_at < datetime.now() - timedelta(hours=1)
+            )
+        )
+        await session.execute(statement)

src/core/tasks/scheduled/loader.py

@@ -4,9 +4,12 @@
 from src.core.tasks.scheduled.enums import IntervalEnum
 from src.core.tasks.scheduled.impl.backlog.operator import PopulateBacklogSnapshotTaskOperator
 from src.core.tasks.scheduled.impl.delete_logs.operator import DeleteOldLogsTaskOperator
+from src.core.tasks.scheduled.impl.delete_stale_screenshots.operator import DeleteStaleScreenshotsTaskOperator
 from src.core.tasks.scheduled.impl.huggingface.operator import PushToHuggingFaceTaskOperator
 from src.core.tasks.scheduled.impl.internet_archives.probe.operator import InternetArchivesProbeTaskOperator
 from src.core.tasks.scheduled.impl.internet_archives.save.operator import InternetArchivesSaveTaskOperator
+from src.core.tasks.scheduled.impl.mark_never_completed.operator import MarkTaskNeverCompletedOperator
+from src.core.tasks.scheduled.impl.mark_never_completed.query import MarkTaskNeverCompletedQueryBuilder
 from src.core.tasks.scheduled.impl.run_url_tasks.operator import RunURLTasksTaskOperator
 from src.core.tasks.scheduled.models.entry import ScheduledTaskEntry
 from src.db.client.async_ import AsyncDatabaseClient
@@ -37,6 +40,9 @@
         self.env = Env()
         self.env.read_env()
 
+    def setup_flag(self, name: str) -> bool:
+        return self.env.bool(name, default=True)
+
 
     async def load_entries(self) -> list[ScheduledTaskEntry]:
         scheduled_task_flag = self.env.bool("SCHEDULED_TASKS_FLAG", default=True)
@@ -52,45 +58,50 @@
                     ia_client=self.ia_client
                 ),
                 interval_minutes=IntervalEnum.TEN_MINUTES.value,
-                enabled=self.env.bool("IA_PROBE_TASK_FLAG", default=True),
+                enabled=self.setup_flag("IA_PROBE_TASK_FLAG"),
             ),
             ScheduledTaskEntry(
                 operator=InternetArchivesSaveTaskOperator(
                     adb_client=self.adb_client,
                     ia_client=self.ia_client
                 ),
                 interval_minutes=IntervalEnum.TEN_MINUTES.value,
-                enabled=self.env.bool("IA_SAVE_TASK_FLAG", default=True),
+                enabled=self.setup_flag("IA_SAVE_TASK_FLAG"),
             ),
             ScheduledTaskEntry(
                 operator=DeleteOldLogsTaskOperator(adb_client=self.adb_client),
                 interval_minutes=IntervalEnum.DAILY.value,
-                enabled=self.env.bool("DELETE_OLD_LOGS_TASK_FLAG", default=True)
+                enabled=self.setup_flag("DELETE_OLD_LOGS_TASK_FLAG")
             ),
             ScheduledTaskEntry(
                 operator=RunURLTasksTaskOperator(async_core=self.async_core),
                 interval_minutes=self.env.int(
                     "URL_TASKS_FREQUENCY_MINUTES",
                     default=IntervalEnum.HOURLY.value
                 ),
-                enabled=self.env.bool("RUN_URL_TASKS_TASK_FLAG", default=True)
-
+                enabled=self.setup_flag("RUN_URL_TASKS_TASK_FLAG")
             ),
             ScheduledTaskEntry(
                 operator=PopulateBacklogSnapshotTaskOperator(adb_client=self.async_core.adb_client),
                 interval_minutes=IntervalEnum.DAILY.value,
-                enabled=self.env.bool("POPULATE_BACKLOG_SNAPSHOT_TASK_FLAG", default=True)
+                enabled=self.setup_flag("POPULATE_BACKLOG_SNAPSHOT_TASK_FLAG")
             ),
             ScheduledTaskEntry(
                 operator=PushToHuggingFaceTaskOperator(
                     adb_client=self.async_core.adb_client,
                     hf_client=self.hf_client
                 ),
                 interval_minutes=IntervalEnum.DAILY.value,
-                enabled=self.env.bool(
-                    "PUSH_TO_HUGGING_FACE_TASK_FLAG",
-                    default=True
-                )
+                enabled=self.setup_flag("PUSH_TO_HUGGING_FACE_TASK_FLAG")
+            ),
+            ScheduledTaskEntry(
+                operator=MarkTaskNeverCompletedOperator(adb_client=self.adb_client),
+                interval_minutes=IntervalEnum.DAILY.value,
+                enabled=self.setup_flag("MARK_TASK_NEVER_COMPLETED_TASK_FLAG")
+            ),
+            ScheduledTaskEntry(
+                operator=DeleteStaleScreenshotsTaskOperator(adb_client=self.adb_client),
+                interval_minutes=IntervalEnum.DAILY.value,
+                enabled=self.setup_flag("DELETE_STALE_SCREENSHOTS_TASK_FLAG")
             )
-
         ]