← back to Dear Bubbe Nextjs
lib/yiddish-dictionary/build_yiddish_db.py
425 lines
#!/usr/bin/env python3
"""
Yiddish Dictionary Builder for Dear Bubbe
Imports multiple Yiddish dictionary sources into a unified SQLite database
with Bubbe-safe filtering and cultural context preservation.
"""
import sqlite3
import json
import csv
from pathlib import Path
from typing import Dict, List, Optional, Any
import re
from datetime import datetime
DB_PATH = "yiddish_dictionary.sqlite"
SCHEMA_PATH = "schema.sql"
# Configuration for input sources
# Update these paths when you download the actual dictionary files
INPUT_SOURCES = [
{
"path": "dictionaries/sample_dictionary.json",
"format": "json",
"name": "Dear Bubbe Core Vocabulary",
"url": "Internal",
"license": "Proprietary",
"field_map": {
"yiddish": "yiddish",
"transliteration": "yiddish_latin",
"english": "english",
"pos": "pos",
"tags": "tags",
"id": "source_entry_id",
},
},
# Uncomment these when you have the actual dictionary files
# {
# "path": "dictionaries/solomon_yiddish.json",
# "format": "json",
# "name": "Solomon Birnbaum Yiddish Dictionary",
# "url": "https://github.com/SolomonBirnbaum/yiddish-dict",
# "license": "MIT",
# "field_map": {
# "yiddish": "yiddish",
# "transliteration": "yiddish_latin",
# "english": "english",
# "pos": "pos",
# "tags": "tags",
# "id": "source_entry_id",
# },
# },
]
# Terms that should be marked as needing review (supplements banned-terms.json)
REVIEW_PATTERNS = [
r'\bgoy\b',
r'\bshiksa\b',
r'\bsheygetz\b',
r'\bshvartze\b',
]
# Bubbe's favorite expressions to pre-populate
BUBBE_FAVORITES = [
{
"yiddish": "אױ װײ",
"english": "oy vey",
"context": "exasperation",
"intensity": 3,
"usage_example": "Oy vey, you spent HOW MUCH on that schmatte?",
},
{
"yiddish": "געװאַלד",
"english": "gevalt",
"context": "alarm",
"intensity": 7,
"usage_example": "Oy gevalt! Still not married at YOUR age?",
},
{
"yiddish": "משוגענער",
"english": "meshugener",
"context": "scolding",
"intensity": 5,
"usage_example": "You meshugener! Who quits a job without another lined up?",
},
{
"yiddish": "שמענדריק",
"english": "shmendrik",
"context": "insult",
"intensity": 4,
"usage_example": "That shmendrik can't even boil water properly!",
},
{
"yiddish": "פֿאַרקאַקטע",
"english": "farkakte",
"context": "complaint",
"intensity": 6,
"usage_example": "This farkakte weather is killing my arthritis!",
},
{
"yiddish": "קװעטש",
"english": "kvetch",
"context": "complaining",
"intensity": 3,
"usage_example": "Stop kvetching and do something about it!",
},
{
"yiddish": "פּלאַצן",
"english": "plotz",
"context": "overwhelming",
"intensity": 5,
"usage_example": "Your mother's gonna plotz when she hears this!",
},
{
"yiddish": "טשעפּען",
"english": "tchepen",
"context": "nagging",
"intensity": 4,
"usage_example": "Don't tchepen me about my weight!",
}
]
def load_schema(conn: sqlite3.Connection, schema_path: str):
"""Load the database schema from SQL file."""
with open(schema_path, "r", encoding="utf-8") as f:
conn.executescript(f.read())
conn.commit()
print(f"✓ Schema loaded from {schema_path}")
def get_or_create_source(conn, name, url=None, license_text=None, notes=None):
"""Get existing source ID or create new source entry."""
cur = conn.cursor()
cur.execute("SELECT id FROM sources WHERE name = ?", (name,))
row = cur.fetchone()
if row:
return row[0]
cur.execute(
"""
INSERT INTO sources (name, url, license, notes)
VALUES (?, ?, ?, ?)
""",
(name, url, license_text, notes),
)
conn.commit()
return cur.lastrowid
def check_needs_review(entry: dict) -> int:
"""
Check if entry contains potentially problematic terms.
Returns: 1 (approved), 0 (needs review), -1 (likely banned)
"""
text_to_check = f"{entry.get('yiddish', '')} {entry.get('english', '')}".lower()
for pattern in REVIEW_PATTERNS:
if re.search(pattern, text_to_check, re.IGNORECASE):
return 0 # Needs review
return 1 # Approved by default
def normalize_entry(raw_entry: dict, field_map: dict, source_id: int) -> Optional[dict]:
"""
Convert raw dictionary entry to normalized database format.
Includes Bubbe-safe filtering.
"""
entry = {
"yiddish": None,
"yiddish_latin": None,
"english": None,
"pos": None,
"gender": None,
"plural": None,
"tags": None,
"usage_example": None,
"cultural_note": None,
"source_id": source_id,
"source_entry_id": None,
"bubbe_approved": 1,
"notes": None,
}
# Map fields from source to database columns
for file_key, db_col in field_map.items():
if file_key in raw_entry and db_col in entry:
value = raw_entry[file_key]
if value and isinstance(value, str):
value = value.strip()
entry[db_col] = value
# Require at minimum Yiddish and English
if not entry["yiddish"] or not entry["english"]:
return None
# Check for problematic content
entry["bubbe_approved"] = check_needs_review(entry)
# Clean up tags if present
if entry["tags"] and isinstance(entry["tags"], list):
entry["tags"] = ", ".join(entry["tags"])
return entry
def import_json_file(conn, source_cfg, source_id):
"""Import entries from JSON file."""
path = Path(source_cfg["path"])
field_map = source_cfg["field_map"]
if not path.exists():
print(f"⚠ Skipping {path} - file not found")
return
with path.open("r", encoding="utf-8") as f:
data = json.load(f)
if not isinstance(data, list):
# Try to extract list from common wrapper formats
if isinstance(data, dict):
data = data.get("entries", data.get("words", data.get("dictionary", [])))
if not isinstance(data, list):
print(f"⚠ JSON file {path} doesn't contain a list of entries")
return
cur = conn.cursor()
count = 0
reviewed = 0
for raw in data:
if not isinstance(raw, dict):
continue
entry = normalize_entry(raw, field_map, source_id)
if not entry:
continue
if entry["bubbe_approved"] == 0:
reviewed += 1
cur.execute(
"""
INSERT OR IGNORE INTO entries
(yiddish, yiddish_latin, english, pos, gender, plural, tags,
usage_example, cultural_note, source_id, source_entry_id,
bubbe_approved, notes)
VALUES (:yiddish, :yiddish_latin, :english, :pos, :gender,
:plural, :tags, :usage_example, :cultural_note,
:source_id, :source_entry_id, :bubbe_approved, :notes)
""",
entry,
)
count += 1
conn.commit()
print(f"✓ Imported {count} entries from JSON: {path}")
if reviewed > 0:
print(f" ⚠ {reviewed} entries marked for review")
def import_delimited_file(conn, source_cfg, source_id, delimiter=","):
"""Import entries from CSV/TSV file."""
path = Path(source_cfg["path"])
field_map = source_cfg["field_map"]
if not path.exists():
print(f"⚠ Skipping {path} - file not found")
return
with path.open("r", encoding="utf-8") as f:
reader = csv.DictReader(f, delimiter=delimiter)
headers = reader.fieldnames or []
# Warn about missing columns
missing = [k for k in field_map.keys() if k not in headers]
if missing:
print(f" ⚠ Warning: {path} missing columns: {missing}")
cur = conn.cursor()
count = 0
reviewed = 0
for raw in reader:
entry = normalize_entry(raw, field_map, source_id)
if not entry:
continue
if entry["bubbe_approved"] == 0:
reviewed += 1
cur.execute(
"""
INSERT OR IGNORE INTO entries
(yiddish, yiddish_latin, english, pos, gender, plural, tags,
usage_example, cultural_note, source_id, source_entry_id,
bubbe_approved, notes)
VALUES (:yiddish, :yiddish_latin, :english, :pos, :gender,
:plural, :tags, :usage_example, :cultural_note,
:source_id, :source_entry_id, :bubbe_approved, :notes)
""",
entry,
)
count += 1
conn.commit()
file_type = "TSV" if delimiter == "\t" else "CSV"
print(f"✓ Imported {count} entries from {file_type}: {path}")
if reviewed > 0:
print(f" ⚠ {reviewed} entries marked for review")
def populate_bubbe_favorites(conn):
"""Add Bubbe's favorite expressions."""
cur = conn.cursor()
for fav in BUBBE_FAVORITES:
cur.execute(
"""
INSERT OR IGNORE INTO bubbe_favorites
(yiddish, english, context, intensity, usage_example)
VALUES (?, ?, ?, ?, ?)
""",
(fav["yiddish"], fav["english"], fav["context"],
fav["intensity"], fav["usage_example"])
)
conn.commit()
print(f"✓ Added {len(BUBBE_FAVORITES)} Bubbe favorite expressions")
def print_statistics(conn):
"""Print database statistics."""
cur = conn.cursor()
# Total entries
cur.execute("SELECT COUNT(*) FROM entries")
total = cur.fetchone()[0]
# Approved entries
cur.execute("SELECT COUNT(*) FROM entries WHERE bubbe_approved = 1")
approved = cur.fetchone()[0]
# Needs review
cur.execute("SELECT COUNT(*) FROM entries WHERE bubbe_approved = 0")
review = cur.fetchone()[0]
# Sources
cur.execute("SELECT COUNT(*) FROM sources")
sources = cur.fetchone()[0]
# Favorites
cur.execute("SELECT COUNT(*) FROM bubbe_favorites")
favorites = cur.fetchone()[0]
print("\n" + "="*50)
print("📚 YIDDISH DICTIONARY STATISTICS")
print("="*50)
print(f"Total entries: {total:,}")
print(f"Approved for Bubbe: {approved:,}")
print(f"Needs review: {review:,}")
print(f"Sources imported: {sources}")
print(f"Bubbe favorites: {favorites}")
print("="*50)
def main():
"""Main import process."""
print("\n🕎 DEAR BUBBE YIDDISH DICTIONARY BUILDER")
print("="*50)
db_path = Path(DB_PATH)
schema_path = Path(SCHEMA_PATH)
if not schema_path.exists():
print(f"❌ Schema file not found: {schema_path}")
print(" Please ensure schema.sql is in the same directory")
return
# Connect to database
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
# Load schema
load_schema(conn, schema_path)
# Add Bubbe's favorites
populate_bubbe_favorites(conn)
# Import each source
print("\n📖 Importing dictionary sources...")
for src in INPUT_SOURCES:
print(f"\n→ Processing: {src['name']}")
source_id = get_or_create_source(
conn,
name=src.get("name"),
url=src.get("url"),
license_text=src.get("license"),
)
fmt = src["format"].lower()
if fmt == "json":
import_json_file(conn, src, source_id)
elif fmt == "tsv":
import_delimited_file(conn, src, source_id, delimiter="\t")
elif fmt == "csv":
import_delimited_file(conn, src, source_id, delimiter=",")
else:
print(f" ❌ Unknown format: {fmt}")
# Print statistics
print_statistics(conn)
conn.close()
print(f"\n✅ Database created: {db_path}")
print(" Run 'sqlite3 yiddish_dictionary.sqlite' to explore")
if __name__ == "__main__":
main()