362 lines
14 KiB
Python
362 lines
14 KiB
Python
"""Веб-панель omni: статика + JSON API поверх той же SQLite.
|
||
|
||
Запуск: python -m omni serve --port 8077
|
||
"""
|
||
import json
|
||
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])
|
||
|
||
|
||
def _job(jid, **kw):
|
||
with _LOCK:
|
||
cur = JOBS.get(jid) or {}
|
||
cur.update(kw)
|
||
JOBS[jid] = cur
|
||
|
||
|
||
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="")
|
||
|
||
|
||
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"])
|
||
return self._json({
|
||
"items": s["items"], "priced": s["priced"], "runs": s["runs"], "runs_ok": s["runs_ok"],
|
||
"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/runs":
|
||
try:
|
||
limit = max(1, min(500, int((q.get("limit") or ["80"])[0])))
|
||
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 = max(1, min(200, int((q.get("limit") or ["40"])[0])))
|
||
except ValueError:
|
||
lim = 40
|
||
return self._json(_ab.scan(limit=lim, tag=tag))
|
||
|
||
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 = min(int((q.get("limit") or ["40"])[0]), 200)
|
||
offset = int((q.get("offset") or ["0"])[0])
|
||
sort = (q.get("sort") or ["score"])[0]
|
||
rows, total = store.browse2(q=query, names=_names_in_category(cat), limit=limit, offset=offset, sort=sort)
|
||
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.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)
|
||
n = int(self.headers.get("Content-Length") or 0)
|
||
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 = max(5, min(120, int((q.get("hosts") or ["40"])[0])))
|
||
add = max(1, min(60, int((q.get("add") or ["25"])[0])))
|
||
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())})")
|
||
srv.serve_forever() |