"""Веб-панель omni: статика + JSON API поверх той же SQLite. Запуск: python -m omni serve --port 8077 """ import json import pathlib import threading 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() def _job(jid, **kw): with _LOCK: cur = JOBS.get(jid) or {} cur.update(kw) JOBS[jid] = cur def _run_feed(job_id, category, limit, tags): picked = sources.pick(category, None, tags) 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), report=sorted(report, key=lambda x: -x[1])[:40]) except Exception as e: _job(job_id, state="error", error=f"{type(e).__name__}: {e}"[:200]) def _check_one(src, limit=5): note = "" 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.1" 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() fl = source_state.flags() all_s = sources.all_sources() 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"]], "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)), "categories": sources.stats(), }) if path == "/api/sources": cat = (q.get("category") or [None])[0] query = (q.get("q") or [""])[0].strip().lower() fl = source_state.flags() custom_names = {d["name"] for d in source_state.custom()} items = [] for s in sources.all_sources(): f = fl.get(s.name, {}) 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 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"), "custom": s.name in custom_names, }) 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": sources.stats(), "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 ["60"])[0]), 300) offset = int((q.get("offset") or ["0"])[0]) rows = store.browse(category=cat, q=query, limit=limit, offset=offset) return self._json({"count": len(rows), "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 = f"feed{len(JOBS)}" _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/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 = None if body.get("all") else body.get("category") jid = f"check{len(JOBS)}" _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()