server.py

#!/usr/bin/env python3
"""
LinkRAG — incarci .txt cu URL-uri sau texte proprii, le proceseaza in paralel
(articole HTML via newspaper4k, PDF via pypdf, DOCX via python-docx),
le sparge in chunk-uri, le embedeaza local via ollama (nomic-embed-text-v2-moe),
le tine in sqlite-vec persistent. La query: top-20 chunks → claude CLI sonnet.
Sesiuni: fiecare batch are session_id; "Sesiune noua" porneste alta; istoricul ramane.
"""
import io
import json
import os
import queue
import re
import shlex
import sqlite3
import struct
import subprocess
import threading
import time
import uuid
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
from pathlib import Path
from urllib.parse import unquote, urlparse, parse_qs

import requests
import sqlite_vec
from flask import Flask, Response, jsonify, request, send_from_directory

import pypdf
import docx as pydocx
from newspaper import Article, Config

APP_DIR = Path(__file__).resolve().parent
DATA_DIR = APP_DIR / "data"
DOCS_DIR = DATA_DIR / "docs"
DB_PATH = DATA_DIR / "linkrag.db"
DOCS_DIR.mkdir(parents=True, exist_ok=True)

PORT = 8770
HOST = "0.0.0.0"
OLLAMA_URL = "http://127.0.0.1:11434"
EMBED_MODEL = os.environ.get("LINKRAG_EMBED", "nomic-embed-text-v2-moe")
EMBED_DIM = 768
CLAUDE_BIN = os.path.expanduser("~/.local/bin/claude")
CLAUDE_MODEL = os.environ.get("LINKRAG_MODEL", "sonnet")

EXTRACT_WORKERS = 24
HTTP_TIMEOUT = 30
MAX_DOC_BYTES = 30 * 1024 * 1024  # 30 MB
CHUNK_SIZE = 500       # cuvinte (~500 tokens)
CHUNK_OVERLAP = 60
TOP_K = 20
HOST_CONCURRENCY = 2          # max requests in parallel per acelasi hostname
HOST_DELAY_SEC = 0.30         # delay min intre hit-uri pe acelasi host

_HOST_SEMS: dict[str, threading.Semaphore] = {}
_HOST_LAST: dict[str, float] = {}
_HOST_LOCK = threading.Lock()


def host_of(url: str) -> str:
    try: return (urlparse(url).hostname or "").lower()
    except Exception: return ""


class HostGate:
    """Limiteaza concurenta per hostname si forteaza un delay minim intre hit-uri."""
    def __init__(self, url: str):
        self.host = host_of(url)
        self.sem = None
    def __enter__(self):
        if not self.host: return self
        with _HOST_LOCK:
            sem = _HOST_SEMS.get(self.host)
            if sem is None:
                sem = threading.Semaphore(HOST_CONCURRENCY)
                _HOST_SEMS[self.host] = sem
        sem.acquire()
        self.sem = sem
        # delay vs ultimul hit pe acest host (sub semafor pentru exclusivitate)
        with _HOST_LOCK:
            last = _HOST_LAST.get(self.host, 0.0)
            wait = HOST_DELAY_SEC - (time.time() - last)
            if wait > 0:
                time.sleep(wait)
            _HOST_LAST[self.host] = time.time()
        return self
    def __exit__(self, *a):
        if self.sem: self.sem.release()

UA = ("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
      "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36")


