changedetection.io/changedetectionio/api/Notifications.py

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)