mirror of
https://gitverse.ru/kpa39l/vesti.git
synced 2026-09-29 09:55:03 +00:00
kpa39l
251 lines
11 KiB
Python
251 lines
11 KiB
Python
# Классификатор постов: словарный фильтр (MUST) + LLM-уточнение (qwen3:8b-nothink).
|
||
# По спеке: сначала keywords.py (без LLM), затем для кандидатов — Ollama.
|
||
# Если модель недоступна — пост остаётся unclassified (не теряется).
|
||
|
||
import json
|
||
import os
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
import httpx
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||
from classifier.keywords import find_direction, DIRECTIONS_CANON
|
||
|
||
OLLAMA_URL = os.getenv("OLLAMA_URL", "http://127.0.0.1:11434")
|
||
OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "qwen3:8b-nothink")
|
||
TIMEOUT = float(os.getenv("CLASSIFY_TIMEOUT", "30"))
|
||
|
||
|
||
def classify_text(post: dict) -> dict:
|
||
"""Классифицирует текст поста. Возвращает dict с direction, relevance, interest, summary, classified.
|
||
|
||
Сначала словарный фильтр (быстрый, без LLM). Если направление найдено —
|
||
пробуем LLM-уточнение (relevance/interest/summary). При недоступности LLM —
|
||
фолбэк: direction из словаря, relevance=low, interest=1, classified=False.
|
||
|
||
Если у поста задан source_slug с правилом SOURCE_RULES (например, lwn → tech) —
|
||
направление принудительное, LLM не вызывается (детерминированно, быстро).
|
||
|
||
СВОЙ контент (is_own=1): «сильный кандидат» — при словарном попадании
|
||
relevance=critical, classified=True даже без LLM (фолбэк 'dict-own').
|
||
Если словарь не дал направление — классификации нет, пост не теряется.
|
||
|
||
post: {"text": str, "views": int, "reactions_total": int, "is_own": int, "source_slug": str|None, ...}
|
||
"""
|
||
text = (post.get("text") or "").strip()
|
||
views = int(post.get("views") or 0)
|
||
reactions = int(post.get("reactions_total") or 0)
|
||
is_own = int(post.get("is_own") or 0) == 1
|
||
source_slug = post.get("source_slug")
|
||
|
||
# правило источника (SOURCE_RULES): принудительное направление, без LLM
|
||
from classifier.keywords import source_directions
|
||
forced = source_directions(source_slug)
|
||
if forced:
|
||
return {
|
||
"direction": forced[0],
|
||
"relevance": "high",
|
||
"interest": 3,
|
||
"summary": "",
|
||
"classified": True,
|
||
"method": "source-rule",
|
||
}
|
||
|
||
direction = find_direction(text)
|
||
if not direction:
|
||
# словарь не дал попадания — всё равно пробуем LLM (MUST: не терять посты)
|
||
try:
|
||
llm = call_ollama(text, views, reactions, None)
|
||
d = llm.get("direction")
|
||
if d and d in DIRECTIONS_CANON:
|
||
return {
|
||
"direction": d,
|
||
"relevance": llm.get("relevance") or "low",
|
||
"interest": clamp_interest(llm.get("interest")),
|
||
"summary": llm.get("summary") or "",
|
||
"classified": True,
|
||
"method": "llm",
|
||
}
|
||
# LLM отработала, но не дала каноническое направление — не классифицируем
|
||
return {
|
||
"direction": None,
|
||
"relevance": "low",
|
||
"interest": 1,
|
||
"summary": "",
|
||
"classified": False,
|
||
"method": "llm-no-direction",
|
||
}
|
||
except Exception as e:
|
||
# LLM недоступна — пост остаётся неклассифицированным (не теряется)
|
||
return {
|
||
"direction": None,
|
||
"relevance": "low",
|
||
"interest": 1,
|
||
"summary": "",
|
||
"classified": False,
|
||
"method": f"dict-miss llm-err ({type(e).__name__})",
|
||
}
|
||
|
||
# словарный фильтр сработал — кандидат; пробуем LLM
|
||
try:
|
||
llm = call_ollama(text, views, reactions, direction)
|
||
return {
|
||
"direction": llm.get("direction") or direction,
|
||
"relevance": llm.get("relevance") or "low",
|
||
"interest": clamp_interest(llm.get("interest")),
|
||
"summary": llm.get("summary") or "",
|
||
"classified": True,
|
||
"method": "llm",
|
||
}
|
||
except Exception as e:
|
||
# СВОЙ контент: сильный кандидат даже без LLM (словарь дал направление)
|
||
if is_own:
|
||
return {
|
||
"direction": direction,
|
||
"relevance": "critical",
|
||
"interest": 1,
|
||
"summary": "",
|
||
"classified": True,
|
||
"method": f"dict-own ({type(e).__name__})",
|
||
}
|
||
# LLM недоступна — фолбэк без LLM (MUST: пост не теряется)
|
||
return {
|
||
"direction": direction,
|
||
"relevance": "low",
|
||
"interest": 1,
|
||
"summary": "",
|
||
"classified": False,
|
||
"method": f"dict-only ({type(e).__name__})",
|
||
}
|
||
|
||
|
||
def call_ollama(text: str, views: int, reactions: int, direction: str) -> dict:
|
||
"""Вызывает qwen3:8b-nothink (OpenAI-совместимый /v1, think:false) и возвращает JSON."""
|
||
prompt = f"""Ты — классификатор новостей. Определи {{
|
||
"direction": "одно из: {', '.join(sorted(DIRECTIONS_CANON))}",
|
||
"relevance": "critical|high|low",
|
||
"interest": 1-5,
|
||
"summary": "краткое резюме 1-2 предложения"
|
||
}} для поста.
|
||
|
||
Пост (направление по словарю: {direction}, views={views}, reactions={reactions}):
|
||
{text[:2000]}
|
||
|
||
Ответь ТОЛЬКО JSON."""
|
||
r = httpx.post(
|
||
f"{OLLAMA_URL}/v1/chat/completions",
|
||
json={
|
||
"model": OLLAMA_MODEL,
|
||
"messages": [{"role": "user", "content": prompt}],
|
||
"temperature": 0.2,
|
||
"stream": False,
|
||
"think": False, # qwen3: отключаем reasoning через extra_body (OpenAI-совместимый)
|
||
},
|
||
timeout=TIMEOUT,
|
||
)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
content = (data.get("choices") or [{}])[0].get("message", {}).get("content") or ""
|
||
# вытаскиваем JSON из ответа (может быть с ```json обёрткой)
|
||
content = content.strip()
|
||
if content.startswith("```"):
|
||
content = content.strip("`")
|
||
if content.startswith("json"):
|
||
content = content[4:].strip()
|
||
return json.loads(content)
|
||
|
||
|
||
def clamp_interest(v) -> int:
|
||
try:
|
||
return max(1, min(5, int(v)))
|
||
except (TypeError, ValueError):
|
||
return 1
|
||
|
||
|
||
def classify_posts_in_db(db_path: str, limit: int = 200, direction: str | None = None):
|
||
"""Обрабатывает неклассифицированные посты в SQLite. Возвращает (processed, classified, llm_ok)."""
|
||
import sqlite3
|
||
|
||
conn = sqlite3.connect(db_path)
|
||
conn.row_factory = sqlite3.Row
|
||
cur = conn.cursor()
|
||
# посты без классификации (classified не задан — status='new' и нет direction?)
|
||
# в схеме пока нет колонок классификации; добавляем по ходу (схема расширяется)
|
||
cols = [r[1] for r in cur.execute("PRAGMA table_info(posts)")]
|
||
if "direction" not in cols:
|
||
cur.execute("ALTER TABLE posts ADD COLUMN direction TEXT")
|
||
cur.execute("ALTER TABLE posts ADD COLUMN relevance TEXT")
|
||
cur.execute("ALTER TABLE posts ADD COLUMN interest INTEGER")
|
||
cur.execute("ALTER TABLE posts ADD COLUMN summary TEXT")
|
||
cur.execute("ALTER TABLE posts ADD COLUMN classified INTEGER DEFAULT 0")
|
||
conn.commit()
|
||
cols = [r[1] for r in cur.execute("PRAGMA table_info(posts)")]
|
||
|
||
where = "WHERE classified IS NULL OR classified=0"
|
||
if direction:
|
||
where += f" AND direction IS NULL" # не классифицированы в этом направлении
|
||
|
||
rows = cur.execute(
|
||
f"SELECT p.id, p.text, p.views, p.reactions_total, p.is_own, s.slug AS source_slug "
|
||
f"FROM posts p LEFT JOIN sources s ON s.id = p.source_id {where} "
|
||
f"ORDER BY p.is_own DESC, p.id LIMIT ?",
|
||
(limit,),
|
||
).fetchall()
|
||
|
||
processed = classified = llm_ok = 0
|
||
for row in rows:
|
||
res = classify_text(dict(row))
|
||
if res["classified"]:
|
||
classified += 1
|
||
if res["method"] == "llm":
|
||
llm_ok += 1
|
||
cur.execute(
|
||
"UPDATE posts SET direction=?, relevance=?, interest=?, summary=?, classified=? WHERE id=?",
|
||
(res["direction"], res["relevance"], res["interest"], res["summary"],
|
||
1 if res["classified"] else 0, row["id"]),
|
||
)
|
||
# мультинаправления: все словарные попадания → classifications (для fan-out)
|
||
from classifier.keywords import find_directions_all
|
||
dirs_all = find_directions_all(row["text"] or "", row["source_slug"])
|
||
if res["direction"] and res["direction"] not in dirs_all:
|
||
dirs_all.append(res["direction"])
|
||
for d in dirs_all:
|
||
exists = cur.execute(
|
||
"SELECT 1 FROM classifications WHERE post_id=? AND direction=? LIMIT 1",
|
||
(row["id"], d),
|
||
).fetchone()
|
||
if not exists:
|
||
cur.execute(
|
||
"""INSERT INTO classifications (post_id, direction, relevance, interest, summary, model)
|
||
VALUES (?,?,?,?,?,?)""",
|
||
(row["id"], d, res["relevance"], res["interest"], res["summary"],
|
||
res.get("method") or "classify"),
|
||
)
|
||
processed += 1
|
||
if processed % 5 == 0 or processed == len(rows):
|
||
conn.commit() # понемногу коммитим — не держать долгую транзакцию
|
||
print(f" ...{processed}/{len(rows)} (LLM: {llm_ok})", flush=True)
|
||
conn.commit()
|
||
conn.close()
|
||
return processed, classified, llm_ok
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import argparse
|
||
|
||
ap = argparse.ArgumentParser(description="VESTI classifier")
|
||
ap.add_argument("--db", default=str(Path(__file__).resolve().parent.parent / "db" / "vesti.db"))
|
||
ap.add_argument("--direction", help="направление (необязательно)")
|
||
ap.add_argument("--limit", type=int, default=200)
|
||
args = ap.parse_args()
|
||
|
||
proc, cls, llm = classify_posts_in_db(args.db, args.limit, args.direction)
|
||
print(f"Обработано: {proc}, классифицировано (LLM+словарь): {cls}, из них через LLM: {llm}", flush=True)
|
||
# показать примеры
|
||
import sqlite3
|
||
conn = sqlite3.connect(args.db)
|
||
conn.row_factory = sqlite3.Row
|
||
for r in conn.execute("SELECT id, direction, relevance, interest, substr(summary,1,60) s FROM posts WHERE classified=1 LIMIT 5"):
|
||
print(f" #{r['id']} {r['direction']} {r['relevance']} interest={r['interest']}: {r['s']}")
|
||
conn.close() |