kopia lustrzana https://github.com/dgtlmoon/changedetection.io
API - Adding "Search" API (#3052)
dgtlmoon
GitHub
rodzic
64075c87ee
commit
5b97c29714
|
|
@ -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
|
||||
|
|
@ -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/<string:uuid>',
|
||||
resource_class_kwargs={'datastore': datastore})
|
||||
|
||||
watch_api.add_resource(Search, '/api/v1/search',
|
||||
resource_class_kwargs={'datastore': datastore})
|
||||
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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]
|
||||
|
||||
Ładowanie…
Reference in New Issue