mc_502_integrity_monitor

ENV.md

@@ -57,29 +57,30 @@ 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.                                              |
-| `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.                         |
-| `TASK_CLEANUP_TASK_FLAG`               | Cleans up tasks that are no longer needed. |
-| `REFRESH_MATERIALIZED_VIEWS_TASK_FLAG` | Refreshes materialized views. |
-| `UPDATE_URL_STATUS_TASK_FLAG` | Updates the status of URLs. |
-| `DS_APP_SYNC_AGENCY_ADD_TASK_FLAG`     | Adds new agencies to the Data Sources App|
-| `DS_APP_SYNC_AGENCY_UPDATE_TASK_FLAG`       | Updates existing agencies in the Data Sources App|
-| `DS_APP_SYNC_AGENCY_DELETE_TASK_FLAG`       | Deletes agencies in the Data Sources App|
-| `DS_APP_SYNC_DATA_SOURCE_ADD_TASK_FLAG`     | Adds new data sources to the Data Sources App|
-| `DS_APP_SYNC_DATA_SOURCE_UPDATE_TASK_FLAG`  | Updates existing data sources in the Data Sources App|
-| `DS_APP_SYNC_DATA_SOURCE_DELETE_TASK_FLAG`  | Deletes data sources in the Data Sources App|
-| `DS_APP_SYNC_META_URL_ADD_TASK_FLAG`        | Adds new meta URLs to the Data Sources App|
-| `DS_APP_SYNC_META_URL_UPDATE_TASK_FLAG`     | Updates existing meta URLs in the Data Sources App|
-| `DS_APP_SYNC_META_URL_DELETE_TASK_FLAG`     | Deletes meta URLs in the Data Sources App|
+| 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.                         |
+| `TASK_CLEANUP_TASK_FLAG`                   | Cleans up tasks that are no longer needed. |
+| `REFRESH_MATERIALIZED_VIEWS_TASK_FLAG`     | Refreshes materialized views. |
+| `UPDATE_URL_STATUS_TASK_FLAG`              | Updates the status of URLs. |
+| `DS_APP_SYNC_AGENCY_ADD_TASK_FLAG`         | Adds new agencies to the Data Sources App|
+| `DS_APP_SYNC_AGENCY_UPDATE_TASK_FLAG`      | Updates existing agencies in the Data Sources App|
+| `DS_APP_SYNC_AGENCY_DELETE_TASK_FLAG`      | Deletes agencies in the Data Sources App|
+| `DS_APP_SYNC_DATA_SOURCE_ADD_TASK_FLAG`    | Adds new data sources to the Data Sources App|
+| `DS_APP_SYNC_DATA_SOURCE_UPDATE_TASK_FLAG` | Updates existing data sources in the Data Sources App|
+| `DS_APP_SYNC_DATA_SOURCE_DELETE_TASK_FLAG` | Deletes data sources in the Data Sources App|
+| `DS_APP_SYNC_META_URL_ADD_TASK_FLAG`       | Adds new meta URLs to the Data Sources App|
+| `DS_APP_SYNC_META_URL_UPDATE_TASK_FLAG`    | Updates existing meta URLs in the Data Sources App|
+| `DS_APP_SYNC_META_URL_DELETE_TASK_FLAG`    | Deletes meta URLs in the Data Sources App|
+| `INTEGRITY_MONITOR_TASK_FLAG`              | Runs integrity checks. |
 
 ### URL Task Flags
 

alembic/versions/2025_11_23_1923-5ac9d50b91c5_add_integrity_monitor.py

@@ -0,0 +1,97 @@
+"""Add integrity monitor
+
+Revision ID: 5ac9d50b91c5
+Revises: 1bb2dfad3275
+Create Date: 2025-11-23 19:23:45.487445
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+
+from src.util.alembic_helpers import add_enum_value
+
+# revision identifiers, used by Alembic.
+revision: str = '5ac9d50b91c5'
+down_revision: Union[str, None] = '1bb2dfad3275'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+def upgrade() -> None:
+    _create_integrity_task()
+    _create_incomplete_data_sources_view()
+    _create_incomplete_meta_urls_view()
+    _create_url_both_data_source_and_meta_url_view()
+    _create_non_federal_agencies_no_location_view()
+
+def _create_non_federal_agencies_no_location_view():
+    op.execute("""
+    create view integrity__non_federal_agencies_no_location_view as
+        select
+            ag.id as agency_id
+        from agencies ag
+        left join link_agencies__locations link on ag.id = link.agency_id
+        where ag.jurisdiction_type != 'federal'
+        and link.location_id is null
+    """)
+
+
+def _create_url_both_data_source_and_meta_url_view():
+    op.execute("""
+    create view integrity__url_both_data_source_and_meta_url_view as
+    select
+        ds.url_id
+    from
+        ds_app_link_data_source ds
+        join ds_app_link_meta_url mu
+             on mu.url_id = ds.url_id
+    """)
+
+
+def _create_incomplete_meta_urls_view():
+    op.execute("""
+    create view integrity__incomplete_meta_urls_view as
+    select
+        mu.url_id,
+        fuv.url_id is not null as has_validated_flag,
+        fuv.type as validated_type,
+        lau.url_id is not null as has_agency_flag
+    from ds_app_link_meta_url mu
+         left join flag_url_validated fuv on fuv.url_id = mu.url_id
+         left join link_agencies__urls lau on lau.url_id = mu.url_id
+    where
+        fuv.url_id is null
+        or fuv.type != 'meta url'
+        or lau.url_id is null
+    """)
+
+
+def _create_incomplete_data_sources_view():
+    op.execute("""
+    create view integrity__incomplete_data_sources_view as
+        select
+            ds.url_id,
+            fuv.url_id is not null as has_validated_flag,
+            fuv.type as validated_type,
+            urt.url_id is not null as has_record_type,
+            lau.url_id is not null as has_agency_flag
+        from ds_app_link_data_source ds
+        left join flag_url_validated fuv on fuv.url_id = ds.url_id
+        left join url_record_type urt on urt.url_id = ds.url_id
+        left join link_agencies__urls lau on lau.url_id = ds.url_id
+        where
+            fuv.url_id is null
+        or fuv.type != 'data source'
+        or urt.url_id is null
+        or lau.url_id is null
+    """)
+
+
+def _create_integrity_task():
+    add_enum_value(
+        enum_name="task_type",
+        enum_value="Integrity Monitor",
+    )
+
+def downgrade() -> None:
+    pass

src/core/tasks/scheduled/impl/integrity/exceptions.py

@@ -0,0 +1,4 @@
+
+
+class IntegrityMonitorTaskException(Exception):
+    pass

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

@@ -0,0 +1,30 @@
+from src.core.tasks.mixins.prereq import HasPrerequisitesMixin
+from src.core.tasks.scheduled.impl.integrity.exceptions import IntegrityMonitorTaskException
+from src.core.tasks.scheduled.impl.integrity.queries.get import GetIntegrityTaskDataQueryBuilder
+from src.core.tasks.scheduled.impl.integrity.queries.prereq import GetIntegrityTaskPrerequisitesQueryBuilder
+from src.core.tasks.scheduled.templates.operator import ScheduledTaskOperatorBase
+from src.db.enums import TaskType
+
+
+class IntegrityMonitorTaskOperator(
+    ScheduledTaskOperatorBase,
+    HasPrerequisitesMixin
+):
+
+    @property
+    def task_type(self) -> TaskType:
+        return TaskType.INTEGRITY_MONITOR
+
+    async def meets_task_prerequisites(self) -> bool:
+        return await self.run_query_builder(
+            query_builder=GetIntegrityTaskPrerequisitesQueryBuilder()
+        )
+
+    async def inner_task_logic(self) -> None:
+        failing_views: list[str] = await self.run_query_builder(
+            query_builder=GetIntegrityTaskDataQueryBuilder()
+        )
+        raise IntegrityMonitorTaskException(
+            f"Integrity Monitor Task failed for the following views {failing_views}",
+        )
+

src/core/tasks/scheduled/impl/integrity/queries/cte.py

@@ -0,0 +1,61 @@
+from sqlalchemy import select, literal, Exists, Label, or_
+
+from src.db.models.templates_.base import Base
+from src.db.models.views.integrity.incomplete_data_sources import IntegrityIncompleteDataSource
+from src.db.models.views.integrity.incomplete_meta_urls import IntegrityIncompleteMetaURL
+from src.db.models.views.integrity.non_federal_agencies_no_location import IntegrityNonFederalAgenciesNoLocation
+from src.db.models.views.integrity.url_both_data_source_and_meta_url import IntegrityURLBothDataSourceAndMetaURL
+
+
+def any_row_exists(
+    model: type[Base]
+) -> Exists:
+    return (
+        select(
+            literal(1)
+        )
+        .select_from(
+            model
+        )
+        .exists()
+    )
+
+class IntegrityTaskCTEContainer:
+
+    def __init__(
+        self,
+    ):
+        self.models: list[type[Base]] = [
+            IntegrityURLBothDataSourceAndMetaURL,
+            IntegrityNonFederalAgenciesNoLocation,
+            IntegrityIncompleteMetaURL,
+            IntegrityIncompleteDataSource,
+        ]
+
+        expressions: list[Label[bool]] = [
+            any_row_exists(model)
+            .label(model.__tablename__)
+            for model in self.models
+        ]
+
+        self.cte = (
+            select(
+                *expressions
+            )
+            .cte(
+                name="integrity_task_cte",
+            )
+        )
+
+    @property
+    def any_rows_exist_query(self) -> select:
+        expression = [
+            getattr(self.cte.c, model.__tablename__)
+            for model in self.models
+        ]
+        return select(or_(*expression))
+
+    @property
+    def select_all_columns_query(self) -> select:
+        return select(self.cte)
+

src/core/tasks/scheduled/impl/integrity/queries/get.py

@@ -0,0 +1,20 @@
+from sqlalchemy import RowMapping
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.core.tasks.scheduled.impl.integrity.queries.cte import IntegrityTaskCTEContainer
+from src.db.queries.base.builder import QueryBuilderBase
+
+
+class GetIntegrityTaskDataQueryBuilder(QueryBuilderBase):
+
+    async def run(self, session: AsyncSession) -> list[str]:
+        cte = IntegrityTaskCTEContainer()
+        mapping: RowMapping = await self.sh.mapping(
+            session=session,
+            query=cte.select_all_columns_query
+        )
+        return [
+            model.__tablename__
+            for model in cte.models
+            if mapping[model.__tablename__]
+        ]

src/core/tasks/scheduled/impl/integrity/queries/prereq.py

@@ -0,0 +1,16 @@
+from typing import Any
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.core.tasks.scheduled.impl.integrity.queries.cte import IntegrityTaskCTEContainer
+from src.db.queries.base.builder import QueryBuilderBase
+
+
+class GetIntegrityTaskPrerequisitesQueryBuilder(QueryBuilderBase):
+
+    async def run(self, session: AsyncSession) -> Any:
+        cte = IntegrityTaskCTEContainer()
+        return await self.sh.scalar(
+            session=session,
+            query=cte.any_rows_exist_query
+        )

src/core/tasks/scheduled/loader.py

@@ -6,6 +6,7 @@
 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.integrity.operator import IntegrityMonitorTaskOperator
 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
@@ -127,6 +128,13 @@ async def load_entries(self) -> list[ScheduledTaskEntry]:
                 interval_minutes=IntervalEnum.DAILY.value,
                 enabled=self.setup_flag("REFRESH_MATERIALIZED_VIEWS_TASK_FLAG")
             ),
+            ScheduledTaskEntry(
+                operator=IntegrityMonitorTaskOperator(
+                    adb_client=self.adb_client
+                ),
+                interval_minutes=IntervalEnum.DAILY.value,
+                enabled=self.setup_flag("INTEGRITY_MONITOR_TASK_FLAG")
+            ),
             # Sync
             ## Adds
             ### Agency

src/db/enums.py

@@ -63,6 +63,7 @@ class TaskType(PyEnum):
     TASK_CLEANUP = "Task Cleanup"
     REFRESH_MATERIALIZED_VIEWS = "Refresh Materialized Views"
     UPDATE_URL_STATUS = "Update URL Status"
+    INTEGRITY_MONITOR = "Integrity Monitor"
 
     # Sync Tasks
     SYNC_AGENCIES_ADD = "Sync Agencies Add"

src/db/models/helpers.py

@@ -46,9 +46,12 @@ def location_id_column() -> Column[int]:
 
 CURRENT_TIME_SERVER_DEFAULT = func.now()
 
+VIEW_ARG = {"info": "view"}
+
 def url_id_primary_key_constraint() -> PrimaryKeyConstraint:
     return PrimaryKeyConstraint('url_id')
 
+
 def county_column(nullable: bool = False) -> Column[int]:
     return Column(
         Integer(),

src/db/models/impl/flag/url_validated/sqlalchemy.py

@@ -1,4 +1,5 @@
 from sqlalchemy import PrimaryKeyConstraint
+from sqlalchemy.orm import Mapped
 
 from src.db.models.helpers import enum_column
 from src.db.models.impl.flag.url_validated.enums import URLType
@@ -19,7 +20,7 @@ class FlagURLValidated(
         ),
     )
 
-    type = enum_column(
+    type: Mapped[URLType] = enum_column(
         enum_type=URLType,
         name="url_type",
     )

src/db/models/mixins.py

@@ -3,7 +3,8 @@
 from sqlalchemy import Column, Integer, ForeignKey, TIMESTAMP, event
 
 from src.db.models.exceptions import WriteToViewError
-from src.db.models.helpers import get_created_at_column, CURRENT_TIME_SERVER_DEFAULT
+from src.db.models.helpers import get_created_at_column, CURRENT_TIME_SERVER_DEFAULT, url_id_primary_key_constraint, \
+    VIEW_ARG
 
 
 class URLDependentMixin:
@@ -90,3 +91,9 @@
     @staticmethod
     def _block_write(mapper, connection, target):
         raise WriteToViewError(f"{type(target).__name__} is a read-only view.")
+
+class URLDependentViewMixin(URLDependentMixin, ViewMixin):
+    __table_args__ = (
+        url_id_primary_key_constraint(),
+        VIEW_ARG
+    )

src/db/models/views/dependent_locations.py

@@ -31,6 +31,7 @@
 """
 from sqlalchemy import Column, Integer, ForeignKey
 
+from src.db.models.helpers import VIEW_ARG
 from src.db.models.mixins import ViewMixin
 from src.db.models.templates_.base import Base
 
@@ -39,7 +40,7 @@ class DependentLocationView(Base, ViewMixin):
 
     __tablename__ = "dependent_locations"
     __table_args__ = (
-        {"info": "view"}
+        VIEW_ARG,
     )
 
     parent_location_id = Column(