API - Search

changedetectionio/api/Search.py

@@ -0,0 +1,51 @@
+from flask_restful import Resource, abort
+from flask import request
+from . import auth
+
+class Search(Resource):
+    def __init__(self, **kwargs):
+        # datastore is a black box dependency
+        self.datastore = kwargs['datastore']
+
+    @auth.check_token
+    def get(self):
+        """
+        @api {get} /api/v1/search Search for watches
+        @apiDescription Search watches by URL or title text
+        @apiExample {curl} Example usage:
+            curl "http://localhost:5000/api/v1/search?q=https://example.com/page1" -H"x-api-key:813031b16330fe25e3780cf0325daa45"
+            curl "http://localhost:5000/api/v1/search?q=https://example.com/page1?tag=Favourites" -H"x-api-key:813031b16330fe25e3780cf0325daa45"
+            curl "http://localhost:5000/api/v1/search?q=https://example.com?partial=true" -H"x-api-key:813031b16330fe25e3780cf0325daa45"
+        @apiName Search
+        @apiGroup Watch Management
+        @apiQuery {String} q Search query to match against watch URLs and titles
+        @apiQuery {String} [tag] Optional name of tag to limit results (name not UUID)
+        @apiQuery {String} [partial] Allow partial matching of URL query
+        @apiSuccess (200) {Object} JSON Object containing matched watches
+        """
+        query = request.args.get('q', '').strip()
+        tag_limit = request.args.get('tag', '').strip()
+        from changedetectionio.strtobool import strtobool
+        partial = bool(strtobool(request.args.get('partial', '0'))) if 'partial' in request.args else False
+
+        # Require a search query
+        if not query:
+            abort(400, message="Search query 'q' parameter is required")
+
+        # Use the search function from the datastore
+        matching_uuids = self.datastore.search_watches_for_url(query=query, tag_limit=tag_limit, partial=partial)
+
+        # Build the response with watch details
+        results = {}
+        for uuid in matching_uuids:
+            watch = self.datastore.data['watching'].get(uuid)
+            results[uuid] = {
+                'last_changed': watch.last_changed,
+                'last_checked': watch['last_checked'],
+                'last_error': watch['last_error'],
+                'title': watch['title'],
+                'url': watch['url'],
+                'viewed': watch.viewed
+            }
+
+        return results, 200

changedetectionio/flask_app.py

@@ -34,6 +34,7 @@
 from changedetectionio import __version__
 from changedetectionio import queuedWatchMetaData
 from changedetectionio.api import Watch, WatchHistory, WatchSingleHistory, CreateWatch, Import, SystemInfo, Tag, Tags
+from changedetectionio.api.Search import Search
 from .time_handler import is_within_schedule
 
 datastore = None
@@ -275,6 +276,9 @@ def check_authentication():
 
     watch_api.add_resource(Tag, '/api/v1/tag', '/api/v1/tag/<string:uuid>',
                            resource_class_kwargs={'datastore': datastore})
+
+    watch_api.add_resource(Search, '/api/v1/search',
+                           resource_class_kwargs={'datastore': datastore})
 
 
 

changedetectionio/store.py

@@ -631,6 +631,41 @@ def any_watches_have_processor_by_name(self, processor_name):
             if watch.get('processor') == processor_name:
                 return True
         return False
+
+    def search_watches_for_url(self, query, tag_limit=None, partial=False):
+        """Search watches by URL, title, or error messages
+
+        Args:
+            query (str): Search term to match against watch URLs, titles, and error messages
+            tag_limit (str, optional): Optional tag name to limit search results
+            partial: (bool, optional): sub-string matching
+
+        Returns:
+            list: List of UUIDs of watches that match the search criteria
+        """
+        matching_uuids = []
+        query = query.lower().strip()
+        tag = self.tag_exists_by_name(tag_limit) if tag_limit else False
+
+        for uuid, watch in self.data['watching'].items():
+            # Filter by tag if requested
+            if tag_limit:
+                if not tag.get('uuid') in watch.get('tags', []):
+                    continue
+
+            # Search in URL, title, or error messages
+            if partial:
+                if ((watch.get('title') and query in watch.get('title').lower()) or
+                    query in watch.get('url', '').lower() or
+                    (watch.get('last_error') and query in watch.get('last_error').lower())):
+                    matching_uuids.append(uuid)
+            else:
+                if ((watch.get('title') and query == watch.get('title').lower()) or
+                        query == watch.get('url', '').lower() or
+                        (watch.get('last_error') and query == watch.get('last_error').lower())):
+                    matching_uuids.append(uuid)
+
+        return matching_uuids
 
     def get_unique_notification_tokens_available(self):
         # Ask each type of watch if they have any extra notification token to add to the validation

