diff --git a/changedetectionio/api/Search.py b/changedetectionio/api/Search.py new file mode 100644 index 00000000..c403e043 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/changedetectionio/flask_app.py b/changedetectionio/flask_app.py index 229ad983..080515e6 100644 --- a/changedetectionio/flask_app.py +++ b/changedetectionio/flask_app.py @@ -34,6 +34,7 @@ from loguru import logger 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 changedetection_app(config=None, datastore_o=None): watch_api.add_resource(Tag, '/api/v1/tag', '/api/v1/tag/', resource_class_kwargs={'datastore': datastore}) + + watch_api.add_resource(Search, '/api/v1/search', + resource_class_kwargs={'datastore': datastore}) diff --git a/changedetectionio/store.py b/changedetectionio/store.py index 0b319cdf..97b4c75c 100644 --- a/changedetectionio/store.py +++ b/changedetectionio/store.py @@ -631,6 +631,41 @@ class ChangeDetectionStore: 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 diff --git a/changedetectionio/tests/test_api_search.py b/changedetectionio/tests/test_api_search.py new file mode 100644 index 00000000..3369905e --- /dev/null +++ b/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] +