Add url, urlPrefix, ts filters and more sort options to crawl pages endpoint

Author: tw4l

backend/btrixcloud/pages.py

@@ -501,6 +501,9 @@ async def list_pages(
         self,
         crawl_id: str,
         org: Optional[Organization] = None,
+        url: Optional[str] = None,
+        url_prefix: Optional[str] = None,
+        ts: Optional[datetime] = None,
         qa_run_id: Optional[str] = None,
         qa_filter_by: Optional[str] = None,
         qa_gte: Optional[float] = None,
@@ -527,6 +530,17 @@ async def list_pages(
         if org:
             query["oid"] = org.id
 
+        if url_prefix:
+            url_prefix = urllib.parse.unquote(url_prefix)
+            regex_pattern = f"^{re.escape(url_prefix)}"
+            query["url"] = {"$regex": regex_pattern, "$options": "i"}
+
+        elif url:
+            query["url"] = urllib.parse.unquote(url)
+
+        if ts:
+            query["ts"] = ts
+
         if reviewed:
             query["$or"] = [
                 {"approved": {"$ne": None}},
@@ -571,7 +585,16 @@ async def list_pages(
             # Sorting options to add:
             # - automated heuristics like screenshot_comparison (dict keyed by QA run id)
             # - Ensure notes sorting works okay with notes in list
-            sort_fields = ("url", "title", "notes", "approved")
+            sort_fields = (
+                "url",
+                "title",
+                "notes",
+                "approved",
+                "ts",
+                "status",
+                "mime",
+                "filename",
+            )
             qa_sort_fields = ("screenshotMatch", "textMatch")
             if sort_by not in sort_fields and sort_by not in qa_sort_fields:
                 raise HTTPException(status_code=400, detail="invalid_sort_by")
@@ -1004,6 +1027,9 @@ async def delete_page_notes(
     async def get_crawl_pages_list(
         crawl_id: str,
         org: Organization = Depends(org_crawl_dep),
+        url: Optional[str] = None,
+        urlPrefix: Optional[str] = None,
+        ts: Optional[datetime] = None,
         reviewed: Optional[bool] = None,
         approved: Optional[str] = None,
         hasNotes: Optional[bool] = None,
@@ -1020,6 +1046,9 @@ async def get_crawl_pages_list(
         pages, total = await ops.list_pages(
             crawl_id=crawl_id,
             org=org,
+            url=url,
+            url_prefix=urlPrefix,
+            ts=ts,
             reviewed=reviewed,
             approved=formatted_approved,
             has_notes=hasNotes,

backend/test/test_run_crawl.py

@@ -682,7 +682,11 @@ def test_crawl_pages(crawler_auth_headers, default_org_id, crawler_crawl_id):
 
     # Test GET page endpoint
     global page_id
-    page_id = pages[0]["id"]
+    test_page = pages[0]
+    page_id = test_page["id"]
+    test_page_url = test_page["url"]
+    test_page_ts = test_page["ts"]
+
     r = requests.get(
         f"{API_PREFIX}/orgs/{default_org_id}/crawls/{crawler_crawl_id}/pages/{page_id}",
         headers=crawler_auth_headers,
@@ -710,6 +714,51 @@ def test_crawl_pages(crawler_auth_headers, default_org_id, crawler_crawl_id):
     assert page.get("modified") is None
     assert page.get("approved") is None
 
+    # Test exact url filter
+    r = requests.get(
+        f"{API_PREFIX}/orgs/{default_org_id}/crawls/{crawler_crawl_id}/pages?url={test_page_url}",
+        headers=crawler_auth_headers,
+    )
+    assert r.status_code == 200
+    data = r.json()
+
+    assert data["total"] >= 1
+    for matching_page in data["items"]:
+        assert matching_page["url"] == test_page_url
+
+    # Test exact url and ts filters together
+    r = requests.get(
+        f"{API_PREFIX}/orgs/{default_org_id}/crawls/{crawler_crawl_id}/pages?url={test_page_url}&ts={test_page_ts}",
+        headers=crawler_auth_headers,
+    )
+    assert r.status_code == 200
+    data = r.json()
+
+    assert data["total"] >= 1
+    for matching_page in data["items"]:
+        assert matching_page["url"] == test_page_url
+        assert matching_page["ts"] == test_page_ts
+
+    # Test urlPrefix filter
+    url_prefix = test_page_url[:8]
+    r = requests.get(
+        f"{API_PREFIX}/orgs/{default_org_id}/crawls/{crawler_crawl_id}/pages?urlPrefix={url_prefix}",
+        headers=crawler_auth_headers,
+    )
+    assert r.status_code == 200
+    data = r.json()
+
+    assert data["total"] >= 1
+
+    found_matching_page = False
+    for page in data["items"]:
+        if page["id"] == page_id and page["url"] == test_page_url:
+            found_matching_page = True
+
+    assert found_matching_page
+
+
+def test_crawl_pages_qa_filters(crawler_auth_headers, default_org_id, crawler_crawl_id):
     # Test reviewed filter (page has no notes or approved so should show up in false)
     r = requests.get(
         f"{API_PREFIX}/orgs/{default_org_id}/crawls/{crawler_crawl_id}/pages?reviewed=False",