"""Живой прогон кандидатов из tools/fresh_candidates.py через штатные адаптеры omni. python tools/probe_fresh.py # прогон + data/fresh_probe.json python tools/probe_fresh.py --emit # + перегенерация omni/sources_fresh.py python tools/probe_fresh.py --only tg_ --workers 4 --limit 5 Критерий живого: feeds.fetch(src, limit) вернул >= 1 запись. Мёртвым добираем диагностику прямым net.get (код/блок/размер), чтобы причина в отчёте была фактической, а не «0 items». Публичные хелперы (для тестов/внешних вызовов): candidate_source(c) -> (Source, marker) — dict кандидата -> Source, url динамического источника пересчитан ДО сети; маркер лежит в extra["_dynamic"]. _registry_sources() -> [Source] — all_sources(include_fresh=False) _registry_keys() -> (set url, set name) — нормализованные ключи реестра без текущей sources_fresh (иначе перегенерация считает свою волну дублем). _registry_urls() -> set url — совместимость. emit(rows, dead, path, dedupe=True) — только известный dynamic-маркер; неизвестный маркер = ValueError, а не замороженный протухший url. """ import argparse import datetime as _dt import json import sys import time from concurrent.futures import ThreadPoolExecutor from pathlib import Path ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT)) sys.path.insert(0, str(ROOT / "tools")) from omni import feeds, net # noqa: E402 from omni.sources import Source # noqa: E402 from fresh_candidates import CANDIDATES # noqa: E402 def _diag(src): """Почему пусто: код ответа, блокировка, размер тела.""" try: r = net.get(src.url, timeout=20) bits = ["http %s" % r.status] if getattr(r, "blocked", False): bits.append("blocked") if getattr(r, "error", ""): bits.append("err:%s" % str(r.error)[:120]) bits.append("%db" % len(r.text or "")) return ", ".join(bits) except Exception as e: # noqa: BLE001 return "exception:%s" % str(e)[:140] def probe_one(src, limit): t0 = time.time() if src.adapter not in feeds.ADAPTERS: return {"name": src.name, "category": src.category, "kind": src.kind, "adapter": src.adapter, "url": src.url, "tags": list(src.tags), "ok": False, "n": 0, "err": "no_adapter:%s" % src.adapter, "ms": 0, "samples": []} try: items = feeds.fetch(src, limit) or [] err = "" except Exception as e: # noqa: BLE001 items, err = [], "raise:%s" % str(e)[:140] ms = int((time.time() - t0) * 1000) if not items and not err: err = _diag(src) return {"name": src.name, "category": src.category, "kind": src.kind, "adapter": src.adapter, "url": src.url, "tags": list(src.tags), "ok": bool(items), "n": len(items), "dynamic": src.extra.get("_dynamic", ""), "err": err, "ms": ms, "samples": [{"title": i.title[:160], "url": i.url[:200]} for i in items[:3]]} def _norm(u): u = str(u or "").strip().lower().rstrip("/") for pre in ("https://", "http://"): if u.startswith(pre): u = u[len(pre):] return u[4:] if u.startswith("www.") else u def _norm_name(n): """Имя источника — первичный ключ source_flags/runs: сравниваем без регистра, пробелов и разделителей, иначе nvd-recent и nvd_recent разъезжаются.""" return "".join(ch for ch in str(n or "").strip().lower() if ch.isalnum()) def _registry_sources(): """Реестр без текущего sources_fresh: перегенерация не должна считать собственную прошлую волну «уже существующей» и стирать свои же находки.""" from omni.sources import all_sources try: return all_sources(include_fresh=False) except TypeError: # реестр без флага — совместимость return all_sources() def _registry_keys(): """(нормализованные url, нормализованные имена) реестра без sources_fresh.""" rows = _registry_sources() return ({_norm(s.url) for s in rows if s.url}, {_norm_name(s.name) for s in rows if s.name}) def _registry_urls(): """Только url-ключи реестра (совместимость с внешними вызовами).""" return _registry_keys()[0] NVD_BASE = "https://services.nvd.nist.gov/rest/json/cves/2.0" KNOWN_DYNAMIC = ("NVD_LAST_WEEK",) def _nvd_week_url(): """Фактический url NVD на момент прогона: окно lastMod <=120 дней, база без окна отдаёт самые старые записи страницы и как «свежая» бесполезна.""" now = _dt.datetime.utcnow() since = (now - _dt.timedelta(days=7)).strftime("%Y-%m-%dT00:00:00.000") return ("%s?resultsPerPage=30&lastModStartDate=%s&lastModEndDate=%s" % (NVD_BASE, since, now.strftime("%Y-%m-%dT%H:%M:%S.000"))) def candidate_source(c): """dict кандидата -> (Source, dynamic-маркер), url пересчитан до сети. Маркер сохраняется в extra["_dynamic"], чтобы отчёт/emit знали, что url нельзя замораживать. Неизвестный маркер — ValueError: молча подставить статический url значит опубликовать заведомо протухший источник. """ row = dict(c) marker = str(row.pop("dynamic", "") or "") extra = dict(row.get("extra") or {}) if marker: if marker not in KNOWN_DYNAMIC: raise ValueError("unknown dynamic marker %r" % marker) if marker == "NVD_LAST_WEEK": row["url"] = _nvd_week_url() extra["_dynamic"] = marker row["extra"] = extra return Source(**row), marker # контент-дубли уже имеющихся в реестре фидов (url отличается, лента та же) CONTENT_DUP = {"elastic_security_labs", "securityweek_alt", "reddit_netsec"} def _extra_lit(extra): """extra-литерал без служебных ключей: _dynamic — метаданные прогона, не источника.""" clean = {k: v for k, v in sorted((extra or {}).items()) if k != "_dynamic"} return "" if not clean else ", extra=%r" % (clean,) def _frozen_window(rows): """Строки, где в url уже стоит вычисленное окно (lastMod), но маркера нет: такой отчёт старее фикса dynamic — при emit окно застынет на дате прогона.""" return [str(r.get("name")) for r in rows if not r.get("dynamic") and "lastmodstartdate=" in str(r.get("url") or "").lower()] def emit(rows, dead, path, dedupe=True): """omni/sources_fresh.py: только живые, без импортов из omni.sources. Литералы пишем через repr: json.dumps даёт true/false/null — это не Python. Дедуп сверяет и url, и имя (нормализованные) и добавляет принятые строки в виденное, поэтому партия уникальна и сама по себе. Динамический url поддерживается только для известного маркера: неизвестный — явная ошибка, а не замороженный url, который молча протухнет. """ rows = list(rows) unknown = sorted({str(r.get("dynamic")) for r in rows if r.get("dynamic") and r.get("dynamic") not in KNOWN_DYNAMIC}) if unknown: raise ValueError("emit: неизвестный dynamic-маркер %s — замораживать url нельзя" % ", ".join(map(repr, unknown))) frozen = _frozen_window(rows) if frozen: print("emit: ВНИМАНИЕ, окно в url без dynamic-маркера (%s): отчёт устарел, " "перепрогони probe_fresh" % ", ".join(frozen)) if dedupe: have_url, have_name = _registry_keys() seen_url, seen_name = set(have_url), set(have_name) kept, skipped, dupes = [], [], [] for r in rows: n, nm = _norm(r.get("url")), _norm_name(r.get("name")) in_registry = (bool(n) and n in have_url) or (bool(nm) and nm in have_name) if r.get("name") in CONTENT_DUP: in_registry = True if in_registry or (bool(n) and n in seen_url) or (bool(nm) and nm in seen_name): (skipped if in_registry else dupes).append(r) continue if n: seen_url.add(n) if nm: seen_name.add(nm) kept.append(r) print("emit: уже есть в реестре -> %d: %s" % (len(skipped), ", ".join(r["name"] for r in skipped))) print("emit: дубли внутри партии -> %d: %s" % (len(dupes), ", ".join(r["name"] for r in dupes))) rows = kept lines = [ '"""Свежая волна источников omni: security/CVE/leaks/CN/RU/tg.', "", "Сгенерировано tools/probe_fresh.py по факту живого fetch (n>0).", "Отклонённые кандидаты и причины — в хвостовом комментарии.", "Импортов из omni.sources нет намеренно (циклический импорт).", '"""', "", "import datetime as _dt", "", "", "def _nvd_week():", ' """NVD 2.0 принимает обе даты и окно <=120 дней: скользящая неделя."""', " now = _dt.datetime.utcnow()", ' since = (now - _dt.timedelta(days=7)).strftime("%Y-%m-%dT00:00:00.000")', ' return ("https://services.nvd.nist.gov/rest/json/cves/2.0?resultsPerPage=30"', ' "&lastModStartDate=%s&lastModEndDate=%s" % (since, now.strftime("%Y-%m-%dT%H:%M:%S.000")))', "", "", "def fresh(S):", ' """S — конструктор Source; возвращает только проверенные сетью источники."""', " return [", ] for r in rows: marker = str(r.get("dynamic") or "") tags = repr(tuple(r.get("tags") or ())) limit = r.get("limit") lim = ", limit=%d" % limit if isinstance(limit, int) and limit != 20 else "" det = ", detail=True" if r.get("detail") else "" if marker == "NVD_LAST_WEEK": lines.append(" S(%r, %r, %r, _nvd_week(), %r, %s%s%s%s)," % ( r["name"], r.get("category", ""), r.get("kind", ""), r.get("adapter", "rss"), tags, lim, det, _extra_lit(r.get("extra")))) continue lines.append(" S(%r, %r, %r, %r, %r, %s%s%s%s)," % ( r["name"], r.get("category", ""), r.get("kind", ""), r.get("url", ""), r.get("adapter", "rss"), tags, lim, det, _extra_lit(r.get("extra")))) lines += [" ]", "", "", "# --- отклонено прогоном (не отдают записи без ключа/логина) ---"] for r in dead: lines.append("# %-26s %-44s %s" % (r.get("name", "?"), str(r.get("url") or "")[:44], (r.get("err") or "0 items")[:90])) text = "\n".join(lines) + "\n" path.write_text(text, encoding="utf-8") return len(rows) def main(): ap = argparse.ArgumentParser() ap.add_argument("--limit", type=int, default=5) ap.add_argument("--workers", type=int, default=6) ap.add_argument("--only", default="") ap.add_argument("--out", default="data/fresh_probe.json") ap.add_argument("--emit", action="store_true") ap.add_argument("--fresh", action="store_true", help="прогонять только то, что уже в sources_fresh.py") ap.add_argument("--emit-only", action="store_true", help="перегенерировать sources_fresh.py из готового JSON") ap.add_argument("--no-dedupe", action="store_true", help="не отбрасывать источники, уже присутствующие в реестре") a = ap.parse_args() if a.emit_only: rep = json.loads((ROOT / a.out).read_text(encoding="utf-8")) rows = rep["results"] alive = [r for r in rows if r["ok"]] dead = [r for r in rows if not r["ok"]] frozen = _frozen_window(alive) if frozen: print("emit-only: отчёт без dynamic-маркера для %s — окно в url застынет на дате прогона; " "перепрогони probe_fresh (сеть) или правь источники вручную" % ", ".join(frozen)) return 2 n = emit(alive, dead, ROOT / "omni" / "sources_fresh.py", dedupe=not a.no_dedupe) print("omni/sources_fresh.py: %d источников" % n) return 0 cands = [c for c in CANDIDATES if not a.only or a.only in c["name"]] srcs = [] for c in cands: try: src, _marker = candidate_source(c) except Exception as e: # noqa: BLE001 print("skip %s: %s" % (c.get("name"), e)) continue srcs.append(src) t0 = time.time() with ThreadPoolExecutor(max_workers=a.workers) as ex: rows = list(ex.map(lambda s: probe_one(s, a.limit), srcs)) alive = [r for r in rows if r["ok"]] dead = [r for r in rows if not r["ok"]] cats = {} for r in alive: cats[r["category"]] = cats.get(r["category"], 0) + 1 out = {"ts": time.strftime("%Y-%m-%dT%H:%M:%S"), "limit": a.limit, "workers": a.workers, "total": len(rows), "alive": len(alive), "dead": len(dead), "seconds": round(time.time() - t0, 1), "by_category": dict(sorted(cats.items(), key=lambda kv: -kv[1])), "results": sorted(rows, key=lambda r: (not r["ok"], r["category"], r["name"]))} p = ROOT / a.out p.parent.mkdir(parents=True, exist_ok=True) p.write_text(json.dumps(out, ensure_ascii=False, indent=1), encoding="utf-8") print("всего %d | живых %d | мёртвых %d | %.1fs" % (len(rows), len(alive), len(dead), time.time() - t0)) print("по категориям:", out["by_category"]) for r in dead: print(" dead %-24s %s" % (r["name"], r["err"][:110])) if a.emit: n = emit(alive, dead, ROOT / "omni" / "sources_fresh.py", dedupe=not a.no_dedupe) print("omni/sources_fresh.py: %d источников" % n) return 0 if __name__ == "__main__": raise SystemExit(main())