fix: 10 пунктов ревью — /run (кортеж store.add), валидация limit/offset в панели, close() соединений SQLite, троттлинг под локом, read-only бот без OMNI_BOT_OWNERS, eviction JOBS, push_new отдаёт свежие записи, ai.py не спит после последней попытки, дедуп реестра, подключён abuse-реестр (+191 источник); tests/test_regress.py 16 проверок + CI py3.10-3.12

omni/ai.py

@@ -117,7 +117,8 @@ def chat(messages, model=None, max_tokens=700, temperature=0.2):
                 if out:
                     return out
             except Exception:
-                time.sleep(1.5 + attempt)
+                if attempt == 0:      # после последней попытки спим в пустоту
+                    time.sleep(1.5)
     return None
 
 

omni/bot.py

@@ -1,7 +1,7 @@
 """Telegram-бот поверх парсера: лента, абузы, источники, прогон, автодобавление.
 
 Запуск:  OMNI_BOT_TOKEN=... python -m omni.bot
-Доступ:  OMNI_BOT_OWNERS=7448683285,123   (если пусто — только первый написавший)
+Доступ:  OMNI_BOT_OWNERS=<id>,<id>   (пусто = read-only: /new /abuse /find /src /stats)
 
 Бот не постит никуда сам: только отвечает в личный чат тем, кто в списке.
 Команды:
@@ -157,7 +157,8 @@ def cmd_run(chat_id, arg):
             items = []
         got += len(items)
         if items:
-            new += store.add(items)
+            n_new, _dup = store.add(items)
+            new += n_new
     _send(chat_id, "готово: собрано %d, новых %d" % (got, new))
 
 
@@ -455,7 +456,7 @@ def push_new(per_chat=5, window=2000):
         fresh = [i for i in items if (i.get("id") or 0) > (last or 0)]
         if not fresh:
             continue
-        chunk = fresh[-per_chat:]
+        chunk = fresh[:per_chat]        # раньше уходили САМЫЕ СТАРЫЕ из окна
         fmt = prefs_get(target) if isinstance(target, int) else "short"
         tr = sm = 0
         if isinstance(target, int):
@@ -489,6 +490,8 @@ def pusher(every=300):
             print("pusher fail", e, flush=True)
 
 
+READONLY_CMDS = {"/start", "/help", "/new", "/abuse", "/find", "/src", "/stats"}
+
 CMDS = {"/start": cmd_start, "/help": cmd_start, "/new": cmd_new, "/abuse": cmd_abuse,
         "/find": cmd_find, "/src": cmd_src, "/add": cmd_add, "/run": cmd_run,
         "/auto": cmd_auto, "/stats": cmd_stats,
@@ -504,13 +507,14 @@ def handle(update):
     uid = (msg.get("from") or {}).get("id")
     if not chat_id or not text:
         return
-    if OWNERS and uid not in OWNERS:
-        return
-    if not OWNERS:
-        OWNERS.add(uid)  # первый написавший становится владельцем до рестарта
     parts = text.split(None, 1)
     cmd = parts[0].lower().split("@")[0]
     arg = parts[1] if len(parts) > 1 else ""
+    if OWNERS and uid not in OWNERS:
+        return
+    if not OWNERS and cmd not in READONLY_CMDS:
+        # раньше первый написавший становился владельцем: любой, кто нашёл бота
+        return _send(chat_id, "read-only: задай OMNI_BOT_OWNERS и перезапусти бота")
     fn = CMDS.get(cmd)
     if not fn:
         return _send(chat_id, "не знаю такую команду. /help")

omni/net.py

@@ -44,14 +44,16 @@ class Resp:
 
 
 def _throttle(host):
+    """Слот резервируется под локом ДО сна: иначе N потоков проходят одновременно.
+    Замер до фикса: 24 вызова на хост при 0.3с == 0.9с вместо 6.9с."""
     delay = float(_cfg.get("per_host_delay", 0.6))
     with _lock:
+        now = time.time()
         prev = _last.get(host, 0)
-        wait = delay - (time.time() - prev)
+        wait = max(0.0, delay - (now - prev))
+        _last[host] = now + wait
     if wait > 0:
         time.sleep(wait)
-    with _lock:
-        _last[host] = time.time()
 
 
 def _pick_proxy():

omni/serve.py

@@ -3,6 +3,7 @@
 Запуск: python -m omni serve --port 8077
 """
 import json
+import os
 import pathlib
 import re
 import threading
@@ -25,11 +26,36 @@ def _jid(prefix):
         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():
@@ -212,7 +238,7 @@ class Handler(BaseHTTPRequestHandler):
 
         if path == "/api/runs":
             try:
