#!/usr/bin/env python3 """ Kabyle Translation Hub - Hugging Face Spaces Edition """ import warnings warnings.filterwarnings("ignore", category=FutureWarning) import os import time import requests import torch from flask import Flask, request, render_template_string, jsonify from flask_cors import CORS from transformers import AutoTokenizer, AutoModelForSeq2SeqLM from concurrent.futures import ThreadPoolExecutor, as_completed # Configuration LIBRETRANSLATE_URL = os.environ.get("LIBRETRANSLATE_URL", "https://imsidag-community-libretranslate-kabyle.hf.space/translate") MODEL_ID = "boffire/marianmt-en-kab" KABYLE_VARIANTS = { "Taqbaylit (Standard)": "kab", "Taqbaylit (Latest)": "kab_kab", "Taqbaylit (Tasenselkimt)": "kab_comp", "Taqbaylit (51000)": "kab_comp2", "Taqbaylit (OS)": "kab_os", "Taqbaylit (Num)": "kab_num", } model = None tokenizer = None device = None # Rate limiting protection _last_libre_request = 0 _MIN_LIBRE_INTERVAL = 0.5 # seconds between LibreTranslate requests def load_model(): global model, tokenizer, device if model is None: print("Loading MarianMT model...", flush=True) device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') tokenizer = AutoTokenizer.from_pretrained(MODEL_ID, use_fast=False) model = AutoModelForSeq2SeqLM.from_pretrained(MODEL_ID).to(device).eval() print(f"Model loaded on {device}", flush=True) return model, tokenizer, device def translate_marian(text): if not text or not text.strip(): return ["Please enter text to translate"] try: model, tokenizer, device = load_model() inputs = tokenizer(text, return_tensors='pt', padding=True, truncation=True, max_length=512) inputs = {k: v.to(device) for k, v in inputs.items()} with torch.no_grad(): outputs = model.generate( **inputs, num_beams=4, num_return_sequences=3, max_length=128, early_stopping=True, do_sample=False, ) translations = [] for output in outputs: trans = tokenizer.decode(output, skip_special_tokens=True) if trans and trans not in translations: translations.append(trans) return translations if translations else ["[Error: No translation generated]"] except Exception as e: print(f"MarianMT error: {e}", flush=True) import traceback traceback.print_exc() return [f"[Error: {str(e)}]"] def translate_libre_variant(text, variant_code): global _last_libre_request # Rate limit protection elapsed = time.time() - _last_libre_request if elapsed < _MIN_LIBRE_INTERVAL: time.sleep(_MIN_LIBRE_INTERVAL - elapsed) _last_libre_request = time.time() try: r = requests.post( LIBRETRANSLATE_URL, headers={"Content-Type": "application/json"}, json={"q": text, "source": "en", "target": variant_code}, timeout=10 ) r.raise_for_status() result = r.json().get("translatedText", "[Error: No translation]") return {"success": True, "text": result} except requests.exceptions.HTTPError as e: if e.response.status_code == 429: return {"success": False, "text": "[Error: Rate limited by LibreTranslate]"} return {"success": False, "text": f"[Error: {str(e)[:50]}]"} except Exception as e: return {"success": False, "text": f"[Error: {str(e)[:50]}]"} def translate_libre_all_variants(text): results = {} # Sequential instead of parallel to avoid rate limiting for name, code in KABYLE_VARIANTS.items(): results[name] = translate_libre_variant(text, code) time.sleep(0.3) # Small delay between requests return results # --- HTML TEMPLATES --- HTML_TEMPLATE = """Kabyle Translation Hub

English → Kabyle

Choose the translation that suits you best

{% if source_text %}
Source Text
{{ source_text }}
MarianMT (Neural)
{{ marian|length }} options
    {% for trans in marian %}
  • {{ trans }}
  • {% endfor %}
