notes-qpid-pro/app/dictionary.py

import json
import re
import urllib.parse
from typing import Optional, Dict, Any, List
import httpx
from app import config
from app.database import get_db

USER_AGENT = "NotesApp/1.0 (https://notes.qpid.pro; admin@qpid.pro)"

async def fetch_wiktionary_wikitext(word: str) -> Dict[str, Any]:
    url = "https://en.wiktionary.org/w/api.php"
    params = {
        "action": "parse",
        "page": word,
        "prop": "wikitext",
        "format": "json"
    }
    try:
        async with httpx.AsyncClient(timeout=4.0) as client:
            resp = await client.get(url, params=params, headers={"User-Agent": USER_AGENT})
            if resp.status_code == 200:
                data = resp.json()
                wikitext = data.get("parse", {}).get("wikitext", {}).get("*", "")
                
                # Extract IPA
                ipa_raw = re.findall(r"\{\{IPA\|en\|([^}]+)\}\}", wikitext)
                ipa_cleaned = []
                for entry in ipa_raw:
                    parts = entry.split("|")
                    for p in parts:
                        if p.startswith("/"):
                            ipa_cleaned.append(p)
                
                # Extract Audio file names
                audio_matches = re.findall(r"\{\{audio\|en\|([^|}]+)", wikitext)
                
                # Extract Russian translations
                ru_raw = re.findall(r"\{\{t[+-]?\|ru\|([^|}]+)", wikitext)
                ru_cleaned = []
                for r in ru_raw:
                    # Clean wiki markup like [[word]] or stress marks
                    clean = r.replace("[[", "").replace("]]", "").strip()
                    if clean and clean not in ru_cleaned:
                        ru_cleaned.append(clean)

                return {
                    "ipa": ipa_cleaned,
                    "audio": audio_matches,
                    "ru_translations": ru_cleaned
                }
    except Exception:
        pass
    return {"ipa": [], "audio": [], "ru_translations": []}

async def fetch_wiktionary_definitions(word: str) -> Dict[str, Any]:
    url = f"https://en.wiktionary.org/api/rest_v1/page/definition/{urllib.parse.quote(word)}"
    try:
        async with httpx.AsyncClient(timeout=4.0) as client:
            resp = await client.get(url, headers={"User-Agent": USER_AGENT})
            if resp.status_code == 200:
                data = resp.json()
                en_entries = data.get("en", [])
                parts_of_speech = []
                definitions = []
                examples = []
                for entry in en_entries:
                    pos = entry.get("partOfSpeech")
                    if pos and pos not in parts_of_speech:
                        parts_of_speech.append(pos)
                    for d in entry.get("definitions", []):
                        raw_def = d.get("definition", "")
                        # strip html tags
                        clean_def = re.sub(r"<[^>]+>", "", raw_def).strip()
                        if clean_def and clean_def not in definitions:
                            definitions.append(clean_def)
                        for ex in d.get("examples", []):
                            clean_ex = re.sub(r"<[^>]+>", "", ex).strip()
                            if clean_ex and clean_ex not in examples:
                                examples.append(clean_ex)
                return {
                    "parts_of_speech": parts_of_speech,
                    "definitions": definitions[:4],
                    "examples": examples[:3]
                }
    except Exception:
        pass
    return {"parts_of_speech": [], "definitions": [], "examples": []}

async def fetch_datamuse(word: str) -> Dict[str, Any]:
    url = "https://api.datamuse.com/words"
    params = {"sp": word, "md": "drp", "ipa": "1"}
    try:
        async with httpx.AsyncClient(timeout=3.0) as client:
            resp = await client.get(url, params=params)
            if resp.status_code == 200:
                items = resp.json()
                if items and isinstance(items, list):
                    first = items[0]
                    tags = first.get("tags", [])
                    ipa = None
                    pos = None
                    for t in tags:
                        if t.startswith("ipa_pron:"):
                            ipa = "/" + t.split(":", 1)[1] + "/"
                        elif t in ("n", "v", "adj", "adv"):
                            pos_map = {"n": "Noun", "v": "Verb", "adj": "Adjective", "adv": "Adverb"}
                            pos = pos_map.get(t)
                    
                    defs = []
                    for d in first.get("defs", []):
                        parts = d.split("\t", 1)
                        defs.append(parts[1].strip() if len(parts) > 1 else parts[0].strip())

                    return {
                        "ipa": ipa,
                        "part_of_speech": pos,
                        "definitions": defs[:3]
                    }
    except Exception:
        pass
    return {"ipa": None, "part_of_speech": None, "definitions": []}

