From 9ccb76a500ea0c6d272c09f1ded62454dac033a2 Mon Sep 17 00:00:00 2001 From: Monitor maintainers Date: Mon, 2 Nov 2026 12:01:25 +0000 Subject: [PATCH] Publish generic monitor source and installation guide --- .env.example | 6 + .github/workflows/test.yml | 24 + .gitignore | 30 + LICENSE | 21 + Makefile | 25 + README.md | 424 ++++ SECURITY.md | 11 + compose.browser.yaml | 19 + compose.yaml | 15 + config.example.json | 79 + deploy/caddy/changedetection.caddy.example | 14 + ignore.example.json | 18 + monitor.py | 2261 +++++++++++++++++ notify_failure.py | 47 + requirements-verification.txt | 1 + scripts/build-release.sh | 30 + scripts/doctor.py | 228 ++ scripts/install-systemd.sh | 63 + scripts/render_systemd.py | 73 + scripts/setup.sh | 43 + scripts/test.sh | 20 + scripts/uninstall-systemd.sh | 25 + secrets.env.example | 4 + systemd/competitor-monitor-collect.service.in | 23 + systemd/competitor-monitor-collect.timer | 13 + systemd/competitor-monitor-digest.service.in | 27 + systemd/competitor-monitor-digest.timer | 12 + .../competitor-monitor-discover.service.in | 23 + systemd/competitor-monitor-discover.timer | 13 + .../competitor-monitor-failure@.service.in | 21 + test_monitor.py | 1303 ++++++++++ tests/test_repository.py | 237 ++ verification/audit_watch_scope.py | 153 ++ verification/e2e_verify.py | 249 ++ verification/verify_persistence.py | 166 ++ verification/verify_representative_scope.py | 170 ++ verification/verify_ui.py | 132 + 37 files changed, 6023 insertions(+) create mode 100644 .env.example create mode 100644 .github/workflows/test.yml create mode 100644 .gitignore create mode 100644 LICENSE create mode 100644 Makefile create mode 100644 README.md create mode 100644 SECURITY.md create mode 100644 compose.browser.yaml create mode 100644 compose.yaml create mode 100644 config.example.json create mode 100644 deploy/caddy/changedetection.caddy.example create mode 100644 ignore.example.json create mode 100755 monitor.py create mode 100755 notify_failure.py create mode 100644 requirements-verification.txt create mode 100755 scripts/build-release.sh create mode 100755 scripts/doctor.py create mode 100755 scripts/install-systemd.sh create mode 100755 scripts/render_systemd.py create mode 100755 scripts/setup.sh create mode 100755 scripts/test.sh create mode 100755 scripts/uninstall-systemd.sh create mode 100644 secrets.env.example create mode 100644 systemd/competitor-monitor-collect.service.in create mode 100644 systemd/competitor-monitor-collect.timer create mode 100644 systemd/competitor-monitor-digest.service.in create mode 100644 systemd/competitor-monitor-digest.timer create mode 100644 systemd/competitor-monitor-discover.service.in create mode 100644 systemd/competitor-monitor-discover.timer create mode 100644 systemd/competitor-monitor-failure@.service.in create mode 100755 test_monitor.py create mode 100755 tests/test_repository.py create mode 100755 verification/audit_watch_scope.py create mode 100755 verification/e2e_verify.py create mode 100755 verification/verify_persistence.py create mode 100755 verification/verify_representative_scope.py create mode 100755 verification/verify_ui.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..eb0a636 --- /dev/null +++ b/.env.example @@ -0,0 +1,6 @@ +# Docker Compose settings. Copy to .env and edit. +TZ=UTC +PUBLIC_BASE_URL=http://localhost:5000 +CHANGEDETECTION_BIND=127.0.0.1 +CHANGEDETECTION_PORT=5000 +CHANGEDETECTION_DATA_DIR=./data/changedetection diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml new file mode 100644 index 0000000..77dfeb2 --- /dev/null +++ b/.github/workflows/test.yml @@ -0,0 +1,24 @@ +name: test + +on: + push: + pull_request: + +permissions: + contents: read + +jobs: + test: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.11", "3.12"] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - name: Verify Docker Compose + run: docker compose version + - name: Run complete repository gate + run: ./scripts/test.sh diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..4334dcf --- /dev/null +++ b/.gitignore @@ -0,0 +1,30 @@ +# Runtime and generated state +.env +secrets.env +config.json +ignore.json +state/ +data/ +build/ +verification/*.json +verification/*.txt +verification/*.headers +verification/*.png +verification/e2e-state-*/ + +# Python/editor artifacts +__pycache__/ +*.py[cod] +.pytest_cache/ +.coverage +.venv/ +venv/ +.idea/ +.vscode/ +.DS_Store + +# Archives and local release outputs +release/ +*.tar.gz +*.zip +*.sha256 diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..574c533 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Kristjan Variksoo + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..b81704f --- /dev/null +++ b/Makefile @@ -0,0 +1,25 @@ +.PHONY: init up up-browser down logs test doctor render-systemd + +init: + ./scripts/setup.sh --no-start + +up: + docker compose up -d + +up-browser: + docker compose -f compose.yaml -f compose.browser.yaml up -d + +down: + docker compose -f compose.yaml -f compose.browser.yaml down --remove-orphans + +logs: + docker compose logs --follow --tail=100 + +test: + ./scripts/test.sh + +doctor: + python3 scripts/doctor.py --mode live + +render-systemd: + python3 scripts/render_systemd.py diff --git a/README.md b/README.md new file mode 100644 index 0000000..707bf24 --- /dev/null +++ b/README.md @@ -0,0 +1,424 @@ +# ChangeDetection competitor-intelligence monitor + +A bounded discovery and intelligence layer for [ChangeDetection.io](https://github.com/dgtlmoon/changedetection.io). It discovers newly published competitor pages, keeps a curated representative watch set, classifies real content changes, and sends immediate or digest notifications to Discord. + +The current example profile uses the reserved `.example` namespace; replace it with your own private `config.json` and `ignore.json` before discovery. Source code and templates are separate from your private targets, credentials, and monitoring history. A generic example at the current revision is not a guarantee that older Git history or test fixtures are generic; see **Publishing safely**. + +## What this repository separates + +These are independent mechanisms and clocks: + +1. **Discovery** — sitemaps, RSS/Atom, configured Certificate Transparency sources, and bounded same-domain links find URLs. +2. **Watch reconciliation** — useful representatives are imported, tagged, paused, or reactivated through the ChangeDetection API. +3. **Page fetching** — ChangeDetection checks existing watches on its own schedule. +4. **Event collection** — this monitor polls retained ChangeDetection history, classifies changes, and delivers notifications. +5. **Snapshot retention** — ChangeDetection history records distinct content versions, not every successful unchanged check. + +A gap between two snapshot timestamps is therefore **not** a check interval or detection delay. Timing claims require raw successful-fetch telemetry identifying the last unchanged check and first changed check. + +## Main behavior + +- recursive sitemap and feed discovery; +- bounded same-domain crawl with canonicalization and hard page/depth/size ceilings; +- optional Certificate Transparency discovery with bounded retries and silent first baseline; +- representative active selection, normally no more than two pages per host and semantic family; +- durable exact/regex exclusions; +- pause-not-delete reconciliation so historical snapshots survive deselection, including pages paused after their removal is confirmed by the configured number of complete inventories; +- exclusion of individual company/person/domain/IP records, routine public-record churn, aggregate lists/statistics, archives, tests, discovery documents, duplicate languages/aliases, and static assets; +- anti-bot challenge detection represented as a coverage gap, never as “no changes”; +- semantic realignment of positional text diffs, so one inserted or removed repeated card is reported as one change rather than a cascade of false replacements, with bounded excerpts that always retain evidence from both old and new sides; +- classification of a broad Estonian-to-English localization with unchanged numeric/commercial values as minor localization rather than as a pricing or policy alert; actual value changes remain important; +- suppression of pure nested navigation-list changes, navigation moves, rendered icon labels, and identical minor fragments repeated across three or more pages; +- source and Certificate Transparency health retained in status/state rather than mixed into competitor-intelligence digests; +- durable immediate and digest outboxes with Discord rate-limit retry; +- global command locking and systemd failure alerts; +- explicit timestamp semantics in event records, plus the last successful unchanged check and first successful changed check captured from consecutive collector observations whenever available. + +The supplied generic template deliberately sets `respect_robots` to `false`. Crawling remains limited by same-domain allowlists, one-second pacing, response timeouts, and hard sitemap/page/depth ceilings. Change that policy in your own deployment if required. + +## Repository layout + +```text +monitor.py discovery, reconciliation, collection, digest +notify_failure.py throttled systemd failure notification +test_monitor.py 45 deterministic engine regressions +config.example.json fictional `.example` scope template +ignore.example.json durable exclusion template +compose.yaml digest-pinned ChangeDetection base stack +compose.browser.yaml optional digest-pinned JS browser override +systemd/ portable unit templates and timers +scripts/setup.sh initialize a checkout and optionally start Compose +scripts/doctor.py repository/config/live readiness checks +scripts/render_systemd.py render host paths and runtime account into units +scripts/install-systemd.sh install and verify timers +scripts/uninstall-systemd.sh remove units but preserve data +verification/ live E2E, scope, persistence, and optional UI checks +deploy/caddy/ optional public reverse-proxy example +``` + +Runtime configuration, credentials, queues, discovery state, snapshots, and generated evidence are intentionally excluded from Git. + +## Requirements + +- Linux with Bash and Python 3.11 or newer (monitor runtime uses only the standard library); systemd is needed only for the supplied recurring timers; +- Git, CA certificates, Docker Engine with Compose v2, and permission to use that Docker daemon; +- Linux `amd64` or `arm64` for the pinned ChangeDetection image and optional browser image; +- a Discord bot that can view and send messages in your own destination channel/thread; +- optional: Caddy for public HTTPS; +- optional UI verifier: a Chrome/Chromium CDP session and `websockets` from `requirements-verification.txt`. + +### Linux / WSL on a work computer + +Use a machine and targets approved by your organization's policy. This is a **new local deployment**, not access to anyone else's existing monitoring installation. A clone is not a runtime backup: you must supply your own private configuration and credentials. Keep the default UI loopback-only; public HTTPS is optional, not a prerequisite. + +For Ubuntu 24.04+ or a comparable Debian-family distribution with Python 3.11+: + +```bash +sudo apt update +sudo apt install -y git python3 python3-venv ca-certificates +``` + +Install Docker Engine and its **Compose plugin** using the [official Linux instructions](https://docs.docker.com/engine/install/). On Windows, use **WSL 2 with an Ubuntu distribution**, run these commands in its Bash terminal, and either enable that distribution in [Docker Desktop's WSL integration](https://docs.docker.com/desktop/features/wsl/) or install a Docker Engine inside the distribution. Do not configure two competing daemons. Native PowerShell and Git Bash are not supported runtimes for these scripts. Store the checkout in the WSL Linux filesystem, such as `~/src`, not a Windows-mounted directory, so Linux permissions work as expected. + +Check prerequisites before cloning or running setup: + +```bash +python3 --version # must be 3.11 or newer +bash --version +git --version +docker compose version # Compose v2, not docker-compose v1 +docker info >/dev/null # must succeed as your normal runtime user +``` + +The scripts call `docker` without `sudo`. If daemon access fails, resolve it with your administrator; do not run the entire setup as root. Docker-group membership grants root-equivalent host access and should not be added casually on a managed computer. No runtime `pip install` is required. + +For recurring timers, check `systemctl is-system-running` and confirm systemd is actually running (a `degraded` result needs investigation; merely having `systemctl` installed is insufficient). For WSL, follow [Microsoft's systemd instructions](https://learn.microsoft.com/en-us/windows/wsl/systemd) if needed; otherwise skip timer installation and run commands manually. Monitoring does not continue while the computer is asleep, shut down, or WSL/Docker is stopped. Use an always-on approved Linux host if continuous monitoring is required. + +The base installation does not run Sockpuppet or add elevated container capabilities. The optional browser override adds `SYS_ADMIN`, which its Chromium sandbox needs. It is not published on a host port, but this remains an elevated capability: review it against the destination host's security policy before enabling the override. + +## Fresh installation + +### 1. Clone and initialize + +Use the repository's advertised **Git clone URL**, not merely the web UI's page URL. Replace the prompt value with the URL supplied by the publisher; no particular hostname or remote hosting service is assumed. + +```bash +mkdir -p "$HOME/src" +cd "$HOME/src" +read -r -p 'Git clone URL: ' REPOSITORY_URL +git clone -- "$REPOSITORY_URL" changedetection-competitor-monitor +cd changedetection-competitor-monitor +git remote get-url origin +./scripts/test.sh +umask 077 +./scripts/setup.sh --no-start +``` + +The local test gate should end with `All repository gates passed.` It does not contact competitor sites or deliver Discord messages. `setup.sh --no-start` requires the Docker CLI/Compose plugin for model validation but does not start containers. Use one deployment per host unless you deliberately isolate both the Compose project and systemd unit names: the defaults use fixed names and a second checkout is not an independent installation. + +`setup.sh` creates, without overwriting existing files: + +- `.env` from `.env.example`; +- `config.json` from `config.example.json`; +- `ignore.json` from `ignore.example.json`; +- `secrets.env` from `secrets.env.example` with mode `0600`; +- local `state/` and `data/changedetection/` directories. + +These local copies are ignored by Git. Edit **the copies**, not the tracked `*.example` templates. `.env` configures Compose; `secrets.env` supplies the monitor's API/bot credentials; `config.json` and `ignore.json` contain your private monitoring scope. Do not reuse a production configuration unless you intentionally want that exact private scope and have authorization. + +Review `.env` in your editor before starting. Retain `CHANGEDETECTION_BIND=127.0.0.1` and `PUBLIC_BASE_URL=http://localhost:5000` for local use. If port 5000 is occupied, choose a free `CHANGEDETECTION_PORT`, then update both `PUBLIC_BASE_URL` and `config.json`'s `changedetection_base` to match. A custom `CHANGEDETECTION_DATA_DIR` must point to private persistent storage. + +### 2. Start and configure ChangeDetection + +```bash +docker compose up -d +docker compose ps +``` + +Open `http://localhost:5000` in your browser (or your configured `PUBLIC_BASE_URL`). In WSL with normal localhost forwarding, use the same URL from the Windows browser. If it fails, first check `docker compose ps` and `docker compose logs --tail=100 changedetection`; do not expose the port publicly as a workaround. In ChangeDetection: + +1. set its native application password; +2. open **Settings → API**; +3. create/copy an API token. + +Compose's startup ordering does not itself prove API readiness; the live doctor in step 4 is the readiness gate before timers are installed. The datastore contains watches, retained snapshots, and native-login/API material; it is private even if its container is stopped. + +The pinned ChangeDetection `0.60.4` release uses a native password-only login; there is no application username. The optional Caddy example intentionally adds no proxy Basic Auth. + +### 3. Configure monitor targets and secrets + +Edit `config.json`: + +- replace `REPLACE_WITH_DISCORD_CHANNEL_OR_THREAD_ID`; +- replace `REPLACE_WITH_DISCORD_USER_ID`; +- review competitor domains, seeds, exclusions, crawl ceilings, and robots policy; +- change the user-agent contact URL. + +The shipped target is intentionally fictional and will not resolve. Add at least one real private target before running discovery. + +Edit `secrets.env`: + +```dotenv +CD_API_TOKEN=REPLACE_WITH_CHANGEDETECTION_API_TOKEN +DISCORD_BOT_TOKEN=REPLACE_WITH_DISCORD_BOT_TOKEN +``` + +Protect all four private configuration files (the secrets file must remain mode `0600`): + +```bash +chmod 600 .env config.json ignore.json secrets.env +python3 scripts/doctor.py --mode configured +``` + +Do not paste tokens into tracked examples, shell commands, screenshots, or issue reports. Check `git status --short` before committing source changes; ignored files are not encrypted or automatically backed up. + +### 4. Prove live prerequisites + +```bash +python3 scripts/doctor.py --mode live +python3 scripts/doctor.py --mode live --verify-discord-send +python3 monitor.py status +python3 monitor.py discover --no-send +python3 monitor.py collect --no-send +``` + +The normal live doctor checks the exact ChangeDetection API token, Discord bot identity, and destination channel/thread without sending a message. The second command explicitly creates, reads back, deletes, and confirms the absence of one probe message to prove the bot can actually deliver there without leaving test noise. The systemd installer repeats that probe as the selected runtime account before installing any units. The first discovery inventory is a silent baseline. + +`--no-send` suppresses Discord-client creation: discovery and collection still need the ChangeDetection API token because they read or reconcile live watches, while `digest --no-send` needs neither API nor Discord credentials. It is not a dry run—discovery can still create, pause, or reactivate watches. + +### 5. Install recurring timers + +```bash +sudo ./scripts/install-systemd.sh --user "$USER" +``` + +Installed schedule: + +- discovery/reconciliation: every 6 hours; +- event collection: every 10 minutes; +- daily digest: 08:30 UTC. + +These timer files are deployment-policy defaults, not engine requirements. Edit the three files under `systemd/` before installation if another cadence or timezone is appropriate. + +Verify: + +```bash +systemctl list-timers 'competitor-monitor*' +systemctl status competitor-monitor-discover.timer competitor-monitor-collect.timer competitor-monitor-digest.timer +``` + +Remove the units without deleting the repository, state, or ChangeDetection datastore: + +```bash +sudo ./scripts/uninstall-systemd.sh +``` + +## Optional browser rendering with Sockpuppet + +Sockpuppet is **not required** for the default `html_requests` fetch backend. Enable it only for watches that require JavaScript rendering, Chromium screenshots, or browser-based selectors: + +```bash +docker compose -f compose.yaml -f compose.browser.yaml up -d +``` + +Then use `"fetch_backend": "html_webdriver"` for only the relevant competitor profile or watch. To return to HTTP-only fetching: + +```bash +docker compose -f compose.yaml -f compose.browser.yaml down --remove-orphans +docker compose up -d +``` + +## Optional public HTTPS with Caddy + +Copy `deploy/caddy/changedetection.caddy.example` into your Caddy site directory, replace `changedetection.example.com`, validate, and reload Caddy. Keep the Compose port bound to loopback. + +```bash +sudo caddy validate --config /etc/caddy/Caddyfile --adapter caddyfile +sudo systemctl reload caddy +``` + +Use ChangeDetection’s native login. Do not add a second Basic Auth layer unless you intentionally want double authentication. + +## Day-to-day commands + +```bash +python3 monitor.py discover # discover, reconcile, and notify +python3 monitor.py discover --no-send # reconcile without Discord delivery +python3 monitor.py collect # classify retained history and notify +python3 monitor.py digest # deliver queued minor topics/gaps +python3 monitor.py status # current watch and run summary +``` + +All paths are overrideable: + +```bash +python3 monitor.py status \ + --config /path/config.json \ + --ignore /path/ignore.json \ + --state-dir /var/lib/competitor-monitor \ + --env-file /run/secrets/competitor-monitor.env +``` + +## Verification + +Run the local, secret-free gate: + +```bash +./scripts/test.sh +``` + +It runs the engine suite, repository/secret-hygiene tests, Python compilation, shell syntax checks, Compose validation, portable systemd rendering, and `systemd-analyze verify` when available. + +Live acceptance tools intentionally require a configured local deployment: + +```bash +python3 verification/e2e_verify.py +python3 verification/verify_representative_scope.py +python3 verification/audit_watch_scope.py +python3 verification/verify_persistence.py before +docker compose restart +python3 verification/verify_persistence.py after +``` + +`e2e_verify.py` creates reversible test watches, proves before/after diffs plus immediate/digest/silent notification paths, reads deliveries back, and removes its fixtures. It writes generated reports under ignored `verification/` paths. + +The optional UI check requires a separately started disposable Chrome CDP session: + +```bash +python3 -m venv .venv +.venv/bin/pip install -r requirements-verification.txt +CD_UI_PASSWORD='your native password' \ + .venv/bin/python verification/verify_ui.py \ + --url https://changedetection.example.com/ \ + --cdp-url http://127.0.0.1:9222 +``` + +## Updating an existing clone + +Run from the same checkout used for installation, as its normal owner. `git pull` updates **tracked source only**; it does not import someone else's targets/history or replace your ignored private files. It also does not refresh an existing local configuration when example schemas change. + +### 1. Inspect and pause this installation + +```bash +cd "$HOME/src/changedetection-competitor-monitor" +git remote get-url origin +git status --short +git branch --show-current +``` + +Expect branch `main` and no tracked edits. Preserve/review any local edits first; do not use `git reset --hard` or `git clean` to force an update. If `origin` is missing, set it to the publisher's advertised Git clone URL, not a credentials-bearing URL. + +If you installed this checkout's systemd timers, pause **both the timers and any in-flight jobs** before updating. Skip this block for manual-only installations. The fixed unit names below must belong to this installation; do not stop units belonging to another checkout. + +```bash +sudo systemctl stop competitor-monitor-discover.timer competitor-monitor-collect.timer competitor-monitor-digest.timer +sudo systemctl stop competitor-monitor-discover.service competitor-monitor-collect.service competitor-monitor-digest.service +``` + +Stop containers using the same file set used to start them (choose one): + +```bash +# Default HTTP-only deployment: +docker compose down --remove-orphans +# If the optional browser override was enabled, use this instead: +# docker compose -f compose.yaml -f compose.browser.yaml down --remove-orphans +``` + +`down` preserves the bind-mounted datastore. Before pulling, make a private backup of `.env`, `config.json`, `ignore.json`, `secrets.env`, `state/`, and the **actual** `CHANGEDETECTION_DATA_DIR` while writers are stopped. Keep it outside the checkout with restricted permissions; include custom/external datastore paths and preserve ownership/modes. Git is not that backup. Do not use `down -v` or delete any data directory as an update step. + +### 2. Pull and validate + +```bash +git pull --ff-only origin main +./scripts/test.sh +./scripts/setup.sh --no-start +``` + +If Git reports divergence, conflicts, or a failed pull, stop and resolve it without discarding local work. If tests fail, leave timers paused and inspect the failure. `setup.sh --no-start` is safe to repeat: it validates configuration/Compose and creates missing files but does **not** overwrite existing configuration. Review the updated README and `git log -5 --oneline`; compare changed example settings with your private files and apply any necessary schema changes manually. + +### 3. Restart and verify + +```bash +# Default deployment (use the browser file set instead if previously enabled): +docker compose up -d +python3 scripts/doctor.py --mode live +python3 monitor.py status +``` + +Require the doctor to return `"ok": true` and status to match your own intended watch inventory. For a timed installation, reinstall the updated units at this same path, then verify all three timers are enabled/active: + +```bash +sudo ./scripts/install-systemd.sh --user "$USER" +systemctl list-timers 'competitor-monitor*' +systemctl is-active competitor-monitor-discover.timer competitor-monitor-collect.timer competitor-monitor-digest.timer +``` + +The installer sends, reads back, and deletes one Discord probe. Manual-only users skip it; the next deliberate monitor command is their next run. If recovery requires older application/data versions, restore the matching private backup as well as source—checking out an older commit alone is not a datastore rollback. + +## Stop or remove this installation + +**Pause monitoring only:** use the two `systemctl stop` commands above if timers were installed. These stops are temporary; enabled timers can return after a reboot. For a persistent pause, disable all three timers with `sudo systemctl disable --now competitor-monitor-discover.timer competitor-monitor-collect.timer competitor-monitor-digest.timer`, then stop the three services. ChangeDetection keeps fetching existing watches until its Compose stack is also stopped. + +**Stop page fetching/UI too:** run `docker compose down --remove-orphans` from this checkout, or the equivalent browser-override `down` command above. This removes this Compose project's containers/network, not the bind-mounted datastore, private configuration, monitor ledger, or historical snapshots. Do not stop the entire Docker daemon. + +**Remove scheduled integration:** first stop the timers and services as above, then run: + +```bash +sudo ./scripts/uninstall-systemd.sh +systemctl list-timers 'competitor-monitor*' +``` + +The remover disables/removes the fixed systemd unit files and reloads systemd. It does not stop an already-running monitor service for you, remove Compose containers/images, revoke tokens, remove an optional Caddy site, or erase source/config/state/datastore. Use the explicit service-stop and Compose commands before deleting or moving the checkout. If Caddy was separately configured, remove only that deployment's site and validate/reload Caddy separately. + +**Erase private data:** this is a separate, irreversible operator decision, not part of uninstall. After stopping every writer, retain any required private backup, identify the actual datastore path from `.env`, and deliberately remove only this installation's files. Deleting `state/` loses cursors/queues and can cause re-baselining or repeated notifications; deleting the datastore loses watches, snapshots, login, and API configuration. Revoke unused API/bot credentials with their providers separately. Never use `git clean -fdx`, a broad wildcard, or an unreviewed recursive deletion to uninstall. + +## Moving to another system + +There are two distinct paths: + +### Fresh baseline + +Clone the repository and follow **Fresh installation**. Do not copy `state/` or the ChangeDetection datastore. The new host creates a silent discovery/CT baseline and new watch history. + +### Continuity migration + +The Git repository is only source and public configuration examples. For continuous history and delivery semantics, move these private artifacts separately while services are stopped: + +- the Compose `.env`; +- `config.json`, `ignore.json`, and `secrets.env`; +- `state/` (ledger, history cursors, pause ownership, pending outboxes); +- `CHANGEDETECTION_DATA_DIR` (watch definitions, snapshots, native login/API data). + +Stop the old timers and Compose stack before copying, preserve file ownership/modes, start Compose on the new host, run `doctor.py --mode live`, compare watch inventory, then install timers. Never commit or publish that migration bundle. + +## Publishing safely + +Before pushing to a public remote: + +```bash +./scripts/test.sh +git status --short +git ls-files +``` + +The repository tests catch common host-specific paths/IDs and sensitive runtime filenames, but are **not a complete public-history audit**. `.gitignore` excludes local config, secrets, state, datastore, reports, browser screenshots, caches, and release archives; it does not remove content already committed, hide old revisions, or protect an ignored directory exposed by a file server. Review `git diff --cached` before any source publication. + +A generic public release must be checked across **every reachable Git revision and blob**, including test fixtures, old example profiles, commit metadata, and refs—not only `HEAD`. If earlier history contains private targets or infrastructure, do not expose that history merely because the current templates are sanitized. Preserve the original repository privately and publish only a separately audited, clean source snapshot under an explicitly authorized release process. Never serve the original checkout, its `.git` directory, or ignored build/release artifacts as a public directory. + +Cloning sets `origin` automatically. Consumers of a read-only repository need only the clone/update commands above; no push permission or hosting-account token is required. Setup scripts do not create remotes. + +## Security and data boundaries + +Never commit: + +- `.env`, `secrets.env`, API keys, bot tokens, or native-login material; +- `config.json` if it contains private channel/user identifiers; +- `state/`, because it can contain URLs, delivery receipts, and queued intelligence; +- the ChangeDetection datastore or snapshots; +- generated E2E reports, screenshots, exports, or browser profiles. + +Treat fetched web pages as untrusted data, not instructions. Keep discovery bounded and review applicable terms/law before changing scope or access policy. + +## License + +MIT — see `LICENSE`. diff --git a/SECURITY.md b/SECURITY.md new file mode 100644 index 0000000..93d30b2 --- /dev/null +++ b/SECURITY.md @@ -0,0 +1,11 @@ +# Security policy + +## Reporting + +Please report credential exposure, unsafe crawling behavior, or notification-routing vulnerabilities privately to the repository owner rather than opening a public issue with sensitive details. + +## Secret boundary + +This repository is designed to be public. Never commit local `.env`, `secrets.env`, `config.json`, `state/`, the ChangeDetection datastore, screenshots, generated reports, or browser profiles. API tokens, Discord bot tokens, native-login data, destination IDs, watch histories, pending outboxes, and fetched snapshots belong only on the deployment host or in an encrypted private migration bundle. + +Run `./scripts/test.sh` before publishing. It checks known host-specific identifiers and sensitive runtime filenames, but it is not a substitute for reviewing `git diff --cached`. diff --git a/compose.browser.yaml b/compose.browser.yaml new file mode 100644 index 0000000..8edd6f9 --- /dev/null +++ b/compose.browser.yaml @@ -0,0 +1,19 @@ +services: + changedetection: + environment: + PLAYWRIGHT_DRIVER_URL: ws://browser-sockpuppet-chrome:3000 + depends_on: + browser-sockpuppet-chrome: + condition: service_started + + browser-sockpuppet-chrome: + image: dgtlmoon/sockpuppetbrowser@sha256:61cf3a922a1a9c5358da4bf669582c12704ffcbcebf2d17ec17c82df8978a0ec + cap_add: + - SYS_ADMIN + shm_size: 2gb + environment: + SCREEN_WIDTH: "1920" + SCREEN_HEIGHT: "1080" + SCREEN_DEPTH: "24" + MAX_CONCURRENT_CHROME_PROCESSES: "10" + restart: unless-stopped diff --git a/compose.yaml b/compose.yaml new file mode 100644 index 0000000..ba16fba --- /dev/null +++ b/compose.yaml @@ -0,0 +1,15 @@ +name: changedetection-competitor-monitor + +services: + changedetection: + image: ghcr.io/dgtlmoon/changedetection.io@sha256:7bb6963b730df6a1070f7a08a607b0eb7a87fe65abfa4742d876ad33bb1267eb + ports: + - "${CHANGEDETECTION_BIND:-127.0.0.1}:${CHANGEDETECTION_PORT:-5000}:5000" + volumes: + - "${CHANGEDETECTION_DATA_DIR:-./data/changedetection}:/datastore" + environment: + TZ: "${TZ:-UTC}" + BASE_URL: "${PUBLIC_BASE_URL:-http://localhost:5000}" + USE_X_SETTINGS: "1" + HIDE_REFERER: "true" + restart: unless-stopped diff --git a/config.example.json b/config.example.json new file mode 100644 index 0000000..39f1747 --- /dev/null +++ b/config.example.json @@ -0,0 +1,79 @@ +{ + "version": 1, + "user_agent": "CompetitorMonitor/1.0 (+https://contact.example/monitor)", + "changedetection_base": "http://127.0.0.1:5000/api/v1", + "discord": { + "thread_id": "REPLACE_WITH_DISCORD_CHANNEL_OR_THREAD_ID", + "important_mention_user_id": "REPLACE_WITH_DISCORD_USER_ID" + }, + "global": { + "max_sitemaps": 40, + "max_urls": 400, + "crawl_depth": 2, + "crawl_pages": 60, + "request_delay_seconds": 1.0, + "watch_check_hours": 6, + "missing_confirmations": 2, + "fetch_timeout_seconds": 25, + "certificate_transparency_attempts": 4, + "certificate_transparency_retry_backoff_seconds": 2, + "max_response_bytes": 3000000, + "watch_selection": { + "enabled": true, + "per_group": 2, + "exclude_categories": ["discovery_source", "routine_data", "archive", "company"] + }, + "exclude_extensions": [ + ".7z", ".avi", ".avif", ".bin", ".bmp", ".css", ".csv", ".doc", ".docx", ".eot", ".exe", ".gif", ".gz", ".ico", ".jpeg", ".jpg", ".js", ".map", ".mov", ".mp3", ".mp4", ".mpeg", ".ogg", ".otf", ".pdf", ".png", ".ppt", ".pptx", ".rar", ".svg", ".tar", ".tgz", ".tif", ".tiff", ".ttf", ".wav", ".webm", ".webp", ".woff", ".woff2", ".xls", ".xlsx", ".zip" + ], + "exclude_regexes": [ + "/(?:wp-admin|wp-login(?:\\.php)?|wp-json)(?:/|$)", + "/(?:login|logout|signin|signup|register|account|cart|checkout)(?:/|$)", + "/search(?:/|$)", + "/(?:ajax|graphql|cdn-cgi)(?:/|$)", + "/page/[0-9]+(?:/|$)", + "/author/[^/]+(?:/|$)", + "/tag/[^/]+/page/[0-9]+(?:/|$)", + "/(?:hello-world|sample-page|thank-you)(?:/|$)", + "/category/uncategorized(?:/|$)", + "/comments(?:/|$)", + "/.+/feed$", + "/index\\.php$", + "/(?:company|companies|person|persons|profile|profiles)/[^/]*[0-9][^/]*(?:/|$)" + ] + }, + "competitors": [ + { + "slug": "example-vendor", + "name": "Example Vendor", + "domains": [ + "competitor.example", + "www.competitor.example", + "docs.competitor.example" + ], + "seeds": [ + "https://competitor.example/", + "https://docs.competitor.example/" + ], + "sitemap_seeds": [ + "https://competitor.example/sitemap.xml" + ], + "allowed_path_prefixes": ["/"], + "exclude_regexes": [ + "/(?:wp-content|wp-includes)/", + "/(?:company|companies|person|persons|profile|profiles)/[^/]*[0-9][^/]*(?:/|$)", + "/sitemaps/(?:companies|people)-" + ], + "exclude_sitemap_regexes": [ + "/sitemaps/(?:companies|people)-" + ], + "max_urls": 300, + "crawl_pages": 60, + "fetch_backend": "html_requests", + "processor": "text_json_diff", + "respect_robots": false, + "certificate_transparency_sources": [], + "policy_watch_urls": [] + } + ] +} diff --git a/deploy/caddy/changedetection.caddy.example b/deploy/caddy/changedetection.caddy.example new file mode 100644 index 0000000..f8d162e --- /dev/null +++ b/deploy/caddy/changedetection.caddy.example @@ -0,0 +1,14 @@ +# Copy this file into your Caddy sites directory and replace the hostname. +# ChangeDetection.io provides the application login; no proxy Basic Auth is used. +changedetection.example.com { + route { + header { + Strict-Transport-Security "max-age=31536000" + X-Content-Type-Options nosniff + Referrer-Policy no-referrer + X-Robots-Tag "noindex, nofollow" + } + + reverse_proxy 127.0.0.1:5000 + } +} diff --git a/ignore.example.json b/ignore.example.json new file mode 100644 index 0000000..b85d2f5 --- /dev/null +++ b/ignore.example.json @@ -0,0 +1,18 @@ +{ + "version": 1, + "description": "Persistent operator-controlled exclusions. Add canonical exact URLs or case-insensitive regular expressions; matching existing watches are paused, not deleted.", + "exact_urls": [], + "regexes": [], + "competitors": { + "example-vendor": { + "exact_urls": [ + "https://competitor.example/legacy-page" + ], + "regexes": [ + "^http://", + "/(?:archive|category|tag)(?:/|$)", + "/(?:company|companies|person|persons|profile|profiles)/[^/]*[0-9][^/]*(?:/|$)" + ] + } + } +} diff --git a/monitor.py b/monitor.py new file mode 100755 index 0000000..7b509ef --- /dev/null +++ b/monitor.py @@ -0,0 +1,2261 @@ +#!/usr/bin/env python3 +"""Continuous competitor discovery and ChangeDetection intelligence. + +Only Python's standard library is required. The service discovers public pages, +reconciles them into ChangeDetection.io, polls stored page histories, classifies +changes, and delivers important alerts or a daily minor-change digest to one +Discord thread. +""" +from __future__ import annotations + +import argparse +import difflib +import fcntl +import gzip +import hashlib +import html +from html.parser import HTMLParser +import json +import os +from pathlib import Path +import re +import sys +import time +from collections import Counter, deque +from datetime import datetime, timezone +from email.utils import parsedate_to_datetime +import urllib.error +import urllib.parse +import urllib.request +import urllib.robotparser +import xml.etree.ElementTree as ET + +TAG_PREFIX = "Competitor • " +DEFAULT_CONFIG = Path(__file__).with_name("config.json") +DEFAULT_IGNORE = Path(__file__).with_name("ignore.json") +DEFAULT_STATE = Path(__file__).with_name("state") +DEFAULT_ENV = Path(__file__).with_name("secrets.env") + +HIGH_SIGNAL_PATH_WORDS = { + "price", "pricing", "package", "packages", "plan", "plans", + "product", "products", "service", "services", + "feature", "features", "api", "integration", "integrations", + "terms", "privacy", "policy", "security", + "changelog", "release", "launch", "news", +} +HIGH_SIGNAL_DIFF_WORDS = { + "launch", "launched", "new product", "new feature", + "api", "integration", "credit score", "credit rating", + "monitoring", "payment default", "acquisition", + "terms", "privacy", "security", +} +ASSET_EXTENSIONS = { + ".7z", ".avi", ".avif", ".bin", ".bmp", ".css", ".csv", ".doc", ".docx", ".eot", + ".exe", ".gif", ".gz", ".ico", ".jpeg", ".jpg", ".js", ".json", ".m4a", ".m4v", + ".map", ".mov", ".mp3", ".mp4", ".mpeg", ".ogg", ".otf", ".pdf", ".png", ".ppt", + ".pptx", ".rar", ".rss", ".svg", ".tar", ".tgz", ".tif", ".tiff", ".ttf", ".wav", + ".webm", ".webp", ".woff", ".woff2", ".xls", ".xlsx", ".xml", ".zip", +} + + +def utc_now() -> str: + return datetime.now(timezone.utc).isoformat(timespec="seconds") + + +def load_json(path: Path, default): + try: + return json.loads(path.read_text(encoding="utf-8")) + except FileNotFoundError: + return default + + +def save_json_atomic(path: Path, value) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temp = path.with_name(path.name + ".new") + temp.write_text(json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8") + os.chmod(temp, 0o640) + os.replace(temp, path) + + +def load_env(path: Path) -> dict[str, str]: + env = {} + try: + lines = path.read_text(encoding="utf-8").splitlines() + except FileNotFoundError: + return env + for line in lines: + line = line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + key, value = line.split("=", 1) + env[key.strip()] = value.strip().strip('"').strip("'") + return env + + +def canonicalize_url(url: str) -> str | None: + try: + p = urllib.parse.urlsplit(html.unescape(url.strip())) + except ValueError: + return None + scheme = p.scheme.lower() + if scheme not in {"http", "https"} or not p.hostname: + return None + host = p.hostname.lower().rstrip(".") + try: + port = p.port + except ValueError: + return None + if port and not ((scheme == "https" and port == 443) or (scheme == "http" and port == 80)): + netloc = f"{host}:{port}" + else: + netloc = host + path = urllib.parse.unquote(p.path or "/", errors="replace") + path = re.sub(r"/{2,}", "/", path) + quoted = urllib.parse.quote(path, safe="/%:@!$&'()*+,;=-._~") + if quoted != "/": + quoted = quoted.rstrip("/") + return urllib.parse.urlunsplit((scheme, netloc, quoted, "", "")) + + +def _compiled_patterns(comp: dict, global_cfg: dict, ignore: dict) -> list[re.Pattern]: + raw = list(global_cfg.get("exclude_regexes", [])) + list(comp.get("exclude_regexes", [])) + raw += list(ignore.get("regexes", [])) + raw += list((ignore.get("competitors", {}).get(comp.get("slug"), {}) or {}).get("regexes", [])) + out = [] + for pattern in raw: + try: + out.append(re.compile(pattern, re.I)) + except re.error as exc: + raise ValueError(f"invalid exclusion regex {pattern!r}: {exc}") from exc + return out + + +def is_ignored(url: str, comp: dict, global_cfg: dict, ignore: dict) -> bool: + canon = canonicalize_url(url) + if not canon: + return True + exact = {canonicalize_url(x) for x in ignore.get("exact_urls", [])} + exact |= {canonicalize_url(x) for x in (ignore.get("competitors", {}).get(comp.get("slug"), {}) or {}).get("exact_urls", [])} + if canon in exact: + return True + return any(p.search(canon) for p in _compiled_patterns(comp, global_cfg, ignore)) + + +def url_allowed(url: str, comp: dict, global_cfg: dict, patterns_or_ignore) -> bool: + canon = canonicalize_url(url) + if not canon: + return False + p = urllib.parse.urlsplit(canon) + domains = {d.lower().rstrip(".") for d in comp.get("domains", [])} + if p.netloc.lower() not in domains: + return False + prefixes = comp.get("allowed_path_prefixes") or ["/"] + if not any(p.path == prefix.rstrip("/") or p.path.startswith(prefix.rstrip("/") + "/") or prefix == "/" for prefix in prefixes): + return False + extensions = set(global_cfg.get("exclude_extensions", [])) | ASSET_EXTENSIONS + if global_cfg.get("_allow_metadata_extensions"): + extensions -= {".xml", ".rss", ".json"} + if any(p.path.lower().endswith(ext.lower()) for ext in extensions): + return False + # Built-in profile guards catch common numeric company/person profile shapes. + profile_patterns = [ + r"/(?:company|companies|organization|organizations|person|persons|profile|profiles)/[^/]*\d{6,}", + r"/(?:[a-z]{2}(?:-[a-z]{2})?/)?\d{6,}[-_/][^/]+/?$", + r"/(?:companies|people)-\d+(?:\.xml)?$", + ] + if any(re.search(x, urllib.parse.unquote(p.path), re.I) for x in profile_patterns): + return False + if isinstance(patterns_or_ignore, dict): + return not is_ignored(canon, comp, global_cfg, patterns_or_ignore) + patterns = patterns_or_ignore or _compiled_patterns(comp, global_cfg, {}) + return not any(pattern.search(canon) for pattern in patterns) + + +NEWS_ROUTE_SEGMENTS = { + "news", "press", "press-releases", "blog", "blogs", "articles", "research", + "case-studies", "customer-stories", "success-stories", "events", "webinars", "insights", +} + + +ROUTINE_DATA_PATH_PATTERNS = ( + r"/(?:lists?|directories|datasets?)/(?:reinstated|new|companies?|people|persons|debtors?|defaults|bankruptcies|liquidations|suspensions|deletions)(?:/|$)", + r"/(?:certificate|certificates|attestations)/(?:companies|people|persons)(?:/|$)", + r"/(?:company|person|profile)-?card(?:/|$)", +) + + +def _is_routine_data_url(url: str) -> bool: + path = urllib.parse.unquote(urllib.parse.urlsplit(url).path).lower() + return any(re.search(pattern, path, re.I) for pattern in ROUTINE_DATA_PATH_PATTERNS) + + +def _watch_category(url: str) -> str: + """Classify a discovered URL for bounded representative watch selection.""" + parsed = urllib.parse.urlsplit(url) + path = urllib.parse.unquote(parsed.path).rstrip("/").lower() + segments = [part for part in path.split("/") if part] + if not segments or (len(segments) == 1 and re.fullmatch(r"[a-z]{2}(?:-[a-z]{2})?", segments[0])): + return "homepage" + if path.endswith("/robots.txt") or re.search( + r"/(?:sitemap[^/]*|feed(?:\.xml)?|rss(?:\.xml)?)$", path + ) or re.search(r"/(?:llms\.txt|master\.md)$", path): + return "discovery_source" + if _is_routine_data_url(url): + return "routine_data" + if re.fullmatch(r"/(?:[a-z]{2}(?:-[a-z]{2})?/)?20\d{2}/\d{2}(?:/\d{2})?", path): + return "archive" + if re.match(r"^/(?:[a-z]{2}(?:-[a-z]{2})?/)?20\d{2}/\d{2}(?:/\d{2})?/[^/]+$", path): + return "news_leaf" + if any(segment in {"category", "tag", "topic", "author"} for segment in segments): + return "discovery_source" + news_position = next( + ((index, segment) for index, segment in enumerate(segments) if segment in NEWS_ROUTE_SEGMENTS), + None, + ) + if news_position: + index, _segment = news_position + return "news_index" if len(segments) <= index + 1 else "news_leaf" + if re.search(r"(?:price|pricing|package|packages|plan|plans)", path): + return "pricing" + if re.search(r"(?:^|/)(?:api|developer|developers|integrations?)(?:/|$)", path): + return "api" + if re.search( + r"(?:terms|privacy|policy|policies|security|cookies?|data-protection)", + path, + ): + return "policy" + if re.search( + r"(?:product|products|service|services|solutions|reports|rating|monitoring|kyc|credit|" + r"payments?|debt|sales|market|audience|target)", + path, + ): + return "product" + if re.search(r"(?:about|contact|management|company|team)", path): + return "company" + return "other" + + +def _watch_family(url: str, category: str) -> str: + parsed = urllib.parse.urlsplit(url) + segments = [part for part in urllib.parse.unquote(parsed.path).strip("/").lower().split("/") if part] + if segments and re.fullmatch(r"[a-z]{2}(?:-[a-z]{2})?", segments[0]): + segments = segments[1:] + if category == "news_leaf": + if segments and re.fullmatch(r"20\d{2}", segments[0]): + return "dated" + for index, segment in enumerate(segments): + if segment not in NEWS_ROUTE_SEGMENTS: + continue + if segment in {"blog", "blogs"}: + return "editorial" + return segment + return "leaf" + return category + + +def _watch_selection_score(url: str, row: dict, category: str) -> tuple: + parsed = urllib.parse.urlsplit(url) + path = urllib.parse.unquote(parsed.path).lower() + depth = len([part for part in path.split("/") if part]) + preferred_language = 0 if re.match(r"^/[a-z]{2}(?:-[a-z]{2})?(?:/|$)", path) else 1 + html_preferred = 0 if path.endswith(".md") else 1 + first_seen = str(row.get("first_seen", "")) + date_match = re.search(r"/(20\d{2})/(\d{2})(?:/(\d{2}))?/", path + "/") + date_key = "".join(part or "00" for part in date_match.groups()) if date_match else "" + if category == "news_leaf": + return preferred_language, date_key or first_seen, html_preferred, -depth, url + return preferred_language, html_preferred, -depth, first_seen, url + + +def select_watch_urls(urls: set[str], comp: dict, global_cfg: dict, previous: dict) -> set[str]: + """Select a small rotating representative subset while retaining full discovery.""" + policy = global_cfg.get("watch_selection") or {} + if not isinstance(policy, dict) or not policy.get("enabled"): + return set(urls) + per_group = max(1, int(policy.get("per_group", 2))) + excluded_categories = set(policy.get("exclude_categories", [])) + candidates = {url for url in urls if canonicalize_url(url)} + mandatory = { + canonical + for raw in comp.get("seeds", []) + if (canonical := canonicalize_url(raw)) in candidates + and _watch_category(canonical) not in excluded_categories + } + + # Prefer the rendered HTML route when a documentation site exposes an + # equivalent .md endpoint, and collapse translated/category copies of the + # same article slug before applying the per-type cap. + deduplicated = [] + news_by_slug: dict[tuple[str, str], str] = {} + previous_rows = previous.get("urls", {}) + for url in sorted(candidates): + if url.lower().endswith(".md") and url[:-3] in candidates: + continue + category = _watch_category(url) + if category in excluded_categories: + continue + if category != "news_leaf": + deduplicated.append(url) + continue + parsed = urllib.parse.urlsplit(url) + slug = parsed.path.rstrip("/").rsplit("/", 1)[-1].lower() + key = (parsed.netloc.lower(), slug) + existing = news_by_slug.get(key) + if existing is None or _watch_selection_score(url, previous_rows.get(url, {}), category) > _watch_selection_score( + existing, previous_rows.get(existing, {}), category + ): + news_by_slug[key] = url + deduplicated.extend(news_by_slug.values()) + + grouped: dict[tuple[str, str, str], list[str]] = {} + for url in deduplicated: + parsed = urllib.parse.urlsplit(url) + category = _watch_category(url) + key = (parsed.netloc.lower(), category, _watch_family(url, category)) + grouped.setdefault(key, []).append(url) + + selected = set(mandatory) + for (_host, category, _family), group in grouped.items(): + ranked = sorted( + group, + key=lambda url: _watch_selection_score(url, previous_rows.get(url, {}), category), + reverse=True, + ) + selected.update(ranked[:per_group]) + return selected + + +class LinkParser(HTMLParser): + def __init__(self): + super().__init__(convert_charrefs=True) + self.links: list[str] = [] + self.feeds: list[str] = [] + self.canonical: str | None = None + + def handle_starttag(self, tag, attrs): + attrs = {k.lower(): (v or "") for k, v in attrs} + if tag.lower() == "a" and attrs.get("href"): + self.links.append(attrs["href"]) + if tag.lower() == "link" and attrs.get("href"): + rel = set(attrs.get("rel", "").lower().split()) + typ = attrs.get("type", "").lower() + if "canonical" in rel: + self.canonical = attrs["href"] + if "alternate" in rel and typ in {"application/rss+xml", "application/atom+xml", "application/feed+json"}: + self.feeds.append(attrs["href"]) + + +class HTTPFetcher: + def __init__(self, user_agent: str, timeout: int = 20, delay: float = 1.0, max_bytes: int = 3_000_000): + self.user_agent = user_agent + self.timeout = timeout + self.delay = delay + self.max_bytes = max_bytes + self.last_request: dict[str, float] = {} + + def get(self, url: str, *, delay: float | None = None) -> tuple[str, bytes, dict[str, str], int]: + host = urllib.parse.urlsplit(url).netloc + wait_for = self.delay if delay is None else delay + elapsed = time.monotonic() - self.last_request.get(host, 0) + if elapsed < wait_for: + time.sleep(wait_for - elapsed) + req = urllib.request.Request( + url, + headers={ + "User-Agent": self.user_agent, + "Accept": "text/html,application/xhtml+xml,application/xml,text/xml,application/rss+xml,application/atom+xml;q=0.9,*/*;q=0.2", + "Accept-Encoding": "gzip", + }, + ) + try: + with urllib.request.urlopen(req, timeout=self.timeout) as response: + final = response.geturl() + status = response.status + headers = {k.lower(): v for k, v in response.headers.items()} + body = response.read(self.max_bytes + 1) + finally: + self.last_request[host] = time.monotonic() + if len(body) > self.max_bytes: + raise ValueError(f"response too large: {url}") + if headers.get("content-encoding", "").lower() == "gzip": + body = gzip.decompress(body) + return final, body, headers, status + + +def _xml_local(tag: str) -> str: + return tag.rsplit("}", 1)[-1].lower() + + +def parse_xml_links(body: bytes) -> tuple[str, list[str]]: + root = ET.fromstring(body) + kind = _xml_local(root.tag) + links: list[str] = [] + if kind in {"sitemapindex", "urlset"}: + for node in root.iter(): + if _xml_local(node.tag) == "loc" and node.text: + links.append(node.text.strip()) + return kind, links + if kind in {"rss", "feed", "rdf"}: + for node in root.iter(): + local = _xml_local(node.tag) + if local == "link": + href = node.attrib.get("href") + text = (node.text or "").strip() + if href: + links.append(href) + elif text: + links.append(text) + elif local == "guid" and (node.text or "").strip().startswith(("http://", "https://")): + links.append(node.text.strip()) + return "feed", links + return kind, links + + +def _certificate_dns_name(raw: str, identity: str) -> str | None: + name = str(raw or "").strip().lower().rstrip(".") + while name.startswith("*."): + name = name[2:] + identity = str(identity or "").strip().lower().rstrip(".") + while identity.startswith("*."): + identity = identity[2:] + if not name or not identity: + return None + if name != identity and not name.endswith("." + identity): + return None + if not re.fullmatch(r"[a-z0-9_*-]+(?:\.[a-z0-9_*-]+)*", name, re.I): + return None + return name + + +def discover_certificate_transparency(comp: dict, global_cfg: dict, fetcher: HTTPFetcher) -> dict: + """Fetch configured crt.sh JSON sources and return normalized in-scope DNS names.""" + configured = comp.get("certificate_transparency_sources") or [] + if not configured: + return { + "status": "disabled", "coverage_gap": False, "errors": [], + "sources": [], "names": [], + } + errors: list[str] = [] + names: set[str] = set() + sources: list[str] = [] + successful = 0 + request_delay = float(global_cfg.get("request_delay_seconds", 1.0)) + retry_attempts = max(1, int(global_cfg.get("certificate_transparency_attempts", 4))) + retry_backoff = max(0.0, float(global_cfg.get("certificate_transparency_retry_backoff_seconds", 2.0))) + for source in configured: + url = str(source.get("url") or "").strip() + identity = str(source.get("identity") or "").strip().lower().rstrip(".") + if not url or not identity: + errors.append("certificate transparency source requires url and identity") + continue + sources.append(url) + try: + last_fetch_error: Exception | None = None + for attempt in range(retry_attempts): + try: + _final, body, _headers, status = fetcher.get(url, delay=request_delay) + if status != 200: + raise ValueError(f"HTTP {status}") + last_fetch_error = None + break + except Exception as exc: + last_fetch_error = exc + if attempt + 1 < retry_attempts and retry_backoff: + time.sleep(retry_backoff * (2 ** attempt)) + if last_fetch_error is not None: + raise last_fetch_error + rows = json.loads(body) + if not isinstance(rows, list): + raise ValueError("crt.sh JSON response was not a list") + for row in rows: + if not isinstance(row, dict): + continue + raw_names = [] + for field in ("common_name", "name_value"): + raw_names.extend(str(row.get(field) or "").splitlines()) + for raw_name in raw_names: + normalized = _certificate_dns_name(raw_name, identity) + if normalized: + names.add(normalized) + successful += 1 + except urllib.error.HTTPError as exc: + errors.append(f"certificate transparency {url}: HTTP {exc.code}") + except Exception as exc: + errors.append(f"certificate transparency {url}: {type(exc).__name__}: {exc}") + if errors: + status = "partial" if successful else "error" + else: + status = "ok" + return { + "status": status, + "coverage_gap": bool(errors), + "errors": errors, + "sources": sources, + "names": sorted(names), + } + + +def _robots_url(seed: str) -> str: + p = urllib.parse.urlsplit(seed) + return urllib.parse.urlunsplit((p.scheme, p.netloc, "/robots.txt", "", "")) + + +def _sitemap_excluded(url: str, comp: dict) -> bool: + return any(re.search(pattern, url, re.I) for pattern in comp.get("exclude_sitemap_regexes", [])) + + +def _is_antibot_challenge(final_url: str, body: bytes) -> bool: + path = urllib.parse.urlsplit(final_url).path.rstrip("/").lower() + if path.endswith(("/captcha", "/browser-check")): + return True + text = body.decode("utf-8", errors="replace").lower() + return any(marker in text for marker in ( + "palun kinnitage, et te ei ole robot", + "kontrollime teie veebilehitsejat", + "verify that you are not a robot", + )) + + +def discover_competitor(comp: dict, global_cfg: dict, ignore: dict, fetcher: HTTPFetcher) -> dict: + seeds = [canonicalize_url(x) for x in comp.get("seeds", [])] + seeds = [x for x in seeds if x] + if not seeds: + return {"status": "error", "coverage_gap": True, "errors": ["no seeds configured"], "urls": []} + + errors: list[str] = [] + urls: set[str] = set() + robot_parsers: dict[str, urllib.robotparser.RobotFileParser] = {} + robot_sitemaps: list[str] = [] + robots_blocked = False + anti_bot_challenge = False + request_delay = float(global_cfg.get("request_delay_seconds", 1.0)) + + robot_targets = {_robots_url(seed) for seed in seeds} + covered_hosts = {urllib.parse.urlsplit(seed).netloc for seed in seeds} + for domain in comp.get("domains", []): + if domain not in covered_hosts: + robot_targets.add(f"https://{domain}/robots.txt") + + for robots_url in sorted(robot_targets): + origin_seed = robots_url + if robots_url in robot_parsers: + continue + rp = urllib.robotparser.RobotFileParser() + rp.set_url(robots_url) + try: + _final, body, _headers, status = fetcher.get(robots_url, delay=request_delay) + if status == 200: + text = body.decode("utf-8", errors="replace") + lines = text.splitlines() + rp.parse(lines) + robot_sitemaps.extend( + match.group(1).strip() + for line in lines + if (match := re.match(r"^\s*Sitemap:\s*(\S+)", line, re.I)) + ) + else: + rp.parse([]) + except urllib.error.HTTPError as exc: + if exc.code == 404: + rp.parse([]) + else: + errors.append(f"robots {robots_url}: HTTP {exc.code}") + rp.parse([]) + except Exception as exc: + errors.append(f"robots {robots_url}: {type(exc).__name__}: {exc}") + rp.parse([]) + robot_parsers[urllib.parse.urlsplit(origin_seed).netloc] = rp + + def robots_allows(url: str) -> bool: + if not comp.get("respect_robots", False): + return True + rp = robot_parsers.get(urllib.parse.urlsplit(url).netloc) + return True if rp is None else rp.can_fetch(global_cfg.get("user_agent_token", "HermesCompetitorMonitor"), url) + + if comp.get("respect_robots", False) and all(not robots_allows(seed) for seed in seeds): + robots_blocked = True + for url in comp.get("policy_watch_urls", []): + canon = canonicalize_url(url) + if canon and url_allowed(canon, comp, global_cfg, ignore): + urls.add(canon) + return { + "status": "robots_blocked", + "coverage_gap": True, + "errors": ["robots.txt disallows crawling for the configured user agent"], + "urls": sorted(urls), + "sitemaps_fetched": 0, + "pages_crawled": 0, + "feeds": [], + } + + sitemap_queue = deque() + for u in comp.get("sitemap_seeds", []) + robot_sitemaps: + canon = canonicalize_url(u) + if canon: + sitemap_queue.append(canon) + for seed in seeds: + p = urllib.parse.urlsplit(seed) + common = urllib.parse.urlunsplit((p.scheme, p.netloc, "/sitemap.xml", "", "")) + sitemap_queue.append(common) + seen_sitemaps: set[str] = set() + feed_urls: set[str] = set() + max_sitemaps = int(comp.get("max_sitemaps", global_cfg.get("max_sitemaps", 40))) + + while sitemap_queue and len(seen_sitemaps) < max_sitemaps: + sitemap = sitemap_queue.popleft() + if sitemap in seen_sitemaps or _sitemap_excluded(sitemap, comp): + continue + if not url_allowed(sitemap, comp, {**global_cfg, "_allow_metadata_extensions": True}, []): + continue + if not robots_allows(sitemap): + continue + seen_sitemaps.add(sitemap) + try: + final, body, headers, _status = fetcher.get(sitemap, delay=request_delay) + ctype = headers.get("content-type", "").lower() + if "html" in ctype and not body.lstrip().startswith(b" tuple[bytes, dict[str, str], int]: + url = self.base + path + if params: + query = urllib.parse.urlencode({k: v for k, v in params.items() if v is not None}) + url += ("&" if "?" in url else "?") + query + headers = {"x-api-key": self.token, "User-Agent": "HermesCompetitorMonitor/1.0"} + if data is not None: + headers["Content-Type"] = "application/json" if isinstance(data, dict) else "text/plain; charset=utf-8" + if isinstance(data, dict): + data = json.dumps(data).encode() + elif isinstance(data, str): + data = data.encode() + request = urllib.request.Request(url, data=data, headers=headers, method=method) + with urllib.request.urlopen(request, timeout=self.timeout) as response: + return response.read(), {k.lower(): v for k, v in response.headers.items()}, response.status + + def request_json(self, method: str, path: str, *, params=None, data=None): + body, _headers, status = self._request(method, path, params=params, data=data) + if status == 204 or not body: + return None + try: + return json.loads(body) + except json.JSONDecodeError: + return body.decode("utf-8", errors="replace") + + def request_text(self, method: str, path: str, *, params=None, data=None) -> str: + body, _headers, _status = self._request(method, path, params=params, data=data) + return body.decode("utf-8", errors="replace") + + +class DiscordClient: + def __init__(self, token: str, channel_id: str, mention_user_id: str | None = None, timeout: int = 30): + if not token: + raise ValueError("DISCORD_BOT_TOKEN is not configured") + self.token = token + self.channel_id = channel_id + self.mention_user_id = mention_user_id + self.timeout = timeout + + @staticmethod + def _chunks(text: str, limit: int = 1900) -> list[str]: + chunks = [] + remaining = text + while len(remaining) > limit: + cut = remaining.rfind("\n", 0, limit) + if cut < limit // 2: + cut = limit + chunks.append(remaining[:cut]) + remaining = remaining[cut:].lstrip("\n") + if remaining: + chunks.append(remaining) + return chunks + + def send(self, content: str, *, mention: bool = False, reply_to: str | None = None): + sent = [] + chunks = self._chunks(content) + for i, chunk in enumerate(chunks): + do_mention = bool(mention and i == 0 and self.mention_user_id) + if do_mention: + chunk = f"<@{self.mention_user_id}> " + chunk + payload = { + "content": chunk, + "allowed_mentions": {"parse": [], "users": [self.mention_user_id] if do_mention else []}, + } + if reply_to and i == 0: + payload["message_reference"] = {"message_id": reply_to, "channel_id": self.channel_id, "fail_if_not_exists": True} + url = f"https://discord.com/api/v10/channels/{self.channel_id}/messages" + for attempt in range(5): + req = urllib.request.Request( + url, + data=json.dumps(payload).encode(), + headers={ + "Authorization": "Bot " + self.token, + "Content-Type": "application/json", + "User-Agent": "HermesCompetitorMonitor/1.0", + }, + method="POST", + ) + try: + with urllib.request.urlopen(req, timeout=self.timeout) as response: + sent.append(json.load(response)) + break + except urllib.error.HTTPError as exc: + if exc.code != 429 or attempt == 4: + raise + raw = exc.read().decode("utf-8", "replace") + try: + retry_after = float(json.loads(raw).get("retry_after", 1.0)) + except (ValueError, TypeError, json.JSONDecodeError): + retry_after = float(exc.headers.get("Retry-After", "1") or "1") + time.sleep(max(0.05, min(retry_after, 65.0))) + return sent[-1] if sent else None + + +def _tag_uuid_by_title(tags: dict, title: str) -> str | None: + for uuid, tag in (tags or {}).items(): + if isinstance(tag, dict) and tag.get("title") == title: + return tag.get("uuid") or uuid + return None + + +def ensure_tag(cd, title: str) -> str: + tags = cd.request_json("GET", "/tags") or {} + found = _tag_uuid_by_title(tags, title) + if found: + return found + result = cd.request_json("POST", "/tag", data={"title": title, "notification_muted": True}) + return result["uuid"] + + +def _watch_tags(watch: dict) -> set[str]: + tags = watch.get("tags") or [] + if isinstance(tags, str): + tags = [tags] + out = {str(x) for x in tags if x} + if watch.get("tag_uuid"): + out.add(str(watch["tag_uuid"])) + if watch.get("tag") and isinstance(watch["tag"], str): + out.add(watch["tag"]) + return out + + +def _watch_by_url(watches: dict) -> dict[str, tuple[str, dict]]: + out = {} + for uuid, watch in (watches or {}).items(): + if not isinstance(watch, dict): + continue + canon = canonicalize_url(watch.get("url", "")) + if canon: + out[canon] = (watch.get("uuid") or uuid, watch) + return out + + +def reconcile_ignored_watches(config: dict, ignore: dict, state: dict, watches: dict, cd) -> dict: + """Apply persistent ignores without deleting history. + + The collection API omits the paused field, so state records only pauses + performed by this monitor. That also prevents undoing a user's manual + pause when an ignore rule is later removed. + """ + global_cfg = dict(config.get("global", {})) + by_url = _watch_by_url(watches) + result = {} + for comp in config.get("competitors", []): + slug = comp["slug"] + domains = {d.lower() for d in comp.get("domains", [])} + comp_state = state.setdefault("competitors", {}).setdefault( + slug, {"urls": {}, "name": comp["name"]} + ) + paused_now = [] + unpaused_now = [] + for url, (uuid, _summary) in by_url.items(): + if urllib.parse.urlsplit(url).netloc.lower() not in domains: + continue + row = comp_state.setdefault("urls", {}).setdefault( + url, {"first_seen": utc_now(), "missing_count": 0} + ) + ignored = is_ignored(url, comp, global_cfg, ignore) + if ignored and not row.get("paused_by_monitor"): + cd.request_json("PUT", f"/watch/{uuid}", data={"paused": True}) + row["paused_by_monitor"] = True + row["ignored"] = True + row["watch_uuid"] = uuid + paused_now.append(url) + elif not ignored and row.get("paused_by_monitor"): + cd.request_json("PUT", f"/watch/{uuid}", data={"paused": False}) + row["paused_by_monitor"] = False + row["ignored"] = False + row["watch_uuid"] = uuid + unpaused_now.append(url) + elif ignored: + row["ignored"] = True + row["watch_uuid"] = uuid + result[slug] = {"paused": paused_now, "unpaused": unpaused_now} + return result + + +def reconcile_selected_watches( + comp: dict, + global_cfg: dict, + ignore: dict, + comp_state: dict, + watches: dict, + cd, + selected_urls: set[str], + tag_uuid: str, + checkpoint=None, +) -> dict: + """Reconcile scope without ever undoing a pre-existing manual pause.""" + selected_urls = set(selected_urls) + paused_now = [] + unpaused_now = [] + ignored_paused = [] + manually_paused_selected = [] + domains = {domain.lower() for domain in comp.get("domains", [])} + for url, (uuid, summary) in _watch_by_url(watches).items(): + if urllib.parse.urlsplit(url).netloc.lower() not in domains: + continue + if tag_uuid not in _watch_tags(summary): + continue + row = comp_state.setdefault("urls", {}).setdefault( + url, {"first_seen": utc_now(), "missing_count": 0} + ) + ignored = is_ignored(url, comp, global_cfg, ignore) + should_watch = url in selected_urls and not ignored + reason = "ignored" if ignored else "representative_scope" + row["ignored"] = ignored + row["selected_for_watch"] = should_watch + row["watch_uuid"] = uuid + + # The collection endpoint omits pause state. Read the authoritative + # detail record before deciding whether this monitor owns a pause. + detail = cd.request_json("GET", f"/watch/{uuid}") or {} + actual_paused = bool(detail.get("paused")) + managed_before = bool(row.get("paused_by_monitor")) + if managed_before and "pre_monitor_paused" not in row: + # Migration for pauses created by earlier monitor versions. + row["pre_monitor_paused"] = False + row["pause_owned_by_monitor"] = True + + if not should_watch: + if not managed_before: + row["pre_monitor_paused"] = actual_paused + row["pause_owned_by_monitor"] = not actual_paused + elif not actual_paused: + # The scope still requires a pause and the watch is currently + # active, so this new transition is monitor-owned. + row["pre_monitor_paused"] = False + row["pause_owned_by_monitor"] = True + row["paused_by_monitor"] = True + row["pause_reason"] = reason + row["monitor_pause_reasons"] = [reason] + if not actual_paused: + row["pause_operation_pending"] = "pause" + if checkpoint: + checkpoint() + cd.request_json("PUT", f"/watch/{uuid}", data={"paused": True}) + row.pop("pause_operation_pending", None) + row["api_paused_confirmed"] = True + if checkpoint: + checkpoint() + paused_now.append(url) + else: + row["api_paused_confirmed"] = True + if row.pop("pause_operation_pending", None) is not None and checkpoint: + checkpoint() + if ignored and (not managed_before or not actual_paused): + ignored_paused.append(url) + continue + + owns_pause = bool(row.get("pause_owned_by_monitor")) and not bool(row.get("pre_monitor_paused")) + if managed_before and owns_pause and actual_paused: + row["pause_operation_pending"] = "unpause" + if checkpoint: + checkpoint() + cd.request_json("PUT", f"/watch/{uuid}", data={"paused": False}) + row.pop("pause_operation_pending", None) + row["api_paused_confirmed"] = False + row["paused_by_monitor"] = False + row["pause_owned_by_monitor"] = False + row["monitor_pause_reasons"] = [] + row.pop("pause_reason", None) + if checkpoint: + checkpoint() + unpaused_now.append(url) + else: + # If it was already paused before monitor ownership, leave the + # application pause untouched when scope rotates back in. + row["paused_by_monitor"] = False + row["pause_owned_by_monitor"] = False + row["monitor_pause_reasons"] = [] + row.pop("pause_reason", None) + row.pop("pause_operation_pending", None) + if actual_paused: + manually_paused_selected.append(url) + return { + "paused": paused_now, + "unpaused": unpaused_now, + "ignored_paused": ignored_paused, + "manually_paused_selected": manually_paused_selected, + } + + +def monitor_paused_watch_ids(state: dict) -> set[str]: + return { + row.get("watch_uuid") + for comp in state.get("competitors", {}).values() + for row in comp.get("urls", {}).values() + if row.get("paused_by_monitor") and row.get("watch_uuid") + } + + +def _watch_interval(hours: int) -> str: + return json.dumps({"weeks": 0, "days": 0, "hours": int(hours), "minutes": 0, "seconds": 0}, separators=(",", ":")) + + +def import_urls(cd, urls: list[str], tag_uuid: str, comp: dict, global_cfg: dict) -> list[str]: + created = [] + params = { + "tag_uuids": tag_uuid, + "dedupe": "true", + "fetch_backend": comp.get("fetch_backend", "html_requests"), + "processor": comp.get("processor", "text_json_diff"), + "notification_muted": "true", + "filter_text_added": "true", + "filter_text_removed": "true", + "filter_text_replaced": "true", + "time_between_check_use_default": "false", + "time_between_check": _watch_interval(int(comp.get("watch_check_hours", global_cfg.get("watch_check_hours", 6)))), + "use_page_title_in_list": "true", + } + for i in range(0, len(urls), 19): + batch = urls[i:i + 19] + result = cd.request_json("POST", "/import", params=params, data="\n".join(batch)) or [] + if isinstance(result, list): + created.extend(result) + return created + + +def _contains_signal(text: str, terms: tuple[str, ...]) -> bool: + return any( + re.search(r"(? str: + changed = [line.strip() for line in diff.splitlines() if line.strip().startswith(_DIFF_MARKERS)] + return "\n".join(changed) if changed else diff + + +def _semantic_diff_text(diff: str) -> str: + """Realign ChangeDetection's positional line replacements. + + When one card is inserted or removed from a repeated list, ChangeDetection's + text renderer can report every following card as ``changed``/``into``. By + reconstructing the old and new line streams and aligning them again, the + alert reports the actual insertion/deletion instead of that cascade. + """ + old_lines: list[str] = [] + new_lines: list[str] = [] + marked = False + for raw in diff.splitlines(): + line = re.sub(r"\s+", " ", raw).strip() + if not line: + continue + match = _DIFF_MARKER_RE.match(line) + if not match: + old_lines.append(line) + new_lines.append(line) + continue + marked = True + marker, text = match.groups() + if marker == "replaced": + # The one-sided representation is not sufficient to reconstruct + # both streams safely; retain ChangeDetection's original evidence. + return _marker_diff_text(diff) + if not text: + continue + if marker in {"removed", "changed"}: + old_lines.append(text) + if marker in {"added", "into"}: + new_lines.append(text) + + if not marked or max(len(old_lines), len(new_lines)) > 4000: + return _marker_diff_text(diff) + + semantic: list[str] = [] + matcher = difflib.SequenceMatcher(a=old_lines, b=new_lines, autojunk=False) + for opcode, old_start, old_end, new_start, new_end in matcher.get_opcodes(): + if opcode == "equal": + continue + old_block = old_lines[old_start:old_end] + new_block = new_lines[new_start:new_end] + if opcode == "delete": + semantic.extend(f"(removed) {line}" for line in old_block) + elif opcode == "insert": + semantic.extend(f"(added) {line}" for line in new_block) + elif len(old_block) == len(new_block): + for before, after in zip(old_block, new_block): + semantic.append(f"(changed) {before}") + semantic.append(f"(into) {after}") + else: + semantic.extend(f"(removed) {line}" for line in old_block) + semantic.extend(f"(added) {line}" for line in new_block) + return "\n".join(semantic) or _marker_diff_text(diff) + + +def _changed_diff_text(diff: str) -> str: + return _semantic_diff_text(diff) + + +_UI_CHROME_TOKENS = { + "arrow_back", "arrow_drop_down", "arrow_forward", "close", "expand_less", + "expand_more", "language", "manage_search", "menu", "more_horiz", + "more_vert", "search", +} + +_ESTONIAN_LOCALIZATION_WORDS = { + "abi", "andmete", "avaleht", "ettevõte", "ettevõtted", "eraisiku", + "firmast", "hinnad", "kasutustingimused", "kontakt", "konto", "kuumakse", + "muu", "müüginimekirjad", "päring", "registreeri", "seire", "sisene", + "taustapäringut", "teenused", "turuanalüüs", "turundus", "tunnustusmärgised", + "töötlemine", "uudised", "valdkondade", "välisraportid", "võlglaste", +} +_ENGLISH_LOCALIZATION_WORDS = { + "about", "account", "additional", "background", "badges", "cards", "companies", + "conditions", "contact", "create", "debtor", "external", "faq", "help", "home", + "information", "inquiry", "items", "lists", "login", "market", "marketing", + "monthly", "monitoring", "overview", "payment", "prices", "processing", "register", + "reports", "sales", "sector", "services", "terms", +} + + +def _is_estonian_to_english_localization(diff: str) -> bool: + """Detect a broad language swap while preserving changed commercial values.""" + old_lines: list[str] = [] + new_lines: list[str] = [] + for line in _semantic_diff_text(diff).splitlines(): + match = _DIFF_MARKER_RE.match(line.strip()) + if not match or not match.group(2).strip(): + continue + marker, text = match.groups() + if marker in {"removed", "changed"}: + old_lines.append(text) + if marker in {"added", "into"}: + new_lines.append(text) + if len(old_lines) < 4 or len(new_lines) < 4: + return False + + def words(lines: list[str]) -> set[str]: + return set(re.findall(r"[a-zõäöüšž]+", " ".join(lines).lower())) + + old_words = words(old_lines) + new_words = words(new_lines) + old_estonian = len(old_words & _ESTONIAN_LOCALIZATION_WORDS) + new_estonian = len(new_words & _ESTONIAN_LOCALIZATION_WORDS) + old_english = len(old_words & _ENGLISH_LOCALIZATION_WORDS) + new_english = len(new_words & _ENGLISH_LOCALIZATION_WORDS) + if not ( + old_estonian >= 3 + and new_english >= 3 + and old_estonian > new_estonian + and new_english > old_english + ): + return False + + def commercial_value_signature(lines: list[str]) -> Counter: + tokens = re.findall( + r"(? bool: + """Return true for pure moves plus known rendered UI-icon labels.""" + entries: list[tuple[str, str]] = [] + for line in _semantic_diff_text(diff).splitlines(): + match = _DIFF_MARKER_RE.match(line.strip()) + if match and match.group(2).strip(): + entries.append((match.group(1), match.group(2).strip())) + if not entries: + return False + old = Counter(text for marker, text in entries if marker in {"removed", "changed"}) + new = Counter(text for marker, text in entries if marker in {"added", "into"}) + shared = old & new + old -= shared + new -= shared + remaining = list(old.elements()) + list(new.elements()) + return ( + not remaining + or all(re.match(r"^[*+]\s+", text) for text in remaining) + or ( + all(text.lower() in _UI_CHROME_TOKENS for text in remaining) + and (bool(shared) or len(entries) > 1) + ) + ) + + +def classify_change(url: str, diff: str) -> tuple[str, str, str]: + path = urllib.parse.unquote(urllib.parse.urlsplit(url).path).lower() + path_words = set(re.findall(r"[a-z0-9]+", path)) + signal_text = _changed_diff_text(diff) + lower = signal_text.lower() + category = "content" + why = "Public positioning or factual wording changed; review whether it affects market messaging." + important = False + if _is_interface_only_diff(diff): + return "ignore", "interface-only churn", "Only navigation order or rendered interface-icon labels changed." + if _is_routine_data_url(url): + return "ignore", "routine public-data churn", "Aggregate company/person list updates are outside competitor-intelligence scope." + if _is_estonian_to_english_localization(diff): + return ( + "minor", + "localization", + "An existing page was translated or its localization was completed; unchanged commercial values do not establish a pricing, packaging, policy, or product change.", + ) + if path.endswith("/robots.txt"): + category = "crawl policy" + why = "robots.txt changed; monitored coverage or machine-use policy may have changed." + important = True + elif "sitemap" in path and signal_text and all( + "sitemap was dynamically generated" in line.lower() + for line in signal_text.splitlines() + ): + return "ignore", "generated sitemap metadata", "Only a generated sitemap timestamp changed." + else: + pricing_path = bool(path_words & {"price", "pricing", "package", "packages", "plan", "plans"}) + pricing_diff = bool( + re.search(r"(?:\b(?:eur|usd|gbp)\b|€|\$|£)\s*\d|\d(?:[.,]\d+)?\s*(?:\b(?:eur|usd|gbp)\b|€|\$|£)", signal_text, re.I) + or re.search(r"\b(?:price|prices|pricing|monthly)\b", signal_text, re.I) + or re.search(r"\bper\s+(?:month|year)\b", signal_text, re.I) + ) + if (not signal_text and pricing_path) or pricing_diff: + category = "pricing" + why = "Pricing or packaging can change competitive positioning and customer switching economics." + important = True + elif path_words & {"api", "developer", "integration", "integrations"} or re.search(r"\b(?:api|integration)\b", lower): + category = "API / integration" + why = "API or integration changes can alter product capability and enterprise adoption friction." + important = True + elif path_words & {"product", "products", "service", "services", "feature", "features"} or _contains_signal(lower, ("new product", "new feature", "launch")): + category = "product / feature" + why = "A product or feature change may affect differentiation and the competitor offer." + important = True + elif path_words & {"terms", "privacy", "policy", "security"}: + category = "policy / security" + why = "Policy, privacy, or security wording can change customer obligations or risk." + important = True + elif path_words & {"news", "press", "blog"} and _contains_signal(lower, ("launch", "partnership", "acquisition", "security")): + category = "material announcement" + why = "This announcement may signal a strategic, product, partnership, or risk change." + important = True + elif _contains_signal(lower, HIGH_SIGNAL_DIFF_WORDS): + category = "high-signal wording" + why = "The changed wording contains product, risk, pricing, or market-significant terms." + important = True + return ("important" if important else "minor", category, why) + + +def classify_new_url(url: str) -> tuple[str, str, str]: + return classify_change(url, "") + + +def _diff_excerpt(diff: str, limit: int = 650) -> str: + semantic_lines = [ + re.sub(r"\s+", " ", raw).strip() + for raw in _semantic_diff_text(diff).splitlines() + if raw.strip() + ] + marked = [line for line in semantic_lines if line.startswith(_DIFF_MARKERS)] + if not marked: + text = "\n".join(semantic_lines) + return text[:limit] or "ChangeDetection recorded a new snapshot; the textual diff was empty." + + old = [line for line in marked if line.startswith(("(changed)", "(removed)"))] + new = [line for line in marked if line.startswith(("(into)", "(added)"))] + + def take(rows: list[str], budget: int) -> list[str]: + selected: list[str] = [] + used = 0 + for row in rows: + needed = len(row) + (1 if selected else 0) + if used + needed <= budget: + selected.append(row) + used += needed + continue + if not selected and budget > 1: + selected.append(row[: budget - 1] + "…") + break + return selected + + if old and new: + old_rows = take(old, max(1, int(limit * 0.45))) + old_text = "\n".join(old_rows) + new_rows = take(new, max(1, limit - len(old_text) - 1)) + lines = old_rows + new_rows + else: + lines = take(marked, limit) + text = "\n".join(lines)[:limit] + return text or "ChangeDetection recorded a new snapshot; the textual diff was empty." + + +def _format_check_time(value) -> str | None: + try: + timestamp = int(float(value)) + except (TypeError, ValueError): + return None + return datetime.fromtimestamp(timestamp, tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") + + +def _event_fetch_window_line(event: dict) -> str: + unchanged = _format_check_time(event.get("last_successful_unchanged_check")) + changed = _format_check_time(event.get("first_successful_changed_check")) + if not unchanged or not changed: + return "" + return f"\n**Observed fetch window:** unchanged at {unchanged}; changed at {changed}." + + +def _append_pending(state_dir: Path, events: list[dict]) -> None: + if not events: + return + path = state_dir / "pending-digest.json" + pending = load_json(path, {"events": []}) + known = {event.get("id") for event in pending.get("events", [])} + for event in events: + if event.get("id") not in known: + pending.setdefault("events", []).append(event) + known.add(event.get("id")) + save_json_atomic(path, pending) + + +def _append_important_outbox(state_dir: Path, events: list[dict]) -> None: + path = state_dir / "pending-important.json" + pending = load_json(path, {"events": []}) + known = {event.get("id") for event in pending.get("events", [])} + for event in events: + if event.get("id") not in known: + pending.setdefault("events", []).append(event) + known.add(event.get("id")) + save_json_atomic(path, pending) + + +def _deliver_important_outbox(config: dict, state_dir: Path, discord) -> int: + path = state_dir / "pending-important.json" + pending = load_json(path, {"events": []}) + events = list(pending.get("events", [])) + delivered = 0 + while events: + event = events[0] + if event.get("type") == "certificate_name": + text = ( + f"🔐 **New certificate name{'s' if event.get('name_count', 1) != 1 else ''} — {event['competitor']}**\n" + f"{event['evidence'][:900]}\n**Certificate Transparency source:**\n{event['url']}\n" + f"**Why it matters:** {event['why']}" + ) + discord.send(text, mention=True) + events.pop(0) + save_json_atomic(path, {"events": events}) + delivered += 1 + continue + if event.get("type") == "new_page": + text = ( + f"🆕 **New high-signal page — {event['competitor']} ({event['category']})**\n" + f"{event['url']}\n**Why it matters:** {event['why']}" + ) + discord.send(text, mention=True) + events.pop(0) + save_json_atomic(path, {"events": events}) + delivered += 1 + continue + corroboration = "" + if len(event.get("urls", [])) > 1: + corroboration = f"\nAlso observed on {len(event['urls']) - 1} matching page(s)." + timing = _event_fetch_window_line(event) + text = ( + f"🔔 **Competitor change — {event['competitor']} ({event['category']})**\n" + f"{event['url']}\n**Before/after evidence**\n```text\n{event['evidence'][:700]}\n```\n" + f"**Why it matters:** {event['why']}{timing}{corroboration}" + ) + discord.send(text, mention=True) + events.pop(0) + save_json_atomic(path, {"events": events}) + delivered += 1 + return delivered + + +def _event_id(*parts: str) -> str: + return hashlib.sha256("\x1f".join(parts).encode()).hexdigest() + + +def _is_shared_minor_content_fragment(event: dict) -> bool: + return ( + event.get("type") == "page_change" + and event.get("category") == "content" + and len(event.get("urls", [])) >= 3 + ) + + +def deduplicate_events(events: list[dict]) -> list[dict]: + """Collapse same-company copies of one observed change. + + This intentionally uses only strong deterministic evidence equivalence; + ambiguous similarities remain separate rather than being falsely merged. + """ + grouped: dict[tuple, dict] = {} + for event in events: + evidence = re.sub(r"\b[0-9a-f]{16,}\b", "", event.get("evidence", "").lower()) + evidence = re.sub(r"\s+", " ", evidence).strip() + if event.get("type") in {"page_change"} and evidence: + key = (event.get("competitor"), event.get("type"), event.get("category"), evidence) + else: + key = (event.get("competitor"), event.get("type"), event.get("category"), event.get("url")) + if key not in grouped: + grouped[key] = dict(event) + grouped[key]["urls"] = list(dict.fromkeys( + event.get("urls") or ([event.get("url")] if event.get("url") else []) + )) + grouped[key]["_source_ids"] = list(dict.fromkeys([ + event.get("id"), + *(event.get("_source_ids") or []), + ])) + continue + row = grouped[key] + if event.get("url") and event["url"] not in row["urls"]: + row["urls"].append(event["url"]) + row["_source_ids"] = list(dict.fromkeys([ + *row["_source_ids"], + event.get("id"), + *(event.get("_source_ids") or []), + ])) + row["id"] = _event_id("deduplicated", *sorted([row.get("id", ""), event.get("id", "")])) + return [ + row for row in grouped.values() + if not _is_shared_minor_content_fragment(row) + ] + + +def _event_is_in_scope( + event: dict, + comp_by_name: dict[str, dict], + global_cfg: dict, + ignore: dict, + challenged_competitors: set[str], +) -> bool: + comp = comp_by_name.get(event.get("competitor")) + url = event.get("url") + if comp and url: + if is_ignored(url, comp, global_cfg, ignore): + return False + excluded_categories = set( + global_cfg.get("watch_selection", {}).get("exclude_categories", []) + ) + if _watch_category(url) in excluded_categories: + return False + if ( + event.get("competitor") in challenged_competitors + and event.get("type") in { + "new_page", + "page_change", + "page_removed_from_discovery", + "coverage_recovered", + } + ): + return False + return True + + +def run_discover(config: dict, ignore: dict, state_dir: Path, cd, discord, *, send: bool = True) -> dict: + global_cfg = dict(config.get("global", {})) + global_cfg["user_agent_token"] = config.get("user_agent", "HermesCompetitorMonitor").split("/", 1)[0] + state_path = state_dir / "discovery.json" + state = load_json(state_path, {"version": 1, "initialized": False, "competitors": {}}) + was_initialized = bool(state.get("initialized")) + now = utc_now() + watches = cd.request_json("GET", "/watch") or {} + watch_by_url = _watch_by_url(watches) + results = {} + all_minor_events = [] + total_new_watches = 0 + total_new_urls = 0 + total_selected_watches = 0 + total_gaps = 0 + fetcher = HTTPFetcher( + config.get("user_agent", "HermesCompetitorMonitor/1.0"), + timeout=int(global_cfg.get("fetch_timeout_seconds", 20)), + delay=float(global_cfg.get("request_delay_seconds", 1.0)), + max_bytes=int(global_cfg.get("max_response_bytes", 3_000_000)), + ) + + for comp in config.get("competitors", []): + slug = comp["slug"] + title = TAG_PREFIX + comp["name"] + tag_uuid = ensure_tag(cd, title) + previous = state.setdefault("competitors", {}).setdefault(slug, {"urls": {}, "name": comp["name"]}) + prior_status = previous.get("last_status") + prior_errors = previous.get("last_errors", []) + prior_ct_status = previous.get("last_certificate_status") + prior_ct_errors = previous.get("last_certificate_errors", []) + try: + discovery = discover_competitor(comp, global_cfg, ignore, fetcher) + except Exception as exc: + discovery = {"status": "error", "coverage_gap": True, "errors": [f"{type(exc).__name__}: {exc}"], "urls": [], "sitemaps_fetched": 0, "pages_crawled": 0, "feeds": []} + try: + certificate_discovery = discover_certificate_transparency(comp, global_cfg, fetcher) + except Exception as exc: + certificate_discovery = { + "status": "error", "coverage_gap": True, + "errors": [f"{type(exc).__name__}: {exc}"], "sources": [], "names": [], + } + current_urls = set(discovery.get("urls", [])) + # Operator exclusions are not removals. Keep their ledger rows and + # paused ChangeDetection history, but never count them as missing. + excluded_categories = set( + global_cfg.get("watch_selection", {}).get("exclude_categories", []) + ) + prior_urls = { + url for url in previous.get("urls", {}) + if not is_ignored(url, comp, global_cfg, ignore) + and _watch_category(url) not in excluded_categories + } + current_event_urls = { + url for url in current_urls + if not is_ignored(url, comp, global_cfg, ignore) + and _watch_category(url) not in excluded_categories + } + new_urls = sorted(current_event_urls - prior_urls) + for url in current_urls: + row = previous.setdefault("urls", {}).setdefault(url, {"first_seen": now, "missing_count": 0}) + row["last_seen"] = now + row["missing_count"] = 0 + + certificate_was_initialized = bool(previous.get("certificate_initialized")) + certificate_rows = previous.setdefault("certificate_names", {}) + current_certificate_names = set(certificate_discovery.get("names", [])) + new_certificate_names = sorted( + current_certificate_names - set(certificate_rows) + ) if certificate_was_initialized else [] + for name in current_certificate_names: + row = certificate_rows.setdefault(name, {"first_seen": now}) + row["last_seen"] = now + if certificate_discovery.get("status") == "ok": + previous["certificate_initialized"] = True + previous["last_successful_certificate_discovery"] = now + elif certificate_discovery.get("status") not in {"disabled", None}: + previous["last_failed_certificate_discovery"] = now + + previous_selected = { + url for url, row in previous.get("urls", {}).items() + if row.get("selected_for_watch") and not is_ignored(url, comp, global_cfg, ignore) + } + discovery_complete = discovery.get("status") == "ok" + if discovery.get("anti_bot_challenge"): + # A CAPTCHA/challenge payload is not competitor content. Pause the + # affected competitor's watches so challenge rotation cannot be + # misreported as intelligence; a later complete run can reactivate + # useful representatives automatically. + selected_current = set() + elif discovery_complete: + selected_current = select_watch_urls(current_urls, comp, global_cfg, previous) + else: + # Incomplete coverage must never rotate or mass-pause a stable + # representative set. Preserve it and only admit explicit source + # anchors or newly discovered high-signal URLs. + anchors = { + canonicalize_url(url) + for url in (comp.get("seeds", []) + comp.get("policy_watch_urls", [])) + } + high_signal_new = { + url for url in new_urls if classify_new_url(url)[0] == "important" + } + selected_current = ( + (previous_selected & current_urls) + | (anchors & current_urls) + | high_signal_new + ) + # A previously selected page that disappears remains watched until the + # configured number of complete inventories confirms the removal. It + # is then paused (never deleted), preserving its history without + # exceeding the representative cap indefinitely. + missing_confirmations = int(global_cfg.get("missing_confirmations", 2)) + selected_missing = { + url for url, row in previous.get("urls", {}).items() + if row.get("selected_for_watch") + and url not in current_urls + and not is_ignored(url, comp, global_cfg, ignore) + and not discovery.get("anti_bot_challenge") + and ( + not discovery_complete + or int(row.get("missing_count", 0)) + 1 < missing_confirmations + ) + and _watch_category(url) not in set( + global_cfg.get("watch_selection", {}).get("exclude_categories", []) + ) + } + selected_urls = selected_current | selected_missing + to_import = sorted(url for url in selected_current if url not in watch_by_url) + created = import_urls(cd, to_import, tag_uuid, comp, global_cfg) if to_import else [] + total_new_watches += len(created) + total_new_urls += len(new_urls) + + # Refresh after import so UUID mappings are authoritative even when the API deduped. + if to_import: + watches = cd.request_json("GET", "/watch") or watches + watch_by_url = _watch_by_url(watches) + for url in current_urls: + row = previous["urls"][url] + row["selected_for_watch"] = url in selected_current + if url in watch_by_url: + row["watch_uuid"] = watch_by_url[url][0] + + confirmed_missing = [] + newly_confirmed_missing = [] + if discovery_complete: + for url in prior_urls - current_urls: + row = previous["urls"][url] + row["missing_count"] = int(row.get("missing_count", 0)) + 1 + if row["missing_count"] >= missing_confirmations: + confirmed_missing.append(url) + if row["missing_count"] == missing_confirmations: + newly_confirmed_missing.append(url) + previous["last_successful_discovery"] = now + else: + previous["last_failed_discovery"] = now + + # Reconcile the representative set only after imports. Excess leaf + # watches are paused, never deleted, and can rotate back in later. + selection_reconciliation = reconcile_selected_watches( + comp, global_cfg, ignore, previous, watches, cd, selected_urls, tag_uuid, + checkpoint=lambda: save_json_atomic(state_path, state), + ) + total_selected_watches += len(selected_urls) + ignored_paused = selection_reconciliation.get("ignored_paused", []) + ignored_unpaused = [] + + important_new = [] + minor_new = [] + if certificate_was_initialized and new_certificate_names and send: + shown = new_certificate_names[:20] + remainder = len(new_certificate_names) - len(shown) + names_text = "\n".join(f"- `{name}`" for name in shown) + if remainder: + names_text += f"\n- …and {remainder} more" + source_url = (certificate_discovery.get("sources") or [""])[0] + _append_important_outbox(state_dir, [{ + "id": _event_id("new-certificate-names", slug, *new_certificate_names), + "type": "certificate_name", + "competitor": comp["name"], + "category": "Certificate Transparency", + "url": source_url, + "evidence": names_text, + "name_count": len(new_certificate_names), + "why": "A newly observed certificate name can reveal a new public product, API, integration, or infrastructure surface. It is evidence of certificate issuance, not proof that the hostname is live.", + "observed_at": now, + }]) + save_json_atomic(state_path, state) + _deliver_important_outbox(config, state_dir, discord) + if was_initialized: + for url in new_urls: + severity, category, why = classify_new_url(url) + event = { + "id": _event_id("new-url", slug, url), + "type": "new_page", + "competitor": comp["name"], + "url": url, + "category": category, + "why": why, + "observed_at": now, + } + if severity == "important": + important_new.append(event) + else: + minor_new.append(event) + if important_new and send: + _append_important_outbox(state_dir, important_new) + save_json_atomic(state_path, state) + _deliver_important_outbox(config, state_dir, discord) + all_minor_events.extend(minor_new) + + if discovery.get("coverage_gap") and ( + prior_status != discovery.get("status") or prior_errors != discovery.get("errors", []) + ): + all_minor_events.append({ + "id": _event_id("discovery-gap", slug, now[:10], "|".join(discovery.get("errors", []))), + "type": "coverage_gap", + "competitor": comp["name"], + "url": comp.get("seeds", [""])[0], + "category": "discovery source failure", + "why": "Discovery coverage is unknown for this source; this is not evidence that no new pages appeared.", + "evidence": "; ".join(discovery.get("errors", []))[:500], + "observed_at": now, + }) + elif prior_status in {"error", "partial", "robots_blocked"} and not discovery.get("coverage_gap"): + all_minor_events.append({ + "id": _event_id("discovery-recovered", slug, now[:10]), + "type": "coverage_recovered", + "competitor": comp["name"], + "url": comp.get("seeds", [""])[0], + "category": "discovery coverage restored", + "why": "Automatic page discovery is succeeding again after a recorded coverage gap.", + "observed_at": now, + }) + if certificate_discovery.get("coverage_gap") and ( + prior_ct_status != certificate_discovery.get("status") + or prior_ct_errors != certificate_discovery.get("errors", []) + ): + all_minor_events.append({ + "id": _event_id("certificate-gap", slug, now[:10], "|".join(certificate_discovery.get("errors", []))), + "type": "coverage_gap", + "competitor": comp["name"], + "url": (certificate_discovery.get("sources") or comp.get("seeds", [""]))[0], + "category": "certificate transparency source failure", + "why": "Certificate Transparency coverage is unknown for this source; this is not evidence that no new certificate names appeared.", + "evidence": "; ".join(certificate_discovery.get("errors", []))[:500], + "observed_at": now, + }) + elif prior_ct_status in {"error", "partial"} and certificate_discovery.get("status") == "ok": + all_minor_events.append({ + "id": _event_id("certificate-recovered", slug, now[:10]), + "type": "coverage_recovered", + "competitor": comp["name"], + "url": (certificate_discovery.get("sources") or comp.get("seeds", [""]))[0], + "category": "certificate transparency coverage restored", + "why": "The configured Certificate Transparency source is succeeding again after a recorded coverage gap.", + "observed_at": now, + }) + for url in newly_confirmed_missing: + all_minor_events.append({ + "id": _event_id("confirmed-missing", slug, url), + "type": "page_removed_from_discovery", + "competitor": comp["name"], + "url": url, + "category": "page removed or unlinked", + "why": "The page disappeared from two successful discovery inventories; its ChangeDetection watch and history were retained.", + "observed_at": now, + }) + + previous.update({ + "name": comp["name"], + "tag_uuid": tag_uuid, + "last_attempt": now, + "last_status": discovery.get("status"), + "last_errors": discovery.get("errors", []), + "last_url_count": len(current_urls), + "last_selected_watch_count": len(selected_urls), + "last_certificate_status": certificate_discovery.get("status"), + "last_certificate_errors": certificate_discovery.get("errors", []), + "last_certificate_name_count": len(current_certificate_names), + }) + if discovery.get("coverage_gap") or certificate_discovery.get("coverage_gap"): + total_gaps += 1 + results[slug] = { + **discovery, + "certificate_transparency": certificate_discovery, + "new_certificate_names": new_certificate_names, + "new_urls": new_urls, + "confirmed_missing_urls": sorted(confirmed_missing), + "created_watch_count": len(created), + "selected_watch_count": len(selected_urls), + "important_new_count": len(important_new), + "minor_new_count": len(minor_new), + "ignored_paused_urls": ignored_paused, + "ignored_unpaused_urls": ignored_unpaused, + "scope_paused_urls": selection_reconciliation.get("paused", []), + "scope_unpaused_urls": selection_reconciliation.get("unpaused", []), + "manually_paused_selected_urls": selection_reconciliation.get("manually_paused_selected", []), + "selection_frozen_for_incomplete_discovery": ( + not discovery_complete and not discovery.get("anti_bot_challenge") + ), + } + + state["initialized"] = True + state["updated_at"] = now + save_json_atomic(state_path, state) + + # Drop queued items that the operator has now explicitly ignored. This + # prevents excluded noise from resurfacing in a later digest. + pending_path = state_dir / "pending-digest.json" + pending = load_json(pending_path, {"events": []}) + comp_by_name = {c["name"]: c for c in config.get("competitors", [])} + challenged_competitors = { + comp["name"] + for comp in config.get("competitors", []) + if results.get(comp["slug"], {}).get("anti_bot_challenge") + } + all_minor_events = [ + event for event in all_minor_events + if _event_is_in_scope( + event, comp_by_name, global_cfg, ignore, challenged_competitors + ) + ] + kept_events = [ + event for event in pending.get("events", []) + if _event_is_in_scope( + event, comp_by_name, global_cfg, ignore, challenged_competitors + ) + ] + if kept_events != pending.get("events", []): + pending["events"] = kept_events + save_json_atomic(pending_path, pending) + _append_pending(state_dir, all_minor_events) + result = { + "at": now, + "competitors": results, + "summary": { + "competitor_count": len(results), + "new_url_count": total_new_urls, + "new_watch_count": total_new_watches, + "selected_watch_count": total_selected_watches, + "coverage_gap_count": total_gaps, + "initial_inventory_suppressed": not was_initialized, + }, + } + save_json_atomic(state_dir / "last-discovery-run.json", result) + return result + + +def _tag_company_map(cd) -> dict[str, str]: + out = {} + for uuid, tag in (cd.request_json("GET", "/tags") or {}).items(): + if not isinstance(tag, dict): + continue + title = tag.get("title", "") + if title.startswith(TAG_PREFIX): + out[tag.get("uuid") or uuid] = title[len(TAG_PREFIX):] + return out + + +def _watch_company(watch: dict, tag_map: dict[str, str]) -> str | None: + for tag in _watch_tags(watch): + if tag in tag_map: + return tag_map[tag] + return None + + +def run_collect(config: dict, state_dir: Path, cd, discord, *, send: bool = True) -> dict: + state_path = state_dir / "collector.json" + state = load_json(state_path, {"version": 1, "last_history": {}, "seen_events": [], "failure_counts": {}, "failure_alerted": {}}) + seen = set(state.get("seen_events", [])) + watches = cd.request_json("GET", "/watch") or {} + tag_map = _tag_company_map(cd) + monitor_paused = monitor_paused_watch_ids(load_json(state_dir / "discovery.json", {})) + important_events = [] + minor_events = [] + coverage_gaps = [] + now = utc_now() + + for uuid, watch in watches.items(): + if not isinstance(watch, dict): + continue + watch_uuid = watch.get("uuid") or uuid + company = _watch_company(watch, tag_map) + if not company or watch.get("paused") or watch_uuid in monitor_paused: + continue + url = canonicalize_url(watch.get("url", "")) or watch.get("url", "") + last_error = watch.get("last_error") or watch.get("last_error_text") + previous_successful_check = state.setdefault("last_successful_checks", {}).get(watch_uuid) + current_successful_check = None + if not last_error and watch.get("last_checked") is not None: + try: + current_successful_check = str(int(float(watch.get("last_checked")))) + state["last_successful_checks"][watch_uuid] = current_successful_check + except (TypeError, ValueError): + current_successful_check = None + if last_error: + count = int(state.setdefault("failure_counts", {}).get(watch_uuid, 0)) + 1 + state["failure_counts"][watch_uuid] = count + gap = { + "id": _event_id("coverage-gap", watch_uuid, str(count // 3)), + "type": "coverage_gap", + "competitor": company, + "url": url, + "category": "source failure", + "why": "This source is currently unknown coverage, not evidence that nothing changed.", + "evidence": str(last_error)[:350], + "observed_at": now, + "consecutive_failures": count, + } + coverage_gaps.append(gap) + if count == 1 or count % 3 == 0: + minor_events.append(gap) + else: + state.setdefault("failure_counts", {})[watch_uuid] = 0 + + try: + history = cd.request_json("GET", f"/watch/{watch_uuid}/history") or {} + except Exception as exc: + gap = { + "id": _event_id("history-gap", watch_uuid, now[:10]), + "type": "coverage_gap", + "competitor": company, + "url": url, + "category": "history API failure", + "why": "Change history could not be read, so coverage is unknown for this source.", + "evidence": f"{type(exc).__name__}: {exc}"[:350], + "observed_at": now, + } + coverage_gaps.append(gap) + minor_events.append(gap) + continue + timestamps = sorted((str(x) for x in history), key=lambda x: int(x) if x.isdigit() else x) + if not timestamps: + continue + last = state.setdefault("last_history", {}).get(watch_uuid) + if not last or last not in timestamps: + state["last_history"][watch_uuid] = timestamps[-1] + continue + start = timestamps.index(last) + 1 + previous = last + for timestamp in timestamps[start:]: + event_id = _event_id("change", watch_uuid, previous, timestamp) + if event_id in seen: + previous = timestamp + continue + try: + diff = cd.request_text( + "GET", + f"/watch/{watch_uuid}/difference/{previous}/{timestamp}", + params={"format": "text", "word_diff": "false"}, + ) + except Exception as exc: + diff = f"Difference unavailable: {type(exc).__name__}: {exc}" + severity, category, why = classify_change(url, diff) + event = { + "id": event_id, + "type": "page_change", + "competitor": company, + "watch_uuid": watch_uuid, + "url": url, + "category": category, + "why": why, + "evidence": _diff_excerpt(diff), + "from_timestamp": previous, + # ChangeDetection history contains distinct stored content + # versions, not every successful fetch. This explicit marker + # prevents callers from misreporting the snapshot gap as the + # check/detection latency or publication window. + "from_timestamp_semantics": "previous_distinct_snapshot_not_previous_check", + "to_timestamp": timestamp, + "to_timestamp_semantics": "first_stored_snapshot_with_changed_content", + "observed_at": now, + } + try: + has_raw_check_window = ( + previous_successful_check is not None + and current_successful_check is not None + and int(previous_successful_check) < int(current_successful_check) + and int(timestamp) == int(current_successful_check) + ) + except (TypeError, ValueError): + has_raw_check_window = False + if has_raw_check_window: + event.update({ + "last_successful_unchanged_check": str(previous_successful_check), + "first_successful_changed_check": str(current_successful_check), + "check_window_semantics": "consecutive_successful_fetches_observed_by_collector", + }) + if severity == "important": + important_events.append(event) + elif severity == "minor": + minor_events.append(event) + seen.add(event_id) + previous = timestamp + state["last_history"][watch_uuid] = timestamps[-1] + + important_events = deduplicate_events(important_events) + minor_events = deduplicate_events(minor_events) + _append_pending(state_dir, minor_events) + if send: + _append_important_outbox(state_dir, important_events) + state["seen_events"] = list(sorted(seen))[-5000:] + state["updated_at"] = now + save_json_atomic(state_path, state) + pending_important_count = len(load_json(state_dir / "pending-important.json", {"events": []}).get("events", [])) + result = { + "at": now, + "important_count": len(important_events), + "minor_count": len(minor_events), + "coverage_gap_count": len(coverage_gaps), + "delivered_important_count": 0, + "pending_important_count": pending_important_count, + "important_events": important_events, + "minor_events": minor_events, + "coverage_gaps": coverage_gaps, + } + # Persist the durable history/outbox state before external delivery. If a + # later send fails, delivered events have already been removed one by one + # and only the unsent tail is retried on the next timer run. + save_json_atomic(state_dir / "last-collector-run.json", result) + if send: + result["delivered_important_count"] = _deliver_important_outbox(config, state_dir, discord) + result["pending_important_count"] = len(load_json(state_dir / "pending-important.json", {"events": []}).get("events", [])) + save_json_atomic(state_dir / "last-collector-run.json", result) + return result + + +def _group_digest_topics(events: list[dict]) -> list[dict]: + grouped: dict[tuple, dict] = {} + for event in events: + key = ( + event.get("competitor"), event.get("type"), event.get("category"), event.get("url") + ) + source_ids = list(dict.fromkeys([ + event.get("id"), + *(event.get("_source_ids") or []), + ])) + if key not in grouped: + row = dict(event) + row["_source_ids"] = source_ids + row["change_count"] = 1 + grouped[key] = row + continue + row = grouped[key] + row["change_count"] += 1 + row["_source_ids"].extend(source_ids) + if event.get("evidence"): + row["evidence"] = event["evidence"] + if event.get("why"): + row["why"] = event["why"] + if event.get("observed_at"): + row["observed_at"] = event["observed_at"] + return list(grouped.values()) + + +def _digest_topic_block(event: dict) -> str: + count_line = f"\n**Observed:** {event['change_count']} changes; latest evidence shown." if event.get("change_count", 1) > 1 else "" + if event.get("type") == "coverage_gap": + return ( + f"\n⚠️ **{event['competitor']} — coverage gap**\n{event['url']}\n" + f"{event.get('evidence', '')[:300]}\n**Meaning:** {event['why']}{count_line}" + ) + evidence = event.get("evidence") + evidence_line = f"\n```text\n{evidence[:300]}\n```" if evidence else "" + timing = _event_fetch_window_line(event) + return ( + f"\n• **{event['competitor']} — {event['category']}**\n{event['url']}" + f"{count_line}{evidence_line}\n**Why it matters:** {event['why']}{timing}" + ) + + +def _digest_batches(topics: list[dict], event_count: int, limit: int = 1800) -> list[tuple[str, set[str]]]: + batches: list[tuple[str, set[str]]] = [] + header = f"🗞️ **Daily competitor intelligence digest — {len(topics)} topic(s) from {event_count} event(s)**" + continuation = "🗞️ **Daily competitor intelligence digest — continued**" + lines = [header] + source_ids: set[str] = set() + for topic in topics: + block = _digest_topic_block(topic) + if len("\n".join(lines)) + len(block) > limit and len(lines) > 1: + batches.append(("\n".join(lines), source_ids)) + lines = [continuation] + source_ids = set() + lines.append(block) + source_ids.update(x for x in topic.get("_source_ids", []) if x) + if len(lines) > 1: + batches.append(("\n".join(lines), source_ids)) + return batches + + +def run_digest(config: dict, state_dir: Path, discord, *, send: bool = True) -> dict: + path = state_dir / "pending-digest.json" + pending = load_json(path, {"events": []}) + all_queued_events = list(pending.get("events", [])) + # Coverage telemetry remains visible in status/state, but it is operational + # health rather than competitor intelligence and must not fill the digest. + queued_events = [ + event for event in all_queued_events + if event.get("type") not in {"coverage_gap", "coverage_recovered"} + and not _is_shared_minor_content_fragment(event) + ] + if send and len(queued_events) != len(all_queued_events): + save_json_atomic(path, {"events": queued_events}) + events = deduplicate_events(queued_events) + topics = _group_digest_topics(events) + now = utc_now() + if not topics: + result = {"at": now, "delivered_count": 0, "silent": True} + save_json_atomic(state_dir / "last-digest-run.json", result) + return result + batches = _digest_batches(topics, len(events)) + receipt = None + if send: + remaining = list(queued_events) + for message, source_ids in batches: + receipt = discord.send(message, mention=False) + remaining = [event for event in remaining if event.get("id") not in source_ids] + save_json_atomic(path, {"events": remaining}) + result = { + "at": now, + "delivered_count": len(events) if send else 0, + "topic_count": len(topics), + "message_count": len(batches) if send else 0, + "silent": False, + "receipt_message_id": receipt.get("id") if isinstance(receipt, dict) else None, + } + save_json_atomic(state_dir / "last-digest-run.json", result) + return result + + +def status_summary(config: dict, state_dir: Path, cd) -> dict: + discovery = load_json(state_dir / "last-discovery-run.json", {}) + collector = load_json(state_dir / "last-collector-run.json", {}) + digest = load_json(state_dir / "last-digest-run.json", {}) + watches = cd.request_json("GET", "/watch") or {} + tag_map = _tag_company_map(cd) + monitor_paused = monitor_paused_watch_ids(load_json(state_dir / "discovery.json", {})) + counts = {name: 0 for name in tag_map.values()} + paused = {name: 0 for name in tag_map.values()} + failures = {name: 0 for name in tag_map.values()} + for uuid, watch in watches.items(): + if not isinstance(watch, dict): + continue + name = _watch_company(watch, tag_map) + if name: + watch_uuid = watch.get("uuid") or uuid + if watch.get("paused") or watch_uuid in monitor_paused: + paused[name] = paused.get(name, 0) + 1 + continue + counts[name] = counts.get(name, 0) + 1 + if watch.get("last_error") or watch.get("last_error_text"): + failures[name] = failures.get(name, 0) + 1 + return { + "at": utc_now(), + "configured_competitors": [x["name"] for x in config.get("competitors", [])], + "watch_counts": counts, + "paused_watch_counts": paused, + "watch_failure_counts": failures, + "last_discovery": discovery.get("at"), + "last_discovery_summary": discovery.get("summary"), + "last_collector": collector.get("at"), + "last_digest": digest, + } + + +def _compact_for_stdout(command: str, result: dict) -> dict: + if command == "discover": + return { + "at": result.get("at"), + "summary": result.get("summary"), + "competitors": { + slug: { + "status": row.get("status"), + "url_count": len(row.get("urls", [])), + "selected_watch_count": row.get("selected_watch_count"), + "new_url_count": len(row.get("new_urls", [])), + "created_watch_count": row.get("created_watch_count"), + "confirmed_missing_count": len(row.get("confirmed_missing_urls", [])), + "error_count": len(row.get("errors", [])), + } + for slug, row in result.get("competitors", {}).items() + }, + } + if command == "collect": + return {k: result.get(k) for k in ( + "at", "important_count", "minor_count", "coverage_gap_count", + "delivered_important_count", "pending_important_count", + )} + return result + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("command", choices=["discover", "collect", "digest", "status"]) + parser.add_argument("--config", type=Path, default=DEFAULT_CONFIG) + parser.add_argument("--ignore", type=Path, default=DEFAULT_IGNORE) + parser.add_argument("--state-dir", type=Path, default=DEFAULT_STATE) + parser.add_argument("--env-file", type=Path, default=DEFAULT_ENV) + parser.add_argument("--no-send", action="store_true") + args = parser.parse_args(argv) + + config = load_json(args.config, None) + if not config: + raise SystemExit(f"configuration not found or invalid: {args.config}") + ignore = load_json(args.ignore, {"exact_urls": [], "regexes": [], "competitors": {}}) + env = load_env(args.env_file) + args.state_dir.mkdir(parents=True, exist_ok=True) + locks = [] + for lock_path, reason in ( + (args.state_dir / "monitor.lock", "another monitor command is running"), + (args.state_dir / f"{args.command}.lock", "already running"), + ): + lock = lock_path.open("w") + try: + flags = fcntl.LOCK_EX if args.command == "digest" and lock_path.name == "monitor.lock" else fcntl.LOCK_EX | fcntl.LOCK_NB + fcntl.flock(lock, flags) + except BlockingIOError: + lock.close() + for held in reversed(locks): + held.close() + print(json.dumps({"status": "skipped", "reason": reason, "command": args.command})) + return 0 + locks.append(lock) + + try: + cd = None + if args.command in {"discover", "collect", "status"}: + cd = ChangeDetectionAPI(config["changedetection_base"], env.get("CD_API_TOKEN", "")) + discord = None + if args.command in {"discover", "collect", "digest"} and not args.no_send: + discord_cfg = config.get("discord", {}) + discord = DiscordClient( + env.get("DISCORD_BOT_TOKEN", ""), + discord_cfg["thread_id"], + discord_cfg.get("important_mention_user_id"), + ) + try: + if args.command == "discover": + result = run_discover(config, ignore, args.state_dir, cd, discord, send=not args.no_send) + elif args.command == "collect": + result = run_collect(config, args.state_dir, cd, discord, send=not args.no_send) + elif args.command == "digest": + result = run_digest(config, args.state_dir, discord, send=not args.no_send) + else: + result = status_summary(config, args.state_dir, cd) + print(json.dumps(_compact_for_stdout(args.command, result), ensure_ascii=False, sort_keys=True)) + return 0 + except Exception as exc: + error = {"at": utc_now(), "command": args.command, "error": f"{type(exc).__name__}: {exc}"} + save_json_atomic(args.state_dir / f"last-{args.command}-error.json", error) + print(json.dumps(error, ensure_ascii=False), file=sys.stderr) + return 1 + finally: + for held in reversed(locks): + held.close() + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/notify_failure.py b/notify_failure.py new file mode 100755 index 0000000..d343953 --- /dev/null +++ b/notify_failure.py @@ -0,0 +1,47 @@ +#!/usr/bin/env python3 +"""Throttled Discord alert for a failed competitor-monitor systemd unit.""" +import hashlib +import json +from pathlib import Path +import subprocess +import sys +import time + +from monitor import DiscordClient, load_env, load_json, save_json_atomic, utc_now + +ROOT = Path(__file__).resolve().parent +unit = sys.argv[1] if len(sys.argv) > 1 else "unknown-unit" +state_path = ROOT / "state" / "service-failure-notifications.json" +state = load_json(state_path, {"units": {}}) +now = int(time.time()) +row = state.setdefault("units", {}).get(unit, {}) +# Avoid a failing 10-minute timer creating notification noise. +if now - int(row.get("last_sent_epoch", 0)) < 21600: + print(json.dumps({"unit": unit, "sent": False, "reason": "throttled"})) + raise SystemExit(0) + +config = load_json(ROOT / "config.json", {}) +env = load_env(ROOT / "secrets.env") +discord_cfg = config["discord"] +client = DiscordClient(env.get("DISCORD_BOT_TOKEN", ""), discord_cfg["thread_id"], discord_cfg.get("important_mention_user_id")) +try: + result = subprocess.run( + ["/usr/bin/systemctl", "show", unit, "--property=Result,ExecMainStatus,ActiveState,SubState", "--no-pager"], + text=True, + capture_output=True, + timeout=15, + ) + evidence = " ".join(x for x in result.stdout.splitlines() if x)[:500] or "systemd status was unavailable" +except Exception as exc: + evidence = f"status lookup failed: {type(exc).__name__}" +receipt = client.send( + f"🚨 **Competitor monitoring coverage gap**\nThe managed unit `{unit}` failed. Until it recovers, coverage is unknown—not ‘no changes’.\n`{evidence}`", + mention=True, +) +state["units"][unit] = { + "last_sent_at": utc_now(), + "last_sent_epoch": now, + "receipt_message_id": receipt.get("id") if isinstance(receipt, dict) else None, +} +save_json_atomic(state_path, state) +print(json.dumps({"unit": unit, "sent": True, "receipt_message_id": state["units"][unit]["receipt_message_id"]})) diff --git a/requirements-verification.txt b/requirements-verification.txt new file mode 100644 index 0000000..fc77dc6 --- /dev/null +++ b/requirements-verification.txt @@ -0,0 +1 @@ +websockets>=12,<16 diff --git a/scripts/build-release.sh b/scripts/build-release.sh new file mode 100755 index 0000000..67c762f --- /dev/null +++ b/scripts/build-release.sh @@ -0,0 +1,30 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +cd "$ROOT" +VERSION="${1:-$(git rev-parse --short=12 HEAD)}" +if [[ ! "$VERSION" =~ ^[A-Za-z0-9][A-Za-z0-9._-]*$ ]]; then + printf 'Invalid release version: %s\n' "$VERSION" >&2 + exit 2 +fi +PREFIX="changedetection-competitor-monitor-${VERSION}" +OUT_DIR="$ROOT/release" +ARCHIVE="$OUT_DIR/${PREFIX}.tar.gz" + +if [[ -n "$(git status --porcelain)" ]]; then + printf 'Refusing to package a dirty or untracked working tree.\n' >&2 + exit 1 +fi +mkdir -p "$OUT_DIR" +SOURCE_DATE_EPOCH="${SOURCE_DATE_EPOCH:-$(git log -1 --format=%ct)}" +export SOURCE_DATE_EPOCH +git archive --format=tar --prefix="${PREFIX}/" HEAD | gzip -n -9 >"$ARCHIVE" +gzip -t "$ARCHIVE" +tar -tzf "$ARCHIVE" >/dev/null +( + cd "$OUT_DIR" + sha256sum "$(basename "$ARCHIVE")" >"$(basename "$ARCHIVE").sha256" + sha256sum -c "$(basename "$ARCHIVE.sha256")" >/dev/null +) +printf '%s\n' "$ARCHIVE" diff --git a/scripts/doctor.py b/scripts/doctor.py new file mode 100755 index 0000000..6fbdce8 --- /dev/null +++ b/scripts/doctor.py @@ -0,0 +1,228 @@ +#!/usr/bin/env python3 +"""Validate repository, local configuration, and optional live integration readiness.""" +from __future__ import annotations + +import argparse +import json +import os +from pathlib import Path +import shutil +import subprocess +import sys +import urllib.error +import urllib.request + +ROOT = Path(__file__).resolve().parents[1] +PLACEHOLDER = "REPLACE_WITH_" + + +def load_json(path: Path) -> dict: + with path.open() as handle: + value = json.load(handle) + if not isinstance(value, dict): + raise ValueError(f"{path.name} must contain a JSON object") + return value + + +def load_env(path: Path) -> dict[str, str]: + result: dict[str, str] = {} + for raw in path.read_text().splitlines(): + line = raw.strip() + if not line or line.startswith("#") or "=" not in line: + continue + key, value = line.split("=", 1) + result[key.strip()] = value.strip().strip('"').strip("'") + return result + + +def http_json( + url: str, + headers: dict[str, str], + *, + method: str = "GET", + data: dict | None = None, +) -> tuple[int, object]: + body = None if data is None else json.dumps(data).encode() + request_headers = dict(headers) + if body is not None: + request_headers["Content-Type"] = "application/json" + request = urllib.request.Request(url, headers=request_headers, data=body, method=method) + try: + with urllib.request.urlopen(request, timeout=20) as response: + body = response.read() + return response.status, json.loads(body) if body else None + except urllib.error.HTTPError as exc: + return exc.code, None + except (urllib.error.URLError, TimeoutError, OSError): + return 0, None + + +def add(checks: dict[str, dict], name: str, ok: bool, detail: object) -> None: + checks[name] = {"ok": bool(ok), "detail": detail} + + +def repository_checks(checks: dict[str, dict]) -> None: + expected = [ + "monitor.py", "notify_failure.py", "test_monitor.py", "compose.yaml", "compose.browser.yaml", + "config.example.json", "ignore.example.json", "secrets.env.example", + "scripts/setup.sh", "scripts/install-systemd.sh", "README.md", + ] + missing = [name for name in expected if not (ROOT / name).exists()] + add(checks, "repository_files", not missing, {"missing": missing}) + try: + cfg = load_json(ROOT / "config.example.json") + ignore = load_json(ROOT / "ignore.example.json") + add(checks, "example_json", bool(cfg.get("competitors")) and ignore.get("version") == 1, "valid") + except Exception as exc: + add(checks, "example_json", False, type(exc).__name__) + docker = shutil.which("docker") + add(checks, "docker_cli", bool(docker), docker or "not found") + if docker: + proc = subprocess.run( + [docker, "compose", "--project-directory", str(ROOT), "-f", str(ROOT / "compose.yaml"), "config", "--quiet"], + text=True, + capture_output=True, + ) + add(checks, "compose_config", proc.returncode == 0, "valid" if proc.returncode == 0 else proc.stderr.strip()[:300]) + browser_proc = subprocess.run( + [ + docker, "compose", "--project-directory", str(ROOT), + "-f", str(ROOT / "compose.yaml"), + "-f", str(ROOT / "compose.browser.yaml"), + "config", "--quiet", + ], + text=True, + capture_output=True, + ) + add( + checks, + "compose_browser_config", + browser_proc.returncode == 0, + "valid" if browser_proc.returncode == 0 else browser_proc.stderr.strip()[:300], + ) + + +def configured_checks(checks: dict[str, dict]) -> tuple[dict, dict]: + required = [ROOT / ".env", ROOT / "config.json", ROOT / "ignore.json", ROOT / "secrets.env"] + missing = [path.name for path in required if not path.exists()] + add(checks, "local_files", not missing, {"missing": missing}) + if missing: + return {}, {} + try: + config = load_json(ROOT / "config.json") + load_json(ROOT / "ignore.json") + env = load_env(ROOT / "secrets.env") + except Exception as exc: + add(checks, "local_configuration_parse", False, type(exc).__name__) + return {}, {} + add(checks, "local_configuration_parse", True, "valid") + values = { + "discord.thread_id": config.get("discord", {}).get("thread_id", ""), + "CD_API_TOKEN": env.get("CD_API_TOKEN", ""), + "DISCORD_BOT_TOKEN": env.get("DISCORD_BOT_TOKEN", ""), + } + unresolved = [name for name, value in values.items() if not value or str(value).startswith(PLACEHOLDER)] + mention = config.get("discord", {}).get("important_mention_user_id", "") + if mention and str(mention).startswith(PLACEHOLDER): + unresolved.append("discord.important_mention_user_id") + add(checks, "credentials_and_targets_configured", not unresolved, {"unresolved_fields": unresolved}) + unreadable = [path.name for path in required if not os.access(path, os.R_OK)] + add(checks, "runtime_files_readable", not unreadable, {"unreadable": unreadable}) + mode = (ROOT / "secrets.env").stat().st_mode & 0o777 + add(checks, "secrets_permissions", mode & 0o077 == 0, oct(mode)) + state_dir = ROOT / "state" + state_ok = state_dir.is_dir() and os.access(state_dir, os.R_OK | os.W_OK | os.X_OK) + inaccessible_state = [] + if state_ok: + for path in state_dir.rglob("*"): + mode = os.R_OK | os.W_OK | (os.X_OK if path.is_dir() else 0) + if not os.access(path, mode): + inaccessible_state.append(str(path.relative_to(state_dir))) + if len(inaccessible_state) >= 20: + break + add( + checks, + "state_directory", + state_ok and not inaccessible_state, + {"path": str(state_dir), "inaccessible_entries": inaccessible_state}, + ) + return config, env + + +def live_checks( + checks: dict[str, dict], + config: dict, + env: dict, + *, + verify_discord_send: bool = False, +) -> None: + if not config or not env or any(not row["ok"] for row in checks.values()): + add(checks, "live_checks", False, "configuration is not ready") + return + base = config["changedetection_base"].rstrip("/") + status, watches = http_json(base + "/watch", {"x-api-key": env["CD_API_TOKEN"]}) + add(checks, "changedetection_api", status == 200 and isinstance(watches, dict), {"status": status, "watch_count": len(watches) if isinstance(watches, dict) else None}) + headers = {"Authorization": "Bot " + env["DISCORD_BOT_TOKEN"], "User-Agent": "CompetitorMonitorDoctor/1.0"} + status, me = http_json("https://discord.com/api/v10/users/@me", headers) + add(checks, "discord_bot", status == 200 and isinstance(me, dict) and bool(me.get("id")), {"status": status}) + channel_id = config["discord"]["thread_id"] + status, channel = http_json(f"https://discord.com/api/v10/channels/{channel_id}", headers) + add(checks, "discord_target", status == 200 and isinstance(channel, dict) and str(channel.get("id")) == str(channel_id), {"status": status, "id_matches": isinstance(channel, dict) and str(channel.get("id")) == str(channel_id)}) + if verify_discord_send: + message_url = f"https://discord.com/api/v10/channels/{channel_id}/messages" + status, message = http_json( + message_url, + headers, + method="POST", + data={"content": "Competitor-monitor installation check — deleting this message automatically."}, + ) + message_id = str(message.get("id", "")) if isinstance(message, dict) else "" + add(checks, "discord_send", status == 200 and bool(message_id), {"status": status, "message_created": bool(message_id)}) + if message_id: + item_url = f"{message_url}/{message_id}" + read_status, readback = http_json(item_url, headers) + add( + checks, + "discord_readback", + read_status == 200 and isinstance(readback, dict) and str(readback.get("id")) == message_id, + {"status": read_status, "id_matches": isinstance(readback, dict) and str(readback.get("id")) == message_id}, + ) + delete_status, _ = http_json(item_url, headers, method="DELETE") + add(checks, "discord_delete", delete_status == 204, {"status": delete_status}) + absent_status, _ = http_json(item_url, headers) + add(checks, "discord_deleted_absent", absent_status == 404, {"status": absent_status}) + else: + add(checks, "discord_readback", False, "not attempted because send failed") + add(checks, "discord_delete", False, "not attempted because send failed") + add(checks, "discord_deleted_absent", False, "not attempted because send failed") + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--mode", choices=("repository", "configured", "live", "systemd-ready"), default="configured") + parser.add_argument( + "--verify-discord-send", + action="store_true", + help="create, read back, and delete a Discord probe message", + ) + args = parser.parse_args() + if args.verify_discord_send and args.mode not in {"live", "systemd-ready"}: + parser.error("--verify-discord-send requires --mode live or --mode systemd-ready") + checks: dict[str, dict] = {} + repository_checks(checks) + config: dict = {} + env: dict = {} + if args.mode != "repository": + config, env = configured_checks(checks) + if args.mode in {"live", "systemd-ready"}: + live_checks(checks, config, env, verify_discord_send=args.verify_discord_send) + if args.mode == "systemd-ready": + add(checks, "install_path_has_no_whitespace", not any(ch.isspace() for ch in str(ROOT)), str(ROOT)) + add(checks, "systemctl", bool(shutil.which("systemctl")), shutil.which("systemctl") or "not found") + ok = all(row["ok"] for row in checks.values()) + print(json.dumps({"ok": ok, "mode": args.mode, "checks": checks}, sort_keys=True)) + return 0 if ok else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/install-systemd.sh b/scripts/install-systemd.sh new file mode 100755 index 0000000..c1d8a3b --- /dev/null +++ b/scripts/install-systemd.sh @@ -0,0 +1,63 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +RUN_USER="${SUDO_USER:-${USER:-}}" +RUN_GROUP="" +PYTHON="$(command -v python3)" +RUNUSER="$(command -v runuser || true)" +SYSTEM_DIR="/etc/systemd/system" + +usage() { + printf 'Usage: sudo %s [--user USER] [--group GROUP] [--python PATH]\n' "$0" +} + +while (($#)); do + case "$1" in + --user) RUN_USER="$2"; shift 2 ;; + --group) RUN_GROUP="$2"; shift 2 ;; + --python) PYTHON="$2"; shift 2 ;; + -h|--help) usage; exit 0 ;; + *) printf 'Unknown argument: %s\n' "$1" >&2; usage >&2; exit 2 ;; + esac +done + +if [[ $EUID -ne 0 ]]; then + printf 'Run this installer with sudo/root.\n' >&2 + exit 1 +fi +if [[ -z "$RUN_USER" ]]; then + printf 'Could not determine runtime user; pass --user.\n' >&2 + exit 1 +fi +if [[ -z "$RUNUSER" ]]; then + printf 'runuser is required to validate the configured runtime account.\n' >&2 + exit 1 +fi +if [[ -z "$RUN_GROUP" ]]; then + RUN_GROUP="$(id -gn "$RUN_USER")" +fi + +install -d -m 0750 -o "$RUN_USER" -g "$RUN_GROUP" "$ROOT/state" +"$RUNUSER" -u "$RUN_USER" -g "$RUN_GROUP" -- \ + "$PYTHON" "$ROOT/scripts/doctor.py" --mode systemd-ready --verify-discord-send +STAGE_DIR="$(mktemp -d /run/competitor-monitor-systemd.XXXXXX)" +trap 'rm -rf "$STAGE_DIR"' EXIT +"$PYTHON" "$ROOT/scripts/render_systemd.py" \ + --install-dir "$ROOT" --run-user "$RUN_USER" --run-group "$RUN_GROUP" \ + --python "$PYTHON" --output-dir "$STAGE_DIR" >/dev/null +systemd-analyze verify "$STAGE_DIR"/*.service "$STAGE_DIR"/*.timer + +install -m 0644 "$STAGE_DIR"/competitor-monitor-*.service "$SYSTEM_DIR/" +install -m 0644 "$STAGE_DIR"/competitor-monitor-*.timer "$SYSTEM_DIR/" +systemctl daemon-reload +systemctl enable --now \ + competitor-monitor-discover.timer \ + competitor-monitor-collect.timer \ + competitor-monitor-digest.timer + +for unit in competitor-monitor-discover.timer competitor-monitor-collect.timer competitor-monitor-digest.timer; do + systemctl is-enabled --quiet "$unit" + systemctl is-active --quiet "$unit" +done +printf 'Installed and verified 3 enabled/active timers for %s:%s.\n' "$RUN_USER" "$RUN_GROUP" diff --git a/scripts/render_systemd.py b/scripts/render_systemd.py new file mode 100755 index 0000000..eec7cf0 --- /dev/null +++ b/scripts/render_systemd.py @@ -0,0 +1,73 @@ +#!/usr/bin/env python3 +"""Render portable systemd unit templates for this checkout.""" +from __future__ import annotations + +import argparse +import grp +import os +from pathlib import Path +import pwd +import shutil +import sys + +ROOT = Path(__file__).resolve().parents[1] +TOKENS = ("@INSTALL_DIR@", "@RUN_USER@", "@RUN_GROUP@", "@PYTHON@") + + +def safe_absolute(value: str, label: str) -> str: + path = str(Path(value).expanduser().resolve()) + if any(ch.isspace() for ch in path): + raise SystemExit(f"{label} cannot contain whitespace: {path}") + return path + + +def render(output_dir: Path, install_dir: str, run_user: str, run_group: str, python: str) -> list[Path]: + pwd.getpwnam(run_user) + grp.getgrnam(run_group) + values = { + "@INSTALL_DIR@": safe_absolute(install_dir, "install directory"), + "@RUN_USER@": run_user, + "@RUN_GROUP@": run_group, + "@PYTHON@": safe_absolute(python, "Python path"), + } + output_dir.mkdir(parents=True, exist_ok=True) + rendered: list[Path] = [] + for source in sorted((ROOT / "systemd").glob("*.service.in")): + text = source.read_text() + for token, value in values.items(): + text = text.replace(token, value) + leftovers = [token for token in TOKENS if token in text] + if leftovers: + raise SystemExit(f"unrendered tokens in {source.name}: {leftovers}") + target = output_dir / source.name.removesuffix(".in") + target.write_text(text) + target.chmod(0o644) + rendered.append(target) + for source in sorted((ROOT / "systemd").glob("*.timer")): + target = output_dir / source.name + shutil.copyfile(source, target) + target.chmod(0o644) + rendered.append(target) + if len(rendered) != 7: + raise SystemExit(f"expected 7 rendered units, got {len(rendered)}") + return rendered + + +def main() -> int: + current_user = pwd.getpwuid(os.getuid()).pw_name + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--install-dir", default=str(ROOT)) + parser.add_argument("--run-user", default=current_user) + parser.add_argument("--run-group") + parser.add_argument("--python", default=shutil.which("python3") or sys.executable) + parser.add_argument("--output-dir", type=Path, default=ROOT / "build" / "systemd") + args = parser.parse_args() + run_group = args.run_group or grp.getgrgid(pwd.getpwnam(args.run_user).pw_gid).gr_name + files = render(args.output_dir, args.install_dir, args.run_user, run_group, args.python) + for path in files: + print(path) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/setup.sh b/scripts/setup.sh new file mode 100755 index 0000000..d4243a2 --- /dev/null +++ b/scripts/setup.sh @@ -0,0 +1,43 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +START=1 + +usage() { + printf 'Usage: %s [--no-start]\n' "$0" +} +while (($#)); do + case "$1" in + --no-start) START=0; shift ;; + -h|--help) usage; exit 0 ;; + *) printf 'Unknown argument: %s\n' "$1" >&2; usage >&2; exit 2 ;; + esac +done + +copy_if_missing() { + local source="$1" target="$2" + if [[ ! -e "$target" ]]; then + cp "$source" "$target" + printf 'Created %s\n' "${target#"$ROOT"/}" + fi +} + +copy_if_missing "$ROOT/.env.example" "$ROOT/.env" +copy_if_missing "$ROOT/secrets.env.example" "$ROOT/secrets.env" +copy_if_missing "$ROOT/config.example.json" "$ROOT/config.json" +copy_if_missing "$ROOT/ignore.example.json" "$ROOT/ignore.json" +install -d -m 0750 "$ROOT/state" "$ROOT/data/changedetection" "$ROOT/verification" +chmod 0600 "$ROOT/secrets.env" + +python3 -m json.tool "$ROOT/config.json" >/dev/null +python3 -m json.tool "$ROOT/ignore.json" >/dev/null +docker compose --project-directory "$ROOT" -f "$ROOT/compose.yaml" config --quiet +docker compose --project-directory "$ROOT" -f "$ROOT/compose.yaml" -f "$ROOT/compose.browser.yaml" config --quiet + +if ((START)); then + docker compose --project-directory "$ROOT" -f "$ROOT/compose.yaml" up -d + printf 'ChangeDetection started. Open the PUBLIC_BASE_URL from .env, set its native password, then copy the API token and Discord bot token into secrets.env.\n' +else + printf 'Local files initialized and Compose validated; containers were not started.\n' +fi diff --git a/scripts/test.sh b/scripts/test.sh new file mode 100755 index 0000000..f07f9cd --- /dev/null +++ b/scripts/test.sh @@ -0,0 +1,20 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +cd "$ROOT" + +python3 -m unittest -v test_monitor.py tests/test_repository.py +python3 -m compileall -q \ + monitor.py notify_failure.py test_monitor.py scripts verification tests +bash -n scripts/*.sh +python3 scripts/doctor.py --mode repository +python3 scripts/render_systemd.py --output-dir build/systemd-test >/dev/null + +if command -v systemd-analyze >/dev/null 2>&1; then + systemd-analyze verify build/systemd-test/*.service build/systemd-test/*.timer +fi + +docker compose --project-directory "$ROOT" -f compose.yaml config --quiet +docker compose --project-directory "$ROOT" -f compose.yaml -f compose.browser.yaml config --quiet +printf 'All repository gates passed.\n' diff --git a/scripts/uninstall-systemd.sh b/scripts/uninstall-systemd.sh new file mode 100755 index 0000000..8ecc5f7 --- /dev/null +++ b/scripts/uninstall-systemd.sh @@ -0,0 +1,25 @@ +#!/usr/bin/env bash +set -euo pipefail + +if [[ $EUID -ne 0 ]]; then + printf 'Run this remover with sudo/root.\n' >&2 + exit 1 +fi + +units=( + competitor-monitor-discover.timer + competitor-monitor-collect.timer + competitor-monitor-digest.timer +) +systemctl disable --now "${units[@]}" 2>/dev/null || true +rm -f \ + /etc/systemd/system/competitor-monitor-discover.service \ + /etc/systemd/system/competitor-monitor-collect.service \ + /etc/systemd/system/competitor-monitor-digest.service \ + /etc/systemd/system/competitor-monitor-failure@.service \ + /etc/systemd/system/competitor-monitor-discover.timer \ + /etc/systemd/system/competitor-monitor-collect.timer \ + /etc/systemd/system/competitor-monitor-digest.timer +systemctl daemon-reload +systemctl reset-failed competitor-monitor-discover.service competitor-monitor-collect.service competitor-monitor-digest.service 2>/dev/null || true +printf 'Removed competitor-monitor systemd units. Repository data and ChangeDetection datastore were preserved.\n' diff --git a/secrets.env.example b/secrets.env.example new file mode 100644 index 0000000..ba810b8 --- /dev/null +++ b/secrets.env.example @@ -0,0 +1,4 @@ +# Local monitor credentials. Copy to secrets.env, replace both values, and chmod 600. +# Obtain CD_API_TOKEN from ChangeDetection.io Settings -> API. +CD_API_TOKEN=REPLACE_WITH_CHANGEDETECTION_API_TOKEN +DISCORD_BOT_TOKEN=REPLACE_WITH_DISCORD_BOT_TOKEN diff --git a/systemd/competitor-monitor-collect.service.in b/systemd/competitor-monitor-collect.service.in new file mode 100644 index 0000000..9d7a1af --- /dev/null +++ b/systemd/competitor-monitor-collect.service.in @@ -0,0 +1,23 @@ +[Unit] +Description=Collect and classify ChangeDetection competitor history +Wants=network-online.target +After=network-online.target docker.service +OnFailure=competitor-monitor-failure@%n.service +ConditionPathExists=@INSTALL_DIR@/config.json + +[Service] +Type=oneshot +User=@RUN_USER@ +Group=@RUN_GROUP@ +WorkingDirectory=@INSTALL_DIR@ +ExecStart=@PYTHON@ @INSTALL_DIR@/monitor.py collect --config @INSTALL_DIR@/config.json --ignore @INSTALL_DIR@/ignore.json --state-dir @INSTALL_DIR@/state --env-file @INSTALL_DIR@/secrets.env +TimeoutStartSec=10min +UMask=0027 +NoNewPrivileges=true +PrivateTmp=true +ProtectSystem=strict +ProtectHome=read-only +ReadWritePaths=@INSTALL_DIR@/state +RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6 +LockPersonality=true +MemoryDenyWriteExecute=true diff --git a/systemd/competitor-monitor-collect.timer b/systemd/competitor-monitor-collect.timer new file mode 100644 index 0000000..a290354 --- /dev/null +++ b/systemd/competitor-monitor-collect.timer @@ -0,0 +1,13 @@ +[Unit] +Description=Poll competitor change history every ten minutes + +[Timer] +OnBootSec=5min +OnUnitActiveSec=10min +RandomizedDelaySec=30s +AccuracySec=30s +Persistent=true +Unit=competitor-monitor-collect.service + +[Install] +WantedBy=timers.target diff --git a/systemd/competitor-monitor-digest.service.in b/systemd/competitor-monitor-digest.service.in new file mode 100644 index 0000000..e905715 --- /dev/null +++ b/systemd/competitor-monitor-digest.service.in @@ -0,0 +1,27 @@ +[Unit] +Description=Deliver the daily minor competitor-change digest +Wants=network-online.target +After=network-online.target +OnFailure=competitor-monitor-failure@%n.service +ConditionPathExists=@INSTALL_DIR@/config.json +StartLimitIntervalSec=20min +StartLimitBurst=3 + +[Service] +Type=oneshot +User=@RUN_USER@ +Group=@RUN_GROUP@ +WorkingDirectory=@INSTALL_DIR@ +ExecStart=@PYTHON@ @INSTALL_DIR@/monitor.py digest --config @INSTALL_DIR@/config.json --ignore @INSTALL_DIR@/ignore.json --state-dir @INSTALL_DIR@/state --env-file @INSTALL_DIR@/secrets.env +TimeoutStartSec=30min +Restart=on-failure +RestartSec=5min +UMask=0027 +NoNewPrivileges=true +PrivateTmp=true +ProtectSystem=strict +ProtectHome=read-only +ReadWritePaths=@INSTALL_DIR@/state +RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6 +LockPersonality=true +MemoryDenyWriteExecute=true diff --git a/systemd/competitor-monitor-digest.timer b/systemd/competitor-monitor-digest.timer new file mode 100644 index 0000000..ca83e1a --- /dev/null +++ b/systemd/competitor-monitor-digest.timer @@ -0,0 +1,12 @@ +[Unit] +Description=Send minor competitor intelligence digest daily at 08:30 UTC + +[Timer] +OnCalendar=*-*-* 08:30:00 UTC +RandomizedDelaySec=2min +AccuracySec=1min +Persistent=true +Unit=competitor-monitor-digest.service + +[Install] +WantedBy=timers.target diff --git a/systemd/competitor-monitor-discover.service.in b/systemd/competitor-monitor-discover.service.in new file mode 100644 index 0000000..89072c7 --- /dev/null +++ b/systemd/competitor-monitor-discover.service.in @@ -0,0 +1,23 @@ +[Unit] +Description=Discover competitor pages and reconcile ChangeDetection watches +Wants=network-online.target +After=network-online.target docker.service +OnFailure=competitor-monitor-failure@%n.service +ConditionPathExists=@INSTALL_DIR@/config.json + +[Service] +Type=oneshot +User=@RUN_USER@ +Group=@RUN_GROUP@ +WorkingDirectory=@INSTALL_DIR@ +ExecStart=@PYTHON@ @INSTALL_DIR@/monitor.py discover --config @INSTALL_DIR@/config.json --ignore @INSTALL_DIR@/ignore.json --state-dir @INSTALL_DIR@/state --env-file @INSTALL_DIR@/secrets.env +TimeoutStartSec=30min +UMask=0027 +NoNewPrivileges=true +PrivateTmp=true +ProtectSystem=strict +ProtectHome=read-only +ReadWritePaths=@INSTALL_DIR@/state +RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6 +LockPersonality=true +MemoryDenyWriteExecute=true diff --git a/systemd/competitor-monitor-discover.timer b/systemd/competitor-monitor-discover.timer new file mode 100644 index 0000000..163cbe6 --- /dev/null +++ b/systemd/competitor-monitor-discover.timer @@ -0,0 +1,13 @@ +[Unit] +Description=Run competitor page discovery every six hours + +[Timer] +OnBootSec=3min +OnUnitActiveSec=6h +RandomizedDelaySec=5min +AccuracySec=1min +Persistent=true +Unit=competitor-monitor-discover.service + +[Install] +WantedBy=timers.target diff --git a/systemd/competitor-monitor-failure@.service.in b/systemd/competitor-monitor-failure@.service.in new file mode 100644 index 0000000..44f6fee --- /dev/null +++ b/systemd/competitor-monitor-failure@.service.in @@ -0,0 +1,21 @@ +[Unit] +Description=Report a competitor-monitor service coverage gap +Wants=network-online.target +After=network-online.target + +[Service] +Type=oneshot +User=@RUN_USER@ +Group=@RUN_GROUP@ +WorkingDirectory=@INSTALL_DIR@ +ExecStart=@PYTHON@ @INSTALL_DIR@/notify_failure.py %i +TimeoutStartSec=2min +UMask=0027 +NoNewPrivileges=true +PrivateTmp=true +ProtectSystem=strict +ProtectHome=read-only +ReadWritePaths=@INSTALL_DIR@/state +RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6 +LockPersonality=true +MemoryDenyWriteExecute=true diff --git a/test_monitor.py b/test_monitor.py new file mode 100755 index 0000000..d4f8db4 --- /dev/null +++ b/test_monitor.py @@ -0,0 +1,1303 @@ +#!/usr/bin/env python3 +import contextlib +import fcntl +import io +import json +import os +import tempfile +import threading +import time +import unittest +from unittest import mock +import urllib.error +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from urllib.parse import parse_qs, urlparse + +import monitor + + +class MutableFixture: + pages = {} + requests = [] + + +class FixtureHandler(BaseHTTPRequestHandler): + def do_GET(self): + MutableFixture.requests.append((self.command, self.path, self.headers.get("User-Agent"))) + path = urlparse(self.path).path + status, content_type, body = MutableFixture.pages.get(path, (404, "text/plain", "missing")) + self.send_response(status) + self.send_header("Content-Type", content_type) + self.end_headers() + self.wfile.write(body.encode()) + + def log_message(self, *_args): + pass + + +class FakeCD: + def __init__(self): + self.tags = {} + self.watches = {} + self.histories = {} + self.diffs = {} + self.import_calls = [] + self.put_calls = [] + self.next_id = 1 + + def request_json(self, method, path, *, params=None, data=None): + if method == "GET" and path == "/tags": + return dict(self.tags) + if method == "POST" and path == "/tag": + tag_id = f"tag-{len(self.tags)+1}" + payload = data if isinstance(data, dict) else json.loads(data.decode()) + self.tags[tag_id] = {"uuid": tag_id, "title": payload["title"]} + return {"uuid": tag_id} + if method == "GET" and path == "/watch": + return dict(self.watches) + if method == "GET" and path.startswith("/watch/") and path.count("/") == 2: + watch_id = path.rsplit("/", 1)[-1] + return dict(self.watches[watch_id]) + if method == "POST" and path == "/import": + body = data.decode() if isinstance(data, (bytes, bytearray)) else str(data) + urls = [u for u in body.splitlines() if u] + self.import_calls.append((urls, dict(params or {}))) + created = [] + existing = {w["url"] for w in self.watches.values()} + for url in urls: + if url in existing: + continue + watch_id = f"watch-{self.next_id}" + self.next_id += 1 + self.watches[watch_id] = { + "uuid": watch_id, + "url": url, + "title": url, + "tags": [params.get("tag_uuids")], + "last_checked": 0, + "last_changed": 0, + "last_error": False, + } + created.append(watch_id) + existing.add(url) + return created + if method == "PUT" and path.startswith("/watch/"): + watch_id = path.rsplit("/", 1)[-1] + payload = data if isinstance(data, dict) else json.loads(data.decode()) + self.put_calls.append((watch_id, dict(payload))) + self.watches[watch_id].update(payload) + return self.watches[watch_id] + if method == "GET" and path.startswith("/watch/") and path.endswith("/history"): + watch_id = path.split("/")[2] + return self.histories.get(watch_id, {}) + raise AssertionError((method, path, params, data)) + + def request_text(self, method, path, *, params=None, data=None): + if method == "GET" and "/difference/" in path: + watch_id = path.split("/")[2] + bits = path.split("/") + return self.diffs[(watch_id, bits[-2], bits[-1])] + raise AssertionError((method, path, params, data)) + + +class CaptureDiscord: + def __init__(self): + self.messages = [] + + def send(self, content, *, mention=False, reply_to=None): + self.messages.append({"content": content, "mention": mention, "reply_to": reply_to}) + return {"id": f"message-{len(self.messages)}", "channel_id": "thread-test"} + + +class FailAfterOneDiscord(CaptureDiscord): + def send(self, content, *, mention=False, reply_to=None): + if len(self.messages) >= 1: + raise RuntimeError("synthetic delivery failure") + return super().send(content, mention=mention, reply_to=reply_to) + + +class AlwaysFailDiscord(CaptureDiscord): + def send(self, content, *, mention=False, reply_to=None): + raise RuntimeError("synthetic delivery failure") + + +class MonitorTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.server = ThreadingHTTPServer(("127.0.0.1", 0), FixtureHandler) + cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True) + cls.thread.start() + cls.origin = f"http://127.0.0.1:{cls.server.server_port}" + + @classmethod + def tearDownClass(cls): + cls.server.shutdown() + cls.server.server_close() + + def setUp(self): + MutableFixture.requests = [] + MutableFixture.pages = { + "/robots.txt": (200, "text/plain", "User-agent: *\nAllow: /\nSitemap: %s/sitemap-index.xml\n" % self.origin), + "/sitemap-index.xml": (200, "application/xml", "%s/nested.xml" % self.origin), + "/nested.xml": (200, "application/xml", "%s/%s/products%s/company/12345678-demo" % (self.origin, self.origin, self.origin)), + "/": (200, "text/html", "PricingProductsAssetSearch"), + "/products": (200, "text/html", "API"), + "/pricing": (200, "text/html", "Plans"), + "/products/api": (200, "text/html", "API"), + "/feed.xml": (200, "application/rss+xml", "%s/blog/new-feature" % self.origin), + "/blog/new-feature": (200, "text/html", "New"), + } + + def make_config(self, root, *, blocked=False): + return { + "version": 1, + "user_agent": "CompetitorMonitorTest/1.0", + "changedetection_base": "http://unused/api/v1", + "discord": {"thread_id": "thread-test", "important_mention_user_id": "user-test"}, + "global": { + "max_sitemaps": 10, + "max_urls": 50, + "crawl_depth": 2, + "crawl_pages": 20, + "request_delay_seconds": 0, + "watch_check_hours": 6, + "missing_confirmations": 2, + "fetch_timeout_seconds": 3, + "exclude_regexes": [r"/assets/", r"/search$", r"/company/[^/]*\\d{6,}"], + "exclude_extensions": [".png", ".jpg", ".pdf"], + }, + "competitors": [{ + "slug": "fixture", + "name": "Fixture Co", + "domains": [urlparse(self.origin).netloc], + "seeds": [self.origin + "/"], + "sitemap_seeds": [], + "allowed_path_prefixes": ["/"], + "exclude_regexes": [], + "exclude_sitemap_regexes": [], + "fetch_backend": "html_requests", + "respect_robots": True, + "policy_watch_urls": [self.origin + "/robots.txt"] if blocked else [], + }], + } + + def test_canonicalize_strips_tracking_query_fragment_and_default_port(self): + self.assertEqual( + monitor.canonicalize_url("HTTPS://Example.COM:443/path/?utm_source=x&a=1#top"), + "https://example.com/path", + ) + + def test_crtsh_source_preserves_query_and_filters_to_identity(self): + class StubFetcher: + def __init__(self): + self.urls = [] + + def get(self, url, *, delay=None): + self.urls.append(url) + body = json.dumps([ + { + "common_name": "competitor.example", + "name_value": "competitor.example\n*.competitor.example\napi.competitor.example\noutside.example", + }, + {"common_name": "API.COMPETITOR.EXAMPLE.", "name_value": "test-api.competitor.example"}, + ]).encode() + return url, body, {"content-type": "application/json"}, 200 + + source = "https://crt.sh/json?identity=competitor.example" + fetcher = StubFetcher() + result = monitor.discover_certificate_transparency( + { + "certificate_transparency_sources": [ + {"url": source, "identity": "competitor.example"}, + ], + }, + {"request_delay_seconds": 0}, + fetcher, + ) + self.assertEqual(fetcher.urls, [source]) + self.assertEqual(result["status"], "ok") + self.assertEqual(result["names"], [ + "api.competitor.example", + "competitor.example", + "test-api.competitor.example", + ]) + self.assertEqual(result["sources"], [source]) + + def test_crtsh_transient_http_error_is_retried_before_coverage_gap(self): + class FlakyFetcher: + def __init__(self): + self.calls = 0 + + def get(self, url, *, delay=None): + self.calls += 1 + if self.calls < 4: + raise urllib.error.HTTPError(url, 404, "transient", None, None) + body = json.dumps([{ + "common_name": "competitor.example", + "name_value": "api.competitor.example", + }]).encode() + return url, body, {"content-type": "application/json"}, 200 + + source = "https://crt.sh/json?identity=competitor.example" + fetcher = FlakyFetcher() + result = monitor.discover_certificate_transparency( + { + "certificate_transparency_sources": [ + {"url": source, "identity": "competitor.example"}, + ], + }, + { + "request_delay_seconds": 0, + "certificate_transparency_attempts": 4, + "certificate_transparency_retry_backoff_seconds": 0, + }, + fetcher, + ) + self.assertEqual(fetcher.calls, 4) + self.assertEqual(result["status"], "ok") + self.assertFalse(result["coverage_gap"]) + self.assertEqual(result["names"], ["api.competitor.example", "competitor.example"]) + + def test_crtsh_persistent_http_error_remains_coverage_gap_after_retries(self): + class BrokenFetcher: + def __init__(self): + self.calls = 0 + + def get(self, url, *, delay=None): + self.calls += 1 + raise urllib.error.HTTPError(url, 502, "persistent", None, None) + + source = "https://crt.sh/json?identity=competitor.example" + fetcher = BrokenFetcher() + result = monitor.discover_certificate_transparency( + { + "certificate_transparency_sources": [ + {"url": source, "identity": "competitor.example"}, + ], + }, + { + "request_delay_seconds": 0, + "certificate_transparency_attempts": 4, + "certificate_transparency_retry_backoff_seconds": 0, + }, + fetcher, + ) + self.assertEqual(fetcher.calls, 4) + self.assertEqual(result["status"], "error") + self.assertTrue(result["coverage_gap"]) + self.assertEqual(result["names"], []) + self.assertIn("HTTP 502", result["errors"][0]) + + def test_scope_excludes_assets_search_and_unbounded_profile(self): + with tempfile.TemporaryDirectory() as td: + cfg = self.make_config(Path(td)) + comp = cfg["competitors"][0] + self.assertTrue(monitor.url_allowed(self.origin + "/products/api", comp, cfg["global"], [])) + self.assertFalse(monitor.url_allowed(self.origin + "/assets/logo.png", comp, cfg["global"], [])) + self.assertFalse(monitor.url_allowed(self.origin + "/search?q=x", comp, cfg["global"], [])) + self.assertFalse(monitor.url_allowed(self.origin + "/company/demo-12345678", comp, cfg["global"], [])) + + def test_example_scope_excludes_numeric_company_and_person_profiles(self): + cfg = monitor.load_json(Path(__file__).with_name("config.example.json"), {}) + ignore = monitor.load_json(Path(__file__).with_name("ignore.example.json"), {}) + comp = cfg["competitors"][0] + for url in ( + "https://competitor.example/company/8957/example-inc", + "https://competitor.example/person/10695948-example", + "https://competitor.example/profile/example-12345678", + ): + self.assertFalse(monitor.url_allowed(url, comp, cfg["global"], ignore), url) + + def test_example_scope_ignores_robots_and_excludes_low_value_watch_categories(self): + cfg = monitor.load_json(Path(__file__).with_name("config.example.json"), {}) + ignore = monitor.load_json(Path(__file__).with_name("ignore.example.json"), {}) + self.assertTrue(cfg.get("competitors")) + self.assertTrue(all(comp.get("respect_robots") is False for comp in cfg["competitors"])) + self.assertTrue(all(comp.get("policy_watch_urls") == [] for comp in cfg["competitors"])) + self.assertEqual( + set(cfg["global"]["watch_selection"].get("exclude_categories", [])), + {"discovery_source", "routine_data", "archive", "company"}, + ) + self.assertEqual(monitor._watch_category("https://docs.competitor.example/llms.txt"), "discovery_source") + self.assertEqual(monitor._watch_category("https://docs.competitor.example/master.md"), "discovery_source") + self.assertEqual(monitor._watch_category("https://competitor.example/topic/market-overview"), "discovery_source") + self.assertEqual(monitor._watch_category("https://competitor.example/category/news"), "discovery_source") + comp = cfg["competitors"][0] + for url in ( + "https://competitor.example/legacy-page", + "http://competitor.example/anything", + "https://competitor.example/archive/2020", + "https://competitor.example/company/12400621/example-inc", + ): + self.assertTrue(monitor.is_ignored(url, comp, cfg["global"], ignore), url) + self.assertTrue(monitor._is_routine_data_url("https://competitor.example/lists/debtors")) + + def test_editorial_subtypes_share_one_bounded_representative_family(self): + urls = { + "https://example.test/blog/news/2026/01/01/news-a", + "https://example.test/blog/articles/2026/02/01/article-b", + "https://example.test/blog/research/2026/03/01/report-c", + "https://example.test/blog/press/2026/04/01/press-d", + } + for url in urls: + self.assertEqual(monitor._watch_category(url), "news_leaf") + self.assertEqual(monitor._watch_family(url, "news_leaf"), "editorial") + global_cfg = { + "watch_selection": { + "enabled": True, + "per_group": 2, + "exclude_categories": ["discovery_source", "routine_data", "archive"], + } + } + selected = monitor.select_watch_urls( + urls, + {"seeds": []}, + global_cfg, + {"urls": {url: {"first_seen": "2026-09-18T00:00:00+00:00"} for url in urls}}, + ) + self.assertEqual(selected, { + "https://example.test/blog/research/2026/03/01/report-c", + "https://example.test/blog/press/2026/04/01/press-d", + }) + + def test_classification_avoids_pricing_false_positives(self): + severity, category, _why = monitor.classify_change( + "https://competitor.example/", + "(changed) New Regional Lead\n(into) Global Economic Outlook Q3 2026", + ) + self.assertEqual((severity, category), ("minor", "content")) + severity, category, _why = monitor.classify_change( + "https://competitor.example/", + "API and integrations\n(added) Prioritizing ESG is not optional anymore", + ) + self.assertEqual((severity, category), ("minor", "content")) + severity, category, _why = monitor.classify_change( + "https://competitor.example/company/95892/example-inc", + "(changed) OUTSTANDING TAX (16.09.2026)\n(into) OUTSTANDING TAX (17.09.2026)", + ) + self.assertNotEqual(category, "pricing") + severity, category, _why = monitor.classify_change( + "https://competitor.example/robots.txt", + "(removed) Content-Signal: ai-train=no", + ) + self.assertEqual((severity, category), ("important", "crawl policy")) + severity, category, _why = monitor.classify_change( + "https://example.test/pricing", + "(changed) EUR 19 per month\n(into) EUR 29 per month", + ) + self.assertEqual((severity, category), ("important", "pricing")) + severity, category, _why = monitor.classify_change( + "https://competitor.example/sitemap.rss", + "(changed) This sitemap was dynamically generated on 16 Sept\n(into) This sitemap was dynamically generated on 17 Sept", + ) + self.assertEqual((severity, category), ("ignore", "generated sitemap metadata")) + severity, category, _why = monitor.classify_change( + "https://apidocs.example.test/api_descriptions", + "(added) IntroductionAPI Documentation\n" + "(removed) manage_search\n(removed) manage_search\n" + "(removed) IntroductionAPI Documentation", + ) + self.assertEqual((severity, category), ("ignore", "interface-only churn")) + severity, category, _why = monitor.classify_change( + "https://apidocs.example.test/api_descriptions", + "(removed) manage_search", + ) + self.assertNotEqual(severity, "ignore") + + def test_wholesale_estonian_to_english_localization_is_not_a_pricing_or_policy_alert(self): + localization_diff = """(changed) Avaleht +(into) Home +(changed) Hinnad +(into) Prices +(changed) Teenused +(into) Services +(changed) Registreeri Sisene +(into) Register Login +(changed) Kuumakse +(into) Monthly payment +(changed) 5 € Üksik päring +(into) 5 € Inquiry +(changed) Loo konto +(into) Create account +(changed) Kontakt +(into) Contact +(changed) Kasutustingimused +(into) Terms & Conditions""" + for url in ("https://competitor.example/prices", "https://competitor.example/terms"): + with self.subTest(url=url): + severity, category, why = monitor.classify_change(url, localization_diff) + self.assertEqual((severity, category), ("minor", "localization")) + self.assertIn("translated", why) + + severity, category, _why = monitor.classify_change( + "https://competitor.example/prices", + "(changed) 5 € Inquiry\n(into) 7 € Inquiry", + ) + self.assertEqual((severity, category), ("important", "pricing")) + + def test_pricing_path_does_not_turn_navigation_or_unrelated_copy_into_pricing(self): + severity, category, _why = monitor.classify_change( + "https://competitor.example/hinnad", + "(added) + Turuanalüüs", + ) + self.assertEqual((severity, category), ("ignore", "interface-only churn")) + + severity, category, _why = monitor.classify_change( + "https://example.test/pricing", + "(added) Our Tallinn office moved to a new address", + ) + self.assertEqual((severity, category), ("minor", "content")) + self.assertEqual(monitor.classify_new_url("https://example.test/pricing")[:2], ("important", "pricing")) + + def test_diff_excerpt_realigns_positional_cascade_to_actual_removed_card(self): + diff = """Products +(changed) Assistant Pro +(into) Data Connect +(changed) Data Connect +(into) Data Blocks +(changed) Data Blocks +(into) Company Number +(removed) Company Number +Footer +""" + evidence = monitor._diff_excerpt(diff) + self.assertEqual(evidence, "(removed) Assistant Pro") + severity, category, _why = monitor.classify_change( + "https://competitor.example/products", diff + ) + self.assertEqual((severity, category), ("important", "product / feature")) + + def test_diff_excerpt_preserves_genuine_replacement(self): + diff = "(changed) EUR 19 per month\n(into) EUR 29 per month" + self.assertEqual( + monitor._diff_excerpt(diff), + "(changed) EUR 19 per month\n(into) EUR 29 per month", + ) + + def test_diff_excerpt_balances_old_and_new_sides_under_limit(self): + diff = "\n".join( + [f"(changed) old proposition line {index} with explanatory wording" for index in range(10)] + + ["(into) New API proposition", "(into) 300 000+ companies", "(into) Request a demo key"] + ) + evidence = monitor._diff_excerpt(diff, limit=240) + self.assertIn("old proposition line 0", evidence) + self.assertIn("New API proposition", evidence) + self.assertLessEqual(len(evidence), 240) + + def test_diff_excerpt_realigns_positional_cascade_to_actual_added_card(self): + diff = """Products +(changed) Data Connect +(into) Assistant Pro +(changed) Data Blocks +(into) Data Connect +(changed) Company Number +(into) Data Blocks +(added) Company Number +Footer +""" + self.assertEqual(monitor._diff_excerpt(diff), "(added) Assistant Pro") + + def test_recursive_sitemap_feed_and_crawl_import_then_dedupe(self): + with tempfile.TemporaryDirectory() as td: + cfg = self.make_config(Path(td)) + state_dir = Path(td) / "state" + cd = FakeCD() + discord = CaptureDiscord() + first = monitor.run_discover(cfg, {"exact_urls": [], "regexes": []}, state_dir, cd, discord, send=True) + self.assertEqual(first["competitors"]["fixture"]["status"], "ok") + urls = set(first["competitors"]["fixture"]["urls"]) + self.assertIn(self.origin + "/products", urls) + self.assertIn(self.origin + "/products/api", urls) + self.assertIn(self.origin + "/pricing", urls) + self.assertIn(self.origin + "/feed.xml", urls) + self.assertIn(self.origin + "/blog/new-feature", urls) + self.assertNotIn(self.origin + "/assets/logo.png", urls) + self.assertFalse(any("12345678" in u for u in urls)) + imported_first = sum(len(c[0]) for c in cd.import_calls) + self.assertEqual(imported_first, len(urls)) + self.assertEqual(discord.messages, []) # initial inventory is silent + second = monitor.run_discover(cfg, {"exact_urls": [], "regexes": []}, state_dir, cd, discord, send=True) + self.assertEqual(second["summary"]["new_watch_count"], 0) + self.assertEqual(sum(len(c[0]) for c in cd.import_calls), imported_first) + + def test_new_page_after_baseline_is_classified_and_sent_immediately(self): + with tempfile.TemporaryDirectory() as td: + cfg = self.make_config(Path(td)) + state_dir = Path(td) / "state" + cd = FakeCD() + discord = CaptureDiscord() + monitor.run_discover(cfg, {"exact_urls": [], "regexes": []}, state_dir, cd, discord, send=True) + MutableFixture.pages["/"] = (200, "text/html", "PricingNew API") + MutableFixture.pages["/products/new-api"] = (200, "text/html", "new API") + result = monitor.run_discover(cfg, {"exact_urls": [], "regexes": []}, state_dir, cd, discord, send=True) + self.assertIn(self.origin + "/products/new-api", result["competitors"]["fixture"]["new_urls"]) + self.assertEqual(len(discord.messages), 1) + self.assertTrue(discord.messages[0]["mention"]) + self.assertIn("New high-signal page", discord.messages[0]["content"]) + + def test_new_page_alert_is_durable_across_delivery_failure(self): + with tempfile.TemporaryDirectory() as td: + cfg = self.make_config(Path(td)) + state_dir = Path(td) / "state" + cd = FakeCD() + monitor.run_discover( + cfg, {"exact_urls": [], "regexes": []}, state_dir, cd, + CaptureDiscord(), send=True, + ) + MutableFixture.pages["/"] = ( + 200, + "text/html", + "New API", + ) + MutableFixture.pages["/products/new-api"] = (200, "text/html", "new API") + with self.assertRaisesRegex(RuntimeError, "synthetic delivery failure"): + monitor.run_discover( + cfg, {"exact_urls": [], "regexes": []}, state_dir, cd, + AlwaysFailDiscord(), send=True, + ) + pending_path = state_dir / "pending-important.json" + pending = monitor.load_json(pending_path, {"events": []}) + self.assertEqual(["new_page"], [event["type"] for event in pending["events"]]) + retry = CaptureDiscord() + self.assertEqual(1, monitor._deliver_important_outbox(cfg, state_dir, retry)) + self.assertIn("New high-signal page", retry.messages[0]["content"]) + self.assertEqual([], monitor.load_json(pending_path, {"events": []})["events"]) + + def test_crtsh_baseline_is_silent_then_new_name_alerts(self): + with tempfile.TemporaryDirectory() as td: + cfg = self.make_config(Path(td)) + cfg["competitors"][0]["certificate_transparency_sources"] = [{ + "url": "https://crt.sh/json?identity=fixture.test", + "identity": "fixture.test", + }] + state_dir = Path(td) / "state" + cd = FakeCD() + discord = CaptureDiscord() + baseline = { + "status": "ok", "coverage_gap": False, "errors": [], + "sources": ["https://crt.sh/json?identity=fixture.test"], + "names": ["fixture.test", "api.fixture.test"], + } + changed = { + **baseline, + "names": ["fixture.test", "api.fixture.test", "new-product.fixture.test"], + } + with mock.patch.object(monitor, "discover_certificate_transparency", side_effect=[baseline, changed, changed]): + first = monitor.run_discover(cfg, {"exact_urls": [], "regexes": []}, state_dir, cd, discord, send=True) + self.assertEqual(first["competitors"]["fixture"]["new_certificate_names"], []) + self.assertEqual(discord.messages, []) + second = monitor.run_discover(cfg, {"exact_urls": [], "regexes": []}, state_dir, cd, discord, send=True) + third = monitor.run_discover(cfg, {"exact_urls": [], "regexes": []}, state_dir, cd, discord, send=True) + self.assertEqual(second["competitors"]["fixture"]["new_certificate_names"], ["new-product.fixture.test"]) + self.assertEqual(third["competitors"]["fixture"]["new_certificate_names"], []) + self.assertEqual(len(discord.messages), 1) + self.assertTrue(discord.messages[0]["mention"]) + self.assertIn("New certificate name", discord.messages[0]["content"]) + self.assertIn("new-product.fixture.test", discord.messages[0]["content"]) + self.assertIn("https://crt.sh/json?identity=fixture.test", discord.messages[0]["content"]) + self.assertEqual(monitor.load_json(state_dir / "pending-important.json", {"events": []})["events"], []) + + def test_representative_selection_prefers_default_language_and_latest_leaf_pages(self): + cfg = self.make_config(Path("unused")) + cfg["global"]["watch_selection"] = { + "enabled": True, + "per_group": 2, + "exclude_categories": ["discovery_source", "routine_data", "archive"], + } + comp = cfg["competitors"][0] + urls = { + self.origin + "/", + self.origin + "/robots.txt", + self.origin + "/sitemap.xml", + self.origin + "/products", + self.origin + "/products/monitoring", + self.origin + "/products/reports", + self.origin + "/en/lists/reinstated", + self.origin + "/datasets/debtors", + self.origin + "/2026/03", + self.origin + "/2026/01/old-announcement", + self.origin + "/2026/02/newer-announcement", + self.origin + "/2026/03/latest-announcement", + self.origin + "/en/2026/03/latest-announcement", + } + previous = {"urls": {url: {"first_seen": "2026-09-14T00:00:00+00:00"} for url in urls}} + selected = monitor.select_watch_urls(urls, comp, cfg["global"], previous) + self.assertIn(self.origin + "/", selected) + self.assertNotIn(self.origin + "/robots.txt", selected) + self.assertNotIn(self.origin + "/sitemap.xml", selected) + self.assertNotIn(self.origin + "/en/lists/reinstated", selected) + self.assertNotIn(self.origin + "/datasets/debtors", selected) + self.assertNotIn(self.origin + "/2026/03", selected) + selected_articles = {url for url in selected if "/2026/" in url} + self.assertEqual(selected_articles, { + self.origin + "/2026/02/newer-announcement", + self.origin + "/2026/03/latest-announcement", + }) + selected_products = {url for url in selected if "/products" in url} + self.assertEqual(len(selected_products), 2) + + def test_representative_mode_discovers_all_but_pauses_excess_leaf_watches(self): + with tempfile.TemporaryDirectory() as td: + cfg = self.make_config(Path(td)) + state_dir = Path(td) / "state" + MutableFixture.pages["/feed.xml"] = ( + 200, + "application/rss+xml", + "" + "".join( + f"{self.origin}/blog/article-{index}" + for index in range(4) + ) + "", + ) + for index in range(4): + MutableFixture.pages[f"/blog/article-{index}"] = (200, "text/html", f"article {index}") + cd = FakeCD() + monitor.run_discover(cfg, {"exact_urls": [], "regexes": []}, state_dir, cd, CaptureDiscord(), send=False) + before_count = len(cd.watches) + cfg["global"]["watch_selection"] = {"enabled": True, "per_group": 2} + result = monitor.run_discover(cfg, {"exact_urls": [], "regexes": []}, state_dir, cd, CaptureDiscord(), send=False) + row = result["competitors"]["fixture"] + self.assertGreater(len(row["urls"]), row["selected_watch_count"]) + self.assertEqual(len(cd.watches), before_count) # no history was deleted + paused_blog = [ + watch for watch in cd.watches.values() + if "/blog/article-" in watch["url"] and watch.get("paused") + ] + self.assertEqual(len(paused_blog), 2) + ledger = monitor.load_json(state_dir / "discovery.json", {}) + self.assertEqual( + sum( + bool(item.get("paused_by_monitor")) + for item in ledger["competitors"]["fixture"]["urls"].values() + ), + 2, + ) + + def test_robots_disallow_all_produces_gap_and_only_policy_watch(self): + with tempfile.TemporaryDirectory() as td: + cfg = self.make_config(Path(td), blocked=True) + state_dir = Path(td) / "state" + MutableFixture.pages["/robots.txt"] = (200, "text/plain", "User-agent: *\nDisallow: /\n") + cd = FakeCD() + result = monitor.run_discover(cfg, {"exact_urls": [], "regexes": []}, state_dir, cd, CaptureDiscord(), send=False) + row = result["competitors"]["fixture"] + self.assertEqual(row["status"], "robots_blocked") + self.assertEqual(row["urls"], [self.origin + "/robots.txt"]) + self.assertTrue(row["coverage_gap"]) + + def test_robots_disallow_all_is_ignored_by_default(self): + with tempfile.TemporaryDirectory() as td: + cfg = self.make_config(Path(td), blocked=True) + cfg["competitors"][0].pop("respect_robots", None) + cfg["competitors"][0]["policy_watch_urls"] = [] + state_dir = Path(td) / "state" + MutableFixture.pages["/robots.txt"] = (200, "text/plain", "User-agent: *\nDisallow: /\n") + cd = FakeCD() + result = monitor.run_discover(cfg, {"exact_urls": [], "regexes": []}, state_dir, cd, CaptureDiscord(), send=False) + row = result["competitors"]["fixture"] + self.assertEqual(row["status"], "ok") + self.assertFalse(row["coverage_gap"]) + self.assertIn(self.origin + "/products", row["urls"]) + + def test_antibot_challenge_is_reported_as_gap_and_not_watched(self): + with tempfile.TemporaryDirectory() as td: + cfg = self.make_config(Path(td)) + cfg["competitors"][0]["respect_robots"] = False + cfg["competitors"][0]["policy_watch_urls"] = [] + state_dir = Path(td) / "state" + MutableFixture.pages["/robots.txt"] = (200, "text/plain", "User-agent: *\nAllow: /\n") + MutableFixture.pages["/sitemap.xml"] = (404, "text/plain", "not found") + MutableFixture.pages["/"] = ( + 200, + "text/html", + "

