{"id":8739,"date":"2025-04-08T11:43:39","date_gmt":"2025-04-08T11:43:39","guid":{"rendered":"https:\/\/pariswells.com\/blog\/?p=8739"},"modified":"2025-04-08T11:43:50","modified_gmt":"2025-04-08T11:43:50","slug":"how-to-update-changedetection-api-via-python","status":"publish","type":"post","link":"https:\/\/pariswells.com\/blog\/research\/how-to-update-changedetection-api-via-python","title":{"rendered":"How to update changedetection API via Python"},"content":{"rendered":"\n
import requests\nimport os\nfrom dotenv import load_dotenv\nimport json\nfrom typing import Dict, Any\nimport time\n\n# Load environment variables\nload_dotenv()\n\nclass ChangeDetectionAPI:\n    def __init__(self):\n        self.base_url = os.getenv('CHANGEDETECTION_BASE_URL', 'https:\/\/lemonade.changedetection.io\/XXXXXXXX\/api')\n        self.api_key = os.getenv('CHANGEDETECTION_API_KEY')\n        if not self.api_key:\n            raise ValueError(\"API key not found in environment variables\")\n        \n        print(f\"Using base URL: {self.base_url}\")\n        print(f\"Using API key: {self.api_key[:4]}...{self.api_key[-4:]}\")\n        \n        self.headers = {\n            \"x-api-key\": self.api_key,\n            \"Content-Type\": \"application\/json\"\n        }\n\n    def create_watcher(self, url: str, tag: str, title: str = None) -> Dict[str, Any]:\n        \"\"\"Create a new watcher with retry mechanism\"\"\"\n        endpoint = f\"{self.base_url}\/v1\/watch\"  # Use v1 endpoint as per example\n        print(f\"Creating watcher at endpoint: {endpoint}\")\n        \n        # Simplified payload based on example\n        payload = {\n            \"url\": url,\n            \"tag\": tag\n        }\n        \n        # Add title if provided\n        if title:\n            payload[\"title\"] = title\n\n        print(f\"Request payload: {json.dumps(payload, indent=2)}\")\n        print(f\"Request headers: {json.dumps(self.headers, indent=2)}\")\n\n        max_retries = 3\n        retry_delay = 5  # seconds\n\n        for attempt in range(max_retries):\n            try:\n                response = requests.post(endpoint, json=payload, headers=self.headers)\n                print(f\"Response status: {response.status_code}\")\n                print(f\"Response headers: {json.dumps(dict(response.headers), indent=2)}\")\n                if response.text:\n                    print(f\"Response body: {response.text}\")\n                response.raise_for_status()\n                return response.json()\n            except requests.exceptions.RequestException as e:\n                if attempt == max_retries - 1:\n                    raise Exception(f\"Failed to create watcher after {max_retries} attempts: {str(e)}\")\n                print(f\"Attempt {attempt + 1} failed, retrying in {retry_delay} seconds...\")\n                time.sleep(retry_delay)\n\n    def get_watcher_status(self, watcher_id: str) -> Dict[str, Any]:\n        \"\"\"Get the status of an existing watcher\"\"\"\n        endpoint = f\"{self.base_url}\/v1\/watch\/{watcher_id}\"\n        response = requests.get(endpoint, headers=self.headers)\n        response.raise_for_status()\n        return response.json()\n\n    def delete_watcher(self, watcher_id: str) -> bool:\n        \"\"\"Delete an existing watcher\"\"\"\n        endpoint = f\"{self.base_url}\/v1\/watch\/{watcher_id}\"\n        response = requests.delete(endpoint, headers=self.headers)\n        return response.status_code == 200\n\ndef main():\n    try:\n        # Initialize API client\n        api = ChangeDetectionAPI()\n\n        # Example usage\n        domain_url = \"https:\/\/www.google.com\"\n        \n        # Create watcher with minimal required fields\n        result = api.create_watcher(\n            url=domain_url,\n            tag=\"stock\",\n            title=\"Title\"\n        )\n        \n        print(\"Watcher created successfully:\", json.dumps(result, indent=2))\n        \n        # Get watcher status\n        if 'uuid' in result:\n            status = api.get_watcher_status(result['uuid'])\n            print(\"Watcher status:\", json.dumps(status, indent=2))\n\n    except Exception as e:\n        print(f\"Error: {str(e)}\")\n\nif __name__ == \"__main__\":\n    main()<\/code><\/pre>\n\n\n\n
<\/p>\n","protected":false},"excerpt":{"rendered":"","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-8739","post","type-post","status-publish","format-standard","hentry","category-research"],"aioseo_notices":[],"aioseo_head":"\n\t\t\n\t\n\t\n\t\n\t\n\t\n\t\t\n\t\t\n\t\t\n\t\t\n\t\t\n\t\t\n\t\t\n\t\t\n\t\t\n\t\t\n\t\t\n\t\t