first deployment of kikomiko with url photos
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
shayan khoshrou
co-authored by
Cursor
Cursor
parent
8d98304137
commit
ae3ed7c7b7
@@ -0,0 +1,420 @@
|
||||
"""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": [],
|
||||
}
|
||||
Reference in New Issue
Block a user