def smart_decode(data: bytes) -> tuple[str, str]:
    """Detecteaza encoding-ul unui .txt. Suporta BOM-uri (UTF-8/16/32),
    UTF-16 fara BOM (heuristic null-bytes), apoi autodetect via
    charset-normalizer, fallback windows-1252. Returneaza (text, encoding)."""
    if not data:
        return "", "empty"

    # BOM sniffing
    if data.startswith(b"\xef\xbb\xbf"):
        return data[3:].decode("utf-8", errors="replace"), "utf-8-sig"
    if data.startswith(b"\xff\xfe\x00\x00"):
        return data[4:].decode("utf-32-le", errors="replace"), "utf-32-le"
    if data.startswith(b"\x00\x00\xfe\xff"):
        return data[4:].decode("utf-32-be", errors="replace"), "utf-32-be"
    if data.startswith(b"\xff\xfe"):
        return data[2:].decode("utf-16-le", errors="replace"), "utf-16-le"
    if data.startswith(b"\xfe\xff"):
        return data[2:].decode("utf-16-be", errors="replace"), "utf-16-be"

    # UTF-8 strict
    try:
        return data.decode("utf-8", errors="strict"), "utf-8"
    except UnicodeDecodeError:
        pass

    # UTF-16 fara BOM: daca sample-ul are multe null-byte-uri (~15%+),
    # probabil e UTF-16 (Windows exports fac asta).
    sample = data[:4096]
    nulls = sample.count(b"\x00")
    if nulls > len(sample) * 0.15:
        # guess endianness: in LE, pozitii impare sunt mai des null pentru text latin
        odd = sum(1 for i in range(1, len(sample), 2) if sample[i] == 0)
        even = sum(1 for i in range(0, len(sample), 2) if sample[i] == 0)
        order = ("utf-16-le", "utf-16-be") if odd > even else ("utf-16-be", "utf-16-le")
        for enc in order:
            try:
                return data.decode(enc, errors="strict"), enc
            except UnicodeDecodeError:
                continue

    # charset-normalizer (vine cu requests)
    try:
        from charset_normalizer import from_bytes
        result = from_bytes(data).best()
        if result:
            return str(result), (result.encoding or "auto")
    except Exception:
        pass

    # ultim: cp1250 (Est-Europa, romana), apoi cp1252 (vest latin)
    for enc in ("cp1250", "cp1252", "cp1251"):
        try:
            return data.decode(enc, errors="strict"), enc
        except UnicodeDecodeError:
            continue
    return data.decode("utf-8", errors="replace"), "utf-8-replace"

app = Flask(__name__)


# ---------- DB ----------

def db_conn():
    con = sqlite3.connect(DB_PATH, check_same_thread=False)
    con.enable_load_extension(True)
    sqlite_vec.load(con)
    con.enable_load_extension(False)
    con.execute("PRAGMA journal_mode=WAL")
    con.execute("PRAGMA foreign_keys=ON")
    return con


def db_init():
    con = db_conn()
    con.executescript(f"""
    CREATE TABLE IF NOT EXISTS sessions (
        id TEXT PRIMARY KEY,
        name TEXT,
        created_at TEXT,
        doc_count INTEGER DEFAULT 0,
        chunk_count INTEGER DEFAULT 0
    );
    CREATE TABLE IF NOT EXISTS documents (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        session_id TEXT NOT NULL,
        source TEXT,
        source_type TEXT,
        title TEXT,
        full_text TEXT,
        bytes INTEGER,
        created_at TEXT,
        FOREIGN KEY(session_id) REFERENCES sessions(id) ON DELETE CASCADE
    );
    CREATE INDEX IF NOT EXISTS idx_doc_session ON documents(session_id);
    CREATE TABLE IF NOT EXISTS chunks (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        doc_id INTEGER NOT NULL,
        session_id TEXT NOT NULL,
        chunk_idx INTEGER,
        text TEXT,
        FOREIGN KEY(doc_id) REFERENCES documents(id) ON DELETE CASCADE
    );
    CREATE INDEX IF NOT EXISTS idx_chunk_session ON chunks(session_id);
    CREATE INDEX IF NOT EXISTS idx_chunk_doc ON chunks(doc_id);
    CREATE VIRTUAL TABLE IF NOT EXISTS vec_chunks USING vec0(
        chunk_id INTEGER PRIMARY KEY,
        embedding FLOAT[{EMBED_DIM}]
    );
    """)
    con.commit()
    con.close()