Palun kinnitage, et Te ei ole robot!

", + ) + cd = FakeCD() + result = monitor.run_discover(cfg, {"exact_urls": [], "regexes": []}, state_dir, cd, CaptureDiscord(), send=False) + row = result["competitors"]["fixture"] + self.assertEqual(row["status"], "error") + self.assertTrue(row["coverage_gap"]) + self.assertEqual(row["urls"], []) + self.assertFalse(cd.watches) + + def test_antibot_challenge_pauses_previous_competitor_watches(self): + with tempfile.TemporaryDirectory() as td: + cfg = self.make_config(Path(td)) + cfg["competitors"][0]["respect_robots"] = False + cfg["competitors"][0]["policy_watch_urls"] = [] + cfg["global"]["watch_selection"] = {"enabled": True, "per_group": 2} + state_dir = Path(td) / "state" + cd = FakeCD() + monitor.run_discover(cfg, {"exact_urls": [], "regexes": []}, state_dir, cd, CaptureDiscord(), send=False) + self.assertTrue(cd.watches) + self.assertTrue(any(not watch.get("paused", False) for watch in cd.watches.values())) + monitor.save_json_atomic(state_dir / "pending-digest.json", {"events": [ + { + "id": "stale-new-page", + "type": "new_page", + "competitor": "Fixture Co", + "url": self.origin + "/products", + "category": "content", + }, + { + "id": "stale-recovered", + "type": "coverage_recovered", + "competitor": "Fixture Co", + "url": self.origin + "/", + "category": "discovery coverage restored", + }, + ]}) + + MutableFixture.pages["/robots.txt"] = (200, "text/plain", "User-agent: *\nAllow: /\n") + MutableFixture.pages["/sitemap.xml"] = (404, "text/plain", "not found") + MutableFixture.pages["/"] = ( + 200, + "text/html", + "

