import os import uuid from copy import deepcopy from loguru import logger import time import json from changedetectionio import strtobool from changedetectionio.notification import default_notification_format_for_watch schema = { # Custom notification content # Re #110, so then if this is set to None, we know to use the default value instead # Requires setting to None on submit if it's the same as the default # Should be all None by default, so we use the system default in this case. 'body': None, 'browser_steps': [], 'browser_steps_last_error_step': None, 'check_count': 0, 'check_unique_lines': False, # On change-detected, compare against all history if its something new 'consecutive_filter_failures': 0, # Every time the CSS/xPath filter cannot be located, reset when all is fine. 'content-type': None, 'date_created': None, 'extract_text': [], # Extract text by regex after filters 'extract_title_as_title': False, 'fetch_backend': 'system', # plaintext, playwright etc 'fetch_time': 0.0, 'filter_failure_notification_send': strtobool(os.getenv('FILTER_FAILURE_NOTIFICATION_SEND_DEFAULT', 'True')), 'filter_text_added': True, 'filter_text_removed': True, 'filter_text_replaced': True, 'follow_price_changes': True, 'has_ldjson_price_data': None, 'headers': {}, # Extra headers to send 'ignore_text': [], # List of text to ignore when calculating the comparison checksum 'in_stock_only': True, # Only trigger change on going to instock from out-of-stock 'include_filters': [], 'last_checked': 0, 'last_error': False, 'last_modified': None, 'last_viewed': 0, # history key value of the last viewed via the [diff] link 'method': 'GET', 'notification_alert_count': 0, 'notification_body': None, 'notification_format': default_notification_format_for_watch, 'notification_muted': False, 'notification_screenshot': False, # Include the latest screenshot if available and supported by the apprise URL 'notification_title': None, 'notification_urls': [], # List of URLs to add to the notification Queue (Usually AppRise) 'paused': False, 'previous_md5': False, 'previous_md5_before_filters': False, # Used for skipping changedetection entirely 'processor': 'text_json_diff', # could be restock_diff or others from .processors 'processor_state': {}, # Extra configs for custom processors/plugins, keyed by processor name 'price_change_threshold_percent': None, 'proxy': None, # Preferred proxy connection 'remote_server_reply': None, # From 'server' reply header 'sort_text_alphabetically': False, 'subtractive_selectors': [], 'tag': '', # Old system of text name for a tag, to be removed 'tags': [], # list of UUIDs to App.Tags 'text_should_not_be_present': [], # Text that should not present 'time_between_check': {'weeks': None, 'days': None, 'hours': None, 'minutes': None, 'seconds': None}, 'time_between_check_use_default': True, "time_schedule_limit": { "enabled": False, "monday": { "enabled": True, "start_time": "00:00", "duration": { "hours": "24", "minutes": "00" } }, "tuesday": { "enabled": True, "start_time": "00:00", "duration": { "hours": "24", "minutes": "00" } }, "wednesday": { "enabled": True, "start_time": "00:00", "duration": { "hours": "24", "minutes": "00" } }, "thursday": { "enabled": True, "start_time": "00:00", "duration": { "hours": "24", "minutes": "00" } }, "friday": { "enabled": True, "start_time": "00:00", "duration": { "hours": "24", "minutes": "00" } }, "saturday": { "enabled": True, "start_time": "00:00", "duration": { "hours": "24", "minutes": "00" } }, "sunday": { "enabled": True, "start_time": "00:00", "duration": { "hours": "24", "minutes": "00" } }, }, 'title': None, 'track_ldjson_price_data': None, 'trim_text_whitespace': False, 'remove_duplicate_lines': False, 'trigger_text': [], # List of text or regex to wait for until a change is detected 'url': '', 'uuid': None, 'webdriver_delay': None, 'webdriver_js_execute_code': None, # Run before change-detection } class watch_base(dict): __data = {} __datastore_path = None __save_enabled = True def __init__(self, *arg, **kw): # Initialize internal data storage self.__data = deepcopy(schema) self.__datastore_path = kw.pop('datastore_path', None) # Initialize as empty dict but maintain dict interface super(watch_base, self).__init__() # Update with provided data if arg or kw: self.update(*arg, **kw) # Generate UUID if needed if not self.__data.get('uuid'): self.__data['uuid'] = str(uuid.uuid4()) if self.__data.get('default'): del(self.__data['default']) @property def watch_data_dir(self): # The base dir of the watch data return os.path.join(self.__datastore_path, self['uuid']) if self.__datastore_path else None def enable_saving(self): self.__save_enabled = True # Dictionary interface methods to use self.__data def __getitem__(self, key): return self.__data[key] def __setitem__(self, key, value): self.__data[key] = value self.__data['last_modified'] = time.time() def __delitem__(self, key): del self.__data[key] def __contains__(self, key): return key in self.__data def __iter__(self): return iter(self.__data) def __len__(self): return len(self.__data) def get(self, key, default=None): return self.__data.get(key, default) def update(self, *args, **kwargs): if args: if len(args) > 1: raise TypeError("update expected at most 1 arguments, got %d" % len(args)) other = dict(args[0]) for key in other: self.__data[key] = other[key] for key in kwargs: self.__data[key] = kwargs[key] self.__data['last_modified'] = time.time() def items(self): return self.__data.items() def keys(self): return self.__data.keys() def values(self): return self.__data.values() def pop(self, key, default=None): return self.__data.pop(key, default) def popitem(self): return self.__data.popitem() def clear(self): self.__data.clear() self.__data['last_modified'] = time.time() def get_data(self): """Returns the internal data dictionary""" return self.__data def save_data(self): if self.__save_enabled: if not self.__data.get('uuid'): # Might have been called when creating the watch return logger.debug(f"Saving watch {self['uuid']}") path = os.path.join(self.__datastore_path, self.get('uuid')) filepath = os.path.join(str(path), "watch.json") if not os.path.exists(path): os.mkdir(path) try: import tempfile with tempfile.NamedTemporaryFile(mode='wb+', delete=False) as tmp: tmp.write(json.dumps(self.get_data(), indent=2).encode('utf-8')) tmp.flush() os.replace(tmp.name, filepath) except Exception as e: logger.error(f"Error writing JSON for {self.get('uuid')}!! (JSON file save was skipped) : {str(e)}")