2d7d509434
- omni/config.py: tg_cfg() читает data/tg.json (вне git), env приоритетнее
- omni/tgsess.py: enabled()/fetch() подтягивают креды из config
- tools/tg_pick.py: перебор пула сессий, проверка is_user_authorized, личные аккаунты исключены
- omni claw pump: 2 канала -> 27 постов через сессию (проверено)
- omni/serve.py: /api/stats += keys{totals}, tg{enabled}
- web/index.html: строка статуса в шапке (ключей / сервисов / TG сессия)
- omni/cli.py radar --apply: пишет полный набор, а не только новые (повторный прогон обнулял файл)
- omni/cli.py radar --import-posts: --since-days + скип известных url до simhash
474 lines
19 KiB
Python
474 lines
19 KiB
Python
"""Веб-панель omni: статика + JSON API поверх той же SQLite.
|
||
|
||
Запуск: python -m omni serve --port 8077
|
||
"""
|
||
import json
|
||
import os
|
||
import pathlib
|
||
import re
|
||
import threading
|
||
import time
|
||
import urllib.parse
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
|
||
from . import feeds, net, sources, source_state, store
|
||
|
||
WEB = pathlib.Path(__file__).resolve().parent.parent / "web"
|
||
JOBS = {}
|
||
_LOCK = threading.Lock()
|
||
SEQ = [0]
|
||
|
||
|
||
def _jid(prefix):
|
||
with _LOCK:
|
||
SEQ[0] += 1
|
||
return "%s%d" % (prefix, SEQ[0])
|
||
|
||
|
||
UI_TOKEN = os.environ.get("OMNI_UI_TOKEN", "")
|
||
|
||
|
||
def _ui_auth(req):
|
||
"""Панель наружу: без токена мутации открыты любому, кто дошёл до порта."""
|
||
if not UI_TOKEN:
|
||
return True
|
||
return req.headers.get("X-Omni-Token") == UI_TOKEN
|
||
|
||
|
||
def _intv(q, key, default, lo, hi):
|
||
"""int из query без падений: ?limit=abc раньше давал трейс и обрыв соединения,
|
||
?limit=-1 отдавал всю базу (SQLite LIMIT -1 = без лимита)."""
|
||
try:
|
||
v = int((q.get(key) or [str(default)])[0])
|
||
except (TypeError, ValueError):
|
||
return default
|
||
return max(lo, min(hi, v))
|
||
|
||
|
||
def _job(jid, **kw):
|
||
with _LOCK:
|
||
cur = JOBS.get(jid) or {}
|
||
cur.update(kw)
|
||
cur["_ts"] = time.time()
|
||
JOBS[jid] = cur
|
||
if len(JOBS) > 400: # раньше росло без границ (500 запусков -> 500 записей)
|
||
dead = sorted(JOBS.items(), key=lambda kv: kv[1].get("_ts", 0))[:len(JOBS) - 300]
|
||
for k, _v in dead:
|
||
JOBS.pop(k, None)
|
||
|
||
|
||
def _cat_health():
|
||
"""Сводка по категориям: всего/включено/живых/мёртвых/непроверенных."""
|
||
fl = source_state.flags()
|
||
out = {}
|
||
for s in sources.all_sources():
|
||
f = fl.get(s.name, {})
|
||
c = out.setdefault(s.category, {"name": s.category, "total": 0, "enabled": 0,
|
||
"alive": 0, "dead": 0, "untested": 0})
|
||
c["total"] += 1
|
||
if f.get("enabled", True):
|
||
c["enabled"] += 1
|
||
if not f.get("checked"):
|
||
c["untested"] += 1
|
||
elif (f.get("items") or 0) > 0:
|
||
c["alive"] += 1
|
||
else:
|
||
c["dead"] += 1
|
||
return sorted(out.values(), key=lambda x: -x["total"])
|
||
|
||
|
||
|
||
def _clean_txt(v, n=0):
|
||
"""Снимает html-теги/энтити из title/text, схлопывает пробелы."""
|
||
import html as _h
|
||
if not v:
|
||
return ""
|
||
t = str(v)
|
||
t = re.sub(r"<[^>]{1,200}>", " ", t)
|
||
for _ in range(2):
|
||
t = _h.unescape(t)
|
||
t = re.sub(r"\s+", " ", t).strip()
|
||
return t[:n].rstrip() if n else t
|
||
|
||
|
||
def _fmt_date(v):
|
||
"""unix-таймстамп -> YYYY-MM-DD, rfc822 оставляет как есть."""
|
||
if not v:
|
||
return ""
|
||
s = str(v).strip()
|
||
if s.isdigit() and len(s) >= 9:
|
||
import datetime as _dt
|
||
try:
|
||
return _dt.datetime.utcfromtimestamp(int(s[:10])).strftime("%Y-%m-%d")
|
||
except Exception:
|
||
return s
|
||
m = re.search(r"(\d{1,2})\s+([A-Za-z]{3})\s+(\d{4})", s)
|
||
if m:
|
||
return "%s %s %s" % (m.group(1), m.group(2), m.group(3))
|
||
return s[:19]
|
||
|
||
|
||
def _names_in_category(cat):
|
||
if not cat or cat in ("all", ""):
|
||
return None
|
||
return [s.name for s in sources.all_sources() if s.category == cat]
|
||
|
||
|
||
def _run_feed(job_id, category, limit, tags):
|
||
picked = sources.pick(category, None, tags)
|
||
_job(job_id, state="running", total=len(picked), done=0)
|
||
try:
|
||
items, report = feeds.fetch_many(picked, limit=limit, workers=8)
|
||
new, dup = store.add(items)
|
||
_job(job_id, state="done", sources=len(picked), fetched=len(items), new=new, dup=dup,
|
||
empty=sum(1 for _, c in report if not c), done=len(picked),
|
||
report=sorted(report, key=lambda x: -x[1])[:60])
|
||
except Exception as e:
|
||
_job(job_id, state="error", error=f"{type(e).__name__}: {e}"[:200])
|
||
|
||
|
||
def _check_one(src, limit=5):
|
||
note = ""
|
||
t0 = time.time()
|
||
try:
|
||
items = feeds.fetch(src, limit)
|
||
except Exception as e:
|
||
items, note = [], type(e).__name__
|
||
if items:
|
||
return source_state.record_check(src.name, 200, len(items), "ok")
|
||
code = 0
|
||
try:
|
||
r = net.get(src.url, timeout=20)
|
||
code = getattr(r, "status", 0) or 0
|
||
note = note or ("пусто" if code == 200 else f"http {code}")
|
||
except Exception as e:
|
||
note = note or type(e).__name__
|
||
return source_state.record_check(src.name, code, 0, note or "нет данных")
|
||
|
||
|
||
def _run_check(job_id, names=None, category=None):
|
||
picked = sources.pick(category, ",".join(names) if names else None, None)
|
||
total = len(picked)
|
||
_job(job_id, state="running", total=total, done=0, alive=0, dead=0, current="")
|
||
alive = dead = 0
|
||
with ThreadPoolExecutor(max_workers=6) as ex:
|
||
futures = {ex.submit(_check_one, s): s for s in picked}
|
||
for i, f in enumerate(futures, 1):
|
||
s = futures[f]
|
||
try:
|
||
res = f.result()
|
||
except Exception:
|
||
res = {}
|
||
if (res.get("items") or 0) > 0:
|
||
alive += 1
|
||
else:
|
||
dead += 1
|
||
_job(job_id, done=i, alive=alive, dead=dead, current=s.name)
|
||
_job(job_id, state="done", done=total, alive=alive, dead=dead, current="")
|
||
|
||
|
||
def _ai_summary(tag=None, limit=14):
|
||
"""ИИ-саммари абузов для панели. None если ИИ не настроен."""
|
||
from . import abuse as abuse_mod, ai as ai_mod
|
||
if not ai_mod.available():
|
||
return {"ok": False, "error": "ai_not_configured"}
|
||
d = abuse_mod.scan(limit=limit, tag=tag, window=6000)
|
||
items = [i for i in d.get("items", []) if [t for t in (i.get("tags") or []) if t != "keys"]]
|
||
if not items:
|
||
return {"ok": True, "summary": "", "n": 0}
|
||
posts = ["%s — %s" % (i.get("title") or "", i.get("source") or "") for i in items]
|
||
txt = ai_mod.summarize(posts, tag=tag)
|
||
return {"ok": bool(txt), "summary": txt or "", "n": len(items)}
|
||
|
||
|
||
class Handler(BaseHTTPRequestHandler):
|
||
server_version = "omni/1.2"
|
||
|
||
def log_message(self, *a):
|
||
pass
|
||
|
||
def _send(self, code, body, ctype="application/json; charset=utf-8"):
|
||
data = body if isinstance(body, bytes) else str(body).encode("utf-8")
|
||
self.send_response(code)
|
||
self.send_header("Content-Type", ctype)
|
||
self.send_header("Content-Length", str(len(data)))
|
||
self.send_header("Cache-Control", "no-store")
|
||
self.end_headers()
|
||
self.wfile.write(data)
|
||
|
||
def _json(self, obj, code=200):
|
||
self._send(code, json.dumps(obj, ensure_ascii=False))
|
||
|
||
def do_GET(self):
|
||
u = urllib.parse.urlsplit(self.path)
|
||
q = urllib.parse.parse_qs(u.query)
|
||
path = u.path
|
||
|
||
if path in ("/", "/index.html"):
|
||
f = WEB / "index.html"
|
||
if not f.exists():
|
||
return self._send(404, "no ui", "text/plain")
|
||
return self._send(200, f.read_bytes(), "text/html; charset=utf-8")
|
||
|
||
if path == "/api/stats":
|
||
s = store.stats()
|
||
per_source = store.count_by_source()
|
||
fl = source_state.flags()
|
||
all_s = sources.all_sources()
|
||
cats = _cat_health()
|
||
for c in cats:
|
||
c["stored"] = sum(per_source.get(x.name, 0) for x in all_s if x.category == c["name"])
|
||
kv = {"services": 0, "keys": 0, "ready": 0}
|
||
tgst = {"enabled": False}
|
||
try:
|
||
from . import vault as _vault, tgsess as _tg
|
||
st_kv = _vault.stats()
|
||
kv = {"services": len(st_kv), "keys": sum(x["keys"] for x in st_kv),
|
||
"ready": sum(x["ready"] for x in st_kv)}
|
||
tgst = {"enabled": bool(_tg.enabled())}
|
||
except Exception:
|
||
pass
|
||
return self._json({
|
||
"items": s["items"], "priced": s["priced"], "runs": s["runs"], "runs_ok": s["runs_ok"],
|
||
"keys": kv, "tg": tgst,
|
||
"by_provider": [{"name": n, "count": c} for n, c in s["by_provider"][:12]],
|
||
"by_kind": [{"name": n, "count": c} for n, c in s["by_kind"]],
|
||
"sources": len(all_s),
|
||
"sources_enabled": sum(1 for x in all_s if fl.get(x.name, {}).get("enabled", True)),
|
||
"alive": sum(1 for x in all_s if (fl.get(x.name, {}).get("items") or 0) > 0),
|
||
"tested": sum(1 for x in all_s if fl.get(x.name, {}).get("checked")),
|
||
"categories": cats,
|
||
})
|
||
|
||
if path == "/api/types":
|
||
from . import taxonomy
|
||
cnt = store.type_counts()
|
||
items = [{"key": k, "label": taxonomy.LABELS.get(k, k), "n": v} for k, v in cnt.items()]
|
||
return self._json({"ok": True, "types": items, "total": sum(cnt.values())})
|
||
|
||
if path == "/api/keys":
|
||
from . import vault as _v
|
||
rows = _v.stats()
|
||
if q.get("service"):
|
||
rows = [x for x in rows if x["service"] == q["service"][0]]
|
||
return self._json({"ok": True, "services": rows,
|
||
"total_keys": sum(x["keys"] for x in rows),
|
||
"total_ready": sum(x["ready"] for x in rows)})
|
||
|
||
if path == "/api/runs":
|
||
try:
|
||
limit = _intv(q, "limit", 80, 1, 500)
|
||
except ValueError:
|
||
limit = 80
|
||
import time as _t
|
||
con = store._con() if hasattr(store, "_con") else None
|
||
rows = []
|
||
try:
|
||
import sqlite3 as _s3
|
||
from .config import DB as _DB
|
||
db = _s3.connect(str(_DB), timeout=30)
|
||
db.row_factory = _s3.Row
|
||
for r in db.execute("SELECT id, ts, provider, source, query, got, kept, status, bytes "
|
||
"FROM runs ORDER BY id DESC LIMIT ?", (limit,)):
|
||
d = dict(r)
|
||
ts = d.get("ts") or 0
|
||
d["when"] = _t.strftime("%Y-%m-%d %H:%M", _t.localtime(ts)) if ts else ""
|
||
rows.append(d)
|
||
total = db.execute("SELECT count(*) FROM runs").fetchone()[0]
|
||
db.close()
|
||
except Exception as e:
|
||
return self._json({"error": "runs: %s" % e})
|
||
return self._json({"items": rows, "total": total, "shown": len(rows)})
|
||
|
||
if path == "/api/abuse":
|
||
from . import abuse as _ab
|
||
tag = (q.get("tag") or [""])[0].strip() or None
|
||
if tag in ("all", "все"):
|
||
tag = None
|
||
try:
|
||
lim = _intv(q, "limit", 40, 1, 200)
|
||
except ValueError:
|
||
lim = 40
|
||
return self._json(_ab.scan(limit=lim, tag=tag))
|
||
|
||
if path == "/api/summary":
|
||
tag = (q.get("tag") or [""])[0].strip() or None
|
||
if tag in ("all", "все"):
|
||
tag = None
|
||
try:
|
||
lim = _intv(q, "limit", 14, 1, 40)
|
||
except ValueError:
|
||
lim = 14
|
||
try:
|
||
return self._json(_ai_summary(tag=tag, limit=lim))
|
||
except Exception as e:
|
||
return self._json({"ok": False, "error": str(e)[:200]})
|
||
|
||
if path == "/api/sources":
|
||
cat = (q.get("category") or [None])[0]
|
||
query = (q.get("q") or [""])[0].strip().lower()
|
||
state = (q.get("state") or [""])[0]
|
||
fl = source_state.flags()
|
||
per_source = store.count_by_source()
|
||
custom_names = {d["name"] for d in source_state.custom()}
|
||
items = []
|
||
for s in sources.all_sources():
|
||
f = fl.get(s.name, {})
|
||
alive = (f.get("items") or 0) > 0
|
||
if cat and cat not in ("all", "") and s.category != cat:
|
||
continue
|
||
if query and query not in s.name.lower() and query not in s.url.lower():
|
||
continue
|
||
if state == "alive" and not alive:
|
||
continue
|
||
if state == "dead" and not (f.get("checked") and not alive):
|
||
continue
|
||
if state == "untested" and f.get("checked"):
|
||
continue
|
||
if state == "off" and f.get("enabled", True):
|
||
continue
|
||
items.append({
|
||
"name": s.name, "category": s.category, "kind": s.kind, "adapter": s.adapter,
|
||
"url": s.url, "tags": list(s.tags),
|
||
"enabled": f.get("enabled", True), "status": f.get("status"),
|
||
"items": f.get("items"), "note": f.get("note"), "checked": f.get("checked"),
|
||
"stored": per_source.get(s.name, 0), "custom": s.name in custom_names,
|
||
})
|
||
items.sort(key=lambda x: (-(x["stored"] or 0), -(x["items"] or 0), x["name"]))
|
||
return self._json({
|
||
"total": len(sources.all_sources()), "shown": len(items),
|
||
"enabled": sum(1 for x in items if x["enabled"]),
|
||
"alive": sum(1 for x in items if (x["items"] or 0) > 0),
|
||
"dead": sum(1 for x in items if x["checked"] and not (x["items"] or 0)),
|
||
"untested": sum(1 for x in items if not x["checked"]),
|
||
"categories": _cat_health(), "items": items,
|
||
})
|
||
|
||
if path == "/api/items":
|
||
cat = (q.get("category") or [None])[0]
|
||
query = (q.get("q") or [""])[0].strip()
|
||
limit = _intv(q, "limit", 40, 1, 200)
|
||
offset = _intv(q, "offset", 0, 0, 1_000_000)
|
||
sort = (q.get("sort") or ["score"])[0]
|
||
priced = (q.get("priced") or ["0"])[0] in ("1", "true", "on")
|
||
typ = (q.get("type") or ["all"])[0]
|
||
rows, total = store.browse2(q=query, names=_names_in_category(cat), limit=limit, offset=offset,
|
||
sort=sort, priced=priced, typ=typ)
|
||
for _r in rows:
|
||
_r["title"] = _clean_txt(_r.get("title"), 180)
|
||
_r["text"] = _clean_txt(_r.get("text"), 260)
|
||
_r["published"] = _fmt_date(_r.get("published"))
|
||
return self._json({"count": len(rows), "total": total, "offset": offset, "items": rows})
|
||
|
||
if path == "/api/export.csv":
|
||
import csv as _csv
|
||
cat = (q.get("category") or [None])[0]
|
||
query = (q.get("q") or [""])[0].strip()
|
||
priced = (q.get("priced") or ["0"])[0] in ("1", "true", "on")
|
||
sort = (q.get("sort") or ["score"])[0]
|
||
typ = (q.get("type") or ["all"])[0]
|
||
rows, total = store.browse2(q=query, names=_names_in_category(cat), limit=5000, offset=0,
|
||
sort=sort, priced=priced, typ=typ)
|
||
import io as _io
|
||
buf = _io.StringIO()
|
||
w = _csv.writer(buf, delimiter=";")
|
||
w.writerow(["source", "kind", "types", "title", "author", "published", "score", "price", "currency", "url"])
|
||
for r in rows:
|
||
w.writerow([r.get("source"), r.get("kind"), r.get("types"), (r.get("title") or "").replace(chr(10), " "),
|
||
r.get("author"), r.get("published"), r.get("score"), r.get("price"),
|
||
r.get("currency"), r.get("url")])
|
||
data = buf.getvalue().encode("utf-8-sig")
|
||
self.send_response(200)
|
||
self.send_header("Content-Type", "text/csv; charset=utf-8")
|
||
self.send_header("Content-Disposition", 'attachment; filename="omni.csv"')
|
||
self.send_header("Content-Length", str(len(data)))
|
||
self.end_headers()
|
||
self.wfile.write(data)
|
||
return
|
||
|
||
if path.startswith("/api/job/"):
|
||
jid = path.rsplit("/", 1)[-1]
|
||
return self._json(JOBS.get(jid) or {"state": "unknown"})
|
||
|
||
return self._send(404, "not found", "text/plain")
|
||
|
||
def do_POST(self):
|
||
u = urllib.parse.urlsplit(self.path)
|
||
if not _ui_auth(self):
|
||
return self._json({"error": "unauthorized"}, 401)
|
||
n = min(int(self.headers.get("Content-Length") or 0), 2_000_000)
|
||
body = {}
|
||
if n:
|
||
try:
|
||
body = json.loads(self.rfile.read(n) or b"{}")
|
||
except Exception:
|
||
body = {}
|
||
|
||
if u.path == "/api/discover":
|
||
from . import discover as _d
|
||
q = urllib.parse.parse_qs(u.query)
|
||
hosts = _intv(q, "hosts", 40, 5, 120)
|
||
add = _intv(q, "add", 25, 1, 60)
|
||
dry = (q.get("dry") or ["0"])[0] not in ("0", "false", "")
|
||
res = _d.discover(limit_hosts=hosts, workers=8, max_add=add, dry=dry)
|
||
return self._json(res)
|
||
|
||
if u.path == "/api/feed":
|
||
jid = _jid("feed")
|
||
_job(jid, state="running")
|
||
threading.Thread(target=_run_feed,
|
||
args=(jid, body.get("category"), int(body.get("limit") or 12), body.get("tags")),
|
||
daemon=True).start()
|
||
return self._json({"job": jid})
|
||
|
||
if u.path == "/api/sources/toggle":
|
||
name = body.get("name")
|
||
if not name:
|
||
return self._json({"error": "name required"}, 400)
|
||
return self._json(source_state.set_enabled(name, bool(body.get("enabled", True))))
|
||
|
||
if u.path == "/api/sources/toggle_all":
|
||
on = bool(body.get("enabled", True))
|
||
cat = body.get("category")
|
||
changed = 0
|
||
for s in sources.all_sources():
|
||
if cat and cat not in ("all", "") and s.category != cat:
|
||
continue
|
||
source_state.set_enabled(s.name, on)
|
||
changed += 1
|
||
return self._json({"ok": True, "changed": changed, "enabled": on})
|
||
|
||
if u.path == "/api/sources/add":
|
||
try:
|
||
res = source_state.add_custom(body.get("name"), body.get("category"), body.get("url"),
|
||
body.get("kind"), body.get("adapter"),
|
||
body.get("tags") or (), int(body.get("limit") or 20),
|
||
bool(body.get("detail")))
|
||
except Exception as e:
|
||
return self._json({"error": str(e)}, 400)
|
||
return self._json(res)
|
||
|
||
if u.path == "/api/sources/remove":
|
||
name = body.get("name")
|
||
if not name:
|
||
return self._json({"error": "name required"}, 400)
|
||
return self._json(source_state.remove_custom(name))
|
||
|
||
if u.path == "/api/sources/check":
|
||
names = body.get("names") or ([body["name"]] if body.get("name") else None)
|
||
category = body.get("category")
|
||
jid = _jid("check")
|
||
_job(jid, state="running")
|
||
threading.Thread(target=_run_check, args=(jid, names, category), daemon=True).start()
|
||
return self._json({"job": jid})
|
||
|
||
return self._json({"error": "not found"}, 404)
|
||
|
||
|
||
def serve(port=8077):
|
||
srv = ThreadingHTTPServer(("127.0.0.1", port), Handler)
|
||
print(f"omni ui: http://127.0.0.1:{port} (источников {len(sources.all_sources())})")
|
||
if not UI_TOKEN:
|
||
print("внимание: OMNI_UI_TOKEN не задан — POST /api/* без авторизации; наружу только за nginx с basic-auth")
|
||
srv.serve_forever() |