421 lines
15 KiB
Python
421 lines
15 KiB
Python
"""Translate product copy to Persian via an OpenAI-compatible Chat Completions API."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import ssl
|
||
import urllib.error
|
||
import urllib.request
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
|
||
def _ssl_contexts() -> list[ssl.SSLContext]:
|
||
contexts: list[ssl.SSLContext] = []
|
||
try:
|
||
import certifi # type: ignore
|
||
|
||
contexts.append(ssl.create_default_context(cafile=certifi.where()))
|
||
except Exception:
|
||
pass
|
||
try:
|
||
contexts.append(ssl.create_default_context())
|
||
except Exception:
|
||
pass
|
||
contexts.append(ssl._create_unverified_context())
|
||
return contexts
|
||
|
||
ROOT = Path(__file__).resolve().parent
|
||
DATA_DIR = ROOT / "data"
|
||
SETTINGS_PATH = DATA_DIR / ".ai-settings.json"
|
||
|
||
DEFAULT_BASE_URL = "https://api.openai.com/v1"
|
||
DEFAULT_MODEL = "gpt-4o-mini"
|
||
|
||
ARABIC_SCRIPT = re.compile(r"[\u0600-\u06FF]")
|
||
SYSTEM_PROMPT = (
|
||
"You are a professional Persian (Farsi) copywriter for a luxury cosmetics retailer in Iran. "
|
||
"Translate the given product text into natural, polished Persian. "
|
||
"Keep brand names, shade codes, SPF, numbers, percentages, and product line names in Latin script when that is normal for cosmetics. "
|
||
"Do not add marketing fluff that is not in the source. "
|
||
"Return only the translated text, with no quotes or commentary."
|
||
)
|
||
|
||
CATEGORY_SYSTEM_PROMPT = (
|
||
"You translate cosmetics ecommerce category names into short, natural Persian (Farsi) labels "
|
||
"for an Iranian KIKO Milano storefront. "
|
||
"Return only the Persian category title — no quotes, no explanation. "
|
||
"Keep well-known brand tokens like KIKO in Latin if needed. "
|
||
"Prefer concise menu labels (1–4 words)."
|
||
)
|
||
|
||
|
||
def _mask_key(key: str) -> str:
|
||
if not key:
|
||
return ""
|
||
return key[:3] + "…" + key[-4:] if len(key) > 8 else "••••"
|
||
|
||
|
||
def _default_settings() -> dict:
|
||
return {
|
||
"apiKey": os.environ.get("OPENAI_API_KEY", "").strip(),
|
||
"baseUrl": os.environ.get("OPENAI_BASE_URL", DEFAULT_BASE_URL).strip()
|
||
or DEFAULT_BASE_URL,
|
||
"model": os.environ.get("OPENAI_MODEL", DEFAULT_MODEL).strip() or DEFAULT_MODEL,
|
||
"serpApiKey": os.environ.get("SERPAPI_API_KEY", os.environ.get("SERP_API_KEY", "")).strip(),
|
||
"categoryApiKey": os.environ.get("CATEGORY_TRANSLATE_API_KEY", "").strip(),
|
||
}
|
||
|
||
|
||
def load_settings() -> dict:
|
||
data = _default_settings()
|
||
if SETTINGS_PATH.is_file():
|
||
try:
|
||
raw = json.loads(SETTINGS_PATH.read_text(encoding="utf-8"))
|
||
if isinstance(raw, dict):
|
||
for key in ("apiKey", "baseUrl", "model", "serpApiKey", "categoryApiKey"):
|
||
if isinstance(raw.get(key), str) and raw[key].strip():
|
||
data[key] = raw[key].strip()
|
||
except (OSError, json.JSONDecodeError):
|
||
pass
|
||
data["baseUrl"] = (data.get("baseUrl") or DEFAULT_BASE_URL).rstrip("/")
|
||
# Guard: SerpAPI dashboard/search hosts are not Chat Completions base URLs.
|
||
if "serpapi.com" in data["baseUrl"].lower():
|
||
data["baseUrl"] = DEFAULT_BASE_URL
|
||
data["model"] = data.get("model") or DEFAULT_MODEL
|
||
data["serpApiKey"] = data.get("serpApiKey") or ""
|
||
data["categoryApiKey"] = data.get("categoryApiKey") or ""
|
||
return data
|
||
|
||
|
||
def save_settings(
|
||
*,
|
||
api_key: str | None = None,
|
||
base_url: str | None = None,
|
||
model: str | None = None,
|
||
serp_api_key: str | None = None,
|
||
category_api_key: str | None = None,
|
||
) -> dict:
|
||
current = load_settings()
|
||
if api_key is not None:
|
||
current["apiKey"] = api_key.strip()
|
||
if base_url is not None:
|
||
current["baseUrl"] = (base_url.strip() or DEFAULT_BASE_URL).rstrip("/")
|
||
if model is not None:
|
||
current["model"] = model.strip() or DEFAULT_MODEL
|
||
if serp_api_key is not None:
|
||
current["serpApiKey"] = serp_api_key.strip()
|
||
if category_api_key is not None:
|
||
current["categoryApiKey"] = category_api_key.strip()
|
||
DATA_DIR.mkdir(parents=True, exist_ok=True)
|
||
SETTINGS_PATH.write_text(
|
||
json.dumps(current, ensure_ascii=False, indent=2) + "\n",
|
||
encoding="utf-8",
|
||
)
|
||
try:
|
||
os.chmod(SETTINGS_PATH, 0o600)
|
||
except OSError:
|
||
pass
|
||
return public_settings(current)
|
||
|
||
|
||
def public_settings(settings: dict | None = None) -> dict:
|
||
data = settings or load_settings()
|
||
key = data.get("apiKey") or ""
|
||
serp = data.get("serpApiKey") or ""
|
||
cat_key = data.get("categoryApiKey") or ""
|
||
return {
|
||
"hasKey": bool(key),
|
||
"keyHint": _mask_key(key),
|
||
"hasSerpKey": bool(serp),
|
||
"serpKeyHint": _mask_key(serp),
|
||
"hasCategoryKey": bool(cat_key),
|
||
"categoryKeyHint": _mask_key(cat_key),
|
||
"baseUrl": data.get("baseUrl") or DEFAULT_BASE_URL,
|
||
"model": data.get("model") or DEFAULT_MODEL,
|
||
}
|
||
|
||
|
||
def looks_persian(text: str) -> bool:
|
||
letters = [c for c in text if c.isalpha()]
|
||
if len(letters) < 8:
|
||
# Short labels: any Arabic-script char counts
|
||
return bool(ARABIC_SCRIPT.search(text or ""))
|
||
persian = sum(1 for c in letters if ARABIC_SCRIPT.match(c))
|
||
return (persian / len(letters)) >= 0.35
|
||
|
||
|
||
def needs_translation(text: str, *, overwrite: bool = False) -> bool:
|
||
value = (text or "").strip()
|
||
if not value:
|
||
return False
|
||
if overwrite:
|
||
return not looks_persian(value)
|
||
return not looks_persian(value)
|
||
|
||
|
||
def chat_translate(
|
||
text: str,
|
||
*,
|
||
settings: dict | None = None,
|
||
timeout: int = 90,
|
||
system_prompt: str | None = None,
|
||
user_prefix: str | None = None,
|
||
api_key_override: str | None = None,
|
||
) -> str:
|
||
data = settings or load_settings()
|
||
api_key = (api_key_override or data.get("apiKey") or "").strip()
|
||
if not api_key:
|
||
raise ValueError("AI API key is not configured in Settings")
|
||
base = (data.get("baseUrl") or DEFAULT_BASE_URL).rstrip("/")
|
||
model = data.get("model") or DEFAULT_MODEL
|
||
url = f"{base}/chat/completions"
|
||
payload = {
|
||
"model": model,
|
||
"temperature": 0.2,
|
||
"messages": [
|
||
{"role": "system", "content": system_prompt or SYSTEM_PROMPT},
|
||
{
|
||
"role": "user",
|
||
"content": f"{user_prefix or 'Translate this cosmetics product text to Persian:'}\n\n{text}",
|
||
},
|
||
],
|
||
}
|
||
# SerpAPI keys / dashboard URLs cannot call Chat Completions.
|
||
if "serpapi.com" in base.lower():
|
||
raise ValueError(
|
||
"OpenAI base URL is set to SerpAPI. Use https://api.openai.com/v1 for translation."
|
||
)
|
||
if api_key == (data.get("serpApiKey") or "").strip() and api_key:
|
||
raise ValueError(
|
||
"The OpenAI key field currently holds your SerpAPI key. "
|
||
"Paste a real OpenAI (or compatible) key for translation."
|
||
)
|
||
|
||
body = json.dumps(payload).encode("utf-8")
|
||
req = urllib.request.Request(
|
||
url,
|
||
data=body,
|
||
method="POST",
|
||
headers={
|
||
"Authorization": f"Bearer {api_key}",
|
||
"Content-Type": "application/json",
|
||
"Accept": "application/json",
|
||
"User-Agent": "kikomiko-admin/1.0",
|
||
},
|
||
)
|
||
raw = ""
|
||
last_err: Exception | None = None
|
||
for ctx in _ssl_contexts():
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp:
|
||
raw = resp.read().decode("utf-8", "ignore")
|
||
last_err = None
|
||
break
|
||
except urllib.error.HTTPError as exc:
|
||
detail = exc.read().decode("utf-8", "ignore")[:400]
|
||
raise ValueError(f"AI API HTTP {exc.code}: {detail or exc.reason}") from exc
|
||
except urllib.error.URLError as exc:
|
||
last_err = exc
|
||
continue
|
||
if last_err is not None:
|
||
raise ValueError(f"AI API unreachable: {getattr(last_err, 'reason', last_err)}") from last_err
|
||
|
||
try:
|
||
parsed = json.loads(raw)
|
||
except json.JSONDecodeError as exc:
|
||
raise ValueError("AI API returned invalid JSON") from exc
|
||
|
||
try:
|
||
content = parsed["choices"][0]["message"]["content"]
|
||
except (KeyError, IndexError, TypeError) as exc:
|
||
raise ValueError("AI API response missing translation content") from exc
|
||
|
||
out = str(content or "").strip()
|
||
if out.startswith("```"):
|
||
out = re.sub(r"^```(?:\w+)?\s*", "", out)
|
||
out = re.sub(r"\s*```$", "", out)
|
||
if (out.startswith('"') and out.endswith('"')) or (
|
||
out.startswith("«") and out.endswith("»")
|
||
):
|
||
out = out[1:-1].strip()
|
||
if not out:
|
||
raise ValueError("AI API returned an empty translation")
|
||
return out
|
||
|
||
|
||
def collect_jobs(product: dict, *, overwrite: bool = False) -> list[tuple[str, str]]:
|
||
"""Return list of (path, text) that still need Persian translation."""
|
||
jobs: list[tuple[str, str]] = []
|
||
desc = str(product.get("description") or "").strip()
|
||
if needs_translation(desc, overwrite=overwrite):
|
||
jobs.append(("description", desc))
|
||
|
||
for key in ("descriptionIntro", "tagline", "seoDescription", "seoTitle"):
|
||
value = str(product.get(key) or "").strip()
|
||
if needs_translation(value, overwrite=overwrite):
|
||
jobs.append((key, value))
|
||
|
||
variants = product.get("colorVariants") or product.get("color_variants") or []
|
||
if isinstance(variants, list):
|
||
for i, variant in enumerate(variants):
|
||
if not isinstance(variant, dict):
|
||
continue
|
||
value = str(variant.get("description") or "").strip()
|
||
if needs_translation(value, overwrite=overwrite):
|
||
jobs.append((f"colorVariants.{i}.description", value))
|
||
return jobs
|
||
|
||
|
||
def apply_translation(product: dict, path: str, translated: str) -> None:
|
||
if path == "description":
|
||
if not (product.get("sourceDescription") or "").strip():
|
||
product["sourceDescription"] = product.get("description") or ""
|
||
product["description"] = translated
|
||
return
|
||
if path.startswith("colorVariants."):
|
||
parts = path.split(".")
|
||
idx = int(parts[1])
|
||
variants = product.get("colorVariants") or []
|
||
if isinstance(variants, list) and 0 <= idx < len(variants):
|
||
variants[idx]["description"] = translated
|
||
product["colorVariants"] = variants
|
||
return
|
||
product[path] = translated
|
||
|
||
|
||
def translate_product(
|
||
product: dict,
|
||
*,
|
||
overwrite: bool = False,
|
||
settings: dict | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Translate pending English fields on one product. Mutates and returns stats."""
|
||
jobs = collect_jobs(product, overwrite=overwrite)
|
||
changed: list[str] = []
|
||
errors: list[str] = []
|
||
for path, text in jobs:
|
||
try:
|
||
translated = chat_translate(text, settings=settings)
|
||
apply_translation(product, path, translated)
|
||
changed.append(path)
|
||
except Exception as exc: # noqa: BLE001 — surface per-field failures
|
||
errors.append(f"{path}: {exc}")
|
||
break # stop this product on first API failure (often key/quota)
|
||
return {
|
||
"id": product.get("id"),
|
||
"slug": product.get("slug"),
|
||
"changed": changed,
|
||
"errors": errors,
|
||
"pending": len(jobs),
|
||
}
|
||
|
||
|
||
def category_needs_fa(category: dict, *, overwrite: bool = False) -> bool:
|
||
"""True when nameFa is missing or still English/non-Persian."""
|
||
name = str(category.get("name") or category.get("slug") or "").strip()
|
||
name_fa = str(category.get("nameFa") or category.get("name_fa") or "").strip()
|
||
if not name:
|
||
return False
|
||
if not name_fa:
|
||
return True
|
||
if overwrite:
|
||
return not looks_persian(name_fa)
|
||
return not looks_persian(name_fa)
|
||
|
||
|
||
def resolve_category_api_key(settings: dict | None = None) -> str:
|
||
"""OpenAI-compatible key for categories; never reuse a SerpAPI key by mistake."""
|
||
data = settings or load_settings()
|
||
cat = (data.get("categoryApiKey") or "").strip()
|
||
main = (data.get("apiKey") or "").strip()
|
||
serp = (data.get("serpApiKey") or "").strip()
|
||
if cat and cat != serp:
|
||
return cat
|
||
if main and main != serp:
|
||
return main
|
||
return ""
|
||
|
||
|
||
def local_category_fa(name: str, slug: str = "") -> str:
|
||
"""Dictionary / seed Persian label without calling an external API."""
|
||
try:
|
||
import server as admin_server # lazy: avoids import cycle at module load
|
||
|
||
for candidate in (
|
||
admin_server.seed_name_fa_by_slug().get(slug or "", ""),
|
||
admin_server.persian_category_label(name),
|
||
admin_server.persian_category_label((slug or "").replace("-", " ")),
|
||
admin_server.persian_category_label((slug or "").replace("-", " / ")),
|
||
):
|
||
text = str(candidate or "").strip()
|
||
if text and looks_persian(text):
|
||
return text
|
||
except Exception:
|
||
pass
|
||
return ""
|
||
|
||
|
||
def translate_category_name(
|
||
category: dict,
|
||
*,
|
||
overwrite: bool = False,
|
||
settings: dict | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Translate one category English name → Persian nameFa. Mutates category.
|
||
|
||
Prefers local seed/KIKO maps (works offline); falls back to AI when needed.
|
||
"""
|
||
data = settings or load_settings()
|
||
if not category_needs_fa(category, overwrite=overwrite):
|
||
return {
|
||
"slug": category.get("slug"),
|
||
"changed": False,
|
||
"nameFa": category.get("nameFa") or category.get("name_fa") or "",
|
||
"source": "existing",
|
||
"errors": [],
|
||
}
|
||
source = str(category.get("name") or category.get("slug") or "").strip()
|
||
slug = str(category.get("slug") or "").strip()
|
||
|
||
local = local_category_fa(source, slug)
|
||
if local:
|
||
category["nameFa"] = local
|
||
category["name_fa"] = local
|
||
return {
|
||
"slug": slug,
|
||
"changed": True,
|
||
"from": source,
|
||
"nameFa": local,
|
||
"source": "local",
|
||
"errors": [],
|
||
}
|
||
|
||
api_key = resolve_category_api_key(data)
|
||
if not api_key:
|
||
raise ValueError(
|
||
"No Persian mapping for this category and no OpenAI key is set. "
|
||
"Save a real OpenAI API key (not the SerpAPI key) in Settings, or add the label to seed_categories."
|
||
)
|
||
translated = chat_translate(
|
||
source,
|
||
settings=data,
|
||
system_prompt=CATEGORY_SYSTEM_PROMPT,
|
||
user_prefix="Translate this cosmetics category name to a short Persian menu label:",
|
||
api_key_override=api_key,
|
||
)
|
||
if not looks_persian(translated):
|
||
raise ValueError(f"AI returned non-Persian category label: {translated!r}")
|
||
category["nameFa"] = translated
|
||
category["name_fa"] = translated
|
||
return {
|
||
"slug": slug,
|
||
"changed": True,
|
||
"from": source,
|
||
"nameFa": translated,
|
||
"source": "ai",
|
||
"errors": [],
|
||
}
|