db_init()
DB_LOCK = threading.Lock()  # serialize writes (sqlite single writer)


# ---------- session helpers ----------

def session_create(name: str = None) -> str:
    sid = uuid.uuid4().hex[:12]
    name = name or f"sesiune {datetime.now().strftime('%Y-%m-%d %H:%M')}"
    with DB_LOCK:
        con = db_conn()
        con.execute("INSERT INTO sessions(id,name,created_at) VALUES(?,?,?)",
                    (sid, name, datetime.now().isoformat()))
        con.commit(); con.close()
    return sid


def session_list():
    con = db_conn()
    rows = con.execute(
        "SELECT id,name,created_at,doc_count,chunk_count FROM sessions ORDER BY created_at DESC"
    ).fetchall()
    con.close()
    return [dict(zip(["id","name","created_at","doc_count","chunk_count"], r)) for r in rows]


def session_update_counts(sid: str):
    with DB_LOCK:
        con = db_conn()
        dc = con.execute("SELECT COUNT(*) FROM documents WHERE session_id=?", (sid,)).fetchone()[0]
        cc = con.execute("SELECT COUNT(*) FROM chunks WHERE session_id=?", (sid,)).fetchone()[0]
        con.execute("UPDATE sessions SET doc_count=?, chunk_count=? WHERE id=?", (dc, cc, sid))
        con.commit(); con.close()


def session_meta(sid: str):
    con = db_conn()
    row = con.execute("SELECT id,name,created_at,doc_count,chunk_count FROM sessions WHERE id=?", (sid,)).fetchone()
    con.close()
    if not row: return None
    return dict(zip(["id","name","created_at","doc_count","chunk_count"], row))


# ---------- URL cleaning ----------

GOOGLE_HOSTS_SKIP = {
    "accounts.google.com", "policies.google.com", "support.google.com",
    "myaccount.google.com", "myactivity.google.com", "ads.google.com",
    "adservice.google.com", "googleadservices.com", "www.googleadservices.com",
    "tpc.googlesyndication.com", "googlesyndication.com",
    "translate.google.com", "maps.google.com", "books.google.com",
    "scholar.google.com",  # search engine itself, not articles
}
GOOGLE_PATH_SKIP_PREFIXES = ("/preferences", "/setprefs", "/advanced_search",
                             "/imgres", "/intl/", "/policies", "/url?sa=X",
                             "/safebrowsing", "/recaptcha")
SOCIAL_SKIP_HOSTS = {
    "facebook.com", "www.facebook.com", "twitter.com", "x.com",
    "instagram.com", "www.instagram.com", "tiktok.com", "www.tiktok.com",
    "youtube.com", "www.youtube.com", "youtu.be",
}
URL_RE = re.compile(r"https?://[^\s<>\"\)\]]+", re.IGNORECASE)


def unwrap_google_redirect(url: str) -> str:
    """ /url?q=https://target&sa=... → https://target """
    try:
        p = urlparse(url)
        if p.netloc.endswith("google.com") and p.path == "/url":
            qs = parse_qs(p.query)
            real = qs.get("q") or qs.get("url")
            if real:
                return unquote(real[0])
    except Exception:
        pass
    return url


def clean_url(raw: str) -> str | None:
    raw = (raw or "").strip().rstrip(".,;:")
    if not raw or not raw.lower().startswith(("http://", "https://")):
        return None
    raw = unwrap_google_redirect(raw)
    try:
        p = urlparse(raw)
    except Exception:
        return None
    host = (p.netloc or "").lower()
    if not host:
        return None
    if host in GOOGLE_HOSTS_SKIP:
        return None
    if host.endswith(".google.com") and host != "www.google.com" and not any(
        host.endswith(x) for x in (".blogspot.google.com",)  # rare, allow
    ):
        return None
    if host == "www.google.com" or host == "google.com":
        # search/results pages — skip unless it's a doc-like static path
        if not p.path or any(p.path.startswith(pre) for pre in GOOGLE_PATH_SKIP_PREFIXES):
            return None
        if p.path == "/search":
            return None
    if host in SOCIAL_SKIP_HOSTS:
        return None
    return raw