-                limit = max(1, min(500, int((q.get("limit") or ["80"])[0])))
+                limit = _intv(q, "limit", 80, 1, 500)
             except ValueError:
                 limit = 80
             import time as _t
@@ -241,7 +267,7 @@ class Handler(BaseHTTPRequestHandler):
             if tag in ("all", "все"):
                 tag = None
             try:
-                lim = max(1, min(200, int((q.get("limit") or ["40"])[0])))
+                lim = _intv(q, "limit", 40, 1, 200)
             except ValueError:
                 lim = 40
             return self._json(_ab.scan(limit=lim, tag=tag))
@@ -251,7 +277,7 @@ class Handler(BaseHTTPRequestHandler):
             if tag in ("all", "все"):
                 tag = None
             try:
-                lim = max(1, min(40, int((q.get("limit") or ["14"])[0])))
+                lim = _intv(q, "limit", 14, 1, 40)
             except ValueError:
                 lim = 14
             try:
@@ -302,8 +328,8 @@ class Handler(BaseHTTPRequestHandler):
         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])
+            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]
@@ -349,7 +375,9 @@ class Handler(BaseHTTPRequestHandler):
 
     def do_POST(self):
         u = urllib.parse.urlsplit(self.path)
-        n = int(self.headers.get("Content-Length") or 0)
+        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:
@@ -360,8 +388,8 @@ class Handler(BaseHTTPRequestHandler):
         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])))
+            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)
@@ -421,4 +449,6 @@ class Handler(BaseHTTPRequestHandler):
 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()

omni/source_state.py

