diff --git a/.dockerignore b/.dockerignore
index 2f88d7d3..14fba462 100644
--- a/.dockerignore
+++ b/.dockerignore
@@ -29,3 +29,35 @@ venv/
# Visual Studio
.vscode/
+
+# Test and development files
+test-datastore/
+tests/
+docs/
+*.md
+!README.md
+
+# Temporary and log files
+*.log
+*.tmp
+tmp/
+temp/
+
+# Training data and large files
+train-data/
+works-data/
+
+# Container files
+Dockerfile*
+docker-compose*.yml
+.dockerignore
+
+# Development certificates and keys
+*.pem
+*.key
+*.crt
+profile_output.prof
+
+# Large binary files that shouldn't be in container
+*.pdf
+chrome.json
\ No newline at end of file
diff --git a/.github/workflows/test-stack-reusable-workflow.yml b/.github/workflows/test-stack-reusable-workflow.yml
index 8f3b9301..af3d0fe2 100644
--- a/.github/workflows/test-stack-reusable-workflow.yml
+++ b/.github/workflows/test-stack-reusable-workflow.yml
@@ -86,10 +86,10 @@ jobs:
run: |
# Playwright via Sockpuppetbrowser fetch
# tests/visualselector/test_fetch_data.py will do browser steps
- docker run --rm -e "FLASK_SERVER_NAME=cdio" -e "PLAYWRIGHT_DRIVER_URL=ws://sockpuppetbrowser:3000" --network changedet-network --hostname=cdio test-changedetectionio bash -c 'cd changedetectionio;pytest --live-server-host=0.0.0.0 --live-server-port=5004 tests/fetchers/test_content.py'
- docker run --rm -e "FLASK_SERVER_NAME=cdio" -e "PLAYWRIGHT_DRIVER_URL=ws://sockpuppetbrowser:3000" --network changedet-network --hostname=cdio test-changedetectionio bash -c 'cd changedetectionio;pytest --live-server-host=0.0.0.0 --live-server-port=5004 tests/test_errorhandling.py'
- docker run --rm -e "FLASK_SERVER_NAME=cdio" -e "PLAYWRIGHT_DRIVER_URL=ws://sockpuppetbrowser:3000" --network changedet-network --hostname=cdio test-changedetectionio bash -c 'cd changedetectionio;pytest --live-server-host=0.0.0.0 --live-server-port=5004 tests/visualselector/test_fetch_data.py'
- docker run --rm -e "FLASK_SERVER_NAME=cdio" -e "PLAYWRIGHT_DRIVER_URL=ws://sockpuppetbrowser:3000" --network changedet-network --hostname=cdio test-changedetectionio bash -c 'cd changedetectionio;pytest --live-server-host=0.0.0.0 --live-server-port=5004 tests/fetchers/test_custom_js_before_content.py'
+ docker run --rm -e "FLASK_SERVER_NAME=cdio" -e "PLAYWRIGHT_DRIVER_URL=ws://sockpuppetbrowser:3000" --network changedet-network --hostname=cdio test-changedetectionio bash -c 'cd changedetectionio;pytest -vv --capture=tee-sys --showlocals --tb=long --live-server-host=0.0.0.0 --live-server-port=5004 tests/fetchers/test_content.py'
+ docker run --rm -e "FLASK_SERVER_NAME=cdio" -e "PLAYWRIGHT_DRIVER_URL=ws://sockpuppetbrowser:3000" --network changedet-network --hostname=cdio test-changedetectionio bash -c 'cd changedetectionio;pytest -vv --capture=tee-sys --showlocals --tb=long --live-server-host=0.0.0.0 --live-server-port=5004 tests/test_errorhandling.py'
+ docker run --rm -e "FLASK_SERVER_NAME=cdio" -e "PLAYWRIGHT_DRIVER_URL=ws://sockpuppetbrowser:3000" --network changedet-network --hostname=cdio test-changedetectionio bash -c 'cd changedetectionio;pytest -vv --capture=tee-sys --showlocals --tb=long --live-server-host=0.0.0.0 --live-server-port=5004 tests/visualselector/test_fetch_data.py'
+ docker run --rm -e "FLASK_SERVER_NAME=cdio" -e "PLAYWRIGHT_DRIVER_URL=ws://sockpuppetbrowser:3000" --network changedet-network --hostname=cdio test-changedetectionio bash -c 'cd changedetectionio;pytest -vv --capture=tee-sys --showlocals --tb=long --live-server-host=0.0.0.0 --live-server-port=5004 tests/fetchers/test_custom_js_before_content.py'
- name: Playwright and SocketPuppetBrowser - Headers and requests
diff --git a/changedetectionio/__init__.py b/changedetectionio/__init__.py
index 0cc72250..c34ab04b 100644
--- a/changedetectionio/__init__.py
+++ b/changedetectionio/__init__.py
@@ -10,10 +10,11 @@ import os
import getopt
import platform
import signal
-import socket
-import sys
-from werkzeug.serving import run_simple
+import sys
+
+# Eventlet completely removed - using threading mode for SocketIO
+# This provides better Python 3.12+ compatibility and eliminates eventlet/asyncio conflicts
from changedetectionio import store
from changedetectionio.flask_app import changedetection_app
from loguru import logger
@@ -28,22 +29,34 @@ def get_version():
# Parent wrapper or OS sends us a SIGTERM/SIGINT, do everything required for a clean shutdown
def sigshutdown_handler(_signo, _stack_frame):
name = signal.Signals(_signo).name
- logger.critical(f'Shutdown: Got Signal - {name} ({_signo}), Saving DB to disk and calling shutdown')
- datastore.sync_to_json()
- logger.success('Sync JSON to disk complete.')
+ logger.critical(f'Shutdown: Got Signal - {name} ({_signo}), Fast shutdown initiated')
- # Shutdown socketio server if available
+ # Set exit flag immediately to stop all loops
+ app.config.exit.set()
+ datastore.stop_thread = True
+
+ # Shutdown workers immediately
+ try:
+ from changedetectionio import worker_handler
+ worker_handler.shutdown_workers()
+ except Exception as e:
+ logger.error(f"Error shutting down workers: {str(e)}")
+
+ # Shutdown socketio server fast
from changedetectionio.flask_app import socketio_server
if socketio_server and hasattr(socketio_server, 'shutdown'):
try:
- logger.info("Shutting down Socket.IO server...")
socketio_server.shutdown()
except Exception as e:
logger.error(f"Error shutting down Socket.IO server: {str(e)}")
- # Set flags for clean shutdown
- datastore.stop_thread = True
- app.config.exit.set()
+ # Save data quickly
+ try:
+ datastore.sync_to_json()
+ logger.success('Fast sync to disk complete.')
+ except Exception as e:
+ logger.error(f"Error syncing to disk: {str(e)}")
+
sys.exit()
def main():
@@ -52,9 +65,9 @@ def main():
datastore_path = None
do_cleanup = False
- host = ''
+ host = "0.0.0.0"
ipv6_enabled = False
- port = os.environ.get('PORT') or 5000
+ port = int(os.environ.get('PORT', 5000))
ssl_mode = False
# On Windows, create and use a default path.
@@ -150,6 +163,11 @@ def main():
app = changedetection_app(app_config, datastore)
+ # Get the SocketIO instance from the Flask app (created in flask_app.py)
+ from changedetectionio.flask_app import socketio_server
+ global socketio
+ socketio = socketio_server
+
signal.signal(signal.SIGTERM, sigshutdown_handler)
signal.signal(signal.SIGINT, sigshutdown_handler)
@@ -174,10 +192,11 @@ def main():
@app.context_processor
- def inject_version():
+ def inject_template_globals():
return dict(right_sticky="v{}".format(datastore.data['version_tag']),
new_version_available=app.config['NEW_VERSION_AVAILABLE'],
- has_password=datastore.data['settings']['application']['password'] != False
+ has_password=datastore.data['settings']['application']['password'] != False,
+ socket_io_enabled=datastore.data['settings']['application']['ui'].get('socket_io_enabled', True)
)
# Monitored websites will not receive a Referer header when a user clicks on an outgoing link.
@@ -201,87 +220,21 @@ def main():
from werkzeug.middleware.proxy_fix import ProxyFix
app.wsgi_app = ProxyFix(app.wsgi_app, x_prefix=1, x_host=1)
- s_type = socket.AF_INET6 if ipv6_enabled else socket.AF_INET
- # Get socketio_server from flask_app
- from changedetectionio.flask_app import socketio_server
+ # SocketIO instance is already initialized in flask_app.py
- if socketio_server and datastore.data['settings']['application']['ui'].get('open_diff_in_new_tab'):
- logger.info("Starting server with Socket.IO support (using threading)...")
-
- # Use Flask-SocketIO's run method with error handling for Werkzeug warning
- # This is the cleanest approach that works with all Flask-SocketIO versions
- # Use '0.0.0.0' as the default host if none is specified
- # This will listen on all available interfaces
- listen_host = '0.0.0.0' if host == '' else host
- logger.info(f"Using host: {listen_host} and port: {port}")
-
- try:
- # First try with the allow_unsafe_werkzeug parameter (newer versions)
- if ssl_mode:
- socketio_server.run(
- app,
- host=listen_host,
- port=int(port),
- certfile='cert.pem',
- keyfile='privkey.pem',
- debug=False,
- use_reloader=False,
- allow_unsafe_werkzeug=True # Only in newer versions
- )
- else:
- socketio_server.run(
- app,
- host=listen_host,
- port=int(port),
- debug=False,
- use_reloader=False,
- allow_unsafe_werkzeug=True # Only in newer versions
- )
- except TypeError:
- # If allow_unsafe_werkzeug is not a valid parameter, try without it
- logger.info("Falling back to basic run method without allow_unsafe_werkzeug")
- # Override the werkzeug safety check by setting an environment variable
- os.environ['WERKZEUG_RUN_MAIN'] = 'true'
- if ssl_mode:
- socketio_server.run(
- app,
- host=listen_host,
- port=int(port),
- certfile='cert.pem',
- keyfile='privkey.pem',
- debug=False,
- use_reloader=False
- )
- else:
- socketio_server.run(
- app,
- host=listen_host,
- port=int(port),
- debug=False,
- use_reloader=False
- )
- else:
- logger.warning("Socket.IO server not initialized, falling back to standard WSGI server")
- # Fallback to standard WSGI server if socketio_server is not available
- listen_host = '0.0.0.0' if host == '' else host
+ # Launch using SocketIO run method for proper integration (if enabled)
+ if socketio_server:
if ssl_mode:
- # Use Werkzeug's run_simple with SSL support
- run_simple(
- hostname=listen_host,
- port=int(port),
- application=app,
- use_reloader=False,
- use_debugger=False,
- ssl_context=('cert.pem', 'privkey.pem')
- )
+ socketio.run(app, host=host, port=int(port), debug=False,
+ certfile='cert.pem', keyfile='privkey.pem', allow_unsafe_werkzeug=True)
else:
- # Use Werkzeug's run_simple for standard HTTP
- run_simple(
- hostname=listen_host,
- port=int(port),
- application=app,
- use_reloader=False,
- use_debugger=False
- )
-
+ socketio.run(app, host=host, port=int(port), debug=False, allow_unsafe_werkzeug=True)
+ else:
+ # Run Flask app without Socket.IO if disabled
+ logger.info("Starting Flask app without Socket.IO server")
+ if ssl_mode:
+ app.run(host=host, port=int(port), debug=False,
+ ssl_context=('cert.pem', 'privkey.pem'))
+ else:
+ app.run(host=host, port=int(port), debug=False)
diff --git a/changedetectionio/api/Watch.py b/changedetectionio/api/Watch.py
index 1a815670..c6011934 100644
--- a/changedetectionio/api/Watch.py
+++ b/changedetectionio/api/Watch.py
@@ -3,6 +3,7 @@ from changedetectionio.strtobool import strtobool
from flask_expects_json import expects_json
from changedetectionio import queuedWatchMetaData
+from changedetectionio import worker_handler
from flask_restful import abort, Resource
from flask import request, make_response
import validators
@@ -47,7 +48,7 @@ class Watch(Resource):
abort(404, message='No watch exists with the UUID of {}'.format(uuid))
if request.args.get('recheck'):
- self.update_q.put(queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': uuid}))
+ worker_handler.queue_item_async_safe(self.update_q, queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': uuid}))
return "OK", 200
if request.args.get('paused', '') == 'paused':
self.datastore.data['watching'].get(uuid).pause()
@@ -236,7 +237,7 @@ class CreateWatch(Resource):
new_uuid = self.datastore.add_watch(url=url, extras=extras, tag=tags)
if new_uuid:
- self.update_q.put(queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': new_uuid}))
+ worker_handler.queue_item_async_safe(self.update_q, queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': new_uuid}))
return {'uuid': new_uuid}, 201
else:
return "Invalid or unsupported URL", 400
@@ -291,7 +292,7 @@ class CreateWatch(Resource):
if request.args.get('recheck_all'):
for uuid in self.datastore.data['watching'].keys():
- self.update_q.put(queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': uuid}))
+ worker_handler.queue_item_async_safe(self.update_q, queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': uuid}))
return {'status': "OK"}, 200
return list, 200
\ No newline at end of file
diff --git a/changedetectionio/async_update_worker.py b/changedetectionio/async_update_worker.py
new file mode 100644
index 00000000..ae0501b0
--- /dev/null
+++ b/changedetectionio/async_update_worker.py
@@ -0,0 +1,449 @@
+from .processors.exceptions import ProcessorException
+import changedetectionio.content_fetchers.exceptions as content_fetchers_exceptions
+from changedetectionio.processors.text_json_diff.processor import FilterNotFoundInResponse
+from changedetectionio import html_tools
+from changedetectionio.flask_app import watch_check_update
+
+import asyncio
+import importlib
+import os
+import time
+
+from loguru import logger
+
+# Async version of update_worker
+# Processes jobs from AsyncSignalPriorityQueue instead of threaded queue
+
+async def async_update_worker(worker_id, q, notification_q, app, datastore):
+ """
+ Async worker function that processes watch check jobs from the queue.
+
+ Args:
+ worker_id: Unique identifier for this worker
+ q: AsyncSignalPriorityQueue containing jobs to process
+ notification_q: Standard queue for notifications
+ app: Flask application instance
+ datastore: Application datastore
+ """
+ # Set a descriptive name for this task
+ task = asyncio.current_task()
+ if task:
+ task.set_name(f"async-worker-{worker_id}")
+
+ logger.info(f"Starting async worker {worker_id}")
+
+ while not app.config.exit.is_set():
+ update_handler = None
+ watch = None
+
+ try:
+ # Use asyncio wait_for to make queue.get() cancellable
+ queued_item_data = await asyncio.wait_for(q.get(), timeout=1.0)
+ except asyncio.TimeoutError:
+ # No jobs available, continue loop
+ continue
+ except Exception as e:
+ logger.error(f"Worker {worker_id} error getting queue item: {e}")
+ await asyncio.sleep(0.1)
+ continue
+
+ uuid = queued_item_data.item.get('uuid')
+ fetch_start_time = round(time.time())
+
+ # Mark this UUID as being processed
+ from changedetectionio import worker_handler
+ worker_handler.set_uuid_processing(uuid, processing=True)
+
+ try:
+ if uuid in list(datastore.data['watching'].keys()) and datastore.data['watching'][uuid].get('url'):
+ changed_detected = False
+ contents = b''
+ process_changedetection_results = True
+ update_obj = {}
+
+ # Clear last errors
+ datastore.data['watching'][uuid]['browser_steps_last_error_step'] = None
+ datastore.data['watching'][uuid]['last_checked'] = fetch_start_time
+
+ watch = datastore.data['watching'].get(uuid)
+
+ logger.info(f"Worker {worker_id} processing watch UUID {uuid} Priority {queued_item_data.priority} URL {watch['url']}")
+
+ try:
+ watch_check_update.send(watch_uuid=uuid)
+
+ # Processor is what we are using for detecting the "Change"
+ processor = watch.get('processor', 'text_json_diff')
+
+ # Init a new 'difference_detection_processor'
+ processor_module_name = f"changedetectionio.processors.{processor}.processor"
+ try:
+ processor_module = importlib.import_module(processor_module_name)
+ except ModuleNotFoundError as e:
+ print(f"Processor module '{processor}' not found.")
+ raise e
+
+ update_handler = processor_module.perform_site_check(datastore=datastore,
+ watch_uuid=uuid)
+
+ # All fetchers are now async, so call directly
+ await update_handler.call_browser()
+
+ # Run change detection (this is synchronous)
+ changed_detected, update_obj, contents = update_handler.run_changedetection(watch=watch)
+
+ except PermissionError as e:
+ logger.critical(f"File permission error updating file, watch: {uuid}")
+ logger.critical(str(e))
+ process_changedetection_results = False
+
+ except ProcessorException as e:
+ if e.screenshot:
+ watch.save_screenshot(screenshot=e.screenshot)
+ if e.xpath_data:
+ watch.save_xpath_data(data=e.xpath_data)
+ datastore.update_watch(uuid=uuid, update_obj={'last_error': e.message})
+ process_changedetection_results = False
+
+ except content_fetchers_exceptions.ReplyWithContentButNoText as e:
+ extra_help = ""
+ if e.has_filters:
+ has_img = html_tools.include_filters(include_filters='img',
+ html_content=e.html_content)
+ if has_img:
+ extra_help = ", it's possible that the filters you have give an empty result or contain only an image."
+ else:
+ extra_help = ", it's possible that the filters were found, but contained no usable text."
+
+ datastore.update_watch(uuid=uuid, update_obj={
+ 'last_error': f"Got HTML content but no text found (With {e.status_code} reply code){extra_help}"
+ })
+
+ if e.screenshot:
+ watch.save_screenshot(screenshot=e.screenshot, as_error=True)
+
+ if e.xpath_data:
+ watch.save_xpath_data(data=e.xpath_data)
+
+ process_changedetection_results = False
+
+ except content_fetchers_exceptions.Non200ErrorCodeReceived as e:
+ if e.status_code == 403:
+ err_text = "Error - 403 (Access denied) received"
+ elif e.status_code == 404:
+ err_text = "Error - 404 (Page not found) received"
+ elif e.status_code == 407:
+ err_text = "Error - 407 (Proxy authentication required) received, did you need a username and password for the proxy?"
+ elif e.status_code == 500:
+ err_text = "Error - 500 (Internal server error) received from the web site"
+ else:
+ extra = ' (Access denied or blocked)' if str(e.status_code).startswith('4') else ''
+ err_text = f"Error - Request returned a HTTP error code {e.status_code}{extra}"
+
+ if e.screenshot:
+ watch.save_screenshot(screenshot=e.screenshot, as_error=True)
+ if e.xpath_data:
+ watch.save_xpath_data(data=e.xpath_data, as_error=True)
+ if e.page_text:
+ watch.save_error_text(contents=e.page_text)
+
+ datastore.update_watch(uuid=uuid, update_obj={'last_error': err_text})
+ process_changedetection_results = False
+
+ except FilterNotFoundInResponse as e:
+ if not datastore.data['watching'].get(uuid):
+ continue
+
+ err_text = "Warning, no filters were found, no change detection ran - Did the page change layout? update your Visual Filter if necessary."
+ datastore.update_watch(uuid=uuid, update_obj={'last_error': err_text})
+
+ # Filter wasnt found, but we should still update the visual selector so that they can have a chance to set it up again
+ if e.screenshot:
+ watch.save_screenshot(screenshot=e.screenshot)
+
+ if e.xpath_data:
+ watch.save_xpath_data(data=e.xpath_data)
+
+ # Only when enabled, send the notification
+ if watch.get('filter_failure_notification_send', False):
+ c = watch.get('consecutive_filter_failures', 0)
+ c += 1
+ # Send notification if we reached the threshold?
+ threshold = datastore.data['settings']['application'].get('filter_failure_notification_threshold_attempts', 0)
+ logger.debug(f"Filter for {uuid} not found, consecutive_filter_failures: {c} of threshold {threshold}")
+ if c >= threshold:
+ if not watch.get('notification_muted'):
+ logger.debug(f"Sending filter failed notification for {uuid}")
+ await send_filter_failure_notification(uuid, notification_q, datastore)
+ c = 0
+ logger.debug(f"Reset filter failure count back to zero")
+
+ datastore.update_watch(uuid=uuid, update_obj={'consecutive_filter_failures': c})
+ else:
+ logger.trace(f"{uuid} - filter_failure_notification_send not enabled, skipping")
+
+ process_changedetection_results = False
+
+ except content_fetchers_exceptions.checksumFromPreviousCheckWasTheSame as e:
+ # Yes fine, so nothing todo, don't continue to process.
+ process_changedetection_results = False
+ changed_detected = False
+
+ except content_fetchers_exceptions.BrowserConnectError as e:
+ datastore.update_watch(uuid=uuid,
+ update_obj={'last_error': e.msg})
+ process_changedetection_results = False
+
+ except content_fetchers_exceptions.BrowserFetchTimedOut as e:
+ datastore.update_watch(uuid=uuid,
+ update_obj={'last_error': e.msg})
+ process_changedetection_results = False
+
+ except content_fetchers_exceptions.BrowserStepsStepException as e:
+ if not datastore.data['watching'].get(uuid):
+ continue
+
+ error_step = e.step_n + 1
+ from playwright._impl._errors import TimeoutError, Error
+
+ # Generally enough info for TimeoutError (couldnt locate the element after default seconds)
+ err_text = f"Browser step at position {error_step} could not run, check the watch, add a delay if necessary, view Browser Steps to see screenshot at that step."
+
+ if e.original_e.name == "TimeoutError":
+ # Just the first line is enough, the rest is the stack trace
+ err_text += " Could not find the target."
+ else:
+ # Other Error, more info is good.
+ err_text += " " + str(e.original_e).splitlines()[0]
+
+ logger.debug(f"BrowserSteps exception at step {error_step} {str(e.original_e)}")
+
+ datastore.update_watch(uuid=uuid,
+ update_obj={'last_error': err_text,
+ 'browser_steps_last_error_step': error_step})
+
+ if watch.get('filter_failure_notification_send', False):
+ c = watch.get('consecutive_filter_failures', 0)
+ c += 1
+ # Send notification if we reached the threshold?
+ threshold = datastore.data['settings']['application'].get('filter_failure_notification_threshold_attempts', 0)
+ logger.error(f"Step for {uuid} not found, consecutive_filter_failures: {c}")
+ if threshold > 0 and c >= threshold:
+ if not watch.get('notification_muted'):
+ await send_step_failure_notification(watch_uuid=uuid, step_n=e.step_n, notification_q=notification_q, datastore=datastore)
+ c = 0
+
+ datastore.update_watch(uuid=uuid, update_obj={'consecutive_filter_failures': c})
+
+ process_changedetection_results = False
+
+ except content_fetchers_exceptions.EmptyReply as e:
+ # Some kind of custom to-str handler in the exception handler that does this?
+ err_text = "EmptyReply - try increasing 'Wait seconds before extracting text', Status Code {}".format(e.status_code)
+ datastore.update_watch(uuid=uuid, update_obj={'last_error': err_text,
+ 'last_check_status': e.status_code})
+ process_changedetection_results = False
+
+ except content_fetchers_exceptions.ScreenshotUnavailable as e:
+ err_text = "Screenshot unavailable, page did not render fully in the expected time or page was too long - try increasing 'Wait seconds before extracting text'"
+ datastore.update_watch(uuid=uuid, update_obj={'last_error': err_text,
+ 'last_check_status': e.status_code})
+ process_changedetection_results = False
+
+ except content_fetchers_exceptions.JSActionExceptions as e:
+ err_text = "Error running JS Actions - Page request - "+e.message
+ if e.screenshot:
+ watch.save_screenshot(screenshot=e.screenshot, as_error=True)
+ datastore.update_watch(uuid=uuid, update_obj={'last_error': err_text,
+ 'last_check_status': e.status_code})
+ process_changedetection_results = False
+
+ except content_fetchers_exceptions.PageUnloadable as e:
+ err_text = "Page request from server didnt respond correctly"
+ if e.message:
+ err_text = "{} - {}".format(err_text, e.message)
+
+ if e.screenshot:
+ watch.save_screenshot(screenshot=e.screenshot, as_error=True)
+
+ datastore.update_watch(uuid=uuid, update_obj={'last_error': err_text,
+ 'last_check_status': e.status_code,
+ 'has_ldjson_price_data': None})
+ process_changedetection_results = False
+
+ except content_fetchers_exceptions.BrowserStepsInUnsupportedFetcher as e:
+ err_text = "This watch has Browser Steps configured and so it cannot run with the 'Basic fast Plaintext/HTTP Client', either remove the Browser Steps or select a Chrome fetcher."
+ datastore.update_watch(uuid=uuid, update_obj={'last_error': err_text})
+ process_changedetection_results = False
+ logger.error(f"Exception (BrowserStepsInUnsupportedFetcher) reached processing watch UUID: {uuid}")
+
+ except Exception as e:
+ logger.error(f"Worker {worker_id} exception processing watch UUID: {uuid}")
+ logger.error(str(e))
+ datastore.update_watch(uuid=uuid, update_obj={'last_error': "Exception: " + str(e)})
+ process_changedetection_results = False
+
+ else:
+ if not datastore.data['watching'].get(uuid):
+ continue
+
+ update_obj['content-type'] = update_handler.fetcher.get_all_headers().get('content-type', '').lower()
+
+ if not watch.get('ignore_status_codes'):
+ update_obj['consecutive_filter_failures'] = 0
+
+ update_obj['last_error'] = False
+ cleanup_error_artifacts(uuid, datastore)
+
+ if not datastore.data['watching'].get(uuid):
+ continue
+
+ if process_changedetection_results:
+ # Extract title if needed
+ if datastore.data['settings']['application'].get('extract_title_as_title') or watch['extract_title_as_title']:
+ if not watch['title'] or not len(watch['title']):
+ try:
+ update_obj['title'] = html_tools.extract_element(find='title', html_content=update_handler.fetcher.content)
+ logger.info(f"UUID: {uuid} Extract
updated title to '{update_obj['title']}")
+ except Exception as e:
+ logger.warning(f"UUID: {uuid} Extract as watch title was enabled, but couldn't find a .")
+
+ try:
+ datastore.update_watch(uuid=uuid, update_obj=update_obj)
+
+ if changed_detected or not watch.history_n:
+ if update_handler.screenshot:
+ watch.save_screenshot(screenshot=update_handler.screenshot)
+
+ if update_handler.xpath_data:
+ watch.save_xpath_data(data=update_handler.xpath_data)
+
+ # Ensure unique timestamp for history
+ if watch.newest_history_key and int(fetch_start_time) == int(watch.newest_history_key):
+ logger.warning(f"Timestamp {fetch_start_time} already exists, waiting 1 seconds")
+ fetch_start_time += 1
+ await asyncio.sleep(1)
+
+ watch.save_history_text(contents=contents,
+ timestamp=int(fetch_start_time),
+ snapshot_id=update_obj.get('previous_md5', 'none'))
+
+ empty_pages_are_a_change = datastore.data['settings']['application'].get('empty_pages_are_a_change', False)
+ if update_handler.fetcher.content or (not update_handler.fetcher.content and empty_pages_are_a_change):
+ watch.save_last_fetched_html(contents=update_handler.fetcher.content, timestamp=int(fetch_start_time))
+
+ # Send notifications on second+ check
+ if watch.history_n >= 2:
+ logger.info(f"Change detected in UUID {uuid} - {watch['url']}")
+ if not watch.get('notification_muted'):
+ await send_content_changed_notification(uuid, notification_q, datastore)
+
+ except Exception as e:
+ logger.critical(f"Worker {worker_id} exception in process_changedetection_results")
+ logger.critical(str(e))
+ datastore.update_watch(uuid=uuid, update_obj={'last_error': str(e)})
+
+ # Always record attempt count
+ count = watch.get('check_count', 0) + 1
+
+ # Record server header
+ try:
+ server_header = update_handler.fetcher.headers.get('server', '').strip().lower()[:255]
+ datastore.update_watch(uuid=uuid, update_obj={'remote_server_reply': server_header})
+ except Exception as e:
+ pass
+
+ datastore.update_watch(uuid=uuid, update_obj={'fetch_time': round(time.time() - fetch_start_time, 3),
+ 'check_count': count})
+
+ except Exception as e:
+ logger.error(f"Worker {worker_id} unexpected error processing {uuid}: {e}")
+ logger.error(f"Worker {worker_id} traceback:", exc_info=True)
+
+ # Also update the watch with error information
+ if datastore and uuid in datastore.data['watching']:
+ datastore.update_watch(uuid=uuid, update_obj={'last_error': f"Worker error: {str(e)}"})
+
+ finally:
+ # Always cleanup - this runs whether there was an exception or not
+ if uuid:
+ try:
+ # Mark UUID as no longer being processed
+ worker_handler.set_uuid_processing(uuid, processing=False)
+
+ # Send completion signal
+ if watch:
+ #logger.info(f"Worker {worker_id} sending completion signal for UUID {watch['uuid']}")
+ watch_check_update.send(watch_uuid=watch['uuid'])
+
+ update_handler = None
+ logger.debug(f"Worker {worker_id} completed watch {uuid} in {time.time()-fetch_start_time:.2f}s")
+ except Exception as cleanup_error:
+ logger.error(f"Worker {worker_id} error during cleanup: {cleanup_error}")
+
+ # Brief pause before continuing to avoid tight error loops (only on error)
+ if 'e' in locals():
+ await asyncio.sleep(1.0)
+ else:
+ # Small yield for normal completion
+ await asyncio.sleep(0.01)
+
+ # Check if we should exit
+ if app.config.exit.is_set():
+ break
+
+ # Check if we're in pytest environment - if so, be more gentle with logging
+ import sys
+ in_pytest = "pytest" in sys.modules or "PYTEST_CURRENT_TEST" in os.environ
+
+ if not in_pytest:
+ logger.info(f"Worker {worker_id} shutting down")
+
+
+def cleanup_error_artifacts(uuid, datastore):
+ """Helper function to clean up error artifacts"""
+ cleanup_files = ["last-error-screenshot.png", "last-error.txt"]
+ for f in cleanup_files:
+ full_path = os.path.join(datastore.datastore_path, uuid, f)
+ if os.path.isfile(full_path):
+ os.unlink(full_path)
+
+
+
+async def send_content_changed_notification(watch_uuid, notification_q, datastore):
+ """Helper function to queue notifications using the new notification service"""
+ try:
+ from changedetectionio.notification_service import create_notification_service
+
+ # Create notification service instance
+ notification_service = create_notification_service(datastore, notification_q)
+
+ notification_service.send_content_changed_notification(watch_uuid)
+ except Exception as e:
+ logger.error(f"Error sending notification for {watch_uuid}: {e}")
+
+
+async def send_filter_failure_notification(watch_uuid, notification_q, datastore):
+ """Helper function to send filter failure notifications using the new notification service"""
+ try:
+ from changedetectionio.notification_service import create_notification_service
+
+ # Create notification service instance
+ notification_service = create_notification_service(datastore, notification_q)
+
+ notification_service.send_filter_failure_notification(watch_uuid)
+ except Exception as e:
+ logger.error(f"Error sending filter failure notification for {watch_uuid}: {e}")
+
+
+async def send_step_failure_notification(watch_uuid, step_n, notification_q, datastore):
+ """Helper function to send step failure notifications using the new notification service"""
+ try:
+ from changedetectionio.notification_service import create_notification_service
+
+ # Create notification service instance
+ notification_service = create_notification_service(datastore, notification_q)
+
+ notification_service.send_step_failure_notification(watch_uuid, step_n)
+ except Exception as e:
+ logger.error(f"Error sending step failure notification for {watch_uuid}: {e}")
\ No newline at end of file
diff --git a/changedetectionio/blueprint/browser_steps/__init__.py b/changedetectionio/blueprint/browser_steps/__init__.py
index f7907c7c..f0de3057 100644
--- a/changedetectionio/blueprint/browser_steps/__init__.py
+++ b/changedetectionio/blueprint/browser_steps/__init__.py
@@ -25,35 +25,53 @@ io_interface_context = None
import json
import hashlib
from flask import Response
+import asyncio
+import threading
+
+def run_async_in_browser_loop(coro):
+ """Run async coroutine using the existing async worker event loop"""
+ from changedetectionio import worker_handler
+
+ # Use the existing async worker event loop instead of creating a new one
+ if worker_handler.USE_ASYNC_WORKERS and worker_handler.async_loop and not worker_handler.async_loop.is_closed():
+ logger.debug("Browser steps using existing async worker event loop")
+ future = asyncio.run_coroutine_threadsafe(coro, worker_handler.async_loop)
+ return future.result()
+ else:
+ # Fallback: create a new event loop (for sync workers or if async loop not available)
+ logger.debug("Browser steps creating temporary event loop")
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ try:
+ return loop.run_until_complete(coro)
+ finally:
+ loop.close()
def construct_blueprint(datastore: ChangeDetectionStore):
browser_steps_blueprint = Blueprint('browser_steps', __name__, template_folder="templates")
- def start_browsersteps_session(watch_uuid):
- from . import nonContext
+ async def start_browsersteps_session(watch_uuid):
from . import browser_steps
import time
- global io_interface_context
+ from playwright.async_api import async_playwright
# We keep the playwright session open for many minutes
keepalive_seconds = int(os.getenv('BROWSERSTEPS_MINUTES_KEEPALIVE', 10)) * 60
browsersteps_start_session = {'start_time': time.time()}
- # You can only have one of these running
- # This should be very fine to leave running for the life of the application
- # @idea - Make it global so the pool of watch fetchers can use it also
- if not io_interface_context:
- io_interface_context = nonContext.c_sync_playwright()
- # Start the Playwright context, which is actually a nodejs sub-process and communicates over STDIN/STDOUT pipes
- io_interface_context = io_interface_context.start()
+ # Create a new async playwright instance for browser steps
+ playwright_instance = async_playwright()
+ playwright_context = await playwright_instance.start()
keepalive_ms = ((keepalive_seconds + 3) * 1000)
base_url = os.getenv('PLAYWRIGHT_DRIVER_URL', '').strip('"')
a = "?" if not '?' in base_url else '&'
base_url += a + f"timeout={keepalive_ms}"
- browsersteps_start_session['browser'] = io_interface_context.chromium.connect_over_cdp(base_url)
+ browser = await playwright_context.chromium.connect_over_cdp(base_url, timeout=keepalive_ms)
+ browsersteps_start_session['browser'] = browser
+ browsersteps_start_session['playwright_context'] = playwright_context
proxy_id = datastore.get_preferred_proxy_for_watch(uuid=watch_uuid)
proxy = None
@@ -75,15 +93,20 @@ def construct_blueprint(datastore: ChangeDetectionStore):
logger.debug(f"Browser Steps: UUID {watch_uuid} selected proxy {proxy_url}")
# Tell Playwright to connect to Chrome and setup a new session via our stepper interface
- browsersteps_start_session['browserstepper'] = browser_steps.browsersteps_live_ui(
- playwright_browser=browsersteps_start_session['browser'],
+ browserstepper = browser_steps.browsersteps_live_ui(
+ playwright_browser=browser,
proxy=proxy,
start_url=datastore.data['watching'][watch_uuid].link,
headers=datastore.data['watching'][watch_uuid].get('headers')
)
+
+ # Initialize the async connection
+ await browserstepper.connect(proxy=proxy)
+
+ browsersteps_start_session['browserstepper'] = browserstepper
# For test
- #browsersteps_start_session['browserstepper'].action_goto_url(value="http://example.com?time="+str(time.time()))
+ #await browsersteps_start_session['browserstepper'].action_goto_url(value="http://example.com?time="+str(time.time()))
return browsersteps_start_session
@@ -92,7 +115,7 @@ def construct_blueprint(datastore: ChangeDetectionStore):
@browser_steps_blueprint.route("/browsersteps_start_session", methods=['GET'])
def browsersteps_start_session():
# A new session was requested, return sessionID
-
+ import asyncio
import uuid
browsersteps_session_id = str(uuid.uuid4())
watch_uuid = request.args.get('uuid')
@@ -104,7 +127,10 @@ def construct_blueprint(datastore: ChangeDetectionStore):
logger.debug("browser_steps.py connecting")
try:
- browsersteps_sessions[browsersteps_session_id] = start_browsersteps_session(watch_uuid)
+ # Run the async function in the dedicated browser steps event loop
+ browsersteps_sessions[browsersteps_session_id] = run_async_in_browser_loop(
+ start_browsersteps_session(watch_uuid)
+ )
except Exception as e:
if 'ECONNREFUSED' in str(e):
return make_response('Unable to start the Playwright Browser session, is sockpuppetbrowser running? Network configuration is OK?', 401)
@@ -169,9 +195,14 @@ def construct_blueprint(datastore: ChangeDetectionStore):
is_last_step = strtobool(request.form.get('is_last_step'))
try:
- browsersteps_sessions[browsersteps_session_id]['browserstepper'].call_action(action_name=step_operation,
- selector=step_selector,
- optional_value=step_optional_value)
+ # Run the async call_action method in the dedicated browser steps event loop
+ run_async_in_browser_loop(
+ browsersteps_sessions[browsersteps_session_id]['browserstepper'].call_action(
+ action_name=step_operation,
+ selector=step_selector,
+ optional_value=step_optional_value
+ )
+ )
except Exception as e:
logger.error(f"Exception when calling step operation {step_operation} {str(e)}")
@@ -185,7 +216,11 @@ def construct_blueprint(datastore: ChangeDetectionStore):
# Screenshots and other info only needed on requesting a step (POST)
try:
- (screenshot, xpath_data) = browsersteps_sessions[browsersteps_session_id]['browserstepper'].get_current_state()
+ # Run the async get_current_state method in the dedicated browser steps event loop
+ (screenshot, xpath_data) = run_async_in_browser_loop(
+ browsersteps_sessions[browsersteps_session_id]['browserstepper'].get_current_state()
+ )
+
if is_last_step:
watch = datastore.data['watching'].get(uuid)
u = browsersteps_sessions[browsersteps_session_id]['browserstepper'].page.url
@@ -199,7 +234,6 @@ def construct_blueprint(datastore: ChangeDetectionStore):
return make_response("Error fetching screenshot and element data - " + str(e), 401)
# SEND THIS BACK TO THE BROWSER
-
output = {
"screenshot": f"data:image/jpeg;base64,{base64.b64encode(screenshot).decode('ascii')}",
"xpath_data": xpath_data,
diff --git a/changedetectionio/blueprint/browser_steps/browser_steps.py b/changedetectionio/blueprint/browser_steps/browser_steps.py
index d380d565..e0a3cb2c 100644
--- a/changedetectionio/blueprint/browser_steps/browser_steps.py
+++ b/changedetectionio/blueprint/browser_steps/browser_steps.py
@@ -63,7 +63,7 @@ class steppable_browser_interface():
self.start_url = start_url
# Convert and perform "Click Button" for example
- def call_action(self, action_name, selector=None, optional_value=None):
+ async def call_action(self, action_name, selector=None, optional_value=None):
if self.page is None:
logger.warning("Cannot call action on None page object")
return
@@ -93,73 +93,74 @@ class steppable_browser_interface():
optional_value = jinja_render(template_str=optional_value)
- action_handler(selector, optional_value)
+ await action_handler(selector, optional_value)
# Safely wait for timeout
- self.page.wait_for_timeout(1.5 * 1000)
+ await self.page.wait_for_timeout(1.5 * 1000)
logger.debug(f"Call action done in {time.time()-now:.2f}s")
- def action_goto_url(self, selector=None, value=None):
+ async def action_goto_url(self, selector=None, value=None):
if not value:
logger.warning("No URL provided for goto_url action")
return None
now = time.time()
- response = self.page.goto(value, timeout=0, wait_until='load')
+ response = await self.page.goto(value, timeout=0, wait_until='load')
logger.debug(f"Time to goto URL {time.time()-now:.2f}s")
return response
# Incase they request to go back to the start
- def action_goto_site(self, selector=None, value=None):
- return self.action_goto_url(value=self.start_url)
+ async def action_goto_site(self, selector=None, value=None):
+ return await self.action_goto_url(value=self.start_url)
- def action_click_element_containing_text(self, selector=None, value=''):
+ async def action_click_element_containing_text(self, selector=None, value=''):
logger.debug("Clicking element containing text")
if not value or not len(value.strip()):
return
elem = self.page.get_by_text(value)
- if elem.count():
- elem.first.click(delay=randint(200, 500), timeout=self.action_timeout)
+ if await elem.count():
+ await elem.first.click(delay=randint(200, 500), timeout=self.action_timeout)
- def action_click_element_containing_text_if_exists(self, selector=None, value=''):
+ async def action_click_element_containing_text_if_exists(self, selector=None, value=''):
logger.debug("Clicking element containing text if exists")
if not value or not len(value.strip()):
return
elem = self.page.get_by_text(value)
- logger.debug(f"Clicking element containing text - {elem.count()} elements found")
- if elem.count():
- elem.first.click(delay=randint(200, 500), timeout=self.action_timeout)
+ count = await elem.count()
+ logger.debug(f"Clicking element containing text - {count} elements found")
+ if count:
+ await elem.first.click(delay=randint(200, 500), timeout=self.action_timeout)
- def action_enter_text_in_field(self, selector, value):
+ async def action_enter_text_in_field(self, selector, value):
if not selector or not len(selector.strip()):
return
- self.page.fill(selector, value, timeout=self.action_timeout)
+ await self.page.fill(selector, value, timeout=self.action_timeout)
- def action_execute_js(self, selector, value):
+ async def action_execute_js(self, selector, value):
if not value:
return None
- return self.page.evaluate(value)
+ return await self.page.evaluate(value)
- def action_click_element(self, selector, value):
+ async def action_click_element(self, selector, value):
logger.debug("Clicking element")
if not selector or not len(selector.strip()):
return
- self.page.click(selector=selector, timeout=self.action_timeout + 20 * 1000, delay=randint(200, 500))
+ await self.page.click(selector=selector, timeout=self.action_timeout + 20 * 1000, delay=randint(200, 500))
- def action_click_element_if_exists(self, selector, value):
+ async def action_click_element_if_exists(self, selector, value):
import playwright._impl._errors as _api_types
logger.debug("Clicking element if exists")
if not selector or not len(selector.strip()):
return
try:
- self.page.click(selector, timeout=self.action_timeout, delay=randint(200, 500))
+ await self.page.click(selector, timeout=self.action_timeout, delay=randint(200, 500))
except _api_types.TimeoutError:
return
except _api_types.Error:
@@ -167,7 +168,7 @@ class steppable_browser_interface():
return
- def action_click_x_y(self, selector, value):
+ async def action_click_x_y(self, selector, value):
if not value or not re.match(r'^\s?\d+\s?,\s?\d+\s?$', value):
logger.warning("'Click X,Y' step should be in the format of '100 , 90'")
return
@@ -177,42 +178,42 @@ class steppable_browser_interface():
x = int(float(x.strip()))
y = int(float(y.strip()))
- self.page.mouse.click(x=x, y=y, delay=randint(200, 500))
+ await self.page.mouse.click(x=x, y=y, delay=randint(200, 500))
except Exception as e:
logger.error(f"Error parsing x,y coordinates: {str(e)}")
- def action__select_by_option_text(self, selector, value):
+ async def action__select_by_option_text(self, selector, value):
if not selector or not len(selector.strip()):
return
- self.page.select_option(selector, label=value, timeout=self.action_timeout)
+ await self.page.select_option(selector, label=value, timeout=self.action_timeout)
- def action_scroll_down(self, selector, value):
+ async def action_scroll_down(self, selector, value):
# Some sites this doesnt work on for some reason
- self.page.mouse.wheel(0, 600)
- self.page.wait_for_timeout(1000)
+ await self.page.mouse.wheel(0, 600)
+ await self.page.wait_for_timeout(1000)
- def action_wait_for_seconds(self, selector, value):
+ async def action_wait_for_seconds(self, selector, value):
try:
seconds = float(value.strip()) if value else 1.0
- self.page.wait_for_timeout(seconds * 1000)
+ await self.page.wait_for_timeout(seconds * 1000)
except (ValueError, TypeError) as e:
logger.error(f"Invalid value for wait_for_seconds: {str(e)}")
- def action_wait_for_text(self, selector, value):
+ async def action_wait_for_text(self, selector, value):
if not value:
return
import json
v = json.dumps(value)
- self.page.wait_for_function(
+ await self.page.wait_for_function(
f'document.querySelector("body").innerText.includes({v});',
timeout=30000
)
- def action_wait_for_text_in_element(self, selector, value):
+ async def action_wait_for_text_in_element(self, selector, value):
if not selector or not value:
return
@@ -220,49 +221,49 @@ class steppable_browser_interface():
s = json.dumps(selector)
v = json.dumps(value)
- self.page.wait_for_function(
+ await self.page.wait_for_function(
f'document.querySelector({s}).innerText.includes({v});',
timeout=30000
)
# @todo - in the future make some popout interface to capture what needs to be set
# https://playwright.dev/python/docs/api/class-keyboard
- def action_press_enter(self, selector, value):
- self.page.keyboard.press("Enter", delay=randint(200, 500))
+ async def action_press_enter(self, selector, value):
+ await self.page.keyboard.press("Enter", delay=randint(200, 500))
- def action_press_page_up(self, selector, value):
- self.page.keyboard.press("PageUp", delay=randint(200, 500))
+ async def action_press_page_up(self, selector, value):
+ await self.page.keyboard.press("PageUp", delay=randint(200, 500))
- def action_press_page_down(self, selector, value):
- self.page.keyboard.press("PageDown", delay=randint(200, 500))
+ async def action_press_page_down(self, selector, value):
+ await self.page.keyboard.press("PageDown", delay=randint(200, 500))
- def action_check_checkbox(self, selector, value):
+ async def action_check_checkbox(self, selector, value):
if not selector:
return
- self.page.locator(selector).check(timeout=self.action_timeout)
+ await self.page.locator(selector).check(timeout=self.action_timeout)
- def action_uncheck_checkbox(self, selector, value):
+ async def action_uncheck_checkbox(self, selector, value):
if not selector:
return
- self.page.locator(selector).uncheck(timeout=self.action_timeout)
+ await self.page.locator(selector).uncheck(timeout=self.action_timeout)
- def action_remove_elements(self, selector, value):
+ async def action_remove_elements(self, selector, value):
"""Removes all elements matching the given selector from the DOM."""
if not selector:
return
- self.page.locator(selector).evaluate_all("els => els.forEach(el => el.remove())")
+ await self.page.locator(selector).evaluate_all("els => els.forEach(el => el.remove())")
- def action_make_all_child_elements_visible(self, selector, value):
+ async def action_make_all_child_elements_visible(self, selector, value):
"""Recursively makes all child elements inside the given selector fully visible."""
if not selector:
return
- self.page.locator(selector).locator("*").evaluate_all("""
+ await self.page.locator(selector).locator("*").evaluate_all("""
els => els.forEach(el => {
el.style.display = 'block'; // Forces it to be displayed
el.style.visibility = 'visible'; // Ensures it's not hidden
@@ -307,21 +308,22 @@ class browsersteps_live_ui(steppable_browser_interface):
self.playwright_browser = playwright_browser
self.start_url = start_url
self._is_cleaned_up = False
- if self.context is None:
- self.connect(proxy=proxy)
+ self.proxy = proxy
+ # Note: connect() is now async and must be called separately
def __del__(self):
# Ensure cleanup happens if object is garbage collected
- self.cleanup()
+ # Note: cleanup is now async, so we can only mark as cleaned up here
+ self._is_cleaned_up = True
# Connect and setup a new context
- def connect(self, proxy=None):
+ async def connect(self, proxy=None):
# Should only get called once - test that
keep_open = 1000 * 60 * 5
now = time.time()
# @todo handle multiple contexts, bind a unique id from the browser on each req?
- self.context = self.playwright_browser.new_context(
+ self.context = await self.playwright_browser.new_context(
accept_downloads=False, # Should never be needed
bypass_csp=True, # This is needed to enable JavaScript execution on GitHub and others
extra_http_headers=self.headers,
@@ -332,7 +334,7 @@ class browsersteps_live_ui(steppable_browser_interface):
user_agent=manage_user_agent(headers=self.headers),
)
- self.page = self.context.new_page()
+ self.page = await self.context.new_page()
# self.page.set_default_navigation_timeout(keep_open)
self.page.set_default_timeout(keep_open)
@@ -342,13 +344,15 @@ class browsersteps_live_ui(steppable_browser_interface):
self.page.on("console", lambda msg: print(f"Browser steps console - {msg.type}: {msg.text} {msg.args}"))
logger.debug(f"Time to browser setup {time.time()-now:.2f}s")
- self.page.wait_for_timeout(1 * 1000)
+ await self.page.wait_for_timeout(1 * 1000)
def mark_as_closed(self):
logger.debug("Page closed, cleaning up..")
- self.cleanup()
+ # Note: This is called from a sync context (event handler)
+ # so we'll just mark as cleaned up and let __del__ handle the rest
+ self._is_cleaned_up = True
- def cleanup(self):
+ async def cleanup(self):
"""Properly clean up all resources to prevent memory leaks"""
if self._is_cleaned_up:
return
@@ -359,7 +363,7 @@ class browsersteps_live_ui(steppable_browser_interface):
if hasattr(self, 'page') and self.page is not None:
try:
# Force garbage collection before closing
- self.page.request_gc()
+ await self.page.request_gc()
except Exception as e:
logger.debug(f"Error during page garbage collection: {str(e)}")
@@ -370,7 +374,7 @@ class browsersteps_live_ui(steppable_browser_interface):
logger.debug(f"Error removing event listeners: {str(e)}")
try:
- self.page.close()
+ await self.page.close()
except Exception as e:
logger.debug(f"Error closing page: {str(e)}")
@@ -379,7 +383,7 @@ class browsersteps_live_ui(steppable_browser_interface):
# Clean up context
if hasattr(self, 'context') and self.context is not None:
try:
- self.context.close()
+ await self.context.close()
except Exception as e:
logger.debug(f"Error closing context: {str(e)}")
@@ -401,12 +405,12 @@ class browsersteps_live_ui(steppable_browser_interface):
return False
- def get_current_state(self):
+ async def get_current_state(self):
"""Return the screenshot and interactive elements mapping, generally always called after action_()"""
import importlib.resources
import json
# because we for now only run browser steps in playwright mode (not puppeteer mode)
- from changedetectionio.content_fetchers.playwright import capture_full_page
+ from changedetectionio.content_fetchers.playwright import capture_full_page_async
# Safety check - don't proceed if resources are cleaned up
if self._is_cleaned_up or self.page is None:
@@ -416,29 +420,29 @@ class browsersteps_live_ui(steppable_browser_interface):
xpath_element_js = importlib.resources.files("changedetectionio.content_fetchers.res").joinpath('xpath_element_scraper.js').read_text()
now = time.time()
- self.page.wait_for_timeout(1 * 1000)
+ await self.page.wait_for_timeout(1 * 1000)
screenshot = None
xpath_data = None
try:
# Get screenshot first
- screenshot = capture_full_page(page=self.page)
+ screenshot = await capture_full_page_async(page=self.page)
logger.debug(f"Time to get screenshot from browser {time.time() - now:.2f}s")
# Then get interactive elements
now = time.time()
- self.page.evaluate("var include_filters=''")
- self.page.request_gc()
+ await self.page.evaluate("var include_filters=''")
+ await self.page.request_gc()
scan_elements = 'a,button,input,select,textarea,i,th,td,p,li,h1,h2,h3,h4,div,span'
MAX_TOTAL_HEIGHT = int(os.getenv("SCREENSHOT_MAX_HEIGHT", SCREENSHOT_MAX_HEIGHT_DEFAULT))
- xpath_data = json.loads(self.page.evaluate(xpath_element_js, {
+ xpath_data = json.loads(await self.page.evaluate(xpath_element_js, {
"visualselector_xpath_selectors": scan_elements,
"max_height": MAX_TOTAL_HEIGHT
}))
- self.page.request_gc()
+ await self.page.request_gc()
# Sort elements by size
xpath_data['size_pos'] = sorted(xpath_data['size_pos'], key=lambda k: k['width'] * k['height'], reverse=True)
@@ -448,13 +452,13 @@ class browsersteps_live_ui(steppable_browser_interface):
logger.error(f"Error getting current state: {str(e)}")
# Attempt recovery - force garbage collection
try:
- self.page.request_gc()
+ await self.page.request_gc()
except:
pass
# Request garbage collection one final time
try:
- self.page.request_gc()
+ await self.page.request_gc()
except:
pass
diff --git a/changedetectionio/blueprint/browser_steps/nonContext.py b/changedetectionio/blueprint/browser_steps/nonContext.py
deleted file mode 100644
index 93abe269..00000000
--- a/changedetectionio/blueprint/browser_steps/nonContext.py
+++ /dev/null
@@ -1,17 +0,0 @@
-from playwright.sync_api import PlaywrightContextManager
-
-# So playwright wants to run as a context manager, but we do something horrible and hacky
-# we are holding the session open for as long as possible, then shutting it down, and opening a new one
-# So it means we don't get to use PlaywrightContextManager' __enter__ __exit__
-# To work around this, make goodbye() act the same as the __exit__()
-#
-# But actually I think this is because the context is opened correctly with __enter__() but we timeout the connection
-# then theres some lock condition where we cant destroy it without it hanging
-
-class c_PlaywrightContextManager(PlaywrightContextManager):
-
- def goodbye(self) -> None:
- self.__exit__()
-
-def c_sync_playwright() -> PlaywrightContextManager:
- return c_PlaywrightContextManager()
diff --git a/changedetectionio/blueprint/imports/__init__.py b/changedetectionio/blueprint/imports/__init__.py
index 2e5fddf5..e6fbf760 100644
--- a/changedetectionio/blueprint/imports/__init__.py
+++ b/changedetectionio/blueprint/imports/__init__.py
@@ -1,6 +1,7 @@
from flask import Blueprint, request, redirect, url_for, flash, render_template
from changedetectionio.store import ChangeDetectionStore
from changedetectionio.auth_decorator import login_optionally_required
+from changedetectionio import worker_handler
from changedetectionio.blueprint.imports.importer import (
import_url_list,
import_distill_io_json,
@@ -24,7 +25,7 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, queuedWatchMe
importer_handler = import_url_list()
importer_handler.run(data=request.values.get('urls'), flash=flash, datastore=datastore, processor=request.values.get('processor', 'text_json_diff'))
for uuid in importer_handler.new_uuids:
- update_q.put(queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': uuid}))
+ worker_handler.queue_item_async_safe(update_q, queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': uuid}))
if len(importer_handler.remaining_data) == 0:
return redirect(url_for('watchlist.index'))
@@ -37,7 +38,7 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, queuedWatchMe
d_importer = import_distill_io_json()
d_importer.run(data=request.values.get('distill-io'), flash=flash, datastore=datastore)
for uuid in d_importer.new_uuids:
- update_q.put(queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': uuid}))
+ worker_handler.queue_item_async_safe(update_q, queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': uuid}))
# XLSX importer
if request.files and request.files.get('xlsx_file'):
@@ -60,7 +61,7 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, queuedWatchMe
w_importer.run(data=file, flash=flash, datastore=datastore)
for uuid in w_importer.new_uuids:
- update_q.put(queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': uuid}))
+ worker_handler.queue_item_async_safe(update_q, queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': uuid}))
# Could be some remaining, or we could be on GET
form = forms.importForm(formdata=request.form if request.method == 'POST' else None)
diff --git a/changedetectionio/blueprint/price_data_follower/__init__.py b/changedetectionio/blueprint/price_data_follower/__init__.py
index 99841d71..c2c6e768 100644
--- a/changedetectionio/blueprint/price_data_follower/__init__.py
+++ b/changedetectionio/blueprint/price_data_follower/__init__.py
@@ -4,6 +4,7 @@ from flask import Blueprint, flash, redirect, url_for
from flask_login import login_required
from changedetectionio.store import ChangeDetectionStore
from changedetectionio import queuedWatchMetaData
+from changedetectionio import worker_handler
from queue import PriorityQueue
PRICE_DATA_TRACK_ACCEPT = 'accepted'
@@ -19,7 +20,7 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q: PriorityQueue
datastore.data['watching'][uuid]['track_ldjson_price_data'] = PRICE_DATA_TRACK_ACCEPT
datastore.data['watching'][uuid]['processor'] = 'restock_diff'
datastore.data['watching'][uuid].clear_watch()
- update_q.put(queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': uuid}))
+ worker_handler.queue_item_async_safe(update_q, queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': uuid}))
return redirect(url_for("watchlist.index"))
@login_required
diff --git a/changedetectionio/blueprint/settings/__init__.py b/changedetectionio/blueprint/settings/__init__.py
index 015cc274..548a5b70 100644
--- a/changedetectionio/blueprint/settings/__init__.py
+++ b/changedetectionio/blueprint/settings/__init__.py
@@ -67,7 +67,32 @@ def construct_blueprint(datastore: ChangeDetectionStore):
del (app_update['password'])
datastore.data['settings']['application'].update(app_update)
+
+ # Handle dynamic worker count adjustment
+ old_worker_count = datastore.data['settings']['requests'].get('workers', 1)
+ new_worker_count = form.data['requests'].get('workers', 1)
+
datastore.data['settings']['requests'].update(form.data['requests'])
+
+ # Adjust worker count if it changed
+ if new_worker_count != old_worker_count:
+ from changedetectionio import worker_handler
+ from changedetectionio.flask_app import update_q, notification_q, app, datastore as ds
+
+ result = worker_handler.adjust_async_worker_count(
+ new_count=new_worker_count,
+ update_q=update_q,
+ notification_q=notification_q,
+ app=app,
+ datastore=ds
+ )
+
+ if result['status'] == 'success':
+ flash(f"Worker count adjusted: {result['message']}", 'notice')
+ elif result['status'] == 'not_supported':
+ flash("Dynamic worker adjustment not supported for sync workers", 'warning')
+ elif result['status'] == 'error':
+ flash(f"Error adjusting workers: {result['message']}", 'error')
if not os.getenv("SALTED_PASS", False) and len(form.application.form.password.encrypted_password):
datastore.data['settings']['application']['password'] = form.application.form.password.encrypted_password
diff --git a/changedetectionio/blueprint/settings/templates/settings.html b/changedetectionio/blueprint/settings/templates/settings.html
index 5f302331..88ebd6de 100644
--- a/changedetectionio/blueprint/settings/templates/settings.html
+++ b/changedetectionio/blueprint/settings/templates/settings.html
@@ -135,6 +135,12 @@
{{ render_field(form.application.form.webdriver_delay) }}
+
+ {{ render_field(form.requests.form.workers) }}
+ {% set worker_info = get_worker_status_info() %}
+ Number of concurrent workers to process watches. More workers = faster processing but higher memory usage.
+ Currently running: {{ worker_info.count }} operational {{ worker_info.type }} workers{% if worker_info.active_workers > 0 %} ({{ worker_info.active_workers }} actively processing){% endif %}.
+
{{ render_field(form.requests.form.default_ua) }}
@@ -247,9 +253,9 @@ nav
Enable this setting to open the diff page in a new tab. If disabled, the diff page will open in the current tab.
- Enable realtime updates in the UI
+ {{ render_checkbox_field(form.application.form.ui.form.socket_io_enabled, class="socket_io_enabled") }}
+ Realtime UI Updates Enabled - (Restart required if this is changed)
-
diff --git a/changedetectionio/blueprint/ui/__init__.py b/changedetectionio/blueprint/ui/__init__.py
index c9061bf7..9ed40554 100644
--- a/changedetectionio/blueprint/ui/__init__.py
+++ b/changedetectionio/blueprint/ui/__init__.py
@@ -1,15 +1,13 @@
import time
from flask import Blueprint, request, redirect, url_for, flash, render_template, session
from loguru import logger
-from functools import wraps
-from changedetectionio.blueprint.ui.ajax import constuct_ui_ajax_blueprint
from changedetectionio.store import ChangeDetectionStore
from changedetectionio.blueprint.ui.edit import construct_blueprint as construct_edit_blueprint
from changedetectionio.blueprint.ui.notification import construct_blueprint as construct_notification_blueprint
from changedetectionio.blueprint.ui.views import construct_blueprint as construct_views_blueprint
-def construct_blueprint(datastore: ChangeDetectionStore, update_q, running_update_threads, queuedWatchMetaData, watch_check_update):
+def construct_blueprint(datastore: ChangeDetectionStore, update_q, worker_handler, queuedWatchMetaData, watch_check_update):
ui_blueprint = Blueprint('ui', __name__, template_folder="templates")
# Register the edit blueprint
@@ -24,9 +22,6 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, running_updat
views_blueprint = construct_views_blueprint(datastore, update_q, queuedWatchMetaData, watch_check_update)
ui_blueprint.register_blueprint(views_blueprint)
- ui_ajax_blueprint = constuct_ui_ajax_blueprint(datastore, update_q, running_update_threads, queuedWatchMetaData, watch_check_update)
- ui_blueprint.register_blueprint(ui_ajax_blueprint)
-
# Import the login decorator
from changedetectionio.auth_decorator import login_optionally_required
@@ -100,7 +95,7 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, running_updat
new_uuid = datastore.clone(uuid)
if not datastore.data['watching'].get(uuid).get('paused'):
- update_q.put(queuedWatchMetaData.PrioritizedItem(priority=5, item={'uuid': new_uuid}))
+ worker_handler.queue_item_async_safe(update_q, queuedWatchMetaData.PrioritizedItem(priority=5, item={'uuid': new_uuid}))
flash('Cloned, you are editing the new watch.')
@@ -116,13 +111,11 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, running_updat
i = 0
- running_uuids = []
- for t in running_update_threads:
- running_uuids.append(t.current_uuid)
+ running_uuids = worker_handler.get_running_uuids()
if uuid:
if uuid not in running_uuids:
- update_q.put(queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': uuid}))
+ worker_handler.queue_item_async_safe(update_q, queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': uuid}))
i += 1
else:
@@ -139,7 +132,7 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, running_updat
if tag != None and tag not in watch['tags']:
continue
- update_q.put(queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': watch_uuid}))
+ worker_handler.queue_item_async_safe(update_q, queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': watch_uuid}))
i += 1
if i == 1:
@@ -197,7 +190,7 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, running_updat
for uuid in uuids:
if datastore.data['watching'].get(uuid):
# Recheck and require a full reprocessing
- update_q.put(queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': uuid}))
+ worker_handler.queue_item_async_safe(update_q, queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': uuid}))
flash("{} watches queued for rechecking".format(len(uuids)))
elif (op == 'clear-errors'):
diff --git a/changedetectionio/blueprint/ui/ajax.py b/changedetectionio/blueprint/ui/ajax.py
deleted file mode 100644
index bbe3464d..00000000
--- a/changedetectionio/blueprint/ui/ajax.py
+++ /dev/null
@@ -1,35 +0,0 @@
-import time
-
-from blinker import signal
-from flask import Blueprint, request, redirect, url_for, flash, render_template, session
-
-
-from changedetectionio.store import ChangeDetectionStore
-
-def constuct_ui_ajax_blueprint(datastore: ChangeDetectionStore, update_q, running_update_threads, queuedWatchMetaData, watch_check_update):
- ui_ajax_blueprint = Blueprint('ajax', __name__, template_folder="templates", url_prefix='/ajax')
-
- # Import the login decorator
- from changedetectionio.auth_decorator import login_optionally_required
-
- @ui_ajax_blueprint.route("/toggle", methods=['POST'])
- @login_optionally_required
- def ajax_toggler():
- op = request.values.get('op')
- uuid = request.values.get('uuid')
- if op and datastore.data['watching'].get(uuid):
- if op == 'pause':
- datastore.data['watching'][uuid].toggle_pause()
- elif op == 'mute':
- datastore.data['watching'][uuid].toggle_mute()
- elif op == 'recheck':
- update_q.put(queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': uuid}))
-
- watch_check_update = signal('watch_check_update')
- if watch_check_update:
- watch_check_update.send(watch_uuid=uuid)
-
- return 'OK'
-
-
- return ui_ajax_blueprint
diff --git a/changedetectionio/blueprint/ui/edit.py b/changedetectionio/blueprint/ui/edit.py
index b491d854..bdee4725 100644
--- a/changedetectionio/blueprint/ui/edit.py
+++ b/changedetectionio/blueprint/ui/edit.py
@@ -9,6 +9,7 @@ from jinja2 import Environment, FileSystemLoader
from changedetectionio.store import ChangeDetectionStore
from changedetectionio.auth_decorator import login_optionally_required
from changedetectionio.time_handler import is_within_schedule
+from changedetectionio import worker_handler
def construct_blueprint(datastore: ChangeDetectionStore, update_q, queuedWatchMetaData):
edit_blueprint = Blueprint('ui_edit', __name__, template_folder="../ui/templates")
@@ -201,7 +202,7 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, queuedWatchMe
#############################
if not datastore.data['watching'][uuid].get('paused') and is_in_schedule:
# Queue the watch for immediate recheck, with a higher priority
- update_q.put(queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': uuid}))
+ worker_handler.queue_item_async_safe(update_q, queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': uuid}))
# Diff page [edit] link should go back to diff page
if request.args.get("next") and request.args.get("next") == 'diff':
diff --git a/changedetectionio/blueprint/ui/views.py b/changedetectionio/blueprint/ui/views.py
index efcdc03a..7954a197 100644
--- a/changedetectionio/blueprint/ui/views.py
+++ b/changedetectionio/blueprint/ui/views.py
@@ -7,6 +7,7 @@ from copy import deepcopy
from changedetectionio.store import ChangeDetectionStore
from changedetectionio.auth_decorator import login_optionally_required
from changedetectionio import html_tools
+from changedetectionio import worker_handler
def construct_blueprint(datastore: ChangeDetectionStore, update_q, queuedWatchMetaData, watch_check_update):
views_blueprint = Blueprint('ui_views', __name__, template_folder="../ui/templates")
@@ -212,7 +213,7 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, queuedWatchMe
return redirect(url_for('ui.ui_edit.edit_page', uuid=new_uuid, unpause_on_save=1, tag=request.args.get('tag')))
else:
# Straight into the queue.
- update_q.put(queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': new_uuid}))
+ worker_handler.queue_item_async_safe(update_q, queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': new_uuid}))
flash("Watch added.")
return redirect(url_for('watchlist.index', tag=request.args.get('tag','')))
diff --git a/changedetectionio/blueprint/watchlist/__init__.py b/changedetectionio/blueprint/watchlist/__init__.py
index bd3b6c98..8cd5423a 100644
--- a/changedetectionio/blueprint/watchlist/__init__.py
+++ b/changedetectionio/blueprint/watchlist/__init__.py
@@ -78,7 +78,6 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, queuedWatchMe
active_tag=active_tag,
active_tag_uuid=active_tag_uuid,
app_rss_token=datastore.data['settings']['application'].get('rss_access_token'),
- ajax_toggle_url=url_for('ui.ajax.ajax_toggler'),
datastore=datastore,
errored_count=errored_count,
form=form,
diff --git a/changedetectionio/blueprint/watchlist/templates/watch-overview.html b/changedetectionio/blueprint/watchlist/templates/watch-overview.html
index 49fd2bd3..728a204c 100644
--- a/changedetectionio/blueprint/watchlist/templates/watch-overview.html
+++ b/changedetectionio/blueprint/watchlist/templates/watch-overview.html
@@ -1,11 +1,15 @@
-{% extends 'base.html' %}
-{% block content %}
-{% from '_helpers.html' import render_simple_field, render_field, render_nolabel_field, sort_by_title %}
+{%- extends 'base.html' -%}
+{%- block content -%}
+{%- from '_helpers.html' import render_simple_field, render_field, render_nolabel_field, sort_by_title -%}
-
-
+