Palun kinnitage, et Te ei ole robot!

", + ) + result = monitor.run_discover(cfg, {"exact_urls": [], "regexes": []}, state_dir, cd, CaptureDiscord(), send=False) + row = result["competitors"]["fixture"] + self.assertTrue(row["anti_bot_challenge"]) + self.assertFalse(row["selection_frozen_for_incomplete_discovery"]) + self.assertTrue(all(watch.get("paused", False) for watch in cd.watches.values())) + pending = monitor.load_json(state_dir / "pending-digest.json", {"events": []})["events"] + self.assertEqual([event["type"] for event in pending], ["coverage_gap"]) + + def test_obsolete_robots_watch_is_paused_after_policy_bypass(self): + with tempfile.TemporaryDirectory() as td: + cfg = self.make_config(Path(td), blocked=True) + state_dir = Path(td) / "state" + MutableFixture.pages["/robots.txt"] = (200, "text/plain", "User-agent: *\nDisallow: /\n") + cd = FakeCD() + monitor.run_discover(cfg, {"exact_urls": [], "regexes": []}, state_dir, cd, CaptureDiscord(), send=False) + robot_watch = next(watch for watch in cd.watches.values() if watch["url"].endswith("/robots.txt")) + self.assertFalse(robot_watch.get("paused", False)) + + cfg["competitors"][0]["respect_robots"] = False + cfg["competitors"][0]["policy_watch_urls"] = [] + cfg["global"]["watch_selection"] = { + "enabled": True, + "per_group": 2, + "exclude_categories": ["discovery_source", "routine_data", "archive"], + } + monitor.run_discover(cfg, {"exact_urls": [], "regexes": []}, state_dir, cd, CaptureDiscord(), send=False) + monitor.run_discover(cfg, {"exact_urls": [], "regexes": []}, state_dir, cd, CaptureDiscord(), send=False) + self.assertTrue(robot_watch.get("paused", False)) + pending = monitor.load_json(state_dir / "pending-digest.json", {"events": []})["events"] + self.assertFalse(any(event.get("url", "").endswith("/robots.txt") for event in pending)) + + def test_routine_data_changes_are_ignored(self): + for url in ( + "https://competitor.example/en/lists/reinstated", + "https://competitor.example/datasets/debtors", + "https://competitor.example/certificates/companies", + ): + severity, category, _why = monitor.classify_change( + url, + "(added) 18.09.2026 EXAMPLE LLC\n(removed) 17.09.2026 OLD LLC", + ) + self.assertEqual((severity, category), ("ignore", "routine public-data churn")) + + def test_missing_page_requires_two_successful_discoveries_then_is_retained_paused(self): + with tempfile.TemporaryDirectory() as td: + cfg = self.make_config(Path(td)) + state_dir = Path(td) / "state" + cd = FakeCD() + monitor.run_discover(cfg, {"exact_urls": [], "regexes": []}, state_dir, cd, CaptureDiscord(), send=False) + MutableFixture.pages["/nested.xml"] = (200, "application/xml", "%s/" % self.origin) + MutableFixture.pages["/"] = (200, "text/html", "") + one = monitor.run_discover(cfg, {"exact_urls": [], "regexes": []}, state_dir, cd, CaptureDiscord(), send=False) + self.assertEqual(one["competitors"]["fixture"]["confirmed_missing_urls"], []) + two = monitor.run_discover(cfg, {"exact_urls": [], "regexes": []}, state_dir, cd, CaptureDiscord(), send=False) + self.assertGreater(len(two["competitors"]["fixture"]["confirmed_missing_urls"]), 0) + self.assertGreater(len(cd.watches), 1) # retained rather than deleted + missing_urls = set(two["competitors"]["fixture"]["confirmed_missing_urls"]) + missing_watches = [watch for watch in cd.watches.values() if watch["url"] in missing_urls] + self.assertTrue(missing_watches) + self.assertTrue(all(watch.get("paused") for watch in missing_watches)) + ledger = monitor.load_json(state_dir / "discovery.json", {}) + for url in missing_urls: + row = ledger["competitors"]["fixture"]["urls"][url] + self.assertFalse(row.get("selected_for_watch")) + self.assertTrue(row.get("paused_by_monitor")) + + def test_manual_pause_is_never_unpaused_by_scope_rotation(self): + cfg = self.make_config(Path("unused")) + comp = cfg["competitors"][0] + cd = FakeCD() + cd.tags["tag-1"] = {"uuid": "tag-1", "title": "Competitor • Fixture Co"} + url = self.origin + "/products" + cd.watches["watch-1"] = { + "uuid": "watch-1", "url": url, "tags": ["tag-1"], "paused": True + } + comp_state = {"urls": {url: {"watch_uuid": "watch-1"}}} + monitor.reconcile_selected_watches( + comp, cfg["global"], {"exact_urls": [], "regexes": []}, + comp_state, cd.watches, cd, set(), "tag-1", + ) + row = comp_state["urls"][url] + self.assertTrue(row["pre_monitor_paused"]) + self.assertFalse(row["pause_owned_by_monitor"]) + self.assertEqual(cd.put_calls, []) + + monitor.reconcile_selected_watches( + comp, cfg["global"], {"exact_urls": [], "regexes": []}, + comp_state, cd.watches, cd, {url}, "tag-1", + ) + self.assertTrue(cd.watches["watch-1"]["paused"]) + self.assertEqual(cd.put_calls, []) + self.assertFalse(comp_state["urls"][url]["paused_by_monitor"]) + + def test_pause_intent_is_checkpointed_before_put_and_untagged_watch_is_untouched(self): + cfg = self.make_config(Path("unused")) + comp = cfg["competitors"][0] + cd = FakeCD() + tagged_url = self.origin + "/products" + untagged_url = self.origin + "/pricing" + cd.watches = { + "watch-1": {"uuid": "watch-1", "url": tagged_url, "tags": ["tag-1"], "paused": False}, + "watch-2": {"uuid": "watch-2", "url": untagged_url, "tags": [], "paused": False}, + } + comp_state = {"urls": {tagged_url: {"watch_uuid": "watch-1"}}} + checkpoints = [] + + def checkpoint(): + checkpoints.append((dict(comp_state["urls"][tagged_url]), cd.watches["watch-1"]["paused"])) + + monitor.reconcile_selected_watches( + comp, cfg["global"], {"exact_urls": [], "regexes": []}, + comp_state, cd.watches, cd, set(), "tag-1", checkpoint=checkpoint, + ) + self.assertEqual(cd.put_calls, [("watch-1", {"paused": True})]) + self.assertEqual(checkpoints[0][0]["pause_operation_pending"], "pause") + self.assertFalse(checkpoints[0][1]) + self.assertNotIn("pause_operation_pending", checkpoints[-1][0]) + self.assertTrue(checkpoints[-1][1]) + self.assertFalse(cd.watches["watch-2"]["paused"]) + + # A crash after a successful pause PUT but before the completion + # checkpoint converges without issuing a second mutation. + comp_state["urls"][tagged_url]["pause_operation_pending"] = "pause" + cd.put_calls.clear() + monitor.reconcile_selected_watches( + comp, cfg["global"], {"exact_urls": [], "regexes": []}, + comp_state, cd.watches, cd, set(), "tag-1", checkpoint=checkpoint, + ) + self.assertEqual(cd.put_calls, []) + self.assertNotIn("pause_operation_pending", comp_state["urls"][tagged_url]) + + def test_partial_discovery_preserves_existing_selection_without_rotation(self): + with tempfile.TemporaryDirectory() as td: + cfg = self.make_config(Path(td)) + cfg["global"]["watch_selection"] = {"enabled": True, "per_group": 2} + state_dir = Path(td) / "state" + cd = FakeCD() + base_urls = { + self.origin + "/", self.origin + "/robots.txt", + self.origin + "/2026/01/article-one", + self.origin + "/2026/02/article-two", + self.origin + "/2026/03/article-three", + } + ok = { + "status": "ok", "coverage_gap": False, "errors": [], + "urls": sorted(base_urls), "sitemaps_fetched": 1, + "pages_crawled": 1, "feeds": [], + } + with mock.patch.object(monitor, "discover_competitor", return_value=ok): + monitor.run_discover(cfg, {"exact_urls": [], "regexes": []}, state_dir, cd, CaptureDiscord(), send=False) + before = monitor.load_json(state_dir / "discovery.json", {}) + before_selected = { + url for url, row in before["competitors"]["fixture"]["urls"].items() + if row.get("selected_for_watch") + } + self.assertIn(self.origin + "/2026/02/article-two", before_selected) + self.assertIn(self.origin + "/2026/03/article-three", before_selected) + + partial_urls = set(base_urls) | {self.origin + "/2026/04/article-four"} + partial = { + "status": "partial", "coverage_gap": True, + "errors": ["synthetic incomplete source"], + "urls": sorted(partial_urls), "sitemaps_fetched": 1, + "pages_crawled": 1, "feeds": [], + } + with mock.patch.object(monitor, "discover_competitor", return_value=partial): + monitor.run_discover(cfg, {"exact_urls": [], "regexes": []}, state_dir, cd, CaptureDiscord(), send=False) + after = monitor.load_json(state_dir / "discovery.json", {}) + after_selected = { + url for url, row in after["competitors"]["fixture"]["urls"].items() + if row.get("selected_for_watch") + } + self.assertEqual(after_selected, before_selected) + self.assertNotIn(self.origin + "/2026/04/article-four", {w["url"] for w in cd.watches.values()}) + for url in before_selected: + self.assertEqual(after["competitors"]["fixture"]["urls"][url].get("missing_count", 0), 0) + + def test_ignore_rules_persistently_pause_existing_watch(self): + with tempfile.TemporaryDirectory() as td: + cfg = self.make_config(Path(td)) + state_dir = Path(td) / "state" + cd = FakeCD() + monitor.run_discover(cfg, {"exact_urls": [], "regexes": []}, state_dir, cd, CaptureDiscord(), send=False) + target = self.origin + "/pricing" + result = monitor.run_discover(cfg, {"exact_urls": [target], "regexes": []}, state_dir, cd, CaptureDiscord(), send=False) + matches = [w for w in cd.watches.values() if w["url"] == target] + self.assertEqual(matches[0].get("paused"), True) + self.assertNotIn(target, result["competitors"]["fixture"]["confirmed_missing_urls"]) + ledger = monitor.load_json(state_dir / "discovery.json", {}) + target_row = ledger["competitors"]["fixture"]["urls"][target] + self.assertTrue(target_row.get("paused_by_monitor")) + pending = monitor.load_json(state_dir / "pending-digest.json", {"events": []}) + self.assertFalse(any(row.get("url") == target for row in pending.get("events", []))) + + def test_change_collection_important_alert_minor_digest_and_dedup(self): + with tempfile.TemporaryDirectory() as td: + cfg = self.make_config(Path(td)) + state_dir = Path(td) / "state" + cd = FakeCD() + discord = CaptureDiscord() + cd.tags["tag-1"] = {"uuid": "tag-1", "title": "Competitor • Fixture Co"} + cd.watches["watch-1"] = {"uuid": "watch-1", "url": self.origin + "/pricing", "title": "Pricing", "tags": ["tag-1"], "last_error": False, "last_checked": 200} + cd.watches["watch-2"] = {"uuid": "watch-2", "url": self.origin + "/about", "title": "About", "tags": ["tag-1"], "last_error": False, "last_checked": 200} + cd.histories["watch-1"] = {"100": "a", "200": "b"} + cd.histories["watch-2"] = {"100": "a", "200": "b"} + cd.diffs[("watch-1", "100", "200")] = "(removed) €19\n(added) €29 Pro plan" + cd.diffs[("watch-2", "100", "200")] = "(removed) Old address\n(added) New address" + monitor.save_json_atomic(state_dir / "collector.json", { + "last_history": {"watch-1": "100", "watch-2": "100"}, + "last_successful_checks": {"watch-1": "150", "watch-2": "150"}, + "seen_events": [], + }) + result = monitor.run_collect(cfg, state_dir, cd, discord, send=True) + self.assertEqual(result["important_count"], 1) + self.assertEqual(result["minor_count"], 1) + self.assertEqual( + result["important_events"][0]["from_timestamp_semantics"], + "previous_distinct_snapshot_not_previous_check", + ) + self.assertEqual( + result["important_events"][0]["to_timestamp_semantics"], + "first_stored_snapshot_with_changed_content", + ) + self.assertEqual( + result["important_events"][0]["last_successful_unchanged_check"], + "150", + ) + self.assertEqual( + result["important_events"][0]["first_successful_changed_check"], + "200", + ) + self.assertEqual(len(discord.messages), 1) + self.assertTrue(discord.messages[0]["mention"]) + self.assertIn("Observed fetch window", discord.messages[0]["content"]) + again = monitor.run_collect(cfg, state_dir, cd, discord, send=True) + self.assertEqual(again["important_count"], 0) + self.assertEqual(again["minor_count"], 0) + digest = monitor.run_digest(cfg, state_dir, discord, send=True) + self.assertEqual(digest["delivered_count"], 1) + self.assertEqual(len(discord.messages), 2) + self.assertFalse(discord.messages[1]["mention"]) + silent = monitor.run_digest(cfg, state_dir, discord, send=True) + self.assertEqual(silent["delivered_count"], 0) + self.assertEqual(len(discord.messages), 2) + + def test_digest_groups_same_topic_and_checkpoints_partial_batches(self): + with tempfile.TemporaryDirectory() as td: + cfg = self.make_config(Path(td)) + state_dir = Path(td) / "state" + events = [] + for index in range(8): + events.append({ + "id": f"minor-{index}", + "type": "page_change", + "competitor": "Fixture Co", + "category": "content", + "url": self.origin + f"/about/{index // 2}", + "evidence": f"change-{index}-" + ("x" * 500), + "why": "Editorial wording changed.", + "observed_at": f"2026-09-17T00:00:{index:02d}+00:00", + }) + monitor.save_json_atomic(state_dir / "pending-digest.json", {"events": events}) + first = FailAfterOneDiscord() + with self.assertRaisesRegex(RuntimeError, "synthetic delivery failure"): + monitor.run_digest(cfg, state_dir, first, send=True) + remaining = monitor.load_json(state_dir / "pending-digest.json", {"events": []})["events"] + self.assertGreater(len(remaining), 0) + self.assertLess(len(remaining), len(events)) + self.assertIn("2 changes", first.messages[0]["content"]) + second = CaptureDiscord() + result = monitor.run_digest(cfg, state_dir, second, send=True) + self.assertEqual(result["delivered_count"], len(remaining)) + self.assertEqual(monitor.load_json(state_dir / "pending-digest.json", {"events": []})["events"], []) + + def test_digest_removes_deduplicated_queue_event_by_its_queue_id(self): + with tempfile.TemporaryDirectory() as td: + cfg = self.make_config(Path(td)) + state_dir = Path(td) / "state" + queued = { + "id": "deduplicated-queue-id", + "_source_ids": ["source-a", "source-b"], + "type": "page_change", + "competitor": "Fixture Co", + "category": "content", + "url": self.origin + "/about", + "evidence": "(added) Actual wording", + "why": "Content changed.", + } + monitor.save_json_atomic(state_dir / "pending-digest.json", {"events": [queued]}) + result = monitor.run_digest(cfg, state_dir, CaptureDiscord(), send=True) + self.assertEqual(result["delivered_count"], 1) + self.assertEqual( + monitor.load_json(state_dir / "pending-digest.json", {"events": []})["events"], + [], + ) + + def test_digest_discards_operational_and_shared_fragment_noise(self): + with tempfile.TemporaryDirectory() as td: + cfg = self.make_config(Path(td)) + state_dir = Path(td) / "state" + events = [ + { + "id": "coverage", "type": "coverage_gap", + "competitor": "Fixture Co", "category": "coverage", "url": self.origin, + }, + { + "id": "shared-fragment", "type": "page_change", + "competitor": "Fixture Co", "category": "content", + "url": self.origin + "/page-1", + "urls": [self.origin + f"/page-{index}" for index in range(3)], + "evidence": "(added) Latest article\n(removed) Previous article", + }, + ] + monitor.save_json_atomic(state_dir / "pending-digest.json", {"events": events}) + discord = CaptureDiscord() + result = monitor.run_digest(cfg, state_dir, discord, send=True) + self.assertTrue(result["silent"]) + self.assertEqual(discord.messages, []) + self.assertEqual( + monitor.load_json(state_dir / "pending-digest.json", {"events": []})["events"], + [], + ) + + def test_important_delivery_outbox_prevents_duplicate_after_partial_failure(self): + with tempfile.TemporaryDirectory() as td: + cfg = self.make_config(Path(td)) + state_dir = Path(td) / "state" + cd = FakeCD() + cd.tags["tag-1"] = {"uuid": "tag-1", "title": "Competitor • Fixture Co"} + for index in (1, 2): + watch_id = f"watch-{index}" + cd.watches[watch_id] = { + "uuid": watch_id, + "url": self.origin + f"/pricing/{index}", + "title": "Pricing", + "tags": ["tag-1"], + "last_error": False, + } + cd.histories[watch_id] = {"100": "a", "200": "b"} + cd.diffs[(watch_id, "100", "200")] = f"(removed) EUR {index}9\n(added) EUR {index}99" + monitor.save_json_atomic( + state_dir / "collector.json", + {"last_history": {"watch-1": "100", "watch-2": "100"}, "seen_events": []}, + ) + first = FailAfterOneDiscord() + with self.assertRaisesRegex(RuntimeError, "synthetic delivery failure"): + monitor.run_collect(cfg, state_dir, cd, first, send=True) + self.assertEqual(len(first.messages), 1) + queue = monitor.load_json(state_dir / "pending-important.json", {"events": []}) + self.assertEqual(len(queue["events"]), 1) + second = CaptureDiscord() + result = monitor.run_collect(cfg, state_dir, cd, second, send=True) + self.assertEqual(result["important_count"], 0) + self.assertEqual(result["delivered_important_count"], 1) + self.assertEqual(len(second.messages), 1) + queue = monitor.load_json(state_dir / "pending-important.json", {"events": []}) + self.assertEqual(queue["events"], []) + + def test_discord_client_retries_rate_limit(self): + error = urllib.error.HTTPError( + "https://discord.test/messages", + 429, + "rate limited", + {"Retry-After": "0"}, + io.BytesIO(b'{"retry_after": 0}'), + ) + response = io.BytesIO(b'{"id": "message-after-retry"}') + client = monitor.DiscordClient("token", "channel", "user") + with mock.patch.object(monitor.urllib.request, "urlopen", side_effect=[error, response]): + receipt = client.send("test") + self.assertEqual(receipt["id"], "message-after-retry") + + def test_no_send_collect_does_not_require_discord_token(self): + with tempfile.TemporaryDirectory() as td: + root = Path(td) + config_path = root / "config.json" + env_path = root / "secrets.env" + state_dir = root / "state" + config_path.write_text(json.dumps(self.make_config(root))) + env_path.write_text("CD_API_TOKEN=dummy\n") + output = io.StringIO() + result = { + "at": "2026-09-29T00:00:00+00:00", + "important_count": 0, + "minor_count": 0, + "coverage_gap_count": 0, + "delivered_important_count": 0, + "pending_important_count": 0, + } + with mock.patch.object(monitor, "run_collect", return_value=result) as run_collect: + with contextlib.redirect_stdout(output): + status = monitor.main([ + "collect", + "--no-send", + "--config", str(config_path), + "--state-dir", str(state_dir), + "--env-file", str(env_path), + ]) + self.assertEqual(status, 0) + self.assertIsNone(run_collect.call_args.args[3]) + self.assertFalse(run_collect.call_args.kwargs["send"]) + + def test_no_send_digest_requires_no_api_or_discord_tokens(self): + with tempfile.TemporaryDirectory() as td: + root = Path(td) + config_path = root / "config.json" + env_path = root / "secrets.env" + state_dir = root / "state" + config_path.write_text(json.dumps(self.make_config(root))) + env_path.write_text("") + output = io.StringIO() + with contextlib.redirect_stdout(output): + status = monitor.main([ + "digest", + "--no-send", + "--config", str(config_path), + "--state-dir", str(state_dir), + "--env-file", str(env_path), + ]) + self.assertEqual(status, 0) + self.assertEqual(json.loads(output.getvalue())["delivered_count"], 0) + + def test_global_command_lock_skips_overlapping_timer_runs(self): + with tempfile.TemporaryDirectory() as td: + root = Path(td) + cfg = self.make_config(root) + config_path = root / "config.json" + env_path = root / "secrets.env" + state_dir = root / "state" + config_path.write_text(json.dumps(cfg)) + env_path.write_text("CD_API_TOKEN=dummy\nDISCORD_BOT_TOKEN=dummy\n") + state_dir.mkdir() + held = (state_dir / "monitor.lock").open("w") + self.addCleanup(held.close) + fcntl.flock(held, fcntl.LOCK_EX | fcntl.LOCK_NB) + output = io.StringIO() + with contextlib.redirect_stdout(output): + status = monitor.main([ + "collect", + "--config", str(config_path), + "--state-dir", str(state_dir), + "--env-file", str(env_path), + ]) + self.assertEqual(status, 0) + self.assertEqual(json.loads(output.getvalue())["reason"], "another monitor command is running") + + def test_daily_digest_waits_for_global_lock_instead_of_skipping(self): + with tempfile.TemporaryDirectory() as td: + root = Path(td) + cfg = self.make_config(root) + config_path = root / "config.json" + env_path = root / "secrets.env" + state_dir = root / "state" + config_path.write_text(json.dumps(cfg)) + env_path.write_text("CD_API_TOKEN=dummy\nDISCORD_BOT_TOKEN=dummy\n") + state_dir.mkdir() + held = (state_dir / "monitor.lock").open("w") + fcntl.flock(held, fcntl.LOCK_EX | fcntl.LOCK_NB) + + def release_lock(): + time.sleep(0.08) + held.close() + + releaser = threading.Thread(target=release_lock) + releaser.start() + started = time.monotonic() + output = io.StringIO() + with contextlib.redirect_stdout(output): + status = monitor.main([ + "digest", + "--no-send", + "--config", str(config_path), + "--state-dir", str(state_dir), + "--env-file", str(env_path), + ]) + releaser.join() + self.assertEqual(status, 0) + self.assertGreaterEqual(time.monotonic() - started, 0.06) + self.assertEqual(json.loads(output.getvalue())["delivered_count"], 0) + + def test_source_failure_is_unknown_gap_not_no_change(self): + with tempfile.TemporaryDirectory() as td: + cfg = self.make_config(Path(td)) + state_dir = Path(td) / "state" + cd = FakeCD() + cd.tags["tag-1"] = {"uuid": "tag-1", "title": "Competitor • Fixture Co"} + cd.watches["watch-1"] = {"uuid": "watch-1", "url": self.origin + "/pricing", "title": "Pricing", "tags": ["tag-1"], "last_error": "Timeout"} + result = monitor.run_collect(cfg, state_dir, cd, CaptureDiscord(), send=False) + self.assertEqual(result["coverage_gap_count"], 1) + self.assertNotIn("no changes", json.dumps(result).lower()) + + def test_identical_cross_url_changes_are_deduplicated(self): + events = [ + {"id": "a", "type": "page_change", "competitor": "Fixture Co", "category": "pricing", "url": self.origin + "/pricing", "evidence": "(changed) EUR 19\n(into) EUR 29"}, + {"id": "b", "type": "page_change", "competitor": "Fixture Co", "category": "pricing", "url": self.origin + "/en/pricing", "evidence": "(changed) EUR 19\n(into) EUR 29"}, + ] + rows = monitor.deduplicate_events(events) + self.assertEqual(len(rows), 1) + self.assertEqual(len(rows[0]["urls"]), 2) + + def test_identical_minor_content_fragment_across_three_pages_is_suppressed(self): + events = [ + { + "id": str(index), "type": "page_change", + "competitor": "Fixture Co", "category": "content", + "url": self.origin + f"/page-{index}", + "evidence": "(added) Latest article\n(removed) Previous article", + } + for index in range(3) + ] + self.assertEqual(monitor.deduplicate_events(events), []) + + def test_daily_digest_keeps_current_source_gap_in_status_but_stays_silent(self): + with tempfile.TemporaryDirectory() as td: + cfg = self.make_config(Path(td)) + state_dir = Path(td) / "state" + monitor.save_json_atomic(state_dir / "last-discovery-run.json", { + "competitors": {"fixture": {"status": "robots_blocked", "errors": ["robots.txt disallows the configured seeds"]}} + }) + discord = CaptureDiscord() + result = monitor.run_digest(cfg, state_dir, discord, send=True) + self.assertTrue(result["silent"]) + self.assertEqual(discord.messages, []) + + def test_daily_digest_keeps_certificate_gap_in_status_but_stays_silent(self): + with tempfile.TemporaryDirectory() as td: + cfg = self.make_config(Path(td)) + state_dir = Path(td) / "state" + monitor.save_json_atomic(state_dir / "last-discovery-run.json", { + "competitors": {"fixture": { + "status": "ok", "errors": [], + "certificate_transparency": { + "status": "error", + "errors": ["certificate transparency fixture: timeout"], + "sources": ["https://crt.sh/json?identity=fixture.test"], + }, + }} + }) + discord = CaptureDiscord() + result = monitor.run_digest(cfg, state_dir, discord, send=True) + self.assertTrue(result["silent"]) + self.assertEqual(discord.messages, []) + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/tests/test_repository.py b/tests/test_repository.py new file mode 100755 index 0000000..6952036 --- /dev/null +++ b/tests/test_repository.py @@ -0,0 +1,237 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import grp +import json +import os +from pathlib import Path +import pwd +import re +import shutil +import subprocess +import tempfile +import unittest +import urllib.parse +from unittest import mock + +from scripts import doctor + +ROOT = Path(__file__).resolve().parents[1] +FORBIDDEN_PATTERNS = ( + ("absolute home path", re.compile(r"/home/[^/\s]+/")), + ("host data mount", re.compile(r"(?:^|[\s\"'(=])/data(?:-[^/\s\"']+)?/", re.MULTILINE)), + ("Discord-style numeric identifier", re.compile(r"\b[0-9]{15,20}\b")), +) +TEXT_SUFFIXES = {"", ".py", ".json", ".md", ".txt", ".yaml", ".yml", ".sh", ".in", ".example"} +RUNTIME_ROOT_FILES = {".env", "secrets.env", "config.json", "ignore.json"} +SKIPPED_PARTS = {".git", "build", "state", "data", "__pycache__", ".venv", "release"} + + +def publishable_files(root: Path = ROOT): + for path in root.rglob("*"): + relative = path.relative_to(root) + if ( + not path.is_file() + or relative.as_posix() in RUNTIME_ROOT_FILES + or any(part in SKIPPED_PARTS for part in relative.parts) + ): + continue + if path.suffix in TEXT_SUFFIXES or path.name in {"Makefile", "LICENSE", ".gitignore"}: + yield path + + +class RepositoryTests(unittest.TestCase): + def test_required_publishable_layout(self): + required = { + ".env.example", ".gitignore", "LICENSE", "README.md", "compose.yaml", "compose.browser.yaml", + "config.example.json", "ignore.example.json", "monitor.py", "notify_failure.py", + "requirements-verification.txt", "secrets.env.example", "test_monitor.py", + "scripts/setup.sh", "scripts/doctor.py", "scripts/render_systemd.py", + "scripts/install-systemd.sh", "scripts/uninstall-systemd.sh", + "scripts/build-release.sh", "verification/e2e_verify.py", + "verification/verify_persistence.py", "verification/verify_ui.py", + } + missing = sorted(name for name in required if not (ROOT / name).is_file()) + self.assertEqual([], missing) + + def test_private_host_identifiers_are_absent(self): + hits = [] + for path in publishable_files(): + if path == Path(__file__).resolve(): + continue + text = path.read_text(errors="replace") + for label, pattern in FORBIDDEN_PATTERNS: + if pattern.search(text): + hits.append((str(path.relative_to(ROOT)), label)) + self.assertEqual([], hits) + + def test_runtime_and_secret_files_are_not_publishable(self): + ignore = (ROOT / ".gitignore").read_text().splitlines() + for entry in sorted(RUNTIME_ROOT_FILES | {"state/", "data/", "build/", "release/"}): + self.assertIn(entry, ignore) + if (ROOT / ".git").is_dir(): + result = subprocess.run( + ["git", "-C", str(ROOT), "ls-files", "-z"], + check=True, + capture_output=True, + ) + tracked = set(result.stdout.decode().rstrip("\0").split("\0")) if result.stdout else set() + forbidden = sorted( + name for name in tracked + if name in RUNTIME_ROOT_FILES + or name.startswith(("state/", "data/", "build/", "release/")) + ) + self.assertEqual([], forbidden) + + def test_examples_are_valid_and_sanitized(self): + config = json.loads((ROOT / "config.example.json").read_text()) + ignore = json.loads((ROOT / "ignore.example.json").read_text()) + self.assertEqual("REPLACE_WITH_DISCORD_CHANNEL_OR_THREAD_ID", config["discord"]["thread_id"]) + self.assertEqual("REPLACE_WITH_DISCORD_USER_ID", config["discord"]["important_mention_user_id"]) + self.assertEqual(1, len(config["competitors"])) + self.assertEqual("Example Vendor", config["competitors"][0]["name"]) + self.assertTrue(all(domain.endswith(".example") for domain in config["competitors"][0]["domains"])) + target_urls = [] + for competitor in config["competitors"]: + target_urls.extend(competitor.get("seeds", [])) + target_urls.extend(competitor.get("sitemap_seeds", [])) + target_urls.extend(competitor.get("policy_watch_urls", [])) + target_urls.extend(source.get("url", "") for source in competitor.get("certificate_transparency_sources", [])) + target_urls.extend(ignore.get("exact_urls", [])) + for competitor_ignore in ignore.get("competitors", {}).values(): + target_urls.extend(competitor_ignore.get("exact_urls", [])) + for url in target_urls: + self.assertTrue( + (urllib.parse.urlsplit(url).hostname or "").endswith(".example"), + f"non-reserved target URL in examples: {url}", + ) + self.assertTrue(all(comp.get("respect_robots") is False for comp in config["competitors"])) + self.assertEqual(1, ignore["version"]) + secrets = (ROOT / "secrets.env.example").read_text() + self.assertIn("CD_API_TOKEN=REPLACE_WITH_", secrets) + self.assertIn("DISCORD_BOT_TOKEN=REPLACE_WITH_", secrets) + + def test_compose_is_digest_pinned_and_portable(self): + base = (ROOT / "compose.yaml").read_text() + browser = (ROOT / "compose.browser.yaml").read_text() + text = base + "\n" + browser + image_lines = [line.strip() for line in text.splitlines() if line.strip().startswith("image:")] + self.assertEqual(2, len(image_lines)) + self.assertTrue(all("@sha256:" in line for line in image_lines)) + self.assertIn("${CHANGEDETECTION_DATA_DIR:-./data/changedetection}", base) + self.assertNotIn("container_name:", text) + self.assertNotIn("sockpuppet", base.lower()) + self.assertNotIn("SYS_ADMIN", base) + self.assertIn("sockpuppet", browser.lower()) + self.assertIn("SYS_ADMIN", browser) + makefile = (ROOT / "Makefile").read_text() + self.assertIn( + "docker compose -f compose.yaml -f compose.browser.yaml down --remove-orphans", + makefile, + ) + readme = (ROOT / "README.md").read_text() + self.assertIn("`amd64` or `arm64`", readme) + self.assertIn("optional browser override adds `SYS_ADMIN`", readme) + self.assertIn("not required", readme) + self.assertIn("deployment-policy defaults", readme) + + def test_readme_documents_first_clone_update_and_removal_boundaries(self): + readme = (ROOT / "README.md").read_text() + for required in ( + "### Linux / WSL on a work computer", + "Python 3.11", "Compose v2", "WSL 2", + 'git clone -- "$REPOSITORY_URL" changedetection-competitor-monitor', + "./scripts/setup.sh --no-start", + "Edit **the copies**, not the tracked `*.example` templates", + "## Updating an existing clone", + "git pull --ff-only origin main", + "both the timers and any in-flight jobs", + "## Stop or remove this installation", + "does not stop an already-running monitor service", + "not a complete public-history audit", + "every reachable Git revision and blob", + "no push permission or hosting-account token is required", + ): + with self.subTest(required=required): + self.assertIn(required, readme) + for line in readme.splitlines(): + if "docker compose" in line and re.search(r"\bdown\b", line): + self.assertIn("--remove-orphans", line) + for label, pattern in FORBIDDEN_PATTERNS: + self.assertIsNone(pattern.search(readme), label) + + def test_live_doctor_proves_discord_send_readback_and_delete(self): + responses = [ + (200, {}), + (200, {"id": "bot"}), + (200, {"id": "thread"}), + (200, {"id": "message"}), + (200, {"id": "message"}), + (204, None), + (404, None), + ] + checks = {} + config = { + "changedetection_base": "http://changedetection.test/api/v1", + "discord": {"thread_id": "thread"}, + } + env = {"CD_API_TOKEN": "cd-token", "DISCORD_BOT_TOKEN": "bot-token"} + with mock.patch.object(doctor, "http_json", side_effect=responses) as request: + doctor.live_checks(checks, config, env, verify_discord_send=True) + self.assertTrue(all(row["ok"] for row in checks.values()), checks) + self.assertEqual("POST", request.call_args_list[3].kwargs["method"]) + self.assertEqual("DELETE", request.call_args_list[5].kwargs["method"]) + + def test_systemd_renderer_replaces_every_token(self): + user = pwd.getpwuid(os.getuid()).pw_name + group = grp.getgrgid(pwd.getpwnam(user).pw_gid).gr_name + with tempfile.TemporaryDirectory(dir=ROOT.parent) as td: + output = Path(td) / "units" + result = subprocess.run( + [ + "python3", str(ROOT / "scripts/render_systemd.py"), + "--install-dir", str(ROOT), "--run-user", user, + "--run-group", group, "--output-dir", str(output), + ], + text=True, + capture_output=True, + ) + self.assertEqual(0, result.returncode, result.stderr) + units = sorted(output.iterdir()) + self.assertEqual(7, len(units)) + merged = "\n".join(path.read_text() for path in units) + self.assertNotIn("@INSTALL_DIR@", merged) + self.assertNotIn("@RUN_USER@", merged) + self.assertIn(str(ROOT), merged) + installer = (ROOT / "scripts/install-systemd.sh").read_text() + self.assertIn("runuser", installer) + self.assertIn("--verify-discord-send", installer) + self.assertLess(installer.index("systemd-analyze verify"), installer.index("install -m 0644")) + + def test_setup_initializes_without_overwriting(self): + with tempfile.TemporaryDirectory(dir=ROOT.parent) as td: + target = Path(td) / "repo" + shutil.copytree( + ROOT, + target, + ignore=shutil.ignore_patterns( + ".git", "build", "state", "data", "__pycache__", "release", + *sorted(RUNTIME_ROOT_FILES), + ), + ) + setup = target / "scripts/setup.sh" + first = subprocess.run(["bash", str(setup), "--no-start"], cwd=target, text=True, capture_output=True) + self.assertEqual(0, first.returncode, first.stderr) + self.assertEqual((target / "config.example.json").read_bytes(), (target / "config.json").read_bytes()) + self.assertEqual(0o600, (target / "secrets.env").stat().st_mode & 0o777) + published = {path.relative_to(target).as_posix() for path in publishable_files(target)} + self.assertTrue(RUNTIME_ROOT_FILES.isdisjoint(published)) + sentinel = '{"sentinel": true}\n' + (target / "config.json").write_text(sentinel) + second = subprocess.run(["bash", str(setup), "--no-start"], cwd=target, text=True, capture_output=True) + self.assertEqual(0, second.returncode, second.stderr) + self.assertEqual(sentinel, (target / "config.json").read_text()) + + +if __name__ == "__main__": + unittest.main() diff --git a/verification/audit_watch_scope.py b/verification/audit_watch_scope.py new file mode 100755 index 0000000..ba6e172 --- /dev/null +++ b/verification/audit_watch_scope.py @@ -0,0 +1,153 @@ +#!/usr/bin/env python3 +"""Summarize active/paused watch scope without encoding any deployment profile.""" + +import collections +import json +import re +import sys +import urllib.parse +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +from monitor import ChangeDetectionAPI, TAG_PREFIX, load_env, load_json # noqa: E402 + +CFG = load_json(ROOT / "config.json", {}) +ENV = load_env(ROOT / "secrets.env") +STATE = load_json(ROOT / "state/discovery.json", {}) + + +def classify(url): + parsed = urllib.parse.urlsplit(url) + low = urllib.parse.unquote(parsed.path).rstrip("/").lower() or "/" + if low.endswith("/robots.txt"): + return "robots_policy" + if re.search(r"/(?:sitemap(?:_index)?(?:\.xml)?|sitemap\.rss)$", low): + return "sitemap_or_feed" + if re.match(r"^/(?:[a-z]{2}(?:-[a-z]{2})?/)?20\d{2}/\d{2}/[^/]+$", low): + return "dated_article_leaf" + if re.search(r"/(?:case-studies|customer-stories|success-stories)/[^/]+$", low): + return "case_study_leaf" + if re.search(r"/(?:category|tag|topic|author)/[^/]+$", low): + return "taxonomy_listing" + if low.endswith(".md"): + return "markdown_duplicate_candidate" + if re.search( + r"/(?:company|companies|organization|organizations|person|persons|profile|profiles)/", + low, + ) or re.search(r"/(?:[a-z]{2}(?:-[a-z]{2})?/)?\d{5,}[-_/]", low): + return "profile_like" + if re.search(r"/(?:news|press|blog|articles|research|reports)(?:/[^/]+)?$", low): + return "news_or_article_surface" + if re.search(r"(?:price|pricing|package|packages|plan|plans)", low): + return "pricing_or_package" + if re.search(r"(?:api|developer|integration)", low): + return "api_or_integration" + if re.search(r"(?:terms|privacy|policy|policies|security|cookie|data-protection)", low): + return "policy_or_security" + if re.search( + r"(?:product|products|service|services|solutions|reports|rating|monitoring|kyc|" + r"credit|payments?|debt|sales|market)", + low, + ): + return "product_or_service" + if re.match(r"^/[a-z]{2}(?:-[a-z]{2})?(?:/|$)", low): + return "translated_other" + if low == "/": + return "homepage" + return "other_leaf_or_section" + + +def norm_duplicate(url): + parsed = urllib.parse.urlsplit(url) + path = urllib.parse.unquote(parsed.path).rstrip("/") or "/" + path = re.sub(r"\.md$", "", path, flags=re.I).lower() + return f"{parsed.netloc.lower()}{path}" + + +def main(): + cd = ChangeDetectionAPI(CFG["changedetection_base"], ENV["CD_API_TOKEN"]) + watches = cd.request_json("GET", "/watch") or {} + tags = cd.request_json("GET", "/tags") or {} + tag_names = { + (row.get("uuid") or uid): row.get("title", "")[len(TAG_PREFIX):] + for uid, row in tags.items() + if isinstance(row, dict) and row.get("title", "").startswith(TAG_PREFIX) + } + paused_ids = { + row.get("watch_uuid") + for competitor in STATE.get("competitors", {}).values() + for row in competitor.get("urls", {}).values() + if row.get("paused_by_monitor") and row.get("watch_uuid") + } + records = [] + for uid, row in watches.items(): + if not isinstance(row, dict): + continue + competitor = next( + (tag_names.get(tag) for tag in row.get("tags", []) if tag_names.get(tag)), + "untagged", + ) + url = row.get("url", "") + path = urllib.parse.urlsplit(url).path + records.append( + { + "uuid": uid, + "competitor": competitor, + "url": url, + "class": classify(url), + "paused": uid in paused_ids, + "host": urllib.parse.urlsplit(url).netloc.lower(), + "first_segment": path.strip("/").split("/")[0] if path.strip("/") else "/", + } + ) + + out = { + "total": len(records), + "active": sum(not row["paused"] for row in records), + "paused": sum(row["paused"] for row in records), + "per_competitor": {}, + } + for competitor in sorted({row["competitor"] for row in records}): + rows = [row for row in records if row["competitor"] == competitor] + active = [row for row in rows if not row["paused"]] + classes = collections.Counter(row["class"] for row in active) + segments = collections.Counter(row["first_segment"] for row in active) + hosts = collections.Counter(row["host"] for row in active) + years = collections.Counter( + match.group(1) + for row in active + if row["class"] == "dated_article_leaf" + and ( + match := re.match( + r"^/(?:[a-z]{2}(?:-[a-z]{2})?/)?(20\d{2})/", + urllib.parse.urlsplit(row["url"]).path.lower(), + ) + ) + ) + groups = collections.defaultdict(list) + for row in active: + groups[norm_duplicate(row["url"])].append(row["url"]) + duplicates = [values for values in groups.values() if len(values) > 1] + out["per_competitor"][competitor] = { + "total": len(rows), + "active": len(active), + "paused": len(rows) - len(active), + "class_counts": dict(classes.most_common()), + "top_first_segments": dict(segments.most_common(20)), + "hosts": dict(hosts.most_common()), + "dated_article_years": dict(sorted(years.items())), + "normalized_duplicate_groups": len(duplicates), + "normalized_duplicate_urls": sum(map(len, duplicates)), + "class_samples": { + cls: [row["url"] for row in active if row["class"] == cls][:5] + for cls in classes + }, + "duplicate_samples": duplicates[:10], + } + print(json.dumps(out, ensure_ascii=False, indent=2, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/verification/e2e_verify.py b/verification/e2e_verify.py new file mode 100755 index 0000000..1014d57 --- /dev/null +++ b/verification/e2e_verify.py @@ -0,0 +1,249 @@ +#!/usr/bin/env python3 +"""Reversible live E2E verification against ChangeDetection and Discord.""" +from __future__ import annotations + +import json +from pathlib import Path +import shutil +import subprocess +import sys +import time +import urllib.error +import urllib.parse +import urllib.request + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +import monitor + +RUN = str(int(time.time())) +TAG_TITLE = "Monitor fixture • E2E" +PRICE_URL = f"https://httpbin.org/anything/pricing/competitor-monitor-{RUN}" +ABOUT_URL = f"https://httpbin.org/anything/about/competitor-monitor-{RUN}" +STATE_DIR = ROOT / "verification" / ("e2e-state-" + RUN) +REPORT_PATH = ROOT / "verification" / "e2e-report.json" + +config = monitor.load_json(ROOT / "config.json", {}) +env = monitor.load_env(ROOT / "secrets.env") +cd = monitor.ChangeDetectionAPI(config["changedetection_base"], env.get("CD_API_TOKEN", ""), timeout=60) +discord_cfg = config["discord"] +discord = monitor.DiscordClient(env.get("DISCORD_BOT_TOKEN", ""), discord_cfg["thread_id"], discord_cfg.get("important_mention_user_id"), timeout=60) + +watch_ids: list[str] = [] +tag_id: str | None = None +message_ids: list[str] = [] +report = {"run": RUN, "started_at": monitor.utc_now(), "checks": {}, "cleanup": {}} + + +def check(name, condition, detail=None): + report["checks"][name] = {"ok": bool(condition), "detail": detail} + if not condition: + raise AssertionError(f"{name}: {detail}") + + +def cd_delete(path): + body, _headers, status = cd._request("DELETE", path) + return status, body.decode(errors="replace") + + +def force_recheck(uuid): + cd._request("GET", f"/watch/{uuid}", params={"recheck": "true"}) + + +def wait_history(uuid, count, timeout=240): + deadline = time.time() + timeout + last = {} + while time.time() < deadline: + last = cd.request_json("GET", f"/watch/{uuid}/history") or {} + if len(last) >= count: + return last + time.sleep(2) + raise TimeoutError(f"watch {uuid} history remained at {len(last)}, wanted {count}") + + +def discord_request(method, path): + req = urllib.request.Request( + "https://discord.com/api/v10" + path, + headers={"Authorization": "Bot " + env["DISCORD_BOT_TOKEN"], "User-Agent": "HermesCompetitorMonitorE2E/1.0"}, + method=method, + ) + with urllib.request.urlopen(req, timeout=30) as response: + body = response.read() + return response.status, json.loads(body) if body else None + + +def recent_marker_messages(marker): + me_status, me = discord_request("GET", "/users/@me") + check("discord_current_user_status", me_status == 200 and bool(me.get("id")), me_status) + status, rows = discord_request("GET", f"/channels/{discord_cfg['thread_id']}/messages?limit=100") + check("discord_fetch_status", status == 200, status) + return [row for row in rows if marker in row.get("content", "") and row.get("author", {}).get("id") == me["id"]] + + +try: + result = cd.request_json("POST", "/tag", data={"title": TAG_TITLE, "notification_muted": True}) + tag_id = result["uuid"] + check("fixture_tag_created", bool(tag_id), tag_id) + + common = { + "tag_uuids": tag_id, + "dedupe": "true", + "fetch_backend": "html_requests", + "processor": "text_json_diff", + "notification_muted": "true", + "filter_text_added": "true", + "filter_text_removed": "true", + "filter_text_replaced": "true", + } + for url, headers in [ + (PRICE_URL, {"X-Hermes-Price": "EUR-19"}), + (ABOUT_URL, {"X-Hermes-Address": "Old address"}), + ]: + params = dict(common) + params["headers"] = json.dumps(headers) + created = cd.request_json("POST", "/import", params=params, data=url) or [] + check("fixture_watch_import_" + ("price" if url == PRICE_URL else "about"), len(created) == 1, created) + watch_ids.extend(created) + + duplicate = cd.request_json("POST", "/import", params=common, data="\n".join([PRICE_URL, ABOUT_URL])) or [] + check("changedetection_import_dedup", duplicate == [], duplicate) + + for uuid in watch_ids: + force_recheck(uuid) + baseline = {uuid: wait_history(uuid, 1) for uuid in watch_ids} + check("baseline_snapshots", all(len(x) == 1 for x in baseline.values()), {k: len(v) for k, v in baseline.items()}) + + cd.request_json("PUT", f"/watch/{watch_ids[0]}", data={"headers": {"X-Hermes-Price": "EUR-29"}}) + cd.request_json("PUT", f"/watch/{watch_ids[1]}", data={"headers": {"X-Hermes-Address": "New address"}}) + time.sleep(2) + for uuid in watch_ids: + force_recheck(uuid) + changed = {uuid: wait_history(uuid, 2) for uuid in watch_ids} + check("changed_snapshots", all(len(x) >= 2 for x in changed.values()), {k: len(v) for k, v in changed.items()}) + + price_ts = sorted(changed[watch_ids[0]], key=int) + diff = cd.request_text("GET", f"/watch/{watch_ids[0]}/difference/{price_ts[-2]}/{price_ts[-1]}", params={"format": "text"}) + check("before_after_diff", "EUR-19" in diff and "EUR-29" in diff, diff[:500]) + + STATE_DIR.mkdir(parents=True, exist_ok=True) + monitor.save_json_atomic(STATE_DIR / "collector.json", { + "version": 1, + "last_history": {uuid: sorted(baseline[uuid], key=int)[-1] for uuid in watch_ids}, + "seen_events": [], + "failure_counts": {}, + "failure_alerted": {}, + }) + original_prefix = monitor.TAG_PREFIX + monitor.TAG_PREFIX = "Monitor fixture • " + collected = monitor.run_collect(config, STATE_DIR, cd, discord, send=True) + check("collector_classification", collected["important_count"] == 1 and collected["minor_count"] == 1, { + "important": collected["important_count"], "minor": collected["minor_count"] + }) + + immediate = recent_marker_messages(RUN) + check("important_discord_delivery", len(immediate) == 1, [x.get("id") for x in immediate]) + immediate_message = immediate[0] + message_ids.append(immediate_message["id"]) + mentions = {x["id"] for x in immediate_message.get("mentions", [])} + check("important_discord_mention", discord_cfg["important_mention_user_id"] in mentions, sorted(mentions)) + + silent = monitor.run_collect(config, STATE_DIR, cd, discord, send=True) + check("collector_silent_dedupe", silent["important_count"] == 0 and silent["minor_count"] == 0, silent) + after_silent = recent_marker_messages(RUN) + check("silent_path_no_delivery", len(after_silent) == 1, [x.get("id") for x in after_silent]) + + digest = monitor.run_digest(config, STATE_DIR, discord, send=True) + check("daily_digest_delivery", digest["delivered_count"] == 1 and bool(digest.get("receipt_message_id")), digest) + message_ids.append(digest["receipt_message_id"]) + status, digest_message = discord_request("GET", f"/channels/{discord_cfg['thread_id']}/messages/{digest['receipt_message_id']}") + check("daily_digest_readback", status == 200 and ABOUT_URL in digest_message.get("content", ""), status) + check("daily_digest_no_mention", digest_message.get("mentions") == [], digest_message.get("mentions")) + digest_silent = monitor.run_digest(config, STATE_DIR, discord, send=True) + check("empty_digest_silent", digest_silent["delivered_count"] == 0 and digest_silent["silent"], digest_silent) + + # Exercise the systemd OnFailure helper and its anti-spam path without + # failing a real production unit. + failure_unit = "hermes-competitor-fixture.service" + failure_state_path = ROOT / "state/service-failure-notifications.json" + failure_state = monitor.load_json(failure_state_path, {"units": {}}) + failure_state.setdefault("units", {}).pop(failure_unit, None) + monitor.save_json_atomic(failure_state_path, failure_state) + failure_cmd = [sys.executable, str(ROOT / "notify_failure.py"), failure_unit] + first_failure = subprocess.run(failure_cmd, cwd=ROOT, text=True, capture_output=True, check=True) + failure_result = json.loads(first_failure.stdout) + failure_message_id = str( + failure_result.get("receipt_message_id") + or failure_result.get("receipt", {}).get("id", "") + ) + check("failure_alert_delivery", bool(failure_message_id), failure_result) + if failure_message_id: + message_ids.append(failure_message_id) + status, failure_message = discord_request( + "GET", f"/channels/{discord_cfg['thread_id']}/messages/{failure_message_id}" + ) + check("failure_alert_readback", status == 200, status) + failure_mentions = {str(row.get("id")) for row in failure_message.get("mentions", [])} + check("failure_alert_mention", discord_cfg["important_mention_user_id"] in failure_mentions, sorted(failure_mentions)) + check("failure_alert_semantics", "coverage gap" in failure_message.get("content", "").lower(), failure_message.get("content", "")[:300]) + second_failure = subprocess.run(failure_cmd, cwd=ROOT, text=True, capture_output=True, check=True) + throttled = json.loads(second_failure.stdout) + check( + "failure_alert_throttled", + throttled.get("sent") is False and throttled.get("reason") == "throttled", + throttled, + ) + monitor.TAG_PREFIX = original_prefix + +finally: + monitor.TAG_PREFIX = monitor.TAG_PREFIX if monitor.TAG_PREFIX != "Monitor fixture • " else "Competitor • " + for message_id in list(dict.fromkeys(message_ids)): + try: + status, _ = discord_request("DELETE", f"/channels/{discord_cfg['thread_id']}/messages/{message_id}") + report["cleanup"]["discord_" + message_id] = status == 204 + except Exception as exc: + report["cleanup"]["discord_" + message_id] = f"{type(exc).__name__}: {exc}" + for uuid in watch_ids: + try: + status, _ = cd_delete(f"/watch/{uuid}") + report["cleanup"]["watch_" + uuid] = status in {200, 204} + except Exception as exc: + report["cleanup"]["watch_" + uuid] = f"{type(exc).__name__}: {exc}" + if tag_id: + try: + status, _ = cd_delete(f"/tag/{tag_id}") + report["cleanup"]["tag_" + tag_id] = status == 204 or status == 200 + except Exception as exc: + report["cleanup"]["tag_" + tag_id] = f"{type(exc).__name__}: {exc}" + try: + watches = cd.request_json("GET", "/watch") or {} + tags = cd.request_json("GET", "/tags") or {} + report["cleanup"]["fixture_watches_absent"] = all(uuid not in watches for uuid in watch_ids) + report["cleanup"]["fixture_tag_absent"] = tag_id not in tags if tag_id else True + except Exception as exc: + report["cleanup"]["readback"] = f"{type(exc).__name__}: {exc}" + if STATE_DIR.exists(): + shutil.rmtree(STATE_DIR) + report["cleanup"]["local_fixture_state_absent"] = not STATE_DIR.exists() + failure_state_path = ROOT / "state/service-failure-notifications.json" + failure_state = monitor.load_json(failure_state_path, {"units": {}}) + failure_state.setdefault("units", {}).pop("hermes-competitor-fixture.service", None) + monitor.save_json_atomic(failure_state_path, failure_state) + report["cleanup"]["failure_fixture_state_absent"] = ( + "hermes-competitor-fixture.service" + not in monitor.load_json(failure_state_path, {"units": {}}).get("units", {}) + ) + report["finished_at"] = monitor.utc_now() + REPORT_PATH.parent.mkdir(parents=True, exist_ok=True) + monitor.save_json_atomic(REPORT_PATH, report) + +failed = [name for name, row in report["checks"].items() if not row.get("ok")] +cleanup_failed = [name for name, ok in report["cleanup"].items() if ok is not True] +print(json.dumps({ + "checks": len(report["checks"]), + "failed": failed, + "cleanup_items": len(report["cleanup"]), + "cleanup_failed": cleanup_failed, + "report": str(REPORT_PATH), +}, sort_keys=True)) +raise SystemExit(1 if failed or cleanup_failed else 0) diff --git a/verification/verify_persistence.py b/verification/verify_persistence.py new file mode 100755 index 0000000..018fa04 --- /dev/null +++ b/verification/verify_persistence.py @@ -0,0 +1,166 @@ +#!/usr/bin/env python3 +"""Snapshot and compare live ChangeDetection state across a restart.""" +from __future__ import annotations + +import argparse +from concurrent.futures import ThreadPoolExecutor, as_completed +import hashlib +import json +import time +import urllib.error +import urllib.request +from collections import Counter +from datetime import datetime, timezone +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +VERIFY = ROOT / "verification" +PRE = VERIFY / "pre-restart.json" +REPORT = VERIFY / "restart-persistence-report.json" + + +def load_env() -> dict[str, str]: + out: dict[str, str] = {} + for line in (ROOT / "secrets.env").read_text().splitlines(): + if line and not line.lstrip().startswith("#") and "=" in line: + key, value = line.split("=", 1) + out[key.strip()] = value.strip().strip('"').strip("'") + return out + + +def api_get(path: str, attempts: int = 30) -> dict: + token = load_env()["CD_API_TOKEN"] + config = json.loads((ROOT / "config.json").read_text()) + base_url = config["changedetection_base"].rstrip("/") + request = urllib.request.Request( + base_url + path, + headers={"x-api-key": token}, + ) + last_error: Exception | None = None + for _ in range(attempts): + try: + with urllib.request.urlopen(request, timeout=15) as response: + return json.load(response) + except (OSError, urllib.error.URLError, json.JSONDecodeError) as exc: + last_error = exc + time.sleep(2) + raise RuntimeError(f"API did not recover after {attempts} attempts: {type(last_error).__name__}") + + +def snapshot() -> dict: + watches = api_get("/watch") or {} + tags = api_get("/tags") or {} + detailed_watches: dict[str, dict] = {} + detail_errors: dict[str, str] = {} + with ThreadPoolExecutor(max_workers=12) as pool: + futures = {pool.submit(api_get, f"/watch/{uuid}", 3): uuid for uuid in watches} + for future in as_completed(futures): + uuid = futures[future] + try: + detailed_watches[uuid] = future.result() + except Exception as exc: + detail_errors[uuid] = f"{type(exc).__name__}: {exc}" + if detail_errors or len(detailed_watches) != len(watches): + raise RuntimeError( + "authoritative watch detail inventory incomplete: " + + json.dumps( + {"expected": len(watches), "received": len(detailed_watches), "errors": detail_errors}, + sort_keys=True, + ) + ) + + tag_titles = {uuid: row.get("title", "") for uuid, row in tags.items()} + per_tag = Counter() + per_tag_active = Counter() + per_tag_paused = Counter() + for watch in detailed_watches.values(): + paused = bool(watch.get("paused")) + for tag_uuid in watch.get("tags", []): + title = tag_titles.get(tag_uuid, f"unknown:{tag_uuid}") + per_tag[title] += 1 + (per_tag_paused if paused else per_tag_active)[title] += 1 + + inventory = [ + { + "uuid": uuid, + "url": watch.get("url"), + "title": watch.get("title"), + "tags": sorted(watch.get("tags", [])), + "paused": bool(watch.get("paused")), + } + for uuid, watch in detailed_watches.items() + ] + inventory.sort(key=lambda row: row["uuid"]) + tag_inventory = [ + {"uuid": uuid, "title": row.get("title")} + for uuid, row in tags.items() + ] + tag_inventory.sort(key=lambda row: row["uuid"]) + digest_input = json.dumps( + {"watches": inventory, "tags": tag_inventory}, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ).encode() + return { + "at": datetime.now(timezone.utc).isoformat(timespec="seconds"), + "watch_count": len(watches), + "authoritative_detail_count": len(detailed_watches), + "tag_count": len(tags), + "per_tag_watch_count": dict(sorted(per_tag.items())), + "per_tag_active_watch_count": dict(sorted(per_tag_active.items())), + "per_tag_paused_watch_count": dict(sorted(per_tag_paused.items())), + "active_watch_count": sum(not bool(w.get("paused")) for w in detailed_watches.values()), + "paused_watch_count": sum(bool(w.get("paused")) for w in detailed_watches.values()), + "watch_error_count": sum(bool(w.get("last_error")) for w in detailed_watches.values()), + "fixture_watch_count": sum( + "httpbin.org/anything" in str(w.get("url", "")) + or "hermes" in str(w.get("title", "")).lower() + for w in detailed_watches.values() + ), + "fixture_tag_count": sum( + "fixture" in str(t.get("title", "")).lower() for t in tags.values() + ), + "inventory_sha256": hashlib.sha256(digest_input).hexdigest(), + } + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("phase", choices=("before", "after")) + args = parser.parse_args() + VERIFY.mkdir(parents=True, exist_ok=True) + current = snapshot() + if args.phase == "before": + PRE.write_text(json.dumps(current, indent=2, sort_keys=True) + "\n") + print(json.dumps(current, indent=2, sort_keys=True)) + return 0 + + before = json.loads(PRE.read_text()) + invariant_keys = ( + "watch_count", + "authoritative_detail_count", + "tag_count", + "per_tag_watch_count", + "per_tag_active_watch_count", + "per_tag_paused_watch_count", + "active_watch_count", + "paused_watch_count", + "fixture_watch_count", + "fixture_tag_count", + "inventory_sha256", + ) + comparisons = {key: before.get(key) == current.get(key) for key in invariant_keys} + report = { + "before": before, + "after": current, + "comparisons": comparisons, + "all_persistence_checks_ok": all(comparisons.values()), + } + REPORT.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n") + print(json.dumps(report, indent=2, sort_keys=True)) + return 0 if report["all_persistence_checks_ok"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/verification/verify_representative_scope.py b/verification/verify_representative_scope.py new file mode 100755 index 0000000..89bff09 --- /dev/null +++ b/verification/verify_representative_scope.py @@ -0,0 +1,170 @@ +#!/usr/bin/env python3 +import collections +from concurrent.futures import ThreadPoolExecutor, as_completed +import json +import re +import sys +import urllib.parse +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) +from monitor import ( # noqa: E402 + ChangeDetectionAPI, + TAG_PREFIX, + _watch_category, + _watch_family, + _watch_tags, + canonicalize_url, + is_ignored, + load_env, + load_json, + monitor_paused_watch_ids, +) + + +def main(): + config = load_json(ROOT / "config.json", {}) + ignore = load_json(ROOT / "ignore.json", {}) + state = load_json(ROOT / "state/discovery.json", {}) + last_discovery = load_json(ROOT / "state/last-discovery-run.json", {}) + env = load_env(ROOT / "secrets.env") + cd = ChangeDetectionAPI(config["changedetection_base"], env["CD_API_TOKEN"]) + watches = cd.request_json("GET", "/watch") or {} + tags = cd.request_json("GET", "/tags") or {} + tag_names = { + row.get("uuid") or uuid: row.get("title", "")[len(TAG_PREFIX):] + for uuid, row in tags.items() + if isinstance(row, dict) and row.get("title", "").startswith(TAG_PREFIX) + } + comp_by_name = {comp["name"]: comp for comp in config.get("competitors", [])} + paused_ids = monitor_paused_watch_ids(state) + tagged_ids = [ + watch.get("uuid") or uuid + for uuid, watch in watches.items() + if isinstance(watch, dict) and any(tag in tag_names for tag in _watch_tags(watch)) + ] + details = {} + detail_errors = [] + with ThreadPoolExecutor(max_workers=16) as executor: + futures = { + executor.submit(cd.request_json, "GET", f"/watch/{uuid}"): uuid + for uuid in tagged_ids + } + for future in as_completed(futures): + uuid = futures[future] + try: + details[uuid] = future.result() or {} + except Exception as exc: + detail_errors.append({"uuid": uuid, "error": type(exc).__name__}) + active = [] + paused = [] + for uuid, watch in watches.items(): + if not isinstance(watch, dict): + continue + company = next((tag_names[tag] for tag in _watch_tags(watch) if tag in tag_names), None) + if not company: + continue + watch_uuid = watch.get("uuid") or uuid + row = { + "uuid": watch_uuid, + "url": canonicalize_url(watch.get("url", "")), + "competitor": company, + } + (paused if bool(details.get(watch_uuid, {}).get("paused")) else active).append(row) + + active_urls = {row["url"] for row in active} + markdown_duplicates = sorted( + url for url in active_urls + if url and url.lower().endswith(".md") and url[:-3] in active_urls + ) + profile_pattern = re.compile( + r"/(?:company|companies|organization|organizations|person|persons|profile|profiles)/[^/]*\d+", + re.I, + ) + active_profile_like = sorted(url for url in active_urls if url and profile_pattern.search(urllib.parse.unquote(url))) + + group_counts = collections.Counter() + ignored_active = [] + excluded_category_active = [] + state_mismatches = [] + manually_paused_selected = [] + per_competitor = collections.Counter() + for row in active: + url = row["url"] + comp = comp_by_name[row["competitor"]] + per_competitor[row["competitor"]] += 1 + category = _watch_category(url) + if category in set(config["global"]["watch_selection"].get("exclude_categories", [])): + excluded_category_active.append({"url": url, "category": category}) + group_counts[(row["competitor"], urllib.parse.urlsplit(url).netloc.lower(), category, _watch_family(url, category))] += 1 + if is_ignored(url, comp, config["global"], ignore): + ignored_active.append(url) + ledger_row = state.get("competitors", {}).get(comp["slug"], {}).get("urls", {}).get(url, {}) + if not ledger_row.get("selected_for_watch"): + state_mismatches.append({"url": url, "reason": "active_not_selected"}) + over_limit = { + " | ".join(key): count + for key, count in group_counts.items() + if count > int(config["global"]["watch_selection"]["per_group"]) + } + configured_competitors = {comp["name"] for comp in config.get("competitors", [])} + zero_active = sorted(configured_competitors - set(per_competitor)) + slug_by_name = {comp["name"]: comp["slug"] for comp in config.get("competitors", [])} + zero_active_with_gap = sorted( + name for name in zero_active + if last_discovery.get("competitors", {}).get(slug_by_name[name], {}).get("coverage_gap") + ) + zero_active_without_gap = sorted(set(zero_active) - set(zero_active_with_gap)) + for row in paused: + comp = comp_by_name[row["competitor"]] + ledger_row = state.get("competitors", {}).get(comp["slug"], {}).get("urls", {}).get(row["url"], {}) + if ledger_row.get("selected_for_watch") and not ledger_row.get("ignored"): + if ledger_row.get("paused_by_monitor"): + state_mismatches.append({"url": row["url"], "reason": "selected_but_monitor_paused"}) + else: + manually_paused_selected.append(row["url"]) + elif not ledger_row.get("paused_by_monitor"): + state_mismatches.append({"url": row["url"], "reason": "paused_out_of_scope_without_monitor_state"}) + for row in active: + comp = comp_by_name[row["competitor"]] + ledger_row = state.get("competitors", {}).get(comp["slug"], {}).get("urls", {}).get(row["url"], {}) + if ledger_row.get("paused_by_monitor") or row["uuid"] in paused_ids: + state_mismatches.append({"url": row["url"], "reason": "monitor_pause_not_applied_in_api"}) + + report = { + "total_tagged_watches": len(active) + len(paused), + "active_representative_watches": len(active), + "paused_retained_watches": len(paused), + "authoritative_detail_records": len(details), + "detail_read_errors": detail_errors, + "manually_paused_selected_urls": sorted(manually_paused_selected), + "per_competitor_active": dict(sorted(per_competitor.items())), + "zero_active_competitors_with_reported_gap": zero_active_with_gap, + "zero_active_competitors_without_reported_gap": zero_active_without_gap, + "active_profile_like_urls": active_profile_like, + "active_html_markdown_duplicates": markdown_duplicates, + "active_excluded_category_urls": sorted(excluded_category_active, key=lambda row: row["url"]), + "ignored_active_urls": sorted(ignored_active), + "state_mismatches": state_mismatches, + "over_limit_groups": over_limit, + } + report["all_checks_ok"] = all( + not report[key] + for key in ( + "active_profile_like_urls", + "active_html_markdown_duplicates", + "active_excluded_category_urls", + "detail_read_errors", + "ignored_active_urls", + "state_mismatches", + "over_limit_groups", + "zero_active_competitors_without_reported_gap", + ) + ) + print(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True)) + raise SystemExit(0 if report["all_checks_ok"] else 1) + + +if __name__ == "__main__": + main() diff --git a/verification/verify_ui.py b/verification/verify_ui.py new file mode 100755 index 0000000..9e8da32 --- /dev/null +++ b/verification/verify_ui.py @@ -0,0 +1,132 @@ +#!/usr/bin/env python3 +"""Exercise the ChangeDetection native-login UI through an existing Chrome CDP session. + +Start Chrome separately with a disposable profile and remote debugging enabled, then +pass its browser endpoint with --cdp-url. The script never launches Chrome and never +stores the password in the repository. +""" +from __future__ import annotations + +import argparse +import asyncio +import base64 +import itertools +import json +import os +from pathlib import Path +import urllib.request + +import websockets + +ROOT = Path(__file__).resolve().parents[1] + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--url", default=os.environ.get("PUBLIC_BASE_URL", "http://127.0.0.1:5000/")) + parser.add_argument("--cdp-url", default=os.environ.get("CHROME_CDP_URL", "http://127.0.0.1:9222")) + parser.add_argument("--password-file", type=Path) + parser.add_argument("--screenshot", type=Path, default=ROOT / "verification" / "ui-smoke.png") + return parser.parse_args() + + +def read_password(path: Path | None) -> str: + direct = os.environ.get("CD_UI_PASSWORD", "") + if direct: + return direct + if path is None: + raise RuntimeError("set CD_UI_PASSWORD or pass --password-file") + text = path.read_text() + for line in text.splitlines(): + if line.lower().startswith("password:"): + return line.split(":", 1)[1].strip() + value = text.strip() + if value and "\n" not in value: + return value + raise RuntimeError("password file has no password value") + + +async def run(args: argparse.Namespace) -> dict: + cdp_base = args.cdp_url.rstrip("/") + version = json.load(urllib.request.urlopen(cdp_base + "/json/version", timeout=10)) + async with websockets.connect(version["webSocketDebuggerUrl"], max_size=20_000_000) as browser: + seq = itertools.count(1) + + async def bcall(method, params=None): + ident = next(seq) + await browser.send(json.dumps({"id": ident, "method": method, "params": params or {}})) + while True: + msg = json.loads(await browser.recv()) + if msg.get("id") == ident: + if "error" in msg: + raise RuntimeError(msg["error"]) + return msg.get("result", {}) + + target = await bcall("Target.createTarget", {"url": args.url}) + + targets = json.load(urllib.request.urlopen(cdp_base + "/json", timeout=10)) + ws_url = next(t["webSocketDebuggerUrl"] for t in targets if t["id"] == target["targetId"]) + async with websockets.connect(ws_url, max_size=50_000_000) as page: + seq = itertools.count(1) + + async def call(method, params=None): + ident = next(seq) + await page.send(json.dumps({"id": ident, "method": method, "params": params or {}})) + while True: + msg = json.loads(await page.recv()) + if msg.get("id") == ident: + if "error" in msg: + raise RuntimeError(msg["error"]) + return msg.get("result", {}) + + await call("Page.enable") + await call("DOM.enable") + await call("Runtime.enable") + await asyncio.sleep(3) + location = (await call("Runtime.evaluate", {"expression": "location.href", "returnByValue": True}))["result"]["value"] + login_exercised = False + if "/login" in location: + password = read_password(args.password_file) + doc = await call("DOM.getDocument") + node = await call("DOM.querySelector", {"nodeId": doc["root"]["nodeId"], "selector": "input[type=password]"}) + if not node.get("nodeId"): + raise RuntimeError("native password field not found") + await call("DOM.focus", {"nodeId": node["nodeId"]}) + await call("Input.insertText", {"text": password}) + button = await call("DOM.querySelector", {"nodeId": doc["root"]["nodeId"], "selector": "button[type=submit], input[type=submit]"}) + box = await call("DOM.getBoxModel", {"nodeId": button["nodeId"]}) + quad = box["model"]["content"] + x = sum(quad[0::2]) / 4 + y = sum(quad[1::2]) / 4 + await call("Input.dispatchMouseEvent", {"type": "mousePressed", "x": x, "y": y, "button": "left", "clickCount": 1}) + await call("Input.dispatchMouseEvent", {"type": "mouseReleased", "x": x, "y": y, "button": "left", "clickCount": 1}) + login_exercised = True + await asyncio.sleep(5) + + expression = "JSON.stringify({url:location.href,title:document.title,body:document.body.innerText.slice(0,4000),rows:document.querySelectorAll('table tbody tr, .watch-table-row').length,tagLinks:[...document.querySelectorAll('a')].filter(a=>a.textContent.includes('Competitor')).slice(0,10).map(a=>a.textContent.trim())})" + result = await call("Runtime.evaluate", {"expression": expression, "returnByValue": True}) + data = json.loads(result["result"]["value"]) + args.screenshot.parent.mkdir(parents=True, exist_ok=True) + shot = await call("Page.captureScreenshot", {"format": "png", "captureBeyondViewport": False}) + args.screenshot.write_bytes(base64.b64decode(shot["data"])) + body = data.pop("body") + final = { + **data, + "login_exercised": login_exercised, + "authenticated_dashboard": "/login" not in data["url"], + "competitor_text": "Competitor" in body, + "screenshot": str(args.screenshot), + } + if not final["authenticated_dashboard"]: + raise RuntimeError("browser remained on the login page") + return final + + +def main() -> int: + result = asyncio.run(run(parse_args())) + print(json.dumps(result, ensure_ascii=False, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) -- 2.43.0