abuse: схлопывание перепечаток (sig в индексе, uniq в API), кэш выдачи 60с; scan 13с -> 1.5с

omni/abuse.py

@@ -8,6 +8,7 @@ from __future__ import annotations
 
 import re
 import sqlite3
+import time
 
 from .config import DB
 
@@ -184,6 +185,9 @@ def _connect():
     con.execute("PRAGMA busy_timeout=60000")
     con.execute("CREATE TABLE IF NOT EXISTS abuse_idx "
                 "(id INTEGER PRIMARY KEY, score INTEGER, tags TEXT)")
+    cols = {r[1] for r in con.execute("PRAGMA table_info(abuse_idx)")}
+    if "sig" not in cols:
+        con.execute("ALTER TABLE abuse_idx ADD COLUMN sig TEXT")
     return con
 
 
@@ -214,10 +218,10 @@ def rebuild(full=False, chunk=4000, auto=False):
             for r in rows:
                 sc, hits = _qualifies(r[1], r[3], r[4], r[5] or "")
                 if sc:
-                    batch.append((r[0], sc, ",".join(hits)))
+                    batch.append((r[0], sc, ",".join(hits), _sig(r[3], r[4])))
                 last = r[0]
             if batch:
-                con.executemany("INSERT OR REPLACE INTO abuse_idx(id, score, tags) VALUES (?,?,?)", batch)
+                con.executemany("INSERT OR REPLACE INTO abuse_idx(id, score, tags, sig) VALUES (?,?,?,?)", batch)
                 added += len(batch)
             con.commit()
         return added
@@ -231,7 +235,7 @@ def _rows(limit=400, tag=None, q=None, since_days=None):
         rebuild(auto=True)
     except Exception:
         pass
-    sql = ("SELECT i.id, i.source, i.url, i.title, i.text, i.published, a.tags, i.score, a.score "
+    sql = ("SELECT i.id, i.source, i.url, i.title, i.text, i.published, a.tags, i.score, a.score, a.sig "
            "FROM abuse_idx a JOIN items i ON i.id = a.id")
     where, args = [], []
     if tag:
@@ -255,17 +259,49 @@ def _rows(limit=400, tag=None, q=None, since_days=None):
             body = (r[4] or "")
             out.append({"id": r[0], "source": r[1], "url": r[2], "title": r[3],
                         "text": body[:600], "published": r[5],
-                        "tags": hits, "abuse_score": r[8], "rank": r[7],
+                        "tags": hits, "abuse_score": r[8], "rank": r[7], "sig": r[9],
                         "find": extract((r[3] or "") + " " + body)})
         return out
     finally:
         con.close()
 
 
-def scan(limit=120, tag=None, q=None, since_days=None, window=6000):
-    """limit — сколько отдать; индекс уже отсортирован по весy, window не режет."""
-    items = _rows(limit=max(int(limit), 400), tag=tag, q=q, since_days=since_days)
+def _sig(title, text):
+    """Подпись для схлопывания перепечаток: нормализованный текст без пунктуации."""
+    blob = re.sub(r"[^\wа-яё]+", "", ((title or "") + " " + (text or ""))[:400].lower())
+    return blob[:150]
+
+
+_CACHE = {"t": 0.0, "key": None, "val": None}
+_CACHE_TTL = 60
+
+
+def scan(limit=120, tag=None, q=None, since_days=None, window=6000, collapse=True):
+    """limit — сколько отдать; индекс уже отсортирован по весy, window не режет.
+
+    collapse=True — перепечатки одного и того же поста в разных каналах схлопываются
+    в одну карточку (поле dups = сколько копий), иначе выдача забивается клонами.
+    Результат кэшируется на минуту: чтение окна в 2000 записей стоит ~10 с.
+    """
+    key = (int(limit), tag, q, since_days, bool(collapse))
+    if _CACHE["key"] == key and (time.time() - _CACHE["t"]) < _CACHE_TTL:
+        return _CACHE["val"]
+    items = _rows(limit=max(int(limit) * 10, 2000), tag=tag, q=q, since_days=since_days)
     items.sort(key=lambda x: (-x["abuse_score"], -(x["id"] or 0)))
+    uniq_total = 0
+    if collapse:
+        seen = {}
+        for it in items:
+            k = it.get("sig") or _sig(it.get("title"), it.get("text"))
+            if len(k) < 40:
+                k = "id:%s" % it.get("id")
+            if k in seen:
+                seen[k]["dups"] = seen[k].get("dups", 0) + 1
+                continue
+            it["dups"] = 0
+            seen[k] = it
+        items = list(seen.values())
+        uniq_total = len(items)
     counts = {}
     for it in items:
         for t in it["tags"]:
@@ -291,8 +327,10 @@ def scan(limit=120, tag=None, q=None, since_days=None, window=6000):
         con.close()
     except Exception:
         pass
-    return {"items": items[:limit], "total": total, "tags": counts,
-            "labels": TAG_LABEL}
+    res = {"items": items[:limit], "total": total, "uniq": uniq_total or total,
+           "tags": counts, "labels": TAG_LABEL}
+    _CACHE.update(t=time.time(), key=key, val=res)
+    return res
 
 _RE_CODE = re.compile(r"(?:промокод|промо-?код|promo\s*code|coupon|купо[нш]|code)[^\n:：]{0,12}[:：\s]+([A-Z0-9][A-Z0-9_\-]{3,23})", re.I)
 _RE_CODE2 = re.compile(r"\b([A-Z]{2,}[0-9]{2,}[A-Z0-9_\-]{0,14})\b")