changedetectionio/tests/test_api_search.py

@@ -0,0 +1,101 @@
+from copy import copy
+
+from flask import url_for
+import json
+import time
+from .util import live_server_setup, wait_for_all_checks
+
+
+def test_api_search(client, live_server):
+    live_server_setup(live_server)
+    api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token')
+
+    watch_data = {}
+    # Add some test watches
+    urls = [
+        'https://example.com/page1',
+        'https://example.org/testing',
+        'https://test-site.com/example'
+    ]
+
+    # Import the test URLs
+    res = client.post(
+        url_for("imports.import_page"),
+        data={"urls": "\r\n".join(urls)},
+        follow_redirects=True
+    )
+    assert b"3 Imported" in res.data
+    wait_for_all_checks(client)
+
+    # Get a listing, it will be the first one
+    watches_response = client.get(
+        url_for("createwatch"),
+        headers={'x-api-key': api_key}
+    )
+
+
+    # Add a title to one watch for title search testing
+    for uuid, watch in watches_response.json.items():
+
+        watch_data = client.get(url_for("watch", uuid=uuid),
+                                follow_redirects=True,
+                                headers={'x-api-key': api_key}
+                                )
+
+        if urls[0] == watch_data.json['url']:
+            # HTTP PUT ( UPDATE an existing watch )
+            client.put(
+                url_for("watch", uuid=uuid),
+                headers={'x-api-key': api_key, 'content-type': 'application/json'},
+                data=json.dumps({'title': 'Example Title Test'}),
+            )
+
+    # Test search by URL
+    res = client.get(url_for("search")+"?q=https://example.com/page1", headers={'x-api-key': api_key, 'content-type': 'application/json'})
+    assert len(res.json) == 1
+    assert list(res.json.values())[0]['url'] == urls[0]
+
+    # Test search by URL - partial should NOT match without ?partial=true flag
+    res = client.get(url_for("search")+"?q=https://example", headers={'x-api-key': api_key, 'content-type': 'application/json'})
+    assert len(res.json) == 0
+
+
+    # Test search by title
+    res = client.get(url_for("search")+"?q=Example Title Test", headers={'x-api-key': api_key, 'content-type': 'application/json'})
+    assert len(res.json) == 1
+    assert list(res.json.values())[0]['url'] == urls[0]
+    assert list(res.json.values())[0]['title'] == 'Example Title Test'
+
+    # Test search that should return multiple results (partial = true)
+    res = client.get(url_for("search")+"?q=https://example&partial=true", headers={'x-api-key': api_key, 'content-type': 'application/json'})
+    assert len(res.json) == 2
+
+    # Test empty search
+    res = client.get(url_for("search")+"?q=", headers={'x-api-key': api_key, 'content-type': 'application/json'})
+    assert res.status_code == 400
+
+    # Add a tag to test search with tag filter
+    tag_name = 'test-tag'
+    res = client.post(
+        url_for("tag"),
+        data=json.dumps({"title": tag_name}),
+        headers={'content-type': 'application/json', 'x-api-key': api_key}
+    )
+    assert res.status_code == 201
+    tag_uuid = res.json['uuid']
+
+    # Add the tag to one watch
+    for uuid, watch in watches_response.json.items():
+        if urls[2] == watch['url']:
+            client.put(
+                url_for("watch", uuid=uuid),
+                headers={'x-api-key': api_key, 'content-type': 'application/json'},
+                data=json.dumps({'tags': [tag_uuid]}),
+            )
+
+
+    # Test search with tag filter and q
+    res = client.get(url_for("search") + f"?q={urls[2]}&tag={tag_name}", headers={'x-api-key': api_key, 'content-type': 'application/json'})
+    assert len(res.json) == 1
+    assert list(res.json.values())[0]['url'] == urls[2]
+