"""Веб-панель 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/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/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()