def parse_links_from_text(text: str) -> list[str]:
    found = URL_RE.findall(text or "")
    seen = set(); out = []
    for u in found:
        c = clean_url(u)
        if c and c not in seen:
            seen.add(c); out.append(c)
    return out


# ---------- extraction ----------

def head_kind(url: str) -> str:
    """ Returns 'pdf' | 'docx' | 'doc' | 'html' based on URL ext or HEAD content-type. """
    low = url.lower().split("?")[0]
    if low.endswith(".pdf"): return "pdf"
    if low.endswith(".docx"): return "docx"
    if low.endswith(".doc"): return "doc"
    try:
        r = requests.head(url, allow_redirects=True, timeout=8,
                          headers={"User-Agent": UA})
        ct = (r.headers.get("Content-Type") or "").lower()
        if "pdf" in ct: return "pdf"
        if "wordprocessingml" in ct or "docx" in ct: return "docx"
        if "msword" in ct: return "doc"
    except Exception:
        pass
    return "html"


def fetch_bytes(url: str) -> bytes:
    r = requests.get(url, timeout=HTTP_TIMEOUT, stream=True,
                     headers={"User-Agent": UA})
    r.raise_for_status()
    buf = io.BytesIO(); total = 0
    for ch in r.iter_content(64 * 1024):
        if not ch: continue
        total += len(ch)
        if total > MAX_DOC_BYTES:
            raise ValueError(f"fisier > {MAX_DOC_BYTES//1024//1024}MB")
        buf.write(ch)
    return buf.getvalue()


def extract_pdf(data: bytes) -> tuple[str, str]:
    reader = pypdf.PdfReader(io.BytesIO(data))
    title = ""
    try:
        title = (reader.metadata.title or "") if reader.metadata else ""
    except Exception:
        title = ""
    parts = []
    for p in reader.pages:
        try:
            parts.append(p.extract_text() or "")
        except Exception:
            continue
    return title, "\n\n".join(parts).strip()


def extract_docx(data: bytes) -> tuple[str, str]:
    d = pydocx.Document(io.BytesIO(data))
    title = ""
    paras = [p.text for p in d.paragraphs if p.text and p.text.strip()]
    if paras:
        title = paras[0][:200]
    return title, "\n\n".join(paras).strip()


def extract_html(url: str) -> tuple[str, str]:
    cfg = Config()
    cfg.browser_user_agent = UA
    cfg.request_timeout = HTTP_TIMEOUT
    cfg.fetch_images = False
    art = Article(url, config=cfg)
    art.download(); art.parse()
    return (art.title or ""), (art.text or "").strip()


def extract_one(url: str) -> dict:
    """Returns { url, kind, title, text, error } """
    out = {"url": url, "kind": None, "title": "", "text": "", "error": None}
    try:
        with HostGate(url):
            kind = head_kind(url)
            out["kind"] = kind
            if kind == "html":
                t, x = extract_html(url)
            elif kind == "pdf":
                t, x = extract_pdf(fetch_bytes(url))
            elif kind == "docx":
                t, x = extract_docx(fetch_bytes(url))
            else:
                out["error"] = f"format {kind} neimplementat"
                return out
        out["title"] = t or ""
        out["text"] = x or ""
        if not out["text"] or len(out["text"]) < 80:
            out["error"] = "text prea scurt"
    except Exception as e:
        out["error"] = f"{type(e).__name__}: {e}"
    return out


# ---------- chunking ----------