LibreTranslate Variants
{{ libre|length }} variants
{% for variant_name, variant_data in libre.items() %}{% if variant_data.success %}
{{ variant_name }}{{ variant_data.code }}
{{ variant_data.text }}
{% else %}
{{ variant_name }}{{ variant_data.code }}
{{ variant_data.text }}
{% endif %}{% endfor %}
{% else %}

Enter text above and click Translate to see results

{% endif %}
API available at POST /api/translate — see full API documentation.
LibreTranslate-compatible endpoints: /translate, /languages, /detect
Copied to clipboard!
""" API_DOCS_TEMPLATE = """Kabyle Translation Hub - API Docs
← Back to translator

Kabyle Translation Hub — API Docs

Base URL: {{ base_url }}

POST/api/translate

Native API for translation with multiple engines and variants.

FieldTypeRequiredDescription
textstringyesEnglish text to translate
enginesarraynoSubset of ["marian","libre"]. Defaults to both.
variantsarraynoSubset of LibreTranslate variant codes. Defaults to all.
curl -X POST "{{ base_url }}/api/translate" \\\n  -H "Content-Type: application/json" \\\n  -d '{"text": "Good morning", "engines": ["marian", "libre"]}'

POST/translate

LibreTranslate-compatible endpoint for Weblate and other tools.

FieldTypeRequiredDescription
qstring or arrayyesText to translate
sourcestringyesSource language code
targetstringyesTarget language code
api_keystringnoIgnored
curl -X POST "{{ base_url }}/translate" \\\n  -H "Content-Type: application/json" \\\n  -d '{"q": "Good morning", "source": "en", "target": "kab"}'

Weblate config:

MT_LIBRETRANSLATE_API_URL = "{{ base_url }}"

GET/languages

Supported language pairs.

curl "{{ base_url }}/languages"

POST/detect

Language detection stub.

curl -X POST "{{ base_url }}/detect" -d "q=Hello world"

GET/health

Health check.