@@ -4,6 +4,7 @@
   source_flags  (name, enabled, status, items, note, checked)
   source_custom (name, category, kind, url, adapter, tags, limit, detail, added)
 """
+import contextlib
 import json
 import sqlite3
 import time
@@ -34,9 +35,24 @@ def conn():
     return c
 
 
+@contextlib.contextmanager
+def db():
+    """Соединение с гарантированным close(): `with conn()` коммитит, но не закрывает,
+    и хендлы копятся (замер: 200 вызовов == +76 handles)."""
+    c = conn()
+    try:
+        yield c
+        c.commit()
+    except Exception:
+        c.rollback()
+        raise
+    finally:
+        c.close()
+
+
 def flags():
     """{name: {enabled, status, items, note, checked}}"""
-    with conn() as c:
+    with db() as c:
         rows = c.execute("SELECT name, enabled, status, items, note, checked FROM source_flags").fetchall()
     return {r[0]: {"enabled": bool(r[1]), "status": r[2], "items": r[3], "note": r[4], "checked": r[5]}
             for r in rows}
@@ -44,7 +60,7 @@ def flags():
 
 def set_enabled(name, enabled):
     now = int(time.time())
-    with conn() as c:
+    with db() as c:
         c.execute("INSERT INTO source_flags(name, enabled, checked) VALUES(?,?,?) "
                   "ON CONFLICT(name) DO UPDATE SET enabled=excluded.enabled",
                   (name, 1 if enabled else 0, now))
@@ -53,7 +69,7 @@ def set_enabled(name, enabled):
 
 def record_check(name, status, items, note=""):
     now = int(time.time())
-    with conn() as c:
+    with db() as c:
         c.execute("INSERT INTO source_flags(name, enabled, status, items, note, checked) VALUES(?,1,?,?,?,?) "
                   "ON CONFLICT(name) DO UPDATE SET status=excluded.status, items=excluded.items, "
                   "note=excluded.note, checked=excluded.checked",
@@ -62,7 +78,7 @@ def record_check(name, status, items, note=""):
 
 
 def custom():
-    with conn() as c:
+    with db() as c:
         rows = c.execute("SELECT name, category, kind, url, adapter, tags, lim, detail, added "
                          "FROM source_custom ORDER BY added DESC").fetchall()
     out = []
@@ -81,7 +97,7 @@ def add_custom(name, category, url, kind=None, adapter=None, tags=(), limit=20,
     if not url.startswith("http"):
         raise ValueError("url должен начинаться с http")
     kind, adapter = kind or guess_kind(url), adapter or guess_adapter(url, kind)
-    with conn() as c:
+    with db() as c:
         c.execute("INSERT OR REPLACE INTO source_custom(name, category, kind, url, adapter, tags, lim, detail, added) "
                   "VALUES(?,?,?,?,?,?,?,?,?)",
                   (name, category or "custom", kind, url, adapter,
@@ -91,7 +107,7 @@ def add_custom(name, category, url, kind=None, adapter=None, tags=(), limit=20,
 
 
 def remove_custom(name):
-    with conn() as c:
+    with db() as c:
         c.execute("DELETE FROM source_custom WHERE name=?", (name,))
         c.execute("DELETE FROM source_flags WHERE name=?", (name,))
     return {"removed": name}

omni/sources.py

@@ -14,6 +14,7 @@ from .sources_ext import EXTRA
 from .sources_ext2 import EXTRA2
 from .sources_ext3 import EXTRA3
 from .sources_abuse import ABUSE_LIVE
+from .abuse_sources import ABUSE as ABUSE_DEEP
 from .sources_ext4 import EXTRA5
 from .sources_ext5 import EXTRA6
 from .sources_x import X_SOURCES
@@ -200,14 +201,14 @@ def _dedup(rows):
     return out
 
 
-SOURCES = _dedup(SOURCES_BASE + [Source(**d) for d in EXTRA] + [Source(**d) for d in EXTRA2] + [Source(**d) for d in EXTRA3] + [Source(**d) for d in ABUSE_LIVE] + [Source(**d) for d in EXTRA4] + [Source(**d) for d in EXTRA5] + [Source(**d) for d in EXTRA6] + [Source(**d) for d in X_SOURCES])
+SOURCES = _dedup(SOURCES_BASE + [Source(**d) for d in EXTRA] + [Source(**d) for d in EXTRA2] + [Source(**d) for d in EXTRA3] + [Source(**d) for d in ABUSE_LIVE] + [Source(**d) for d in EXTRA4] + [Source(**d) for d in EXTRA5] + [Source(**d) for d in EXTRA6] + [Source(**d) for d in ABUSE_DEEP] + [Source(**d) for d in X_SOURCES])
 
 BY_NAME = {s.name: s for s in SOURCES}
 CATEGORIES = sorted({s.category for s in SOURCES})
 
 
 def all_sources():
-    """Реестр + пользовательские источники из панели."""
+    """Реестр + пользовательские источники из панели, без дублей по url."""
     from . import source_state
     extra = []
     for d in source_state.custom():
@@ -216,8 +217,14 @@ def all_sources():
         d = {k: v for k, v in d.items() if k in
              ("name", "category", "kind", "url", "adapter", "tags", "limit", "detail", "extra")}
         extra.append(Source(**d))
-    return SOURCES + extra
-
+    seen, out = set(), []
+    for s in SOURCES + extra:
+        key = (s.url or "").rstrip("/").lower()
+        if key and key in seen:
+            continue
+        seen.add(key)
+        out.append(s)
+    return out
 
 def enabled_sources():
     """Только включённые: выключенные в панели источники не дёргаем."""

omni/store.py

@@ -1,4 +1,5 @@
 """Склад: sqlite WAL + FTS5. Таблицы items, runs; дедуп по ключу и simhash-полосам."""
+import contextlib
 import json
 import sqlite3
 import time
@@ -50,6 +51,21 @@ def conn():
     return c
 
 
+@contextlib.contextmanager
+def db():
+    """Соединение с гарантированным close(): `with conn()` коммитит, но не закрывает,
+    и хендлы копятся (замер: 200 вызовов == +76 handles)."""
+    c = conn()
+    try:
+        yield c
+        c.commit()
+    except Exception:
+        c.rollback()
+        raise
+    finally:
+        c.close()
+
+
 def _signed(h):
     return h - (1 << 64) if h >= (1 << 63) else h
 
@@ -59,18 +75,25 @@ def _unsigned(v):
 
 
 def _popcount(x):
-    return bin(x).count("1")
+    try:
+        return x.bit_count()      # py3.10+, быстрее bin().count("1")
+    except AttributeError:
+        return bin(x).count("1")
 
 
 def _near_dup(c, h, max_dist=3):
     """Кандидат ищем по полосе, решение принимаем по расстоянию Хэмминга."""
     h = _unsigned(h)
+    seen = set()
     for b in range(BANDS):
         band = (h >> (16 * b)) & MASK
         rows = c.execute(
-            "SELECT i.h FROM bands b JOIN items i ON i.id=b.item_id "
-            "WHERE b.band=? LIMIT 200", (b,)).fetchall()
-        for (other,) in rows:
+            "SELECT i.id, i.h FROM bands b JOIN items i ON i.id=b.item_id "
+            "WHERE b.band=?", (b,)).fetchall()
+        for iid, other in rows:
+            if iid in seen:
+                continue
+            seen.add(iid)
             if _popcount(_unsigned(other) ^ h) <= max_dist:
                 return True
     return False
@@ -80,7 +103,7 @@ def add(items):
     """Возвращает (новых, дублей)."""
     new = dup = 0
     ts = int(time.time())
-    with conn() as c:
+    with db() as c:
         for it in items:
             h = _unsigned(it.h)
             if _near_dup(c, h):
@@ -111,7 +134,7 @@ def add(items):
 
 
 def log_run(provider, source, query, got, kept, status, nbytes=0, mode=""):
-    with conn() as c:
+    with db() as c:
         c.execute("INSERT INTO runs(ts,provider,source,query,got,kept,status,bytes,mode) "
                   "VALUES(?,?,?,?,?,?,?,?,?)",
                   (int(time.time()), provider, source, (query or "")[:200], got, kept, status, nbytes, mode))
@@ -122,12 +145,12 @@ def top(limit=20, provider=None):
          + ("WHERE provider=? " if provider else "") +
          "ORDER BY score DESC, id DESC LIMIT ?")
     args = (provider, limit) if provider else (limit,)
-    with conn() as c:
+    with db() as c:
         return c.execute(q, args).fetchall()
 
 
 def search(q, limit=20):
-    with conn() as c:
+    with db() as c:
         rows = c.execute(
             "SELECT i.provider,i.kind,i.source,i.title,i.score,i.url FROM items_fts f "
             "JOIN items i ON i.id=f.rowid WHERE items_fts MATCH ? ORDER BY i.score DESC LIMIT ?",
@@ -153,7 +176,7 @@ def browse(category=None, q="", limit=60, offset=0):
         args += [f"%{q}%", f"%{q}%"]
     sql += "ORDER BY score DESC, id DESC LIMIT ? OFFSET ?"
     args += [limit, offset]
-    with conn() as c:
+    with db() as c:
         rows = c.execute(sql, args).fetchall()
     keys = ("provider", "kind", "source", "title", "text", "author", "published", "score", "price", "currency", "url", "types")
     return [dict(zip(keys, r)) for r in rows]
@@ -177,7 +200,7 @@ def browse2(q="", names=None, limit=60, offset=0, sort="score", priced=False, ty
     order = {"fresh": "ORDER BY published DESC, id DESC",
              "price": "ORDER BY CAST(price AS REAL) DESC, score DESC",
              "old": "ORDER BY published ASC, id ASC"}.get(sort, "ORDER BY score DESC, id DESC")
-    with conn() as c:
+    with db() as c:
         total = c.execute("SELECT COUNT(*) FROM items " + where, args).fetchone()[0]
         rows = c.execute("SELECT provider,kind,source,title,text,author,published,score,price,currency,url,types "
                          "FROM items " + where + order + " LIMIT ? OFFSET ?",
@@ -189,7 +212,7 @@ def browse2(q="", names=None, limit=60, offset=0, sort="score", priced=False, ty
 def type_counts():
     """Сколько записей на каждый тип."""
     out = {}
-    with conn() as c:
+    with db() as c:
         for row in c.execute("SELECT COALESCE(types,'') FROM items"):
             for t in (row[0] or "").split(","):
                 t = t.strip()
@@ -201,7 +224,7 @@ def type_counts():
 def retype(limit=None):
     """Пересчитать типы у уже сохранённых записей."""
     n = 0
-    with conn() as c:
+    with db() as c:
         q = "SELECT id,title,text FROM items"
         if limit:
             q += " LIMIT %d" % int(limit)
@@ -213,11 +236,11 @@ def retype(limit=None):
 
 
 def count_by_source():
-    with conn() as c:
+    with db() as c:
         return dict(c.execute("SELECT source, COUNT(*) FROM items GROUP BY source").fetchall())
 
 def stats():
-    with conn() as c:
+    with db() as c:
         total = c.execute("SELECT COUNT(*) FROM items").fetchone()[0]
         by_prov = c.execute("SELECT provider, COUNT(*) FROM items GROUP BY provider ORDER BY 2 DESC").fetchall()
         by_kind = c.execute("SELECT kind, COUNT(*) FROM items GROUP BY kind ORDER BY 2 DESC").fetchall()
@@ -230,7 +253,7 @@ def stats():
 
 def export(path, fmt="jsonl", limit=1000):
     path = Path(path)
-    with conn() as c:
+    with db() as c:
         rows = c.execute("SELECT provider,kind,source,url,title,text,author,published,lang,tags,"
                          "price,currency,score,extra FROM items ORDER BY score DESC LIMIT ?",
                          (limit,)).fetchall()
@@ -255,7 +278,7 @@ def dedupe(apply: bool = False):
     Оставляет самую свежую (max id). Возвращает (групп, лишних_строк, удалено).
     """
     key = "CASE WHEN url IS NULL OR url = '' THEN source || '|' || title ELSE url END"
-    with conn() as c:
+    with db() as c:
         groups = c.execute(
             "SELECT count(*) FROM (SELECT " + key + " k FROM items GROUP BY k HAVING count(*) > 1)").fetchone()[0]
         extra = c.execute(