#!/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()