def chunk_text(text: str, size: int = CHUNK_SIZE, overlap: int = CHUNK_OVERLAP) -> list[str]:
    words = (text or "").split()
    if not words: return []
    chunks = []
    i = 0
    step = max(1, size - overlap)
    while i < len(words):
        chunk = " ".join(words[i:i+size])
        chunks.append(chunk)
        if i + size >= len(words): break
        i += step
    return chunks


# ---------- embedding ----------

def embed_batch(texts: list[str]) -> list[list[float]]:
    if not texts: return []
    r = requests.post(f"{OLLAMA_URL}/api/embed",
                      json={"model": EMBED_MODEL, "input": texts},
                      timeout=180)
    r.raise_for_status()
    return r.json()["embeddings"]


def vec_blob(v: list[float]) -> bytes:
    return struct.pack(f"{len(v)}f", *v)


# ---------- storage ----------

def store_doc(session_id: str, source: str, source_type: str,
              title: str, full_text: str) -> int:
    chunks = chunk_text(full_text)
    if not chunks:
        return 0
    embs = embed_batch(chunks)
    with DB_LOCK:
        con = db_conn()
        cur = con.execute(
            "INSERT INTO documents(session_id,source,source_type,title,full_text,bytes,created_at) "
            "VALUES(?,?,?,?,?,?,?)",
            (session_id, source, source_type, title[:300],
             full_text, len(full_text.encode("utf-8", errors="ignore")),
             datetime.now().isoformat()))
        doc_id = cur.lastrowid
        for idx, (txt, emb) in enumerate(zip(chunks, embs)):
            cc = con.execute(
                "INSERT INTO chunks(doc_id,session_id,chunk_idx,text) VALUES(?,?,?,?)",
                (doc_id, session_id, idx, txt))
            chunk_id = cc.lastrowid
            con.execute("INSERT INTO vec_chunks(chunk_id,embedding) VALUES(?,?)",
                        (chunk_id, vec_blob(emb)))
        con.commit(); con.close()
    return len(chunks)


def search_chunks(session_id: str, query: str, k: int = TOP_K) -> list[dict]:
    qemb = embed_batch([query])[0]
    con = db_conn()
    rows = con.execute(
        """
        SELECT c.id, c.doc_id, c.chunk_idx, c.text, d.title, d.source, d.source_type, v.distance
        FROM vec_chunks v
        JOIN chunks c ON c.id = v.chunk_id
        JOIN documents d ON d.id = c.doc_id
        WHERE c.session_id = ? AND v.embedding MATCH ? AND k = ?
        ORDER BY v.distance
        """,
        (session_id, vec_blob(qemb), k),
    ).fetchall()
    con.close()
    return [
        {"chunk_id": r[0], "doc_id": r[1], "chunk_idx": r[2], "text": r[3],
         "title": r[4], "source": r[5], "source_type": r[6], "distance": r[7]}
        for r in rows
    ]


# ---------- claude CLI ----------

CLAUDE_SYSTEM = (
    "Esti un asistent care raspunde STRICT pe baza fragmentelor de context oferite. "
    "Daca raspunsul nu e in context, spune clar ca nu il gasesti. "
    "Scrii in romana, concis, factual.\n\n"
    "REGULA DE CITARE (obligatorie):\n"
    "Cand afirmi ceva, citeaza sursa scriind EXACT asa: [#N] — paranteza patrata, diez, "
    "numarul documentului, paranteza patrata inchisa. Nimic altceva intre paranteze.\n"
    "EXEMPLE CORECTE: [#3], [#12]\n"
    "EXEMPLE GRESITE (nu le folosi): [#3 - titlu], [3], [#3/4], [#3, #4]\n"
    "Daca vrei sa citezi mai multe surse, scrie-le separat: [#3] [#5] [#7]."
)


