"""Склад: sqlite WAL + FTS5. Таблицы items, runs; дедуп по ключу и simhash-полосам.""" import json import sqlite3 import time from pathlib import Path from . import config 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) 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 try: cur = c.execute( "INSERT INTO items(k,provider,kind,source,url,title,text,author,published,lang," "tags,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), 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 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)