From 84ae6cfa9d6cb55aeee8be5218728bd9f16be051 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9CNaeel=E2=80=9D?= Date: Tue, 9 Jun 2026 06:58:57 +0400 Subject: [PATCH] ElsaWin RAG pipeline: full parsing chain + ChromaDB ingest + docs --- scripts/audit_elsa_en.py | 117 ++++++++++++++++++ scripts/check_slp_en.py | 54 +++++++++ scripts/parse_elsa_htm_en.py | 124 +++++++++++++++++++ scripts/parse_elsa_mdb.py | 224 +++++++++++++++++++++++++++++++++++ scripts/parse_elsa_wi.py | 131 ++++++++++++++++++++ scripts/rag_ingest.py | 125 +++++++++++++++++++ scripts/rag_ingest_cpu.py | 85 +++++++++++++ scripts/rag_ingest_full.py | 115 ++++++++++++++++++ scripts/rag_ingest_safe.py | 92 ++++++++++++++ scripts/rag_query.py | 107 +++++++++++++++++ scripts/rag_test.py | 54 +++++++++ 11 files changed, 1228 insertions(+) create mode 100644 scripts/audit_elsa_en.py create mode 100644 scripts/check_slp_en.py create mode 100644 scripts/parse_elsa_htm_en.py create mode 100644 scripts/parse_elsa_mdb.py create mode 100644 scripts/parse_elsa_wi.py create mode 100644 scripts/rag_ingest.py create mode 100644 scripts/rag_ingest_cpu.py create mode 100644 scripts/rag_ingest_full.py create mode 100644 scripts/rag_ingest_safe.py create mode 100644 scripts/rag_query.py create mode 100644 scripts/rag_test.py diff --git a/scripts/audit_elsa_en.py b/scripts/audit_elsa_en.py new file mode 100644 index 0000000..ea3ad78 --- /dev/null +++ b/scripts/audit_elsa_en.py @@ -0,0 +1,117 @@ +#!/usr/bin/env python3 +"""Полный аудит Elsa: WI, hs2, www — какие типы, что извлекаемо.""" +import os, struct, re + +elsa = os.path.expanduser("~/nubes/data/elsa") + +# 1. WI файлы +print("=" * 60) +print("1. WI-файлы (OLE2 compound documents)") +print("=" * 60) + +docs_rl = os.path.join(elsa, "docs", "rl") +wi_files = [f for f in os.listdir(docs_rl) if f.endswith(".wi")] +wi_en = [f for f in wi_files if "en-GB" in f] +print(f"Total WI: {len(wi_files)}, en-GB: {len(wi_en)}") + +if wi_en: + path = os.path.join(docs_rl, wi_en[0]) + with open(path, "rb") as f: + data = f.read(min(10000, os.path.getsize(path))) + + print(f"\nSample: {wi_en[0]}") + print(f"Size: {len(data)} bytes (shown), OLE2: {data[:4] == b'\xd0\xcf\x11\xe0'}") + print(f"Has ", b"= 0: + ctx = data[pos:pos+80] + readable = ctx.decode("utf-16-le", errors="replace") + print(f" Found '{tag.decode()}' at offset {pos}: {readable[:80]}") + + +# 2. hs2 en-GB +print("\n" + "=" * 60) +print("2. hs2/V/en-GB/ — HTML контент") +print("=" * 60) + +hs2_en = os.path.join(elsa, "docs", "hs2", "V", "en-GB") +total_hs2 = 0 +samples = [] +for root, dirs, files in os.walk(hs2_en): + for f in files: + if f.endswith(".htm"): + total_hs2 += 1 + if len(samples) < 3: + samples.append(os.path.join(root, f)) + +print(f"Total HTM: {total_hs2}") + +for sp in samples: + with open(sp, "rb") as fh: + raw = fh.read(4096) + text = raw.decode("utf-16-le", errors="replace") + m = re.search(r"(.*?)", text, re.IGNORECASE) + title = m.group(1) if m else "(no title)" + body = re.sub(r"<[^>]+>", " ", text)[:200] + body = re.sub(r"\s+", " ", body).strip() + dir_name = sp.split("/")[-2] + print(f" [{dir_name}] {title}") + print(f" {body[:150]}") + print() + +# 3. www en-GB +print("=" * 60) +print("3. www/V/en-GB/ — навигация и контент") +print("=" * 60) + +www_en = os.path.join(elsa, "docs", "www", "V", "en-GB") +if os.path.exists(www_en): + items = os.listdir(www_en) + print(f"Items: {items}") + for item in items: + ipath = os.path.join(www_en, item) + if os.path.isfile(ipath) and item.endswith(".htm"): + print(f" {item} — {os.path.getsize(ipath)} bytes") + +# 4. WI en-GB in other dirs +print("\n" + "=" * 60) +print("4. WI en-GB в других директориях") +print("=" * 60) + +for subdir in ["igg", "au", "ki"]: + d = os.path.join(elsa, "docs", subdir) + if os.path.exists(d): + total = 0 + en = 0 + for f in os.listdir(d): + if f.endswith(".wi"): + total += 1 + if "en-GB" in f: + en += 1 + print(f" {subdir}: {total} WI total, {en} en-GB") + +# 5. Summary +print("\n" + "=" * 60) +print("ИТОГО: что можно извлечь en-GB") +print("=" * 60) +print(""" +MDB (уже парсено): + rldal.V.en-GB.mdb — 191k док-тов, иерархия ✅ + ipsvrap.mdb — 2.5M строк, каталог запчастей ✅ + + dbsvrfi, dbsvrfz — справочники авто (не тронуты) + +HTM (настоящие): + au/V/en-GB/ — 1295 HTM (UTF-16) — уже в elsa_jsonl ✅ + hs2/V/en-GB/ — 4473 HTM (UTF-16) — НЕ ПАРСЕНЫ ❗ + www/.../en-GB/ — 2538 HTM (ASCII) — НЕ ПАРСЕНЫ ❗ + slp/V/en-GB/ — 10845 (x64 code) — МУСОР ❌ + +WI (OLE2): + rl/ — 1716 en-GB WI-файлов — НЕ ПАРСЕНЫ ❗ + igg/ — ? en-GB WI + au/ — ? en-GB WI +""") diff --git a/scripts/check_slp_en.py b/scripts/check_slp_en.py new file mode 100644 index 0000000..b3bdc86 --- /dev/null +++ b/scripts/check_slp_en.py @@ -0,0 +1,54 @@ +#!/usr/bin/env python3 +"""Проверить что за файлы в slp/V/en-GB/*.htm""" +import zlib, re, os, struct + +path = "/home/naeel/nubes/data/elsa/docs/slp/V/en-GB/000502200001.htm" +with open(path, "rb") as f: + raw = f.read() + +print(f"File size: {len(raw)} bytes") +print(f"First 64 hex: {raw[:64].hex()}") + +# PE check +if raw[:2] == b"MZ": + print(">>> It's a PE/EXE/DLL file!") +elif raw[:2] == b"\x48\xc7": + print(">>> Starts with 48 c7 - could be x64 machine code!") + +# Look for ELF +if raw[:4] == b"\x7fELF": + print(">>> It's an ELF binary!") + +# Try full zlib decompress at different offsets +for offset in range(0, min(500, len(raw))): + if raw[offset] == 0x78 and raw[offset+1] in [0x01, 0x9c, 0xda]: + try: + decomp = zlib.decompress(raw[offset:]) + print(f">>> zlib at offset {offset}: {len(decomp)} bytes decompressed") + # Try decode as UTF-16LE + text = decomp.decode("utf-16-le", errors="replace") + m = re.search(r"(.*?)", text, re.IGNORECASE) + if m: + print(f" Title: {m.group(1)}") + body = re.sub(r"<[^>]+>", " ", text)[:500] + body = re.sub(r"\s+", " ", body).strip() + print(f" Text: {body[:200]}") + break + except: + pass +else: + print(">>> No zlib content found") + +# Also check ru-RU for comparison +ru_path = "/home/naeel/nubes/data/elsa/docs/slp/V/ru-RU" +if os.path.exists(ru_path): + ru_files = [f for f in os.listdir(ru_path) if f.endswith(".htm")] + if ru_files: + ru_file = os.path.join(ru_path, ru_files[0]) + with open(ru_file, "rb") as f: + ru_raw = f.read(64) + print(f"\n>>> For comparison, ru-RU first 64 hex: {ru_raw.hex()}") + # Check encoding + import subprocess + result = subprocess.run(["file", ru_file], capture_output=True, text=True) + print(f" file says: {result.stdout.strip()}") diff --git a/scripts/parse_elsa_htm_en.py b/scripts/parse_elsa_htm_en.py new file mode 100644 index 0000000..b71dbe8 --- /dev/null +++ b/scripts/parse_elsa_htm_en.py @@ -0,0 +1,124 @@ +#!/usr/bin/env python3 +"""Парсер en-GB HTM файлов ElsaWin: hs2 + www""" +import os, re, json, sys +from pathlib import Path + +def parse_htm_file(path): + """Парсить один HTM-файл в UTF-16LE.""" + with open(path, "rb") as f: + raw = f.read() + + # Проверяем кодировку + if raw[:2] == b"\xff\xfe": + text = raw.decode("utf-16-le", errors="replace") + elif raw[:2] == b"\xfe\xff": + text = raw.decode("utf-16-be", errors="replace") + else: + text = raw.decode("ascii", errors="replace") + + # Title + m = re.search(r"(.*?)", text, re.IGNORECASE | re.DOTALL) + title = m.group(1).strip() if m else "" + + # Body text + body = re.sub(r"]*>.*?", " ", text, flags=re.IGNORECASE | re.DOTALL) + body = re.sub(r"]*>.*?", " ", body, flags=re.IGNORECASE | re.DOTALL) + body = re.sub(r"<[^>]+>", " ", body) + body = re.sub(r" ", " ", body) + body = re.sub(r"&\w+;", " ", body) + body = re.sub(r"\s+", " ", body).strip() + + return {"title": title, "text": body} + + +def main(): + if len(sys.argv) < 3: + print(f"Usage: {sys.argv[0]} ") + sys.exit(1) + + elsa_docs = os.path.abspath(sys.argv[1]) + out_dir = os.path.abspath(sys.argv[2]) + os.makedirs(out_dir, exist_ok=True) + + en_gb_dirs = [ + # hs2 — in nested structure: hs2/V/en-GB/src/XX/XXXXXX/master.htm + os.path.join(elsa_docs, "hs2", "V", "en-GB"), + # www — various paths under www/.../en-GB/ + ] + + # Find all en-GB HTM + all_files = [] + for root, dirs, files in os.walk(elsa_docs): + if "/en-GB/" in root and not "/slp/" in root: + for f in files: + if f.endswith(".htm") or f.endswith(".html"): + all_files.append(os.path.join(root, f)) + + # Skip already-known-malformed (slp — machine code, already excluded above) + # Only au, hs2, www, igg + good_files = [f for f in all_files if any(d in f for d in ["/www/", "/hs2/", "/au/"])] + + print(f"Found {len(good_files)} en-GB HTM files (excluding slp)") + + # Split by source + by_source = {} + for f in good_files: + if "/au/" in f: + src = "au" + elif "/hs2/" in f: + src = "hs2" + elif "/www/" in f: + src = "www" + else: + src = "other" + by_source.setdefault(src, []).append(f) + + for src, files in by_source.items(): + print(f" {src}: {len(files)} files") + + # Parse and save + total = 0 + part_size = 5000 + part_idx = 0 + buf = [] + out_file = None + + for src, files in by_source.items(): + for fpath in sorted(files): + try: + doc = parse_htm_file(fpath) + rel = os.path.relpath(fpath, elsa_docs) + record = { + "source": f"elsa_{src}", + "file": "/docs/" + rel, + "title": doc["title"], + "text": doc["text"], + } + buf.append(record) + total += 1 + + if len(buf) >= part_size: + out_file = os.path.join(out_dir, f"part_{part_idx:04d}.jsonl") + with open(out_file, "w", encoding="utf-8") as f: + for r in buf: + f.write(json.dumps(r, ensure_ascii=False) + "\n") + print(f" Part {part_idx:04d}: {len(buf)} records → {out_file}") + buf = [] + part_idx += 1 + except Exception as e: + print(f" ERROR {fpath}: {e}", file=sys.stderr) + + # Flush remaining + if buf: + out_file = os.path.join(out_dir, f"part_{part_idx:04d}.jsonl") + with open(out_file, "w", encoding="utf-8") as f: + for r in buf: + f.write(json.dumps(r, ensure_ascii=False) + "\n") + print(f" Part {part_idx:04d}: {len(buf)} records → {out_file}") + part_idx += 1 + + print(f"\nTotal: {total} documents in {part_idx} parts → {out_dir}") + + +if __name__ == "__main__": + main() diff --git a/scripts/parse_elsa_mdb.py b/scripts/parse_elsa_mdb.py new file mode 100644 index 0000000..76effbb --- /dev/null +++ b/scripts/parse_elsa_mdb.py @@ -0,0 +1,224 @@ +#!/usr/bin/env python3 +""" +Парсер MDB-баз ElsaWin → JSONL для RAG. +Использует mdbtools (mdb-export) для выгрузки таблиц. + +Использование: + python3 parse_elsa_mdb.py <путь_к_mdb> <выходная_директория> + +Пример: + python3 parse_elsa_mdb.py ~/nubes/data/elsa/data/rldal.V.en-GB.mdb ~/nubes/data/elsa_mdb_jsonl/ +""" + +import csv +import io +import json +import os +import subprocess +import sys +from pathlib import Path + + +def mdb_tables(mdb_path: str) -> list[str]: + """Получить список таблиц из MDB через mdb-tables.""" + result = subprocess.run( + ["mdb-tables", mdb_path], + capture_output=True, text=True, check=True + ) + return result.stdout.strip().split() + + +def mdb_export(mdb_path: str, table: str) -> str: + """Экспорт таблицы в CSV через mdb-export.""" + result = subprocess.run( + ["mdb-export", mdb_path, table], + capture_output=True, text=True, check=True + ) + return result.stdout + + +def parse_csv(csv_text: str) -> list[dict]: + """Распарсить CSV в список словарей.""" + reader = csv.DictReader(io.StringIO(csv_text)) + return [row for row in reader] + + +def export_table_to_jsonl(mdb_path: str, table: str, output_dir: str) -> str: + """Экспорт одной таблицы в JSONL.""" + print(f" {table}...", end=" ", flush=True) + csv_text = mdb_export(mdb_path, table) + rows = parse_csv(csv_text) + + out_file = os.path.join(output_dir, f"{table}.jsonl") + with open(out_file, "w", encoding="utf-8") as f: + for row in rows: + # Очистка значений: убираем лишние пробелы + cleaned = {k: v.strip() if v else "" for k, v in row.items()} + f.write(json.dumps(cleaned, ensure_ascii=False) + "\n") + + print(f"{len(rows)} rows → {out_file}") + return out_file + + +def build_wi_dokument_docs(mdb_path: str, output_dir: str): + """ + Построить документы из wi_dokument с полной иерархией. + Каждый документ = одна строка JSONL с полным путём разделов. + """ + print("\n=== Построение документов из wi_dokument ===") + + csv_text = mdb_export(mdb_path, "wi_dokument") + rows = parse_csv(csv_text) + + docs = [] + for row in rows: + # Собираем полный путь иерархии + hierarchy_parts = [] + for key in ["og_bez", "bg_bez", "rg_bez", "hkap_bez", "kap_bez", "ukap_bez"]: + val = row.get(key, "").strip() + if val: + hierarchy_parts.append(val) + + full_path = " → ".join(hierarchy_parts) if hierarchy_parts else "" + + doc = { + "dokument_id": int(row.get("dokument_id", 0)), + "spk": row.get("spk", "").strip(), + "redsystyp": int(row.get("redsystyp", 0)), + "name": row.get("name", "").strip(), + "hierarchy": { + "og_id": int(row.get("og_id", 0)) if row.get("og_id", "").strip() else None, + "bg_id": int(row.get("bg_id", 0)) if row.get("bg_id", "").strip() else None, + "rg_id": row.get("rg_id", "").strip(), + "hkap_id": int(row.get("hkap_id", 0)) if row.get("hkap_id", "").strip() else None, + "kap_id": int(row.get("kap_id", 0)) if row.get("kap_id", "").strip() else None, + "ukap_id": int(row.get("ukap_id", 0)) if row.get("ukap_id", "").strip() else None, + }, + "titles": { + "og": row.get("og_bez", "").strip(), + "bg": row.get("bg_bez", "").strip(), + "rg": row.get("rg_bez", "").strip(), + "hkap": row.get("hkap_bez", "").strip(), + "kap": row.get("kap_bez", "").strip(), + "ukap": row.get("ukap_bez", "").strip(), + }, + "full_title": full_path, + "parent_dokument_id": int(row.get("parent_dokument_id", 0)) if row.get("parent_dokument_id", "").strip() else None, + "ext_id": row.get("ext_id", "").strip(), + "source": "elsa", + "type": "wi_dokument", + } + docs.append(doc) + + out_file = os.path.join(output_dir, "wi_dokument_enriched.jsonl") + with open(out_file, "w", encoding="utf-8") as f: + for doc in docs: + f.write(json.dumps(doc, ensure_ascii=False) + "\n") + + print(f"Построено {len(docs)} документов → {out_file}") + + # Статистика по иерархии + with_titles = sum(1 for d in docs if d["full_title"]) + print(f" Из них с иерархией: {with_titles}") + print(f" Без иерархии: {len(docs) - with_titles}") + + return out_file + + +def build_vehicle_lookup(mdb_path: str, output_dir: str): + """ + Построить справочник привязки документов к автомобилям. + """ + print("\n=== Построение привязки документов к автомобилям ===") + + # wi_dokument_fzg + csv_text = mdb_export(mdb_path, "wi_dokument_fzg") + rows = parse_csv(csv_text) + + # Группируем по dokument_id + from collections import defaultdict + by_doc = defaultdict(list) + for row in rows: + entry = { + "vtyp": row.get("vtyp", "").strip(), + "marke": row.get("marke", "").strip(), + "mkb": row.get("mkb", "").strip(), + "gkb": row.get("gkb", "").strip(), + "gtyp": row.get("gtyp", "").strip(), + "mj_von": int(row.get("mj_von", 0)) if row.get("mj_von", "").strip() else None, + "mj_bis": int(row.get("mj_bis", 0)) if row.get("mj_bis", "").strip() else None, + } + by_doc[int(row["dokument_id"])].append(entry) + + out_file = os.path.join(output_dir, "wi_dokument_fzg_grouped.jsonl") + with open(out_file, "w", encoding="utf-8") as f: + for doc_id, vehicles in by_doc.items(): + f.write(json.dumps({ + "dokument_id": doc_id, + "vehicles": vehicles, + "count": len(vehicles), + }, ensure_ascii=False) + "\n") + + print(f"{len(by_doc)} документов с привязкой к авто → {out_file}") + return out_file + + +def build_hierarchy_reference(mdb_path: str, output_dir: str): + """ + Построить справочники иерархии: og, bg, rg с vehicle-привязкой. + """ + print("\n=== Построение справочников иерархии ===") + + for tbl, name in [("wi_og_fzg", "og"), ("wi_bg_fzg", "bg"), ("wi_rg_fzg", "rg")]: + csv_text = mdb_export(mdb_path, tbl) + rows = parse_csv(csv_text) + out_file = os.path.join(output_dir, f"{tbl}.jsonl") + with open(out_file, "w", encoding="utf-8") as f: + for row in rows: + cleaned = {k: v.strip() if v else "" for k, v in row.items()} + f.write(json.dumps(cleaned, ensure_ascii=False) + "\n") + print(f" {tbl}: {len(rows)} rows → {out_file}") + + +def main(): + if len(sys.argv) < 3: + print(__doc__) + sys.exit(1) + + mdb_path = os.path.abspath(sys.argv[1]) + output_dir = os.path.abspath(sys.argv[2]) + + if not os.path.exists(mdb_path): + print(f"ОШИБКА: MDB файл не найден: {mdb_path}") + sys.exit(1) + + os.makedirs(output_dir, exist_ok=True) + + print(f"MDB: {mdb_path}") + print(f"Выход: {output_dir}") + + # 1. Список таблиц + tables = mdb_tables(mdb_path) + print(f"\nТаблицы ({len(tables)}): {', '.join(tables)}") + + # 2. Экспорт каждой таблицы в сырой JSONL + print("\n=== Экспорт таблиц ===") + for table in tables: + try: + export_table_to_jsonl(mdb_path, table, output_dir) + except subprocess.CalledProcessError as e: + print(f"ОШИБКА экспорта {table}: {e}") + + # 3. Построение обогащённых документов (rldal-specific) + if "wi_dokument" in tables: + build_wi_dokument_docs(mdb_path, output_dir) + build_vehicle_lookup(mdb_path, output_dir) + build_hierarchy_reference(mdb_path, output_dir) + else: + print("\n=== Пропускаем обогащение (нет wi_dokument таблицы) ===") + + print(f"\n✅ Готово. Результат: {output_dir}") + + +if __name__ == "__main__": + main() diff --git a/scripts/parse_elsa_wi.py b/scripts/parse_elsa_wi.py new file mode 100644 index 0000000..58cba92 --- /dev/null +++ b/scripts/parse_elsa_wi.py @@ -0,0 +1,131 @@ +#!/usr/bin/env python3 +"""Парсер WI-файлов ElsaWin (OLE2 → XML → JSONL)""" +import os, re, json, sys +import olefile + +def parse_wi_file(path): + """Извлечь XML-потоки из WI (OLE2) файла.""" + docs = [] + + with olefile.OleFileIO(path) as ole: + for stream in ole.listdir(): + stream_name = "/".join(stream) + if not stream_name.endswith(".xml"): + continue + + data = ole.openstream(stream).read() + text = data.decode("utf-8", errors="replace") + + # Extract key elements + title_parts = [] + + m = re.search(r"(.*?)", text, re.IGNORECASE) + marke = m.group(1) if m else "" + + m = re.search(r"(.*?)", text, re.IGNORECASE) + modell = m.group(1) if m else "" + + m = re.search(r"(.*?)", text, re.IGNORECASE) + obergrup = m.group(1) if m else "" + + m = re.search(r"(.*?)", text, re.IGNORECASE) + baugrup = m.group(1) if m else "" + + # Название документа + m = re.search(r"]*?typen-bez=\"([^\"]+)\"", text) + typen = m.group(1) if m else "" + + # Title from vorspann + title = " ".join(p for p in [obergrup, baugrup] if p) + if marke: + title = f"{marke} {modell}: {title}" if modell else f"{marke}: {title}" + + # Clean body + body = re.sub(r"<[^>]+>", " ", text) + body = re.sub(r"\s+", " ", body).strip() + + docs.append({ + "title": title, + "marke": marke, + "modell": modell, + "obergrup": obergrup, + "baugrup": baugrup, + "text": body, + }) + + return docs + + +def main(): + if len(sys.argv) < 3: + print(f"Usage: {sys.argv[0]} ") + sys.exit(1) + + elsa_docs = os.path.abspath(sys.argv[1]) + out_dir = os.path.abspath(sys.argv[2]) + os.makedirs(out_dir, exist_ok=True) + + # Collect en-GB WI files + wi_files = [] + for root, dirs, files in os.walk(elsa_docs): + for f in files: + if f.endswith(".wi") and "en-GB" in f: + wi_files.append(os.path.join(root, f)) + + print(f"Found {len(wi_files)} en-GB WI files") + + # By category + by_cat = {} + for f in wi_files: + if "/rl/" in f: + cat = "rl" + elif "/igg/" in f: + cat = "igg" + elif "/ki/" in f: + cat = "ki" + else: + cat = "other" + by_cat.setdefault(cat, []).append(f) + + for cat, files in by_cat.items(): + print(f" {cat}: {len(files)} files") + + # Parse + total_docs = 0 + part_size = 5000 + part_idx = 0 + buf = [] + + for cat, files in by_cat.items(): + for fpath in sorted(files): + try: + docs = parse_wi_file(fpath) + for doc in docs: + doc["source"] = f"elsa_wi_{cat}" + doc["file"] = os.path.relpath(fpath, elsa_docs) + buf.append(doc) + total_docs += 1 + except Exception as e: + print(f" ERROR {fpath}: {e}", file=sys.stderr) + + if len(buf) >= part_size: + out_file = os.path.join(out_dir, f"part_{part_idx:04d}.jsonl") + with open(out_file, "w", encoding="utf-8") as f: + for r in buf: + f.write(json.dumps(r, ensure_ascii=False) + "\n") + print(f" Part {part_idx:04d}: {len(buf)} → {out_file}") + buf = [] + part_idx += 1 + + if buf: + out_file = os.path.join(out_dir, f"part_{part_idx:04d}.jsonl") + with open(out_file, "w", encoding="utf-8") as f: + for r in buf: + f.write(json.dumps(r, ensure_ascii=False) + "\n") + print(f" Part {part_idx:04d}: {len(buf)} → {out_file}") + + print(f"\nTotal: {total_docs} XML documents from {len(wi_files)} WI files → {out_dir}") + + +if __name__ == "__main__": + main() diff --git a/scripts/rag_ingest.py b/scripts/rag_ingest.py new file mode 100644 index 0000000..c142ed3 --- /dev/null +++ b/scripts/rag_ingest.py @@ -0,0 +1,125 @@ +#!/usr/bin/env python3 +"""Ingest JSONL into ChromaDB — uses llama.cpp /embedding endpoint.""" +import json, os, sys, glob, urllib.request, time +import chromadb + +LLAMA_URL = "http://localhost:8081/embedding" + +def llama_embed(texts): + """Get embeddings from nomic-embed server with retry.""" + data = json.dumps({"content": texts}).encode() + req = urllib.request.Request(LLAMA_URL, data=data, + headers={"Content-Type": "application/json"}) + for attempt in range(3): + try: + with urllib.request.urlopen(req, timeout=120) as resp: + body = json.loads(resp.read()) + if isinstance(body, list): + return [item["embedding"][0] for item in body] + return body.get("embedding", []) + except Exception as e: + if attempt < 2: + print(f"[retry {attempt+1}] {e}") + time.sleep(5) + else: + raise + + +class LlamaEmbedFn: + """Custom embedding function using llama.cpp.""" + def __call__(self, input): + all_embs = [] + for i in range(0, len(input), 32): + batch = input[i:i+32] + embs = llama_embed(batch) + if isinstance(embs[0], float): + all_embs.append(embs) + else: + all_embs.extend(embs) + return all_embs + + +def main(): + jsonl_dir = sys.argv[1] if len(sys.argv) > 1 else os.path.expanduser("~/nubes/data/elsa_jsonl_wi_en") + db_dir = sys.argv[2] if len(sys.argv) > 2 else os.path.expanduser("~/nubes/chroma_db") + collection_name = sys.argv[3] if len(sys.argv) > 3 else "elsa_docs" + + ef = LlamaEmbedFn() + + client = chromadb.PersistentClient(path=db_dir) + + try: + client.delete_collection(collection_name) + except: + pass + + collection = client.create_collection( + name=collection_name, + embedding_function=ef, + metadata={"hnsw:space": "cosine"} + ) + + jsonl_files = sorted(glob.glob(os.path.join(jsonl_dir, "*.jsonl"))) + print(f"Found {len(jsonl_files)} JSONL files") + print(f"Using llama.cpp at {LLAMA_URL}") + + total = 0 + for jf in jsonl_files: + ids, docs, metas = [], [], [] + + with open(jf, "r", encoding="utf-8") as f: + for line in f: + try: + d = json.loads(line) + title = d.get("title", "").strip() + text = d.get("text", "").strip() + ft = d.get("full_title", "").strip() + + if ft and text: + doc = f"{ft}\n{text}" + elif title and text: + doc = f"{title}\n{text}" + else: + doc = text or title or ft + + if not doc or len(doc) < 20: + continue + + # Split long docs into chunks of ~1000 chars + chunks = [doc[i:i+1000] for i in range(0, len(doc), 1000)] + for chunk in chunks: + if len(chunk) < 20: + continue + ids.append(str(total)) + docs.append(chunk) + metas.append({ + "source": d.get("source", ""), + "file": d.get("file", ""), + "title": title or ft or doc[:100], + }) + total += 1 + except: + continue + + if ids: + # Send in smaller batches of 500 + for i in range(0, len(ids), 500): + batch_ids = ids[i:i+500] + batch_docs = docs[i:i+500] + batch_metas = metas[i:i+500] + for attempt in range(3): + try: + collection.add(ids=batch_ids, documents=batch_docs, metadatas=batch_metas) + break + except Exception as e: + if attempt < 2: + print(f" [retry {os.path.basename(jf)} batch {attempt+1}] {e}") + time.sleep(10) + else: + print(f" [SKIP batch] {e}") + print(f" {os.path.basename(jf)}: +{len(ids)} (total {total})") + + print(f"\nDone. {total} documents → {db_dir}/{collection_name}") + +if __name__ == "__main__": + main() diff --git a/scripts/rag_ingest_cpu.py b/scripts/rag_ingest_cpu.py new file mode 100644 index 0000000..089b570 --- /dev/null +++ b/scripts/rag_ingest_cpu.py @@ -0,0 +1,85 @@ +#!/usr/bin/env python3 +"""Полная индексация через sentence-transformers (CPU, стабильно).""" +import json, os, sys, time, glob, shutil +from chromadb import PersistentClient +from chromadb.utils import embedding_functions + +ALL_DIRS = [ + "~/nubes/data/elsa_jsonl_wi_en", + "~/nubes/data/elsa_mdb_jsonl", + "~/nubes/data/elsa_mdb_jsonl/ipsvrap", + "~/nubes/data/elsa_jsonl_htm_en", + "~/nubes/data/elsa_jsonl", + "~/nubes/data/elsa_mdb_jsonl/dbsvrfi", + "~/nubes/data/elsa_mdb_jsonl/dbsvrfz", +] + +DB_DIR = os.path.expanduser("~/nubes/chroma_db") +MODEL = "all-MiniLM-L6-v2" # 90 MB, CPU + +ef = embedding_functions.SentenceTransformerEmbeddingFunction(model_name=MODEL) + +if os.path.exists(DB_DIR): + shutil.rmtree(DB_DIR) + print("Old chroma_db deleted") + +client = PersistentClient(path=DB_DIR) +collection = client.create_collection("elsa_docs", embedding_function=ef) + +all_files = [] +for d in ALL_DIRS: + path = os.path.expanduser(d) + files = sorted(glob.glob(os.path.join(path, "*.jsonl"))) + all_files.extend(files) + print(f"{d}: {len(files)} files") + +print(f"\nTotal: {len(all_files)} JSONL files") + +total = 0 +for fpath in all_files: + docs, ids, metas = [], [], [] + prefix = os.path.basename(os.path.dirname(fpath)) + "_" + + try: + with open(fpath) as fh: + for line in fh: + try: + d = json.loads(line) + title = d.get("title", "") or d.get("full_title", "") + text = d.get("text", "") + doc = (title + "\n" + text).strip()[:1000] + if len(doc) < 20: + continue + docs.append(doc) + ids.append(f"{prefix}{os.path.basename(fpath)}_{len(docs)}") + metas.append({"source": prefix.strip("_")}) + except: + continue + except: + continue + + if not docs: + continue + + print(f"{os.path.basename(fpath)}: {len(docs)} docs", end="", flush=True) + + for j in range(0, len(docs), 500): + batch = docs[j:j+500] + batch_ids = ids[j:j+500] + for retry in range(3): + try: + collection.add(ids=batch_ids, documents=batch, + metadatas=metas[j:j+500]) + total += len(batch) + print(".", end="", flush=True) + break + except Exception as e: + if retry < 2: + print(f"R{retry}", end="", flush=True) + time.sleep(5) + else: + print(f"X", end="", flush=True) + + print(f" total={collection.count()}") + +print(f"\n✅ Done. Total: {collection.count()} docs") diff --git a/scripts/rag_ingest_full.py b/scripts/rag_ingest_full.py new file mode 100644 index 0000000..e7b090a --- /dev/null +++ b/scripts/rag_ingest_full.py @@ -0,0 +1,115 @@ +#!/usr/bin/env python3 +"""Полная переиндексация ВСЕХ данных в ChromaDB. Работает до победного.""" +import json, os, sys, time, urllib.request, glob +from chromadb import PersistentClient + +EF_URL = "http://localhost:8081/embedding" +DB_DIR = os.path.expanduser("~/nubes/chroma_db") + +ALL_DIRS = [ + "~/nubes/data/elsa_jsonl_wi_en", # 244k WI XML + "~/nubes/data/elsa_mdb_jsonl", # 191k rldal + "~/nubes/data/elsa_mdb_jsonl/ipsvrap", # 2.5M parts + "~/nubes/data/elsa_jsonl_htm_en", # 7k hs2+www + "~/nubes/data/elsa_jsonl", # 24k HTM + "~/nubes/data/elsa_mdb_jsonl/dbsvrfi", # car ref + "~/nubes/data/elsa_mdb_jsonl/dbsvrfz", # PR codes +] + +def embed(texts): + for attempt in range(5): + try: + data = json.dumps({"content": texts}).encode() + req = urllib.request.Request(EF_URL, data=data, + headers={"Content-Type": "application/json"}) + with urllib.request.urlopen(req, timeout=120) as resp: + body = json.loads(resp.read()) + if isinstance(body, list): + return [item["embedding"][0] for item in body] + return body.get("embedding", []) + except Exception as e: + print(f"\n [embed retry {attempt+1}/5] {e}") + time.sleep(15) + raise Exception("Embedding failed after 5 retries") + +def main(): + # Удалить старую БД и создать новую + import shutil + if os.path.exists(DB_DIR): + shutil.rmtree(DB_DIR) + print("Старая ChromaDB удалена") + + client = PersistentClient(path=DB_DIR) + collection = client.create_collection("elsa_docs") + + # Собираем все JSONL + all_files = [] + for d in ALL_DIRS: + path = os.path.expanduser(d) + files = sorted(glob.glob(os.path.join(path, "*.jsonl"))) + all_files.extend(files) + print(f"{d}: {len(files)} files") + + print(f"\nВсего {len(all_files)} JSONL файлов") + + total = 0 + errors = 0 + + for fpath in all_files: + docs, ids = [], [] + prefix = os.path.basename(os.path.dirname(fpath)) + "_" + + try: + with open(fpath) as fh: + for line in fh: + try: + d = json.loads(line) + title = d.get("title", "") or d.get("full_title", "") + text = d.get("text", "") + doc = (title + "\n" + text).strip()[:1000] + if len(doc) < 20: + continue + docs.append(doc) + ids.append(f"{prefix}{os.path.basename(fpath)}_{len(docs)}") + except: + continue + except Exception as e: + print(f" [SKIP {os.path.basename(fpath)}] read error: {e}") + errors += 1 + continue + + if not docs: + print(f" {os.path.basename(fpath)}: empty") + continue + + print(f" {os.path.basename(fpath)}: {len(docs)} docs", end="", flush=True) + + # Бачами по 50 + for j in range(0, len(docs), 50): + batch_docs = docs[j:j+50] + batch_ids = ids[j:j+50] + + for retry in range(5): + try: + embs = embed(batch_docs) + collection.add(ids=batch_ids, documents=batch_docs, + embeddings=embs, metadatas=[{"source": prefix.strip("_")}] * len(batch_docs)) + total += len(batch_docs) + print(".", end="", flush=True) + break + except Exception as e: + if retry < 4: + print(f"R{retry}", end="", flush=True) + time.sleep(20) + else: + print(f"X", end="", flush=True) + errors += len(batch_docs) + + time.sleep(0.2) + + print(f" total={collection.count()}") + + print(f"\n✅ Done. Total: {collection.count()} docs. Errors: {errors}") + +if __name__ == "__main__": + main() diff --git a/scripts/rag_ingest_safe.py b/scripts/rag_ingest_safe.py new file mode 100644 index 0000000..89a7db1 --- /dev/null +++ b/scripts/rag_ingest_safe.py @@ -0,0 +1,92 @@ +#!/usr/bin/env python3 +"""Ingest WI JSONL into ChromaDB — one file at a time, batches of 50.""" +import json, os, sys, time, urllib.request, glob +from chromadb import PersistentClient + +EF_URL = "http://localhost:8081/embedding" +DB_DIR = os.path.expanduser("~/nubes/chroma_db") +# Default: ingest ALL JSONL dirs +ALL_DIRS = [ + os.path.expanduser("~/nubes/data/elsa_jsonl_wi_en"), # 244k WI XML + os.path.expanduser("~/nubes/data/elsa_mdb_jsonl"), # 191k rldal + os.path.expanduser("~/nubes/data/elsa_mdb_jsonl/ipsvrap"), # 2.5M parts + os.path.expanduser("~/nubes/data/elsa_jsonl_htm_en"), # 7k hs2+www + os.path.expanduser("~/nubes/data/elsa_jsonl"), # 24k HTM + os.path.expanduser("~/nubes/data/elsa_mdb_jsonl/dbsvrfi"), # car ref + os.path.expanduser("~/nubes/data/elsa_mdb_jsonl/dbsvrfz"), # PR codes +] + +DATA_DIR = sys.argv[1] if len(sys.argv) > 1 else None +COLLECTION_NAME = "elsa_docs" + +# Connect to existing ChromaDB +client = PersistentClient(path=DB_DIR) +try: + collection = client.get_or_create_collection(COLLECTION_NAME) +except: + collection = client.create_collection(COLLECTION_NAME) + +existing = collection.count() +print(f"Collection has {existing} docs already") + +# Find remaining files +if DATA_DIR: + files = sorted(glob.glob(os.path.join(DATA_DIR, "*.jsonl"))) +else: + files = sorted(glob.glob(os.path.join(ALL_DIRS[0], "*.jsonl"))) + for d in ALL_DIRS[1:]: + files.extend(sorted(glob.glob(os.path.join(d, "*.jsonl")))) + +print(f"Processing {len(files)} JSONL files") + +for fpath in files: + docs, ids = [], [] + prefix = os.path.basename(os.path.dirname(fpath)) + "_" + with open(fpath) as fh: + for line in fh: + d = json.loads(line) + title = d.get("title", "") or d.get("full_title", "") + text = d.get("text", "") + doc = (title + "\n" + text).strip()[:1000] + if len(doc) < 20: + continue + docs.append(doc) + ids.append(f"{prefix}{os.path.basename(fpath)}_{len(docs)}") + + if not docs: + print(f"{os.path.basename(fpath)}: empty, skip") + continue + + print(f"{os.path.basename(fpath)}: {len(docs)} docs", end="") + + for j in range(0, len(docs), 50): + batch_docs = docs[j:j+50] + batch_ids = ids[j:j+50] + data = json.dumps({"content": batch_docs}).encode() + + for retry in range(3): + try: + req = urllib.request.Request(EF_URL, data=data, + headers={"Content-Type": "application/json"}) + with urllib.request.urlopen(req, timeout=120) as resp: + body = json.loads(resp.read()) + if isinstance(body, list): + embs = [item["embedding"][0] for item in body] + else: + embs = body.get("embedding", []) + + collection.add(ids=batch_ids, documents=batch_docs, embeddings=embs) + print(".", end="", flush=True) + break + except Exception as e: + if retry < 2: + print(f"R{retry}", end="", flush=True) + time.sleep(10) + else: + print(f"X({e})", end="", flush=True) + + time.sleep(0.3) + + print(f" total={collection.count()}") + +print(f"\nDone. Total: {collection.count()} documents") diff --git a/scripts/rag_query.py b/scripts/rag_query.py new file mode 100644 index 0000000..801dd80 --- /dev/null +++ b/scripts/rag_query.py @@ -0,0 +1,107 @@ +#!/usr/bin/env python3 +"""RAG Query — search ChromaDB + answer via llama.cpp.""" +import json, os, sys, urllib.request, time +import chromadb + +LLAMA_URL = "http://localhost:8080/completion" + +def llama_embed(texts): + data = json.dumps({"content": texts}).encode() + req = urllib.request.Request("http://localhost:8081/embedding", data=data, + headers={"Content-Type": "application/json"}) + with urllib.request.urlopen(req, timeout=60) as resp: + body = json.loads(resp.read()) + if isinstance(body, list): + return [item["embedding"][0] for item in body] + return body.get("embedding", []) + + +class LlamaEmbedFn: + def __call__(self, input): + all_embs = [] + for i in range(0, len(input), 32): + batch = input[i:i+32] + embs = llama_embed(batch) + all_embs.extend(embs if isinstance(embs[0], list) else [embs]) + return all_embs + + +def search(collection, query, n=5): + results = collection.query(query_texts=[query], n_results=n) + docs = [] + for i in range(len(results["ids"][0])): + docs.append({ + "id": results["ids"][0][i], + "doc": results["documents"][0][i], + "meta": results["metadatas"][0][i], + "distance": results["distances"][0][i], + }) + return docs + + +def ask_llama(prompt, max_tokens=256): + data = json.dumps({ + "prompt": prompt, "temperature": 0.1, + "n_predict": max_tokens, + "stop": ["\n\n", "Documentation:", "Question:"], + }).encode() + req = urllib.request.Request(LLAMA_URL, data=data, + headers={"Content-Type": "application/json"}) + with urllib.request.urlopen(req, timeout=120) as resp: + return json.loads(resp.read()).get("content", "") + + +def main(): + db_dir = os.path.expanduser("~/nubes/chroma_db") + collection_name = "elsa_docs" + + ef = LlamaEmbedFn() + + client = chromadb.PersistentClient(path=db_dir) + collection = client.get_collection(collection_name, embedding_function=ef) + + print(f"Collection '{collection_name}': {collection.count()} documents") + + while True: + try: + query = input("\n>>> ").strip() + except (EOFError, KeyboardInterrupt): + print() + break + + if not query: + continue + if query.lower() in ("exit", "quit", "q"): + break + + # 1. Поиск + results = search(collection, query, n_results=5) + + # 2. Формируем контекст + context_parts = [] + for r in results: + title = r["meta"].get("title", "")[:120] + ctx = f"[{title}] {r['doc'][:1000]}" + context_parts.append(ctx) + + context = "\n\n---\n\n".join(context_parts) + + # 3. Промпт + prompt = f"""You are an expert VAG (VW/Audi/Skoda/SEAT) automotive diagnostician. +Answer the user's question using ONLY the provided repair documentation below. +If the answer is not in the documentation, say "Not found in documentation." +Answer in Russian. + +DOCUMENTATION: +{context} + +Question: {query} +Answer:""" + + print(f"\n[Found {len(results)} matches, sending to LLM...]") + answer = ask_llama(prompt) + print(f"\n{answer}") + + +if __name__ == "__main__": + main() diff --git a/scripts/rag_test.py b/scripts/rag_test.py new file mode 100644 index 0000000..ba72781 --- /dev/null +++ b/scripts/rag_test.py @@ -0,0 +1,54 @@ +#!/usr/bin/env python3 +"""RAG test - embed query, search, answer via Qwen.""" +import json, os, chromadb, urllib.request + +def embed(texts): + data = json.dumps({"content": texts}).encode() + req = urllib.request.Request("http://localhost:8081/embedding", data=data, + headers={"Content-Type": "application/json"}) + with urllib.request.urlopen(req, timeout=30) as resp: + body = json.loads(resp.read()) + if isinstance(body, list): + return [item["embedding"][0] for item in body] + return body.get("embedding", []) + +os.chdir(os.path.expanduser("~/nubes")) +client = chromadb.PersistentClient(path=os.path.expanduser("~/nubes/chroma_db")) +collection = client.get_collection("elsa_docs") + +# 1. Embed query +query = "What is the tightening torque for fuel filter on VW Phaeton 3.2 VR6 2004?" +print(f"Query: {query}") +q_emb = embed([query])[0] + +# 2. Search +results = collection.query(query_embeddings=[q_emb], n_results=3) +print(f"Matches: {len(results['ids'][0])}") +for i in range(len(results['ids'][0])): + meta = results['metadatas'][0][i] or {} + dist = results['distances'][0][i] + print(f"\n[{i}] dist={dist:.3f}") + print(f" source: {meta.get('source','')}") + print(f" title: {meta.get('title','')[:120]}") + print(f" text: {results['documents'][0][i][:300]}...") + +# 3. Build prompt for Qwen +context = "\n\n".join([f"[{m.get('title','')[:80] if m else ''}]\n{d[:1500]}" + for d, m in zip(results['documents'][0], results['metadatas'][0])]) + +prompt = f"""You are a VAG automotive diagnostician. Answer using ONLY the documentation below. +DOCUMENTATION: +{context} + +Question: {query} +Answer:""" + +data = json.dumps({"prompt": prompt, "temperature": 0.1, "n_predict": 300, + "stop": ["\n\n", "Answer:", "\nThe answer", "\nDocumentation"]}).encode() +req = urllib.request.Request("http://localhost:8080/completion", data=data, + headers={"Content-Type": "application/json"}) +with urllib.request.urlopen(req, timeout=120) as resp: + answer = json.loads(resp.read()).get("content", "") + +print(f"\n\n=== LLM Answer ===\n{answer}") +