89 lines
No EOL
3.8 KiB
Python
89 lines
No EOL
3.8 KiB
Python
# core/bulk_importer.py
|
|
import re
|
|
from database.connection import get_connection
|
|
from docling.document_converter import DocumentConverter
|
|
|
|
class BulkImporter:
|
|
def __init__(self):
|
|
# Converter engine for handling multi-column documents
|
|
self.converter = DocumentConverter()
|
|
# Matches textbook unit codes (e.g., U5_3D, U7 4A, UG_11B)
|
|
self.unit_regex = re.compile(r'U([0-9]+)')
|
|
|
|
def clean_text(self, text: str) -> str:
|
|
"""Cleans syntax breaks, structural commas, and quotes from strings."""
|
|
if not text:
|
|
return ""
|
|
return text.replace('\n', ' ').replace('"', '').replace("'", "").strip()
|
|
|
|
def parse_unit(self, context_str: str) -> int:
|
|
"""Extracts the exact Unit integer from codes like U5_3D or U2 1A."""
|
|
match = self.unit_regex.search(context_str)
|
|
if match:
|
|
return int(match.group(1))
|
|
return None # General reference markers like 'UT LEX' or 'UG'
|
|
|
|
def import_pdf_glossary(self, file_path: str, textbook_name: str):
|
|
"""Converts multi-column PDF via Docling and maps items to SQLite."""
|
|
print(f"🔄 Docling is analyzing structural layout for '{file_path}'...")
|
|
|
|
try:
|
|
# Render layout-aware structural conversion
|
|
result = self.converter.convert(file_path)
|
|
document_text = result.document.export_to_markdown()
|
|
except Exception as e:
|
|
print(f"❌ Docling processing failed: {e}")
|
|
return 0
|
|
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
|
|
print("📥 Parsing text structures and writing to database...")
|
|
count = 0
|
|
|
|
# Regex targeted to catch both standard blocks and tabular lines safely
|
|
# Matches patterns like: "Spanish Word", "English Translation", "Unit Code"
|
|
pattern = re.compile(r'([^,\n"\[]+?)\s*,\s*([^,\n"\[]+?)\s*,\s*(U[G|T|0-9][^\n,]+)')
|
|
matches = pattern.findall(document_text)
|
|
|
|
try:
|
|
for es_raw, en_raw, ctx_raw in matches:
|
|
spanish_text = self.clean_text(es_raw)
|
|
english_text = self.clean_text(en_raw)
|
|
context_tag = self.clean_text(ctx_raw)
|
|
unit_number = self.parse_unit(context_tag)
|
|
|
|
# Avoid table headers or non-lexical entries
|
|
if spanish_text.lower() in ["spanish", "word", "alphabetical glossary"] or len(spanish_text) <= 1:
|
|
continue
|
|
|
|
# 1. Insert Spanish Term Entry
|
|
cursor.execute("""
|
|
INSERT INTO phrases (text, language, textbook, unit, source_context)
|
|
VALUES (?, 'es', ?, ?, ?)
|
|
""", (spanish_text, textbook_name, unit_number, context_tag))
|
|
es_id = cursor.lastrowid
|
|
|
|
# 2. Insert English Translation Entry
|
|
cursor.execute("""
|
|
INSERT INTO phrases (text, language, textbook, unit, source_context)
|
|
VALUES (?, 'en', ?, ?, ?)
|
|
""", (english_text, textbook_name, unit_number, context_tag))
|
|
en_id = cursor.lastrowid
|
|
|
|
# 3. Create Bidirectional Cross-Reference Records
|
|
cursor.execute("INSERT INTO translations (source_phrase_id, target_phrase_id) VALUES (?, ?)", (es_id, en_id))
|
|
cursor.execute("INSERT INTO translations (source_phrase_id, target_phrase_id) VALUES (?, ?)", (en_id, es_id))
|
|
|
|
count += 1
|
|
|
|
conn.commit()
|
|
print(f"🎉 Bulk ingestion complete! Registered {count} matched glossary items.")
|
|
return count
|
|
|
|
except Exception as e:
|
|
conn.rollback()
|
|
print(f"❌ Error during database transaction: {e}")
|
|
return 0
|
|
finally:
|
|
conn.close() |