async def fetch_mymemory(query: str) -> Dict[str, Any]:
    url = "https://api.mymemory.translated.net/get"
    params = {
        "q": query,
        "langpair": "en|ru",
        "de": config.MYMEMORY_EMAIL
    }
    try:
        async with httpx.AsyncClient(timeout=4.0) as client:
            resp = await client.get(url, params=params)
            if resp.status_code == 200:
                data = resp.json()
                main_trans = data.get("responseData", {}).get("translatedText", "").strip()
                matches = data.get("matches", [])
                alt_list = []
                if main_trans:
                    alt_list.append(main_trans)
                for m in matches:
                    t = m.get("translation", "").strip()
                    if t and t not in alt_list and t.lower() != query.lower():
                        alt_list.append(t)
                return {
                    "translation": main_trans,
                    "matches": alt_list[:6]
                }
    except Exception:
        pass
    return {"translation": "", "matches": []}

async def fetch_free_dictionary_api(word: str) -> Dict[str, Any]:
    """Optional tier: api.dictionaryapi.dev with fast timeout."""
    url = f"https://api.dictionaryapi.dev/api/v2/entries/en/{urllib.parse.quote(word)}"
    try:
        async with httpx.AsyncClient(timeout=1.5) as client:
            resp = await client.get(url)
            if resp.status_code == 200:
                data = resp.json()[0]
                phonetic = data.get("phonetic")
                audio_url = None
                for p in data.get("phonetics", []):
                    if p.get("audio"):
                        audio_url = p.get("audio")
                        break
                return {
                    "phonetic": phonetic,
                    "audio_url": audio_url
                }
    except Exception:
        pass
    return {"phonetic": None, "audio_url": None}

async def lookup_term(query: str) -> Dict[str, Any]:
    query_clean = query.strip()
    if not query_clean:
        return {}

    # Check cache first
    with get_db() as conn:
        row = conn.execute(
            "SELECT response_json FROM dictionary_cache WHERE query = ?",
            (query_clean,)
        ).fetchone()
        if row:
            try:
                return json.loads(row["response_json"])
            except Exception:
                pass

    words = query_clean.split()
    is_sentence = len(words) > 4

    result: Dict[str, Any] = {
        "query": query_clean,
        "is_sentence": is_sentence,
        "translation": "",
        "translations_list": [],
        "phonetic": "",
        "audio_file": "",
        "audio_url": "",
        "part_of_speech": "",
        "definitions": [],
        "examples": []
    }

    # Parallel queries via asyncio
    import asyncio
    
    if is_sentence:
        # For long sentences, primary focus is translation
        mm_data = await fetch_mymemory(query_clean)
        result["translation"] = mm_data.get("translation", "")
        result["translations_list"] = mm_data.get("matches", [])
    else:
        # For words and short phrases
        tasks = [
            fetch_wiktionary_wikitext(query_clean),
            fetch_wiktionary_definitions(query_clean),
            fetch_datamuse(query_clean),
            fetch_mymemory(query_clean),
            fetch_free_dictionary_api(query_clean)
        ]
        w_wiki, w_defs, d_muse, mm_data, free_dict = await asyncio.gather(*tasks)

        # 1. Phonetic IPA
        if free_dict.get("phonetic"):
            result["phonetic"] = free_dict["phonetic"]
        elif w_wiki.get("ipa"):
            result["phonetic"] = w_wiki["ipa"][0]
        elif d_muse.get("ipa"):
            result["phonetic"] = d_muse["ipa"]

        # 2. Audio URL / File
        if free_dict.get("audio_url"):
            result["audio_url"] = free_dict["audio_url"]
        elif w_wiki.get("audio"):
            result["audio_file"] = w_wiki["audio"][0]
            result["audio_url"] = f"/api/audio?file={urllib.parse.quote(result['audio_file'])}"

        # 3. Part of Speech
        if w_defs.get("parts_of_speech"):
            result["part_of_speech"] = ", ".join(w_defs["parts_of_speech"][:2])
        elif d_muse.get("part_of_speech"):
            result["part_of_speech"] = d_muse["part_of_speech"]

        # 4. Definitions & Examples
        if w_defs.get("definitions"):
            result["definitions"] = w_defs["definitions"]
        elif d_muse.get("definitions"):
            result["definitions"] = d_muse["definitions"]

        if w_defs.get("examples"):
            result["examples"] = w_defs["examples"]

        # 5. Translations
        candidates = []
        # Wiktionary curated Russian translations
        for t in w_wiki.get("ru_translations", []):
            if t not in candidates:
                candidates.append(t)

        # MyMemory translation
        for t in mm_data.get("matches", []):
            if t and t not in candidates and t.lower() != query_clean.lower():
                candidates.append(t)

        if candidates:
            result["translation"] = ", ".join(candidates[:3])
            result["translations_list"] = candidates
        else:
            result["translation"] = mm_data.get("translation", "")
            result["translations_list"] = [result["translation"]] if result["translation"] else []

    # Store into SQLite cache
    with get_db() as conn:
        conn.execute(
            "INSERT OR REPLACE INTO dictionary_cache (query, response_json) VALUES (?, ?)",
            (query_clean, json.dumps(result, ensure_ascii=False))
        )

    return result