# Классификатор постов: словарный фильтр (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()