225 lines
8.5 KiB
Python
225 lines
8.5 KiB
Python
#!/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()
|