561 lines
22 KiB
Python
561 lines
22 KiB
Python
import json
|
||
import re
|
||
import html
|
||
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)"
|
||
|
||
_http_client: Optional[httpx.AsyncClient] = None
|
||
|
||
|
||
def get_http_client() -> httpx.AsyncClient:
|
||
"""
|
||
Returns a shared singleton httpx.AsyncClient instance with connection pooling and Keep-Alive.
|
||
Reuses persistent TCP/TLS connections to dictionary providers (Wiktionary, MyMemory)
|
||
eliminating 300-600ms of repeated connection handshakes per word lookup.
|
||
"""
|
||
global _http_client
|
||
if _http_client is None or _http_client.is_closed:
|
||
limits = httpx.Limits(max_keepalive_connections=10, max_connections=20, keepalive_expiry=60.0)
|
||
timeout = httpx.Timeout(connect=3.0, read=4.0, write=3.0, pool=3.0)
|
||
_http_client = httpx.AsyncClient(
|
||
limits=limits,
|
||
timeout=timeout,
|
||
headers={"User-Agent": USER_AGENT},
|
||
follow_redirects=True
|
||
)
|
||
return _http_client
|
||
|
||
|
||
async def close_http_client():
|
||
"""
|
||
Closes the shared singleton HTTP client on application shutdown.
|
||
"""
|
||
global _http_client
|
||
if _http_client is not None:
|
||
client = _http_client
|
||
_http_client = None
|
||
if not client.is_closed:
|
||
try:
|
||
await client.aclose()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
|
||
async def fetch_wiktionary_wikitext(word: str, client: Optional[httpx.AsyncClient] = None) -> Dict[str, Any]:
|
||
"""
|
||
Fetches wikitext via Wiktionary modern REST API (fallback to action=parse)
|
||
and extracts clean IPA phonetics and Russian translations from the English section.
|
||
"""
|
||
http = client or get_http_client()
|
||
words_to_try = [word] if word == word.lower() else [word, word.lower()]
|
||
|
||
for w in words_to_try:
|
||
wikitext = ""
|
||
# 1. Try modern Wikimedia REST API (faster, lightweight source)
|
||
url_rest = f"https://en.wiktionary.org/w/rest.php/v1/page/{urllib.parse.quote(w)}"
|
||
try:
|
||
resp = await http.get(url_rest, timeout=4.0)
|
||
if resp.status_code == 200:
|
||
data = resp.json()
|
||
wikitext = data.get("source", "")
|
||
except Exception:
|
||
pass
|
||
|
||
# 2. Fallback to legacy action=parse if REST returned empty
|
||
if not wikitext:
|
||
url_legacy = "https://en.wiktionary.org/w/api.php"
|
||
params = {
|
||
"action": "parse",
|
||
"page": w,
|
||
"prop": "wikitext",
|
||
"format": "json"
|
||
}
|
||
try:
|
||
resp = await http.get(url_legacy, params=params, timeout=4.0)
|
||
if resp.status_code == 200:
|
||
data = resp.json()
|
||
wikitext = data.get("parse", {}).get("wikitext", {}).get("*", "")
|
||
except Exception:
|
||
pass
|
||
|
||
if not wikitext:
|
||
continue
|
||
|
||
# Isolate English section to avoid leakage from other languages on the page
|
||
en_match = re.search(r"==English==\s*(.*?)(?=\n==[A-Za-z]|\Z)", wikitext, re.DOTALL)
|
||
section = en_match.group(1) if en_match else wikitext
|
||
|
||
# Extract IPA
|
||
ipa_raw = re.findall(r"\{\{IPA\|en\|([^}]+)\}\}", section)
|
||
ipa_cleaned = []
|
||
for entry in ipa_raw:
|
||
for p in entry.split("|"):
|
||
p = p.strip()
|
||
if p.startswith("/") and p.endswith("/"):
|
||
if p not in ipa_cleaned:
|
||
ipa_cleaned.append(p)
|
||
elif p.startswith("/"):
|
||
p_slash = p + "/"
|
||
if p_slash not in ipa_cleaned:
|
||
ipa_cleaned.append(p_slash)
|
||
elif p.startswith("[") and p.endswith("]"):
|
||
if p not in ipa_cleaned:
|
||
ipa_cleaned.append(p)
|
||
|
||
# Extract Russian translations
|
||
ru_raw = re.findall(r"\{\{t[+-]?\|ru\|([^|}]+)", section)
|
||
ru_cleaned = []
|
||
for r in ru_raw:
|
||
clean = r.replace("[[", "").replace("]]", "").replace("́", "").strip()
|
||
if clean and clean not in ru_cleaned:
|
||
ru_cleaned.append(clean)
|
||
|
||
if ipa_cleaned or ru_cleaned:
|
||
return {
|
||
"ipa": ipa_cleaned,
|
||
"audio": [],
|
||
"ru_translations": ru_cleaned
|
||
}
|
||
|
||
return {"ipa": [], "audio": [], "ru_translations": []}
|
||
|
||
|
||
async def fetch_wiktionary_definitions(word: str, client: Optional[httpx.AsyncClient] = None) -> Dict[str, Any]:
|
||
"""
|
||
Fetches structured definitions from Wiktionary REST endpoint
|
||
(https://en.wiktionary.org/api/rest_v1/page/definition/{word}) with HTML unescaping and whitespace normalization.
|
||
"""
|
||
http = client or get_http_client()
|
||
words_to_try = [word] if word == word.lower() else [word, word.lower()]
|
||
|
||
for w in words_to_try:
|
||
url = f"https://en.wiktionary.org/api/rest_v1/page/definition/{urllib.parse.quote(w)}"
|
||
try:
|
||
resp = await http.get(url, timeout=4.0)
|
||
if resp.status_code == 200:
|
||
data = resp.json()
|
||
en_entries = data.get("en", [])
|
||
if not en_entries:
|
||
continue
|
||
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", "")
|
||
clean_def = html.unescape(re.sub(r"<[^>]+>", "", raw_def)).strip()
|
||
clean_def = re.sub(r"\s+", " ", clean_def).strip()
|
||
if clean_def and clean_def not in definitions:
|
||
definitions.append(clean_def)
|
||
for ex in d.get("examples", []):
|
||
clean_ex = html.unescape(re.sub(r"<[^>]+>", "", ex)).strip()
|
||
clean_ex = re.sub(r"\s+", " ", clean_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_mymemory(query: str, langpair: str = "en|ru", client: Optional[httpx.AsyncClient] = None) -> Dict[str, Any]:
|
||
url = "https://api.mymemory.translated.net/get"
|
||
params = {
|
||
"q": query,
|
||
"langpair": langpair,
|
||
"de": config.MYMEMORY_EMAIL
|
||
}
|
||
try:
|
||
http = client or get_http_client()
|
||
resp = await http.get(url, params=params, timeout=4.0)
|
||
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, client: Optional[httpx.AsyncClient] = None) -> 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:
|
||
http = client or get_http_client()
|
||
resp = await http.get(url, timeout=1.5)
|
||
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}
|
||
|
||
def clean_russian_translation(text: str) -> str:
|
||
"""
|
||
Cleans up Russian translations by stripping Latin romanization / phonetic debris
|
||
(e.g. 'át’sjaпорхать át’/', 'krýl’jami/', 'ževáto-koríčnevyj/', '] ]/]')
|
||
and restoring clean, deduplicated Cyrillic phrases, while preserving grammatical notes.
|
||
"""
|
||
if not text:
|
||
return ""
|
||
|
||
# Protect grammatical notes like (прош. вр. от go), (3-е л. ед. ч. от flutter)
|
||
notes = []
|
||
def save_note(m):
|
||
notes.append(m.group(0))
|
||
return f"«ПРИМЕЧАНИЕ_{len(notes)-1}»"
|
||
|
||
s = re.sub(r"\([а-яА-ЯёЁ0-9\s.,-]*\bот\s+[a-zA-Z\x27-]+\)", save_note, text)
|
||
|
||
# 1. Clean out brackets junk and stray backslashes
|
||
s = re.sub(r"\[+[\s\]/\\,;]*\]*", " ", s)
|
||
s = re.sub(r"[\]\[\\]+", " ", s)
|
||
|
||
# 2. Match Latin romanization fragments with accents/slashes/apostrophes
|
||
latin_rom = r"[a-zA-Z\u00C0-\u024F\u1E00-\u1EFF\x27\u2019/]+"
|
||
s = re.sub(rf"\s*{latin_rom}\s*", ", ", s)
|
||
|
||
# 3. Clean up multiple commas/semicolons
|
||
s = re.sub(r"[,;\s]*,[,;\s]*", ", ", s)
|
||
|
||
# 4. Clean and deduplicate phrases
|
||
items = []
|
||
seen = set()
|
||
for part in s.split(","):
|
||
p = part.strip(" /\\,;-\x27\u2019")
|
||
if p.count("(") > p.count(")"):
|
||
p += ")"
|
||
elif p.count(")") > p.count("("):
|
||
p = re.sub(r"\)+$", "", p).strip()
|
||
p = re.sub(r"\s+", " ", p)
|
||
for idx, n in enumerate(notes):
|
||
token = f"«ПРИМЕЧАНИЕ_{idx}»"
|
||
if token in p:
|
||
p = p.replace(token, n).strip()
|
||
if p and re.search(r"[\u0400-\u04FF0-9]", p):
|
||
p_lower = p.lower()
|
||
if p_lower not in seen:
|
||
seen.add(p_lower)
|
||
items.append(p)
|
||
|
||
res = ", ".join(items) if items else text.strip()
|
||
for idx, n in enumerate(notes):
|
||
res = res.replace(f"«ПРИМЕЧАНИЕ_{idx}»", n)
|
||
return res
|
||
|
||
|
||
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
|
||
|
||
is_cyrillic = bool(re.search(r"[\u0400-\u04FF]", query_clean))
|
||
words = query_clean.split()
|
||
is_sentence = len(words) > 4
|
||
|
||
# -----------------------------------------------------------------------
|
||
# Reverse Translation (Russian -> English)
|
||
# -----------------------------------------------------------------------
|
||
if is_cyrillic:
|
||
matched_term = None
|
||
phonetic = ""
|
||
pos = ""
|
||
clean_trans = query_clean
|
||
tr_list = [query_clean]
|
||
is_single = len(words) == 1
|
||
|
||
# 1. Search offline dictionary for exact reverse matches
|
||
if not is_sentence:
|
||
try:
|
||
with get_db() as conn:
|
||
cur = conn.execute(
|
||
"SELECT term, translation, phonetic, pos FROM offline_dictionary WHERE translation LIKE ?",
|
||
(f"%{query_clean}%",)
|
||
)
|
||
rows = cur.fetchall()
|
||
exact_matches = []
|
||
for r in rows:
|
||
raw_t = r["translation"] or ""
|
||
parts = [p.strip().lower() for p in raw_t.split(",")]
|
||
if query_clean.lower() in parts:
|
||
w_cnt = len(r["term"].split())
|
||
is_multi = 1 if (is_single and w_cnt > 1) else 0
|
||
is_inflection = 1 if " от " in raw_t else 0
|
||
exact_matches.append((is_multi, is_inflection, w_cnt, len(r["term"]), r))
|
||
if exact_matches:
|
||
exact_matches.sort(key=lambda x: (x[0], x[1], x[2], x[3]))
|
||
# Prefer single-word offline match if query was single word
|
||
if exact_matches[0][0] == 0:
|
||
best_row = exact_matches[0][4]
|
||
matched_term = best_row["term"]
|
||
phonetic = best_row["phonetic"] or ""
|
||
pos = best_row["pos"] or ""
|
||
clean_trans = clean_russian_translation(best_row["translation"] or "")
|
||
tr_list = [t.strip() for t in clean_trans.split(",") if t.strip()]
|
||
except Exception:
|
||
pass
|
||
|
||
# 2. If no direct offline match, query MyMemory with langpair="ru|en"
|
||
http_client = get_http_client()
|
||
if not matched_term:
|
||
try:
|
||
mm_data = await fetch_mymemory(query_clean, langpair="ru|en", client=http_client)
|
||
raw_cands = []
|
||
if mm_data.get("translation"):
|
||
raw_cands.append(mm_data["translation"])
|
||
for m in mm_data.get("matches", []):
|
||
if m and m not in raw_cands and m.lower() != query_clean.lower():
|
||
raw_cands.append(m)
|
||
|
||
cleaned_cands = []
|
||
for c in raw_cands:
|
||
c_clean = re.sub(r"^(?:to|an?|the)\s+", "", c.strip(), flags=re.IGNORECASE).strip(" *.,!?\"'")
|
||
if c_clean and c_clean not in cleaned_cands and not re.search(r"[\u0400-\u04FF]", c_clean):
|
||
cleaned_cands.append(c_clean)
|
||
|
||
if cleaned_cands:
|
||
matched_term = cleaned_cands[0]
|
||
# Enrich from offline dictionary if candidate exists
|
||
with get_db() as conn:
|
||
for cand in cleaned_cands:
|
||
off_row = conn.execute(
|
||
"SELECT term, translation, phonetic, pos FROM offline_dictionary WHERE term = ? COLLATE NOCASE",
|
||
(cand,)
|
||
).fetchone()
|
||
if off_row:
|
||
matched_term = off_row["term"]
|
||
phonetic = off_row["phonetic"] or ""
|
||
pos = off_row["pos"] or ""
|
||
if not clean_trans or clean_trans == query_clean:
|
||
clean_trans = clean_russian_translation(off_row["translation"] or "") or query_clean
|
||
tr_list = [t.strip() for t in clean_trans.split(",") if t.strip()]
|
||
break
|
||
|
||
# If phonetic still missing and candidate is a single word, fetch from free dictionary API
|
||
if not phonetic and len(matched_term.split()) == 1:
|
||
fd = await fetch_free_dictionary_api(matched_term, client=http_client)
|
||
if fd.get("phonetic"):
|
||
phonetic = fd["phonetic"]
|
||
except Exception:
|
||
pass
|
||
|
||
if not matched_term:
|
||
matched_term = query_clean
|
||
|
||
res = {
|
||
"query": query_clean,
|
||
"term": matched_term,
|
||
"english_term": matched_term,
|
||
"russian_query": query_clean,
|
||
"is_sentence": is_sentence,
|
||
"is_reverse": True,
|
||
"translation": clean_trans,
|
||
"translations_list": tr_list if tr_list else [clean_trans],
|
||
"translation_ru": clean_trans,
|
||
"phonetic": phonetic,
|
||
"audio_file": "",
|
||
"audio_url": "",
|
||
"part_of_speech": pos or ("Sentence" if is_sentence else "Word"),
|
||
"definitions": [],
|
||
"examples": [],
|
||
"source": "reverse_dictionary",
|
||
}
|
||
|
||
with get_db() as conn:
|
||
conn.execute(
|
||
"INSERT OR REPLACE INTO dictionary_cache (query, response_json) VALUES (?, ?)",
|
||
(query_clean, json.dumps(res, ensure_ascii=False))
|
||
)
|
||
return res
|
||
|
||
# -----------------------------------------------------------------------
|
||
# Forward Translation (English -> Russian)
|
||
# -----------------------------------------------------------------------
|
||
# Check offline dictionary first (Zero Daemon, < 0.1ms via SQLite mmap)
|
||
if not is_sentence:
|
||
try:
|
||
with get_db() as conn:
|
||
off_row = conn.execute(
|
||
"SELECT term, translation, phonetic, pos FROM offline_dictionary WHERE term = ? COLLATE NOCASE",
|
||
(query_clean,)
|
||
).fetchone()
|
||
|
||
lemma_found = None
|
||
if not off_row and len(words) == 1:
|
||
try:
|
||
from app.phrasal import lemmatize_word
|
||
for cand in lemmatize_word(query_clean):
|
||
if cand.lower() != query_clean.lower():
|
||
row = conn.execute(
|
||
"SELECT term, translation, phonetic, pos FROM offline_dictionary WHERE term = ? COLLATE NOCASE",
|
||
(cand,)
|
||
).fetchone()
|
||
if row:
|
||
off_row = row
|
||
lemma_found = row["term"]
|
||
break
|
||
except Exception:
|
||
pass
|
||
|
||
if off_row:
|
||
raw_trans = off_row["translation"] or ""
|
||
clean_trans = clean_russian_translation(raw_trans)
|
||
if lemma_found and lemma_found.lower() != query_clean.lower():
|
||
clean_trans = f"{clean_trans} (от {lemma_found})"
|
||
tr_list = [t.strip() for t in clean_trans.split(",") if t.strip()]
|
||
res = {
|
||
"query": query_clean,
|
||
"term": off_row["term"] or query_clean,
|
||
"is_sentence": False,
|
||
"is_reverse": False,
|
||
"translation": clean_trans,
|
||
"translations_list": tr_list,
|
||
"translation_ru": clean_trans,
|
||
"phonetic": off_row["phonetic"] or "",
|
||
"audio_file": "",
|
||
"audio_url": "",
|
||
"part_of_speech": off_row["pos"] or "",
|
||
"definitions": [],
|
||
"examples": [],
|
||
"source": "offline",
|
||
}
|
||
conn.execute(
|
||
"INSERT OR REPLACE INTO dictionary_cache (query, response_json) VALUES (?, ?)",
|
||
(query_clean, json.dumps(res, ensure_ascii=False))
|
||
)
|
||
return res
|
||
except Exception:
|
||
pass
|
||
|
||
result: Dict[str, Any] = {
|
||
"query": query_clean,
|
||
"term": query_clean,
|
||
"is_sentence": is_sentence,
|
||
"is_reverse": False,
|
||
"translation": "",
|
||
"translations_list": [],
|
||
"phonetic": "",
|
||
"audio_file": "",
|
||
"audio_url": "",
|
||
"part_of_speech": "",
|
||
"definitions": [],
|
||
"examples": []
|
||
}
|
||
|
||
# Parallel queries via asyncio
|
||
import asyncio
|
||
|
||
http_client = get_http_client()
|
||
|
||
if is_sentence:
|
||
# For long sentences, primary focus is translation
|
||
mm_data = await fetch_mymemory(query_clean, client=http_client)
|
||
result["translation"] = mm_data.get("translation", "")
|
||
result["translations_list"] = mm_data.get("matches", [])
|
||
else:
|
||
# For words and short phrases: reuse keep-alive connections across providers
|
||
tasks = [
|
||
fetch_wiktionary_wikitext(query_clean, client=http_client),
|
||
fetch_wiktionary_definitions(query_clean, client=http_client),
|
||
fetch_mymemory(query_clean, client=http_client),
|
||
fetch_free_dictionary_api(query_clean, client=http_client)
|
||
]
|
||
w_wiki, w_defs, 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]
|
||
|
||
# 2. Audio URL / File
|
||
result["audio_file"] = ""
|
||
result["audio_url"] = free_dict.get("audio_url") or ""
|
||
|
||
# 3. Part of Speech
|
||
if w_defs.get("parts_of_speech"):
|
||
result["part_of_speech"] = ", ".join(w_defs["parts_of_speech"][:2])
|
||
|
||
# 4. Definitions & Examples
|
||
if w_defs.get("definitions"):
|
||
result["definitions"] = w_defs["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 []
|
||
|
||
# Alias translation_ru for client compatibility
|
||
result["translation_ru"] = result["translation"]
|
||
|
||
# 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
|