207 lines
8.5 KiB
Python
207 lines
8.5 KiB
Python
"""Склад: 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 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")
|
|
return [dict(zip(keys, r)) for r in rows]
|
|
|
|
|
|
def browse2(q="", names=None, limit=60, offset=0, sort="score"):
|
|
"""Записи для панели: фильтр по списку источников (категория), подстрока, пагинация + 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}%"]
|
|
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 "
|
|
"FROM items " + where + ("ORDER BY published DESC, id DESC LIMIT ? OFFSET ?" if sort == "fresh" else "ORDER BY score DESC, id DESC LIMIT ? OFFSET ?"),
|
|
args + [limit, offset]).fetchall()
|
|
keys = ("provider", "kind", "source", "title", "text", "author", "published", "score", "price", "currency", "url")
|
|
return [dict(zip(keys, r)) for r in rows], total
|
|
|
|
|
|
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) |