295 lines
12 KiB
Python
295 lines
12 KiB
Python
"""Склад: sqlite WAL + FTS5. Таблицы items, runs; дедуп по ключу и simhash-полосам."""
|
||
import contextlib
|
||
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
|
||
|
||
|
||
@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
|
||
|
||
|
||
def _unsigned(v):
|
||
return v & 0xFFFFFFFFFFFFFFFF
|
||
|
||
|
||
def _popcount(x):
|
||
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.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
|
||
|
||
|
||
def add(items):
|
||
"""Возвращает (новых, дублей)."""
|
||
new = dup = 0
|
||
ts = int(time.time())
|
||
with db() 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 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))
|
||
|
||
|
||
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 db() as c:
|
||
return c.execute(q, args).fetchall()
|
||
|
||
|
||
def search(q, limit=20):
|
||
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 ?",
|
||
(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 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]
|
||
|
||
|
||
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 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 ?",
|
||
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 db() 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 db() 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 db() as c:
|
||
return dict(c.execute("SELECT source, COUNT(*) FROM items GROUP BY source").fetchall())
|
||
|
||
def stats():
|
||
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()
|
||
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 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()
|
||
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 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(
|
||
"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
|