def build_query_prompt(question: str, hits: list[dict]) -> str:
    parts = [f"# Intrebare\n{question}\n\n# Context (top {len(hits)} fragmente relevante)\n"]
    # grupez chunk-urile dupa doc_id (sa vada claude ca mai multe chunk-uri = acelasi document)
    seen_docs = {}
    for h in hits:
        seen_docs.setdefault(h["doc_id"], {"title": h["title"] or h["source"], "chunks": []})
        seen_docs[h["doc_id"]]["chunks"].append(h)
    for doc_id, info in seen_docs.items():
        parts.append(f"\n### Sursa #{doc_id} — {info['title'][:140]}\n")
        for h in info["chunks"]:
            parts.append(f"(chunk {h['chunk_idx']})\n{h['text']}\n")
    parts.append("\n# Sarcina\nRaspunde la intrebare folosind DOAR contextul de mai sus. "
                 "Citeaza sursele in format [#N] conform regulilor de sistem. "
                 "Daca informatia lipseste sau e contradictorie, spune-o.")
    return "".join(parts)


def stream_claude(prompt: str, q: queue.Queue):
    cmd = [
        CLAUDE_BIN, "-p",
        "--model", CLAUDE_MODEL,
        "--permission-mode", "bypassPermissions",
        "--output-format", "stream-json",
        "--include-partial-messages",
        "--verbose",
        "--append-system-prompt", CLAUDE_SYSTEM,
        prompt,
    ]
    try:
        proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                text=True, bufsize=1)
    except Exception as e:
        q.put(("error", {"message": f"claude pornire: {e}"})); q.put(("done", {})); return

    stderr_buf = []
    threading.Thread(target=lambda: stderr_buf.extend(proc.stderr.readlines()), daemon=True).start()
    full = []
    try:
        for line in proc.stdout:
            line = line.strip()
            if not line: continue
            try: ev = json.loads(line)
            except json.JSONDecodeError: continue
            t = ev.get("type")
            if t == "stream_event":
                se = ev.get("event", {})
                if se.get("type") == "content_block_delta":
                    d = se.get("delta", {})
                    if d.get("type") == "text_delta":
                        full.append(d.get("text", ""))
                        q.put(("text", {"text": d.get("text", "")}))
            elif t == "result":
                q.put(("meta", {
                    "cost_usd": ev.get("total_cost_usd"),
                    "duration_ms": ev.get("duration_ms"),
                    "num_turns": ev.get("num_turns"),
                }))
        proc.wait(timeout=5)
        if proc.returncode not in (0, None):
            q.put(("error", {"message": f"claude exit {proc.returncode}",
                             "stderr": "".join(stderr_buf)[-1500:]}))
    except Exception as e:
        q.put(("error", {"message": str(e)}))
    finally:
        q.put(("done", {}))


# ---------- SSE ----------

def sse(event: str, data) -> str:
    return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"


# ---------- routes ----------

@app.route("/")
def index():
    return send_from_directory(str(APP_DIR), "index.html")


@app.route("/health")
def health():
    return jsonify({"ok": True, "port": PORT,
                    "embed_model": EMBED_MODEL, "claude_model": CLAUDE_MODEL,
                    "embed_dim": EMBED_DIM, "top_k": TOP_K})


@app.route("/sessions")
def sessions_route():
    return jsonify(session_list())


@app.route("/sessions/new", methods=["POST"])
def session_new_route():
    name = (request.json or {}).get("name") if request.is_json else None
    sid = session_create(name)
    return jsonify(session_meta(sid))


@app.route("/sessions/<sid>")
def session_get_route(sid):
    m = session_meta(sid)
    if not m: return jsonify({"error": "not found"}), 404
    return jsonify(m)


@app.route("/sessions/<sid>/docs")
def session_docs_route(sid):
    con = db_conn()
    rows = con.execute(
        "SELECT id,source,source_type,title,bytes,created_at FROM documents WHERE session_id=? ORDER BY id",
        (sid,)).fetchall()
    con.close()
    return jsonify([dict(zip(["id","source","source_type","title","bytes","created_at"], r))
                    for r in rows])


