"""Склад: sqlite WAL + FTS5. Таблицы items, runs; дедуп по ключу и simhash-полосам.""" import json import sqlite3 import time from pathlib import Path from . import config, taxonomy SCHEMA = """ CREATE TABLE IF NOT EXISTS items ( id INTEGER PRIMARY KEY, k TEXT UNIQUE, provider TEXT, kind TEXT, source TEXT, url TEXT, title TEXT, text TEXT, author TEXT, published TEXT, lang TEXT, tags TEXT, price REAL, currency TEXT, score REAL, h INTEGER, extra TEXT, seen INTEGER ); CREATE INDEX IF NOT EXISTS ix_items_prov ON items(provider, kind); CREATE INDEX IF NOT EXISTS ix_items_seen ON items(seen); CREATE INDEX IF NOT EXISTS ix_items_h ON items(h); CREATE TABLE IF NOT EXISTS bands (band INTEGER, h INTEGER, item_id INTEGER, PRIMARY KEY(band,h,item_id)); CREATE INDEX IF NOT EXISTS ix_bands_h ON bands(h); CREATE TABLE IF NOT EXISTS runs ( id INTEGER PRIMARY KEY, ts INTEGER, provider TEXT, source TEXT, query TEXT, got INTEGER, kept INTEGER, status TEXT, bytes INTEGER, mode TEXT ); CREATE VIRTUAL TABLE IF NOT EXISTS items_fts USING fts5( title, text, content='items', content_rowid='id', tokenize='unicode61 remove_diacritics 2' ); CREATE TRIGGER IF NOT EXISTS items_ai AFTER INSERT ON items BEGIN INSERT INTO items_fts(rowid,title,text) VALUES (new.id,new.title,new.text); END; CREATE TRIGGER IF NOT EXISTS items_ad AFTER DELETE ON items BEGIN INSERT INTO items_fts(items_fts,rowid,title,text) VALUES('delete',old.id,old.title,old.text); END; """ BANDS = 4 # 4 полосы по 16 бит MASK = 0xFFFF def conn(): c = sqlite3.connect(str(config.DB), timeout=30) c.execute("PRAGMA journal_mode=WAL") c.execute("PRAGMA synchronous=NORMAL") c.executescript(SCHEMA) cols = {r[1] for r in c.execute("PRAGMA table_info(items)")} if "types" not in cols: c.execute("ALTER TABLE items ADD COLUMN types TEXT") c.commit() return c def _signed(h): return h - (1 << 64) if h >= (1 << 63) else h def _unsigned(v): return v & 0xFFFFFFFFFFFFFFFF def _popcount(x): return bin(x).count("1") def _near_dup(c, h, max_dist=3): """Кандидат ищем по полосе, решение принимаем по расстоянию Хэмминга.""" h = _unsigned(h) 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: if _popcount(_unsigned(other) ^ h) <= max_dist: return True return False def add(items): """Возвращает (новых, дублей).""" new = dup = 0 ts = int(time.time()) with conn() as c: for it in items: h = _unsigned(it.h) if _near_dup(c, h): dup += 1 continue u = (getattr(it, "url", "") or "").strip() if u and c.execute("SELECT 1 FROM items WHERE url = ? LIMIT 1", (u,)).fetchone(): dup += 1 continue try: cur = c.execute( "INSERT INTO items(k,provider,kind,source,url,title,text,author,published,lang," "tags,types,price,currency,score,h,extra,seen) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", (it.key(), it.provider, it.kind, it.source, it.url, it.title, it.text, it.author, it.published, it.lang, ",".join(it.tags), ",".join(taxonomy.classify(it.title, it.text)), it.price, it.currency, it.score, _signed(h), json.dumps(it.extra, ensure_ascii=False)[:2000], ts)) except sqlite3.IntegrityError: dup += 1 continue iid = cur.lastrowid for b in range(BANDS): band = (h >> (16 * b)) & MASK c.execute("INSERT OR IGNORE INTO bands(band,h,item_id) VALUES(?,?,?)", (band, _signed(h), iid)) new += 1 return new, dup def log_run(provider, source, query, got, kept, status, nbytes=0, mode=""): with conn() 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)) def top(limit=20, provider=None): q = ("SELECT provider,kind,source,title,score,price,currency,url FROM items " + ("WHERE provider=? " if provider else "") + "ORDER BY score DESC, id DESC LIMIT ?") args = (provider, limit) if provider else (limit,) with conn() as c: return c.execute(q, args).fetchall() def search(q, limit=20): with conn() 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 ?", (q, limit)).fetchall() if rows: return rows like = f"%{q}%" return c.execute( "SELECT provider,kind,source,title,score,url FROM items WHERE title LIKE ? OR text LIKE ? " "ORDER BY score DESC LIMIT ?", (like, like, limit)).fetchall() def browse(category=None, q="", limit=60, offset=0): """Список записей для веб-панели: фильтр по категории источника и подстроке.""" sql = ("SELECT provider,kind,source,title,text,author,published,score,price,currency,url " "FROM items WHERE 1=1 ") args = [] if category and category != "all": sql += "AND source=? " args.append(category) if q: sql += "AND (title LIKE ? OR text LIKE ?) " args += [f"%{q}%", f"%{q}%"] sql += "ORDER BY score DESC, id DESC LIMIT ? OFFSET ?" args += [limit, offset] with conn() 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] def browse2(q="", names=None, limit=60, offset=0, sort="score", priced=False, typ=None): """Записи для панели: фильтр по списку источников (категория), подстрока, пагинация + total.""" where = "WHERE 1=1 " args = [] if names: where += "AND source IN (%s) " % ",".join("?" * len(names)) args += list(names) if q: where += "AND (title LIKE ? OR text LIKE ?) " args += [f"%{q}%", f"%{q}%"] if priced: where += "AND price IS NOT NULL AND price != '' " if typ and typ != "all": where += "AND (',' || COALESCE(types,'') || ',') LIKE ? " args += [f"%,{typ},%"] 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: 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 ?", args + [limit, offset]).fetchall() keys = ("provider", "kind", "source", "title", "text", "author", "published", "score", "price", "currency", "url", "types") return [dict(zip(keys, r)) for r in rows], total def type_counts(): """Сколько записей на каждый тип.""" out = {} with conn() as c: for row in c.execute("SELECT COALESCE(types,'') FROM items"): for t in (row[0] or "").split(","): t = t.strip() if t: out[t] = out.get(t, 0) + 1 return dict(sorted(out.items(), key=lambda kv: -kv[1])) def retype(limit=None): """Пересчитать типы у уже сохранённых записей.""" n = 0 with conn() as c: q = "SELECT id,title,text FROM items" if limit: q += " LIMIT %d" % int(limit) for iid, title, text in c.execute(q).fetchall(): c.execute("UPDATE items SET types=? WHERE id=?", (",".join(taxonomy.classify(title, text)), iid)) n += 1 return n def count_by_source(): with conn() as c: return dict(c.execute("SELECT source, COUNT(*) FROM items GROUP BY source").fetchall()) def stats(): with conn() 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() priced = c.execute("SELECT COUNT(*) FROM items WHERE price IS NOT NULL").fetchone()[0] runs = c.execute("SELECT COUNT(*) FROM runs").fetchone()[0] ok_runs = c.execute("SELECT COUNT(*) FROM runs WHERE status='ok'").fetchone()[0] return {"items": total, "priced": priced, "runs": runs, "runs_ok": ok_runs, "by_provider": by_prov, "by_kind": by_kind} def export(path, fmt="jsonl", limit=1000): path = Path(path) with conn() 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() cols = ["provider", "kind", "source", "url", "title", "text", "author", "published", "lang", "tags", "price", "currency", "score", "extra"] if fmt == "csv": import csv with path.open("w", encoding="utf-8", newline="") as f: w = csv.DictWriter(f, fieldnames=cols) w.writeheader() for r in rows: w.writerow(dict(zip(cols, r))) else: with path.open("w", encoding="utf-8") as f: for r in rows: f.write(json.dumps(dict(zip(cols, r)), ensure_ascii=False) + "\n") return len(rows), str(path) def dedupe(apply: bool = False): """Схлопывает записи с одинаковым адресом (или source+title при пустом адресе). Оставляет самую свежую (max id). Возвращает (групп, лишних_строк, удалено). """ key = "CASE WHEN url IS NULL OR url = '' THEN source || '|' || title ELSE url END" with conn() as c: groups = c.execute( "SELECT count(*) FROM (SELECT " + key + " k FROM items GROUP BY k HAVING count(*) > 1)").fetchone()[0] extra = c.execute( "SELECT coalesce(sum(n - 1), 0) FROM (SELECT count(*) n FROM items GROUP BY " + key + " HAVING n > 1)").fetchone()[0] killed = 0 if apply and extra: ids = [r[0] for r in c.execute( "SELECT id FROM items WHERE id NOT IN (SELECT max(id) FROM items GROUP BY " + key + ")")] for chunk in [ids[i:i + 400] for i in range(0, len(ids), 400)]: q = "DELETE FROM items WHERE id IN (" + ",".join("?" * len(chunk)) + ")" c.execute(q, chunk) killed += len(chunk) return groups, extra, killed