Compare commits

..

3 Commits

Author SHA1 Message Date
dgtlmoon
78bc6ae0d3 Small useless improvements 2025-04-11 12:23:30 +02:00
dgtlmoon
c07ab75837 Make playwright work the same way 2025-04-11 12:04:42 +02:00
dgtlmoon
0c7689fbd5 Use bigger chunks to speed up page screenshot 2025-04-11 11:59:36 +02:00
10 changed files with 5 additions and 314 deletions

View File

@@ -2,7 +2,7 @@
# Read more https://github.com/dgtlmoon/changedetection.io/wiki
__version__ = '0.49.13'
__version__ = '0.49.12'
from changedetectionio.strtobool import strtobool
from json.decoder import JSONDecodeError

View File

@@ -1,145 +0,0 @@
from flask_expects_json import expects_json
from flask_restful import Resource
from . import auth
from flask_restful import abort, Resource
from flask import request
from . import auth
from . import schema_create_notification_urls, schema_delete_notification_urls
class Notifications(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/notifications Return Notification URL List
@apiDescription Return the Notification URL List from the configuration
@apiExample {curl} Example usage:
curl http://localhost:5000/api/v1/notifications -H"x-api-key:813031b16330fe25e3780cf0325daa45"
HTTP/1.0 200
{
'notification_urls': ["notification-urls-list"]
}
@apiName Get
@apiGroup Notifications
"""
notification_urls = self.datastore.data.get('settings', {}).get('application', {}).get('notification_urls', [])
return {
'notification_urls': notification_urls,
}, 200
@auth.check_token
@expects_json(schema_create_notification_urls)
def post(self):
"""
@api {post} /api/v1/notifications Create Notification URLs
@apiDescription Add one or more notification URLs from the configuration
@apiExample {curl} Example usage:
curl http://localhost:5000/api/v1/notifications/batch -H"x-api-key:813031b16330fe25e3780cf0325daa45" -H "Content-Type: application/json" -d '{"notification_urls": ["url1", "url2"]}'
@apiName CreateBatch
@apiGroup Notifications
@apiSuccess (201) {Object[]} notification_urls List of added notification URLs
@apiError (400) {String} Invalid input
"""
json_data = request.get_json()
notification_urls = json_data.get("notification_urls", [])
from wtforms import ValidationError
try:
validate_notification_urls(notification_urls)
except ValidationError as e:
return str(e), 400
added_urls = []
for url in notification_urls:
clean_url = url.strip()
added_url = self.datastore.add_notification_url(clean_url)
if added_url:
added_urls.append(added_url)
if not added_urls:
return "No valid notification URLs were added", 400
return {'notification_urls': added_urls}, 201
@auth.check_token
@expects_json(schema_create_notification_urls)
def put(self):
"""
@api {put} /api/v1/notifications Replace Notification URLs
@apiDescription Replace all notification URLs with the provided list (can be empty)
@apiExample {curl} Example usage:
curl -X PUT http://localhost:5000/api/v1/notifications -H"x-api-key:813031b16330fe25e3780cf0325daa45" -H "Content-Type: application/json" -d '{"notification_urls": ["url1", "url2"]}'
@apiName Replace
@apiGroup Notifications
@apiSuccess (200) {Object[]} notification_urls List of current notification URLs
@apiError (400) {String} Invalid input
"""
json_data = request.get_json()
notification_urls = json_data.get("notification_urls", [])
from wtforms import ValidationError
try:
validate_notification_urls(notification_urls)
except ValidationError as e:
return str(e), 400
if not isinstance(notification_urls, list):
return "Invalid input format", 400
clean_urls = [url.strip() for url in notification_urls if isinstance(url, str)]
self.datastore.data['settings']['application']['notification_urls'] = clean_urls
self.datastore.needs_write = True
return {'notification_urls': clean_urls}, 200
@auth.check_token
@expects_json(schema_delete_notification_urls)
def delete(self):
"""
@api {delete} /api/v1/notifications Delete Notification URLs
@apiDescription Deletes one or more notification URLs from the configuration
@apiExample {curl} Example usage:
curl http://localhost:5000/api/v1/notifications -X DELETE -H"x-api-key:813031b16330fe25e3780cf0325daa45" -H "Content-Type: application/json" -d '{"notification_urls": ["url1", "url2"]}'
@apiParam {String[]} notification_urls The notification URLs to delete.
@apiName Delete
@apiGroup Notifications
@apiSuccess (204) {String} OK Deleted
@apiError (400) {String} No matching notification URLs found.
"""
json_data = request.get_json()
urls_to_delete = json_data.get("notification_urls", [])
if not isinstance(urls_to_delete, list):
abort(400, message="Expected a list of notification URLs.")
notification_urls = self.datastore.data['settings']['application'].get('notification_urls', [])
deleted = []
for url in urls_to_delete:
clean_url = url.strip()
if clean_url in notification_urls:
notification_urls.remove(clean_url)
deleted.append(clean_url)
if not deleted:
abort(400, message="No matching notification URLs found.")
self.datastore.data['settings']['application']['notification_urls'] = notification_urls
self.datastore.needs_write = True
return 'OK', 204
def validate_notification_urls(notification_urls):
from changedetectionio.forms import ValidateAppRiseServers
validator = ValidateAppRiseServers()
class DummyForm: pass
dummy_form = DummyForm()
field = type("Field", (object,), {"data": notification_urls, "gettext": lambda self, x: x})()
validator(dummy_form, field)

View File

@@ -19,15 +19,8 @@ schema_create_tag['required'] = ['title']
schema_update_tag = copy.deepcopy(schema_tag)
schema_update_tag['additionalProperties'] = False
schema_notification_urls = copy.deepcopy(schema)
schema_create_notification_urls = copy.deepcopy(schema_notification_urls)
schema_create_notification_urls['required'] = ['notification_urls']
schema_delete_notification_urls = copy.deepcopy(schema_notification_urls)
schema_delete_notification_urls['required'] = ['notification_urls']
# Import all API resources
from .Watch import Watch, WatchHistory, WatchSingleHistory, CreateWatch
from .Tags import Tags, Tag
from .Import import Import
from .SystemInfo import SystemInfo
from .Notifications import Notifications

View File

@@ -19,20 +19,6 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, queuedWatchMe
if tag_uuid in watch.get('tags', []) and (tag.get('include_filters') or tag.get('subtractive_selectors')):
return True
def levenshtein_ratio_recent_history(watch):
try:
from Levenshtein import ratio, distance
k = list(watch.history.keys())
if len(k) >= 2:
a = watch.get_history_snapshot(timestamp=k[0])
b = watch.get_history_snapshot(timestamp=k[1])
distance = distance(a, b)
return distance
except Exception as e:
logger.warning("Unable to calc similarity", e)
return "Unable to calc similarity"
return ''
@edit_blueprint.route("/edit/<string:uuid>", methods=['GET', 'POST'])
@login_optionally_required
# https://stackoverflow.com/questions/42984453/wtforms-populate-form-with-data-if-data-exists
@@ -261,15 +247,14 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, queuedWatchMe
'has_default_notification_urls': True if len(datastore.data['settings']['application']['notification_urls']) else False,
'has_extra_headers_file': len(datastore.get_all_headers_in_textfile_for_watch(uuid=uuid)) > 0,
'has_special_tag_options': _watch_has_tag_options_set(watch=watch),
'watch_uses_webdriver': watch_uses_webdriver,
'jq_support': jq_support,
'lev_info': levenshtein_ratio_recent_history(watch),
'playwright_enabled': os.getenv('PLAYWRIGHT_DRIVER_URL', False),
'settings_application': datastore.data['settings']['application'],
'timezone_default_config': datastore.data['settings']['application'].get('timezone'),
'using_global_webdriver_wait': not default['webdriver_delay'],
'uuid': uuid,
'watch': watch,
'watch_uses_webdriver': watch_uses_webdriver,
'watch': watch
}
included_content = None

View File

@@ -33,7 +33,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, Notifications
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
@@ -285,8 +285,7 @@ def changedetection_app(config=None, datastore_o=None):
watch_api.add_resource(Search, '/api/v1/search',
resource_class_kwargs={'datastore': datastore})
watch_api.add_resource(Notifications, '/api/v1/notifications',
resource_class_kwargs={'datastore': datastore})
@login_manager.user_loader
def user_loader(email):

View File

@@ -964,25 +964,3 @@ class ChangeDetectionStore:
f_d.write(zlib.compress(f_j.read()))
os.unlink(json_path)
def add_notification_url(self, notification_url):
logger.debug(f">>> Adding new notification_url - '{notification_url}'")
notification_urls = self.data['settings']['application'].get('notification_urls', [])
if notification_url in notification_urls:
return notification_url
with self.lock:
notification_urls = self.__data['settings']['application'].get('notification_urls', [])
if notification_url in notification_urls:
return notification_url
# Append and update the datastore
notification_urls.append(notification_url)
self.__data['settings']['application']['notification_urls'] = notification_urls
self.needs_write = True
return notification_url

View File

@@ -443,10 +443,6 @@ Math: {{ 1 + 1 }}") }}
</tr>
</tbody>
</table>
<h4>Text similarity</h4>
<p><strong>Levenshtein Distance</strong> - Last 2 snapshots: {{ lev_info }}</p>
<p style="max-width: 80%; font-size: 80%"><strong>Levenshtein Distance</strong> Calculates the minimum number of insertions, deletions, and substitutions required to change one text into the other.</p>
{% if watch.history_n %}
<p>
<a href="{{url_for('ui.ui_edit.watch_get_latest_html', uuid=uuid)}}" class="pure-button button-small">Download latest HTML snapshot</a>

View File

@@ -1,108 +0,0 @@
#!/usr/bin/env python3
from flask import url_for
from .util import live_server_setup
import json
def test_api_notifications_crud(client, live_server):
live_server_setup(live_server)
api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token')
# Confirm notifications are initially empty
res = client.get(
url_for("notifications"),
headers={'x-api-key': api_key}
)
assert res.status_code == 200
assert res.json == {"notification_urls": []}
# Add notification URLs
test_urls = ["posts://example.com/notify1", "posts://example.com/notify2"]
res = client.post(
url_for("notifications"),
data=json.dumps({"notification_urls": test_urls}),
headers={'content-type': 'application/json', 'x-api-key': api_key}
)
assert res.status_code == 201
for url in test_urls:
assert url in res.json["notification_urls"]
# Confirm the notification URLs were added
res = client.get(
url_for("notifications"),
headers={'x-api-key': api_key}
)
assert res.status_code == 200
for url in test_urls:
assert url in res.json["notification_urls"]
# Delete one notification URL
res = client.delete(
url_for("notifications"),
data=json.dumps({"notification_urls": [test_urls[0]]}),
headers={'content-type': 'application/json', 'x-api-key': api_key}
)
assert res.status_code == 204
# Confirm it was removed and the other remains
res = client.get(
url_for("notifications"),
headers={'x-api-key': api_key}
)
assert res.status_code == 200
assert test_urls[0] not in res.json["notification_urls"]
assert test_urls[1] in res.json["notification_urls"]
# Try deleting a non-existent URL
res = client.delete(
url_for("notifications"),
data=json.dumps({"notification_urls": ["posts://nonexistent.com"]}),
headers={'content-type': 'application/json', 'x-api-key': api_key}
)
assert res.status_code == 400
res = client.post(
url_for("notifications"),
data=json.dumps({"notification_urls": test_urls}),
headers={'content-type': 'application/json', 'x-api-key': api_key}
)
assert res.status_code == 201
# Replace with a new list
replacement_urls = ["posts://new.example.com"]
res = client.put(
url_for("notifications"),
data=json.dumps({"notification_urls": replacement_urls}),
headers={'content-type': 'application/json', 'x-api-key': api_key}
)
assert res.status_code == 200
assert res.json["notification_urls"] == replacement_urls
# Replace with an empty list
res = client.put(
url_for("notifications"),
data=json.dumps({"notification_urls": []}),
headers={'content-type': 'application/json', 'x-api-key': api_key}
)
assert res.status_code == 200
assert res.json["notification_urls"] == []
# Provide an invalid AppRise URL to trigger validation error
invalid_urls = ["ftp://not-app-rise"]
res = client.post(
url_for("notifications"),
data=json.dumps({"notification_urls": invalid_urls}),
headers={'content-type': 'application/json', 'x-api-key': api_key}
)
assert res.status_code == 400
assert "is not a valid AppRise URL." in res.data.decode()
res = client.put(
url_for("notifications"),
data=json.dumps({"notification_urls": invalid_urls}),
headers={'content-type': 'application/json', 'x-api-key': api_key}
)
assert res.status_code == 400
assert "is not a valid AppRise URL." in res.data.decode()

View File

@@ -74,11 +74,6 @@ def test_check_basic_change_detection_functionality(client, live_server, measure
res = client.get(url_for("ui.ui_edit.watch_get_latest_html", uuid=uuid))
assert b'which has this one new line' in res.data
# Check the 'levenshtein' distance calc showed something useful
res = client.get(url_for("ui.ui_edit.edit_page", uuid=uuid))
assert b'Last 2 snapshots: 17' in res.data
# Now something should be ready, indicated by having a 'unviewed' class
res = client.get(url_for("watchlist.index"))
assert b'unviewed' in res.data

View File

@@ -68,8 +68,6 @@ openpyxl
jq~=1.3; python_version >= "3.8" and sys_platform == "darwin"
jq~=1.3; python_version >= "3.8" and sys_platform == "linux"
levenshtein
# playwright is installed at Dockerfile build time because it's not available on all platforms
pyppeteer-ng==2.0.0rc9