chris/AV-Watch
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
660d5b2d73df72e47957213c3e98b2b0f7382155
AV-Watch/avwatch/risk/collectors.py
T

367 lines
13 KiB
Python
Raw Blame History

"""Evidence collectors: where community bug reports actually come from.
Verified availability as of 2026-09-08:
DuckDuckGo HTML works, no key. Soft-blocks with an empty HTTP 202 after
roughly two rapid queries, so the shared HttpClient holds
a 6 s/host floor and backs off on 202.
Reddit JSON unauthenticated .json endpoints now return 403. A free
"script" app at reddit.com/prefs/apps gives 100 req/min
via client-credentials OAuth -- strongly recommended,
because r/livesound, r/VIDEOENGINEERING and r/churchtech
are the highest-signal sources for this fleet.
GitHub issues works, best signal available for the open-source items.
HN Algolia works, no key. Low yield for AV gear; off by default.
Gearspace and most vendor forums sit behind Cloudflare and are not fetchable
from a script; they are reachable through the DuckDuckGo path instead.
"""
from __future__ import annotations
import logging
import re
import time
import urllib.parse
from datetime import datetime, timedelta, timezone
from ..http import FetchError, HttpClient
from ..models import Evidence
from .classify import classify, relevance
log = logging.getLogger(__name__)
# --------------------------------------------------------------------------
# DuckDuckGo
# --------------------------------------------------------------------------
_DDG_RESULT = re.compile(
r'<a[^>]+class="result__a"[^>]+href="([^"]+)"[^>]*>(.*?)</a>', re.S
)
_DDG_SNIPPET = re.compile(
r'<a[^>]+class="result__snippet"[^>]*>(.*?)</a>', re.S
)
_TAGS = re.compile(r"<[^>]+>")
def _clean(fragment: str) -> str:
text = _TAGS.sub("", fragment)
for a, b in (("&amp;", "&"), ("&quot;", '"'), ("&#x27;", "'"),
("&#39;", "'"), ("&lt;", "<"), ("&gt;", ">"), ("&nbsp;", " ")):
text = text.replace(a, b)
return re.sub(r"\s+", " ", text).strip()
def _undirect(href: str) -> str:
m = re.search(r"[?&]uddg=([^&]+)", href)
return urllib.parse.unquote(m.group(1)) if m else href
# GitHub issue-template boilerplate. Without this, every snippet in the
# report is "### Operating System Info ### Other OS _No response_ ###..."
# instead of what the person actually reported.
_TEMPLATE_NOISE = re.compile(
r"(?:^|\s)###\s*[^#\n]{0,60}?(?=\s|$)" # markdown headings
r"|_No response_"
r"|<!--.*?-->" # HTML comments
r"|https?://\S*obsproject\.com/logs/\S*" # pasted log links
r"|\|[-: ]+\|", # table rules
re.S,
)
def clean_body(text: str, limit: int = 600) -> str:
"""Turn an issue or post body into something readable in a digest."""
if not text:
return ""
out = _TEMPLATE_NOISE.sub(" ", text)
out = re.sub(r"```.*?```", " [code] ", out, flags=re.S)
out = re.sub(r"[*_`>#]+", "", out)
out = re.sub(r"\s+", " ", out).strip()
return out[:limit]
def duckduckgo(http: HttpClient, query: str, limit: int = 6) -> list[Evidence]:
url = "https://html.duckduckgo.com/html/?q=" + urllib.parse.quote(query)
try:
resp = http.get(url, ttl=12 * 3600)
except FetchError as exc:
log.debug("ddg offline: %s", exc)
return []
if resp.status == 202 or not resp.text.strip():
log.info("DuckDuckGo throttled us (status %s) for %r", resp.status, query)
return []
if not resp.ok:
log.info("DuckDuckGo HTTP %s for %r", resp.status, query)
return []
titles = _DDG_RESULT.findall(resp.text)
snippets = [_clean(s) for s in _DDG_SNIPPET.findall(resp.text)]
out: list[Evidence] = []
for i, (href, raw_title) in enumerate(titles[:limit]):
title = _clean(raw_title)
if not title:
continue
link = _undirect(href)
snippet = snippets[i] if i < len(snippets) else ""
sev, sent, terms = classify(title, snippet)
source = "reddit" if "reddit.com" in link else "web"
out.append(Evidence(
source=source, title=title, url=link, snippet=snippet[:600],
sentiment=sent, severity=sev, matched_terms=terms,
))
return out
# --------------------------------------------------------------------------
# Reddit (OAuth client-credentials)
# --------------------------------------------------------------------------
class RedditClient:
"""Client-credentials OAuth. Needs a free 'script' app:
reddit.com/prefs/apps -> create app -> type 'script'.
Set AVWATCH_REDDIT_CLIENT_ID and AVWATCH_REDDIT_CLIENT_SECRET.
"""
TOKEN_URL = "https://www.reddit.com/api/v1/access_token"
API = "https://oauth.reddit.com"
def __init__(self, http: HttpClient, client_id: str, client_secret: str):
self.http = http
self.client_id = client_id
self.client_secret = client_secret
self._token: str | None = None
self._expires: float = 0.0
@property
def configured(self) -> bool:
return bool(self.client_id and self.client_secret)
def _ensure_token(self) -> str | None:
if not self.configured:
return None
if self._token and time.time() < self._expires - 60:
return self._token
try:
resp = self.http.post_form(
self.TOKEN_URL,
data={"grant_type": "client_credentials"},
auth=(self.client_id, self.client_secret),
)
except FetchError as exc:
log.warning("reddit token request failed: %s", exc)
return None
if not resp.ok:
log.warning("reddit token HTTP %s -- check your app credentials", resp.status)
return None
try:
blob = resp.json()
except ValueError:
return None
self._token = blob.get("access_token")
self._expires = time.time() + float(blob.get("expires_in", 3600))
return self._token
def search(self, query: str, subreddits: list[str], limit: int = 8,
lookback_days: int = 400) -> list[Evidence]:
token = self._ensure_token()
if not token:
return []
# One multireddit request beats N per-subreddit requests.
multi = "+".join(s.strip() for s in subreddits if s.strip())
if not multi:
return []
params = urllib.parse.urlencode({
"q": query, "restrict_sr": "1", "sort": "relevance",
"t": "year", "limit": str(limit), "type": "link",
})
url = f"{self.API}/r/{multi}/search?{params}"
resp = self.http.get(
url, headers={"Authorization": f"Bearer {token}"}, ttl=12 * 3600
)
if not resp.ok:
log.info("reddit search HTTP %s for %r", resp.status, query)
return []
try:
children = resp.json().get("data", {}).get("children", [])
except (ValueError, AttributeError):
return []
cutoff = datetime.now(timezone.utc) - timedelta(days=lookback_days)
out: list[Evidence] = []
for child in children:
d = child.get("data") or {}
title = (d.get("title") or "").strip()
if not title:
continue
created = d.get("created_utc")
published = (
datetime.fromtimestamp(created, tz=timezone.utc)
if isinstance(created, (int, float)) else None
)
if published and published < cutoff:
continue
raw = d.get("selftext") or ""
body = clean_body(raw)
sev, sent, terms = classify(title, raw[:1500])
out.append(Evidence(
source="reddit",
title=title,
url="https://www.reddit.com" + (d.get("permalink") or ""),
snippet=body,
published=published,
score=d.get("score"),
sentiment=sent, severity=sev, matched_terms=terms,
))
return out
# --------------------------------------------------------------------------
# GitHub issues
# --------------------------------------------------------------------------
def github_issues(http: HttpClient, repo: str, version: str | None,
limit: int = 8, lookback_days: int = 400) -> list[Evidence]:
"""Open issues mentioning trouble, newest first. For OBS, Companion and
RustDesk this is the most direct bug signal that exists."""
from ..sources.github_releases import auth_headers
since = (datetime.now(timezone.utc) - timedelta(days=lookback_days)).date().isoformat()
terms = "crash OR freeze OR regression OR broken OR bug"
if version:
terms = f'"{version}" OR ({terms})'
q = f"repo:{repo} is:issue created:>{since} {terms}"
url = (
"https://api.github.com/search/issues?q="
+ urllib.parse.quote(q)
+ f"&sort=created&order=desc&per_page={limit}"
)
resp = http.get(url, headers=auth_headers(), ttl=6 * 3600)
if not resp.ok:
log.info("github issue search HTTP %s for %s", resp.status, repo)
return []
try:
items = resp.json().get("items", [])
except (ValueError, AttributeError):
return []
out: list[Evidence] = []
for it in items:
title = (it.get("title") or "").strip()
if not title:
continue
raw_body = it.get("body") or ""
# Classify against the raw body so template fields like
# "OBS Studio Version 32.2.2" still drive the version match.
body = clean_body(raw_body)
sev, sent, terms_hit = classify(title, raw_body[:1500])
published = None
if raw := it.get("created_at"):
try:
published = datetime.fromisoformat(raw.replace("Z", "+00:00"))
except ValueError:
pass
out.append(Evidence(
source="github-issues", title=title, url=it.get("html_url", ""),
snippet=body, published=published,
score=(it.get("reactions") or {}).get("total_count"),
sentiment=sent, severity=sev, matched_terms=terms_hit,
))
return out
# --------------------------------------------------------------------------
# Hacker News
# --------------------------------------------------------------------------
def hackernews(http: HttpClient, query: str, limit: int = 5) -> list[Evidence]:
url = (
"https://hn.algolia.com/api/v1/search_by_date?query="
+ urllib.parse.quote(query)
+ f"&tags=(story,comment)&hitsPerPage={limit}"
)
resp = http.get(url, ttl=12 * 3600)
if not resp.ok:
return []
try:
hits = resp.json().get("hits", [])
except (ValueError, AttributeError):
return []
out: list[Evidence] = []
for h in hits:
title = (h.get("title") or h.get("story_title") or "").strip()
text = (h.get("comment_text") or h.get("story_text") or "")[:600]
if not title and not text:
continue
sev, sent, terms = classify(title, text)
published = None
if raw := h.get("created_at"):
try:
published = datetime.fromisoformat(raw.replace("Z", "+00:00"))
except ValueError:
pass
oid = h.get("objectID")
out.append(Evidence(
source="hackernews", title=title or text[:80],
url=f"https://news.ycombinator.com/item?id={oid}" if oid else "",
snippet=_clean(text), published=published,
score=h.get("points"), sentiment=sent, severity=sev,
matched_terms=terms,
))
return out
# --------------------------------------------------------------------------
# Ranking
# --------------------------------------------------------------------------
_SEV_WEIGHT = {"brick": 5.0, "crash": 3.0, "regression": 2.0, "annoyance": 0.6, "info": 0.2}
_SOURCE_WEIGHT = {"reddit": 1.2, "github-issues": 1.3, "web": 1.0, "hackernews": 0.7}
def rank_and_filter(
evidence: list[Evidence],
item_terms: list[str],
version: str | None,
limit: int,
min_relevance: float = 0.34,
) -> list[Evidence]:
"""Drop off-topic hits, de-duplicate by URL, keep the most alarming and
most relevant."""
seen: set[str] = set()
scored: list[tuple[float, Evidence]] = []
for e in evidence:
key = (e.url or e.title).lower().rstrip("/")
if key in seen:
continue
seen.add(key)
blob = f"{e.title} {e.snippet}"
rel = relevance(blob, item_terms, version)
if rel < min_relevance:
continue
e.mentions_version = bool(version and version.lower() in blob.lower())
weight = _SEV_WEIGHT.get(e.severity, 0.2) * _SOURCE_WEIGHT.get(e.source, 1.0)
if e.sentiment == "negative":
weight *= 1.6
elif e.sentiment == "positive":
weight *= 0.5
if e.score and e.score > 25:
weight *= 1.25
# Prefer posts that actually name the release being considered, so
# they surface above a project's permanent background of open bugs.
if e.mentions_version:
weight *= 2.0
scored.append((weight * (0.5 + rel), e))
scored.sort(key=lambda t: t[0], reverse=True)
return [e for _, e in scored[:limit]]
Reference in New Issue View Git Blame Copy Permalink