#!/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{% if source_text %}
Source Text
{{ source_text }}
{% for trans in marian %}- {{ trans }}
{% endfor %}
{% for variant_name, variant_data in libre.items() %}{% if variant_data.success %}
{% else %}
{% 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,
/detectCopied to clipboard!
"""
API_DOCS_TEMPLATE = """Kabyle Translation Hub - API Docs← Back to translatorKabyle Translation Hub — API Docs
Base URL: {{ base_url }}
POST/api/translate
Native API for translation with multiple engines and variants.
| Field | Type | Required | Description |
|---|
text | string | yes | English text to translate |
engines | array | no | Subset of ["marian","libre"]. Defaults to both. |
variants | array | no | Subset 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.
| Field | Type | Required | Description |
|---|
q | string or array | yes | Text to translate |
source | string | yes | Source language code |
target | string | yes | Target language code |
api_key | string | no | Ignored |
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)