curl "{{ base_url }}/health"
""" app = Flask(__name__) CORS(app, resources={r"/*": {"origins": "*"}}) # Global error handler - NEVER return 500 @app.errorhandler(Exception) def handle_exception(e): print(f"[GLOBAL ERROR] {type(e).__name__}: {e}", flush=True) import traceback traceback.print_exc() return jsonify({"translatedText": "", "error": str(e)}), 200 # Pre-load model try: load_model() except Exception as e: print(f"Pre-load failed: {e}", flush=True) @app.route("/", methods=["GET", "POST"]) def index(): marian = [] libre = {} source_text = "" if request.method == "POST": source_text = request.form.get("text", "").strip() if source_text: marian = translate_marian(source_text) libre_results = translate_libre_all_variants(source_text) for name, data in libre_results.items(): data['code'] = KABYLE_VARIANTS[name] libre[name] = data return render_template_string(HTML_TEMPLATE, marian=marian, libre=libre, source_text=source_text) def _extract_text(data): q = data.get("q", "") if isinstance(q, list): return q[0] if q else "" return str(q).strip() @app.route("/translate", methods=["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"]) def libretranslate_translate(): """ Weblate-compatible translation endpoint. Uses ONLY MarianMT - no LibreTranslate fallback to avoid 429 errors. NEVER returns 500. """ print(f"[WEblate] === /translate called ({request.method}) ===", flush=True) if request.method == "GET": return jsonify({"translatedText": ""}), 200 try: raw_data = request.get_data(as_text=True) print(f"[WEblate] Raw body: {raw_data[:500]}", flush=True) except Exception as e: print(f"[WEblate] Could not read raw body: {e}", flush=True) raw_data = "" data = {} try: json_data = request.get_json(silent=True, force=True) if json_data: data = json_data except Exception as e: print(f"[WEblate] JSON parse failed: {e}", flush=True) if not data: try: form_data = request.form if form_data: data = dict(form_data) except Exception as e: print(f"[WEblate] Form parse failed: {e}", flush=True) if not data and raw_data: try: import json data = json.loads(raw_data) except Exception as e: print(f"[WEblate] Manual JSON parse failed: {e}", flush=True) text = _extract_text(data) source = str(data.get("source", "en")).strip() if data else "en" target = str(data.get("target", "kab")).strip() if data else "kab" print(f"[WEblate] text={text[:50]}, source={source}, target={target}", flush=True) if not text: return jsonify({"translatedText": ""}), 200 if source == "auto": kab_chars = set("ɛƐɣƔṭḍčǧḥṣẓ") source = "kab" if any(c in text.lower() for c in kab_chars) else "en" if source != "en": return jsonify({"translatedText": ""}), 200 # Only use MarianMT - no LibreTranslate fallback to avoid 429s try: marian_results = translate_marian(text) if marian_results and not marian_results[0].startswith("[Error"): return jsonify({"translatedText": marian_results[0]}), 200 except Exception as e: print(f"[WEblate] MarianMT failed: {e}", flush=True) return jsonify({"translatedText": ""}), 200 @app.route("/languages", methods=["GET"]) def libretranslate_languages(): try: return jsonify([ {"code": "en", "name": "English", "targets": ["kab"]}, {"code": "kab", "name": "Kabyle", "targets": ["en"]} ]) except Exception as e: print(f"[WEblate] /languages error: {e}", flush=True) return jsonify([]), 200 @app.route("/detect", methods=["POST"]) def libretranslate_detect(): try: data = request.get_json(silent=True) or {} if not data: data = dict(request.form) text = _extract_text(data) kab_chars = set("ɛƐɣƔṭḍčǧḥṣẓ") detected = "kab" if any(c in text.lower() for c in kab_chars) else "en" return jsonify([{"confidence": 0.99, "language": detected}]) except Exception as e: print(f"[WEblate] /detect error: {e}", flush=True) return jsonify([]), 200 @app.route("/frontend/settings", methods=["GET"]) def libretranslate_settings(): return jsonify({ "apiKeys": False, "charLimit": 5000, "dailyLimit": -1, "frontendTimeout": 5000, "keyRequired": False, "language": {"source": {"code": "en", "name": "English"}, "target": {"code": "kab", "name": "Kabyle"}}, "suggestions": False, "supportedFilesFormat": ["text", "html"], "theme": "auto" }) @app.route("/api/docs", methods=["GET"]) def api_docs(): base_url = request.url_root.rstrip("/") return render_template_string(API_DOCS_TEMPLATE, base_url=base_url) @app.route("/api/translate", methods=["GET", "POST"]) def api_translate(): if request.method == "POST": payload = request.get_json(silent=True) or {} text = payload.get("text") or request.form.get("text", "") engines = payload.get("engines") or ["marian", "libre"] variant_filter = payload.get("variants") else: text = request.args.get("text", "") engines_param = request.args.get("engines", "marian,libre") engines = [e.strip() for e in engines_param.split(",") if e.strip()] variants_param = request.args.get("variants") variant_filter = [v.strip() for v in variants_param.split(",")] if variants_param else None text = (text or "").strip() if not text: return jsonify({"error": "Missing 'text' parameter"}), 400 engines = [e.lower() for e in engines] result = {"source_text": text} if "marian" in engines: result["marian"] = translate_marian(text) if "libre" in engines: variants_to_use = KABYLE_VARIANTS if variant_filter: variants_to_use = { name: code for name, code in KABYLE_VARIANTS.items() if code in variant_filter or name in variant_filter } if not variants_to_use: return jsonify({"error": f"No matching variants for: {variant_filter}"}), 400 libre_results = {} for name, code in variants_to_use.items(): libre_results[name] = translate_libre_variant(text, code) time.sleep(0.3) for name, data in libre_results.items(): data["code"] = code result["libre"] = libre_results return jsonify(result) @app.route("/api/variants", methods=["GET"]) def api_variants(): return jsonify(KABYLE_VARIANTS) @app.route("/health") def health(): return jsonify({ "status": "healthy", "model_loaded": model is not None }) if __name__ == "__main__": port = int(os.environ.get("PORT", 7860)) app.run(host="0.0.0.0", port=port, debug=False)