312 lines
15 KiB
Python
312 lines
15 KiB
Python
"""Живой прогон кандидатов из 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()) |