@app.route("/sessions/<sid>", methods=["DELETE"])
def session_delete_route(sid):
    if not session_meta(sid):
        return jsonify({"error": "not found"}), 404
    with DB_LOCK:
        con = db_conn()
        # sterge vec_chunks prin chunk_ids (virtual table, nu are FK cascade)
        ids = [r[0] for r in con.execute(
            "SELECT id FROM chunks WHERE session_id=?", (sid,)).fetchall()]
        for cid in ids:
            con.execute("DELETE FROM vec_chunks WHERE chunk_id=?", (cid,))
        con.execute("DELETE FROM chunks WHERE session_id=?", (sid,))
        con.execute("DELETE FROM documents WHERE session_id=?", (sid,))
        con.execute("DELETE FROM sessions WHERE id=?", (sid,))
        con.commit(); con.close()
    return jsonify({"ok": True, "deleted": sid, "chunks_removed": len(ids)})


@app.route("/docs/<int:doc_id>")
def doc_get_route(doc_id):
    con = db_conn()
    row = con.execute(
        "SELECT id,session_id,source,source_type,title,full_text,bytes,created_at "
        "FROM documents WHERE id=?", (doc_id,)).fetchone()
    con.close()
    if not row:
        return jsonify({"error": "not found"}), 404
    keys = ["id","session_id","source","source_type","title","full_text","bytes","created_at"]
    return jsonify(dict(zip(keys, row)))


@app.route("/process-links", methods=["POST"])
def process_links_route():
    """Body: { session_id, urls: [...] OR raw_text: "..." }"""
    body = request.get_json(force=True, silent=True) or {}
    sid = body.get("session_id")
    if not sid or not session_meta(sid):
        return jsonify({"error": "session_id invalid"}), 400
    raw = body.get("raw_text") or ""
    urls = body.get("urls") or []
    if raw and not urls:
        urls = parse_links_from_text(raw)
    # final clean + dedupe
    seen = set(); cleaned = []
    for u in urls:
        c = clean_url(u)
        if c and c not in seen:
            seen.add(c); cleaned.append(c)

    def gen():
        yield sse("status", {"message": f"{len(cleaned)} URL-uri valide dupa filtrare"})
        if not cleaned:
            yield sse("done", {"docs": 0, "chunks": 0}); return
        yield sse("urls", {"urls": cleaned})

        results_q: queue.Queue = queue.Queue()
        done_count = [0]

        def work(u):
            res = extract_one(u)
            results_q.put(res)

        executor = ThreadPoolExecutor(max_workers=EXTRACT_WORKERS)
        futs = [executor.submit(work, u) for u in cleaned]

        ok = err = 0
        total_chunks = 0
        while done_count[0] < len(cleaned):
            try:
                res = results_q.get(timeout=1)
            except queue.Empty:
                yield ": keepalive\n\n"
                continue
            done_count[0] += 1
            if res.get("error"):
                err += 1
                yield sse("doc_err", {"url": res["url"], "kind": res.get("kind"),
                                      "error": res["error"], "i": done_count[0],
                                      "total": len(cleaned)})
                continue
            try:
                n = store_doc(sid, res["url"], res["kind"], res["title"], res["text"])
                ok += 1; total_chunks += n
                yield sse("doc_ok", {
                    "url": res["url"], "kind": res["kind"],
                    "title": res["title"][:140], "chunks": n,
                    "i": done_count[0], "total": len(cleaned),
                })
            except Exception as e:
                err += 1
                yield sse("doc_err", {"url": res["url"], "error": f"store: {e}",
                                      "i": done_count[0], "total": len(cleaned)})
        executor.shutdown(wait=False)
        session_update_counts(sid)
        yield sse("done", {"docs_ok": ok, "docs_err": err, "chunks": total_chunks})

    return Response(gen(), mimetype="text/event-stream",
                    headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})


