+#!/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"<?xml"):
+ continue
+ kind, found = parse_xml_links(body)
+ if kind == "sitemapindex":
+ for child in found:
+ canon = canonicalize_url(urllib.parse.urljoin(final, child))
+ if canon and canon not in seen_sitemaps and not _sitemap_excluded(canon, comp):
+ sitemap_queue.append(canon)
+ elif kind == "urlset":
+ for page in found:
+ canon = canonicalize_url(urllib.parse.urljoin(final, page))
+ if canon and url_allowed(canon, comp, global_cfg, ignore) and robots_allows(canon):
+ urls.add(canon)
+ elif kind == "feed":
+ feed_urls.add(canonicalize_url(final) or final)
+ for page in found:
+ canon = canonicalize_url(urllib.parse.urljoin(final, page))
+ if canon and url_allowed(canon, comp, global_cfg, ignore) and robots_allows(canon):
+ urls.add(canon)
+ except urllib.error.HTTPError as exc:
+ if exc.code != 404:
+ errors.append(f"sitemap {sitemap}: HTTP {exc.code}")
+ except Exception as exc:
+ errors.append(f"sitemap {sitemap}: {type(exc).__name__}: {exc}")
+
+ max_urls = int(comp.get("max_urls", global_cfg.get("max_urls", 500)))
+ crawl_limit = int(comp.get("crawl_pages", global_cfg.get("crawl_pages", 100)))
+ max_depth = int(comp.get("crawl_depth", global_cfg.get("crawl_depth", 2)))
+ crawl_queue = deque((u, 0) for u in seeds)
+ for u in sorted(urls)[: max(0, crawl_limit - len(crawl_queue))]:
+ crawl_queue.append((u, 0))
+ crawled: set[str] = set()
+
+ while crawl_queue and len(crawled) < crawl_limit and len(urls) < max_urls:
+ url, depth = crawl_queue.popleft()
+ url = canonicalize_url(url)
+ if not url or url in crawled or not url_allowed(url, comp, global_cfg, ignore) or not robots_allows(url):
+ continue
+ crawled.add(url)
+ try:
+ final, body, headers, _status = fetcher.get(url, delay=request_delay)
+ if _is_antibot_challenge(final, body):
+ anti_bot_challenge = True
+ errors.append(f"page {url}: anti-bot challenge at {final}")
+ continue
+ ctype = headers.get("content-type", "").lower()
+ if "xml" in ctype or body.lstrip().startswith(b"<?xml"):
+ try:
+ kind, found = parse_xml_links(body)
+ except ET.ParseError:
+ continue
+ if kind == "feed":
+ feed_urls.add(url)
+ urls.add(url)
+ for target in found:
+ canon = canonicalize_url(urllib.parse.urljoin(final, target))
+ if canon and url_allowed(canon, comp, global_cfg, ignore) and robots_allows(canon):
+ urls.add(canon)
+ continue
+ if "html" not in ctype and not body.lstrip().lower().startswith((b"<!doctype html", b"<html")):
+ continue
+ parser = LinkParser()
+ parser.feed(body.decode("utf-8", errors="replace"))
+ canonical = canonicalize_url(urllib.parse.urljoin(final, parser.canonical)) if parser.canonical else url
+ if canonical and url_allowed(canonical, comp, global_cfg, ignore) and robots_allows(canonical):
+ urls.add(canonical)
+ for feed in parser.feeds:
+ canon = canonicalize_url(urllib.parse.urljoin(final, feed))
+ if canon and url_allowed(canon, comp, {**global_cfg, "_allow_metadata_extensions": True}, ignore) and robots_allows(canon):
+ feed_urls.add(canon)
+ if depth < max_depth:
+ crawl_queue.append((canon, depth + 1))
+ if depth < max_depth:
+ for link in parser.links:
+ canon = canonicalize_url(urllib.parse.urljoin(final, link))
+ if canon and url_allowed(canon, comp, global_cfg, ignore) and robots_allows(canon):
+ urls.add(canon)
+ if canon not in crawled:
+ crawl_queue.append((canon, depth + 1))
+ except urllib.error.HTTPError as exc:
+ errors.append(f"page {url}: HTTP {exc.code}")
+ except Exception as exc:
+ errors.append(f"page {url}: {type(exc).__name__}: {exc}")
+
+ # Parse feed item links even when crawl depth prevented a normal crawl.
+ for feed in list(feed_urls):
+ if feed in crawled:
+ continue
+ try:
+ final, body, _headers, _status = fetcher.get(feed, delay=request_delay)
+ kind, found = parse_xml_links(body)
+ if kind == "feed":
+ urls.add(feed)
+ for target in found:
+ canon = canonicalize_url(urllib.parse.urljoin(final, target))
+ if canon and url_allowed(canon, comp, global_cfg, ignore) and robots_allows(canon):
+ urls.add(canon)
+ except urllib.error.HTTPError as exc:
+ # Auto-discovered feed links are frequently stale; a 404 simply
+ # rejects that optional candidate rather than failing site coverage.
+ if exc.code != 404:
+ errors.append(f"feed {feed}: HTTP {exc.code}")
+ except Exception as exc:
+ errors.append(f"feed {feed}: {type(exc).__name__}: {exc}")
+
+ for url in comp.get("policy_watch_urls", []):
+ canon = canonicalize_url(url)
+ if canon and url_allowed(canon, comp, {**global_cfg, "_allow_metadata_extensions": True}, ignore):
+ urls.add(canon)
+
+ status = "ok" if not errors else ("partial" if urls else "error")
+ return {
+ "status": status,
+ "coverage_gap": bool(errors),
+ "errors": errors,
+ "urls": sorted(urls)[:max_urls],
+ "sitemaps_fetched": len(seen_sitemaps),
+ "pages_crawled": len(crawled),
+ "feeds": sorted(feed_urls),
+ "anti_bot_challenge": anti_bot_challenge,
+ }
+
+
+class ChangeDetectionAPI:
+ def __init__(self, base: str, token: str, timeout: int = 30):
+ if not token:
+ raise ValueError("CD_API_TOKEN is not configured")
+ self.base = base.rstrip("/")
+ self.token = token
+ self.timeout = timeout
+
+ def _request(self, method: str, path: str, *, params=None, data=None) -> 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"(?<!\w)" + re.escape(term.lower()) + r"(?!\w)", text)
+ for term in terms
+ )
+
+
+_DIFF_MARKERS = ("(added)", "(removed)", "(replaced)", "(changed)", "(into)")
+_DIFF_MARKER_RE = re.compile(r"^\((added|removed|replaced|changed|into)\)\s*(.*)$")
+
+
+def _marker_diff_text(diff: str) -> 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"(?<!\w)\d+(?:[.,]\d+)?|[€$£]|\b(?:eur|usd|gbp)\b",
+ " ".join(lines).lower(),
+ )
+ return Counter(token.replace(",", ".") for token in tokens)
+
+ return commercial_value_signature(old_lines) == commercial_value_signature(new_lines)
+
+
+def _is_interface_only_diff(diff: str) -> 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())
Setup guide / monitor.git / commitdiff