121 lines
No EOL
4.8 KiB
Python
121 lines
No EOL
4.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):
|
|
self.converter = DocumentConverter()
|
|
self.unit_regex = re.compile(r'U([0-9]+)')
|
|
|
|
def clean_field(self, text: str) -> str:
|
|
"""Removes layout spacing, structural artifacts, and noise."""
|
|
if not text:
|
|
return ""
|
|
# Remove markdown bold/italics markers if any, and compress spaces
|
|
cleaned = text.replace('*', '').replace('_', '').replace('\\', '')
|
|
cleaned = re.sub(r'\s+', ' ', cleaned)
|
|
return cleaned.strip()
|
|
|
|
def parse_unit(self, context_str: str) -> int:
|
|
"""Extracts the integer unit ID out of strings like 'U3_8B' or 'U7'."""
|
|
match = self.unit_regex.search(context_str)
|
|
if match:
|
|
return int(match.group(1))
|
|
return None
|
|
|
|
def insert_pair(self, cursor, es_text, en_text, context_tag, textbook_name):
|
|
"""Helper to cleanly insert a single Spanish-English translation pair."""
|
|
es_clean = self.clean_field(es_text)
|
|
en_clean = self.clean_field(en_text)
|
|
tag_clean = self.clean_field(context_tag)
|
|
|
|
# Guard clause against column headers or empty layout cells
|
|
if not es_clean or not en_clean:
|
|
return False
|
|
if es_clean.lower() in ["spanish", "word", "alphabetical glossary", "es", "en"] or len(es_clean) <= 1:
|
|
return False
|
|
|
|
unit_number = self.parse_unit(tag_clean)
|
|
|
|
try:
|
|
# 1. Insert Spanish Phrase
|
|
cursor.execute("""
|
|
INSERT INTO phrases (text, language, textbook, unit, source_context)
|
|
VALUES (?, 'es', ?, ?, ?)
|
|
""", (es_clean, textbook_name, unit_number, tag_clean))
|
|
es_id = cursor.lastrowid
|
|
|
|
# 2. Insert English Translation
|
|
cursor.execute("""
|
|
INSERT INTO phrases (text, language, textbook, unit, source_context)
|
|
VALUES (?, 'en', ?, ?, ?)
|
|
""", (en_clean, textbook_name, unit_number, tag_clean))
|
|
en_id = cursor.lastrowid
|
|
|
|
# 3. Create Bidirectional Cross-Reference Entries
|
|
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))
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
def import_pdf_glossary(self, file_path: str, textbook_name: str):
|
|
print(f"🔄 Docling is analyzing structural layout for '{file_path}'...")
|
|
|
|
try:
|
|
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 6-column text grid structure and writing to database...")
|
|
count = 0
|
|
|
|
# Process the markdown line-by-line
|
|
lines = document_text.split('\n')
|
|
for line in lines:
|
|
line_str = line.strip()
|
|
|
|
# Target lines containing table formatting row data
|
|
if not line_str.startswith('|') or not line_str.endswith('|'):
|
|
continue
|
|
|
|
# Split by markdown table pipes
|
|
# e.g., "| años60 m | 1960s | U8_7A | bañ osmpl | bath | U1_1A |"
|
|
parts = [p.strip() for p in line_str.split('|')]
|
|
|
|
# A valid row split will include empty items at the ends due to leading/trailing pipes
|
|
# For a 6-column table, len(parts) should be at least 8 elements
|
|
if len(parts) < 7:
|
|
continue
|
|
|
|
# Skip Markdown table header separator rows: |---|---|...
|
|
if '---' in parts[1]:
|
|
continue
|
|
|
|
# Extract Left-Hand Column Group (Columns 1, 2, 3)
|
|
es_left = parts[1]
|
|
en_left = parts[2]
|
|
unit_left = parts[3] if len(parts) >= 4 else ""
|
|
|
|
if self.insert_pair(cursor, es_left, en_left, unit_left, textbook_name):
|
|
count += 1
|
|
|
|
# Extract Right-Hand Column Group (Columns 4, 5, 6)
|
|
if len(parts) >= 7:
|
|
es_right = parts[4]
|
|
en_right = parts[5]
|
|
unit_right = parts[6]
|
|
|
|
if self.insert_pair(cursor, es_right, en_right, unit_right, textbook_name):
|
|
count += 1
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
print(f"🎉 Bulk ingestion complete! Registered {count} matched glossary items.")
|
|
return count |