@app.route("/process-texts", methods=["POST"])
def process_texts_route():
    """multipart: session_id + files[] (oricate .txt)."""
    sid = request.form.get("session_id")
    if not sid or not session_meta(sid):
        return jsonify({"error": "session_id invalid"}), 400
    files = request.files.getlist("files")
    items = []
    for f in files:
        try:
            data = f.read()
            text, enc = smart_decode(data)
            # normalizare: drop NULs reziduale + replacement chars
            text = text.replace("\x00", "").replace("", "")
            items.append({"name": f.filename, "text": text, "encoding": enc,
                          "bytes": len(data)})
        except Exception as e:
            items.append({"name": f.filename, "error": str(e)})

    def gen():
        yield sse("status", {"message": f"{len(items)} fisier(e) primite"})
        ok = err = 0; total_chunks = 0
        for i, it in enumerate(items, 1):
            if it.get("error"):
                err += 1
                yield sse("doc_err", {"url": it["name"], "error": it["error"],
                                      "i": i, "total": len(items)}); continue
            text = (it.get("text") or "").strip()
            if len(text) < 80:
                err += 1
                yield sse("doc_err", {"url": it["name"], "error": "text prea scurt",
                                      "i": i, "total": len(items)}); continue
            # avertisment daca au ramas multe replacement chars (decodare imperfecta)
            repl = text.count("�")
            if repl > 0 and repl > len(text) * 0.01:
                yield sse("status", {"message": f"atentie: {it['name']} are {repl} caractere neidentificate (enc={it.get('encoding')})"})
            try:
                n = store_doc(sid, it["name"], "manual", it["name"], text)
                ok += 1; total_chunks += n
                yield sse("doc_ok", {"url": it["name"], "kind": "manual",
                                     "title": it["name"], "chunks": n,
                                     "encoding": it.get("encoding"),
                                     "i": i, "total": len(items)})
            except Exception as e:
                err += 1
                yield sse("doc_err", {"url": it["name"], "error": str(e),
                                      "i": i, "total": len(items)})
        session_update_counts(sid)
        yield sse("done", {"docs_ok": ok, "docs_err": err, "chunks": total_chunks})

    return Response(gen(), mimetype="text/event-stream",
                    headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})


@app.route("/query", methods=["POST"])
def query_route():
    body = request.get_json(force=True, silent=True) or {}
    sid = body.get("session_id"); q = (body.get("question") or "").strip()
    if not sid or not session_meta(sid):
        return jsonify({"error": "session_id invalid"}), 400
    if not q:
        return jsonify({"error": "intrebare goala"}), 400

    def gen():
        yield sse("status", {"message": "caut chunk-uri relevante..."})
        try:
            hits = search_chunks(sid, q, TOP_K)
        except Exception as e:
            yield sse("error", {"message": f"search: {e}"})
            yield sse("done", {}); return
        if not hits:
            yield sse("error", {"message": "niciun chunk in sesiune"})
            yield sse("done", {}); return
        yield sse("hits", {"hits": [
            {"doc_id": h["doc_id"], "chunk_idx": h["chunk_idx"],
             "title": h["title"], "source": h["source"],
             "source_type": h["source_type"],
             "distance": h["distance"],
             "preview": h["text"][:240],
             "text": h["text"]}
            for h in hits]})
        prompt = build_query_prompt(q, hits)
        yield sse("status", {"message": f"trimit la claude {CLAUDE_MODEL} ({len(hits)} fragmente)..."})
        out_q: queue.Queue = queue.Queue()
        threading.Thread(target=stream_claude, args=(prompt, out_q), daemon=True).start()
        while True:
            try:
                ev, data = out_q.get(timeout=5)
            except queue.Empty:
                yield ": keepalive\n\n"; continue
            yield sse(ev, data)
            if ev == "done": break

    return Response(gen(), mimetype="text/event-stream",
                    headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})


if __name__ == "__main__":
    print(f"LinkRAG pornit pe http://{HOST}:{PORT}")
    app.run(host=HOST, port=PORT, threaded=True, debug=False)