import os from functools import wraps from flask import Flask, abort, jsonify, render_template, request from flask_swagger import swagger from flask_swagger_ui import get_swaggerui_blueprint from app import flood from app.language import detect_languages, transliterate from .api_keys import Database from translatehtml import translate_html def get_json_dict(request): d = request.get_json() if not isinstance(d, dict): abort(400, description="Invalid JSON format") return d def get_remote_address(): if request.headers.getlist("X-Forwarded-For"): ip = request.headers.getlist("X-Forwarded-For")[0].split(",")[0] else: ip = request.remote_addr or "127.0.0.1" return ip def get_req_limits(default_limit, api_keys_db, multiplier = 1): req_limit = default_limit if api_keys_db: if request.is_json: json = get_json_dict(request) api_key = json.get("api_key") else: api_key = request.values.get("api_key") if api_key: db_req_limit = api_keys_db.lookup(api_key) if db_req_limit is not None: req_limit = db_req_limit * multiplier return req_limit def get_routes_limits(default_req_limit, daily_req_limit, api_keys_db): if default_req_limit == -1: # TODO: better way? default_req_limit = 9999999999999 def minute_limits(): return "%s per minute" % get_req_limits(default_req_limit, api_keys_db) def daily_limits(): return "%s per day" % get_req_limits(daily_req_limit, api_keys_db, 1440) res = [minute_limits] if daily_req_limit > 0: res.append(daily_limits) return res def create_app(args): from app.init import boot boot(args.load_only) from app.language import languages app = Flask(__name__) if args.debug: app.config["TEMPLATES_AUTO_RELOAD"] = True # Map userdefined frontend languages to argos language object. if args.frontend_language_source == "auto": frontend_argos_language_source = type( "obj", (object,), {"code": "auto", "name": "Auto Detect"} ) else: frontend_argos_language_source = next( iter([l for l in languages if l.code == args.frontend_language_source]), None, ) frontend_argos_language_target = next( iter([l for l in languages if l.code == args.frontend_language_target]), None ) # Raise AttributeError to prevent app startup if user input is not valid. if frontend_argos_language_source is None: raise AttributeError( f"{args.frontend_language_source} as frontend source language is not supported." ) if frontend_argos_language_target is None: raise AttributeError( f"{args.frontend_language_target} as frontend target language is not supported." ) api_keys_db = None if args.req_limit > 0 or args.api_keys or args.daily_req_limit > 0: api_keys_db = Database() if args.api_keys else None from flask_limiter import Limiter limiter = Limiter( app, key_func=get_remote_address, default_limits=get_routes_limits( args.req_limit, args.daily_req_limit, api_keys_db ), ) else: from .no_limiter import Limiter limiter = Limiter() if args.req_flood_threshold > 0: flood.setup(args.req_flood_threshold) def access_check(f): @wraps(f) def func(*a, **kw): if flood.is_banned(get_remote_address()): abort(403, description="Too many request limits violations") if args.api_keys and args.require_api_key_origin: if request.is_json: json = get_json_dict(request) ak = json.get("api_key") else: ak = request.values.get("api_key") if ( api_keys_db.lookup(ak) is None and request.headers.get("Origin") != args.require_api_key_origin ): abort( 403, description="Please contact the server operator to obtain an API key", ) return f(*a, **kw) return func @app.errorhandler(400) def invalid_api(e): return jsonify({"error": str(e.description)}), 400 @app.errorhandler(500) def server_error(e): return jsonify({"error": str(e.description)}), 500 @app.errorhandler(429) def slow_down_error(e): flood.report(get_remote_address()) return jsonify({"error": "Slowdown: " + str(e.description)}), 429 @app.errorhandler(403) def denied(e): return jsonify({"error": str(e.description)}), 403 @app.route("/") @limiter.exempt def index(): return render_template( "index.html", gaId=args.ga_id, frontendTimeout=args.frontend_timeout, api_keys=args.api_keys, web_version=os.environ.get("LT_WEB") is not None, ) @app.route("/javascript-licenses", methods=["GET"]) @limiter.exempt def javascript_licenses(): return render_template("javascript-licenses.html") @app.route("/languages", methods=["GET", "POST"]) @limiter.exempt def langs(): """ Retrieve list of supported languages --- tags: - translate responses: 200: description: List of languages schema: id: languages type: array items: type: object properties: code: type: string description: Language code name: type: string description: Human-readable language name (in English) 429: description: Slow down schema: id: error-slow-down type: object properties: error: type: string description: Reason for slow down """ return jsonify([{"code": l.code, "name": l.name} for l in languages]) # Add cors @app.after_request def after_request(response): response.headers.add("Access-Control-Allow-Origin", "*") response.headers.add( "Access-Control-Allow-Headers", "Authorization, Content-Type" ) response.headers.add("Access-Control-Expose-Headers", "Authorization") response.headers.add("Access-Control-Allow-Methods", "GET, POST") response.headers.add("Access-Control-Allow-Credentials", "true") response.headers.add("Access-Control-Max-Age", 60 * 60 * 24 * 20) return response @app.route("/translate", methods=["POST"]) @access_check def translate(): """ Translate text from a language to another --- tags: - translate parameters: - in: formData name: q schema: oneOf: - type: string example: Hello world! - type: array example: ['Hello world!'] required: true description: Text(s) to translate - in: formData name: source schema: type: string example: en required: true description: Source language code - in: formData name: target schema: type: string example: es required: true description: Target language code - in: formData name: format schema: type: string example: html or text required: false description: The format of the source text - in: formData name: api_key schema: type: string example: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx required: false description: API key responses: 200: description: Translated text schema: id: translate type: object properties: translatedText: oneOf: - type: string - type: array description: Translated text(s) 400: description: Invalid request schema: id: error-response type: object properties: error: type: string description: Error message 500: description: Translation error schema: id: error-response type: object properties: error: type: string description: Error message 429: description: Slow down schema: id: error-slow-down type: object properties: error: type: string description: Reason for slow down 403: description: Banned schema: id: error-response type: object properties: error: type: string description: Error message """ if request.is_json: json = get_json_dict(request) q = json.get("q") source_lang = json.get("source") target_lang = json.get("target") text_format = json.get("format") else: q = request.values.get("q") source_lang = request.values.get("source") target_lang = request.values.get("target") text_format = request.values.get("format") if not q: abort(400, description="Invalid request: missing q parameter") if not source_lang: abort(400, description="Invalid request: missing source parameter") if not target_lang: abort(400, description="Invalid request: missing target parameter") batch = isinstance(q, list) if batch and args.batch_limit != -1: batch_size = len(q) if args.batch_limit < batch_size: abort( 400, description="Invalid request: Request (%d) exceeds text limit (%d)" % (batch_size, args.batch_limit), ) if args.char_limit != -1: if batch: chars = sum([len(text) for text in q]) else: chars = len(q) if args.char_limit < chars: abort( 400, description="Invalid request: Request (%d) exceeds character limit (%d)" % (chars, args.char_limit), ) if source_lang == "auto": source_langs = [] if batch: auto_detect_texts = q else: auto_detect_texts = [q] overall_candidates = detect_languages(q) for text_to_check in auto_detect_texts: if len(text_to_check) > 40: candidate_langs = detect_languages(text_to_check) else: # Unable to accurately detect languages for short texts candidate_langs = overall_candidates source_langs.append(candidate_langs[0]["language"]) if args.debug: print(text_to_check, candidate_langs) print("Auto detected: %s" % candidate_langs[0]["language"]) else: if batch: source_langs = [source_lang for text in q] else: source_langs = [source_lang] src_langs = [next(iter([l for l in languages if l.code == source_lang]), None) for source_lang in source_langs] for idx, lang in enumerate(src_langs): if lang is None: abort(400, description="%s is not supported" % source_langs[idx]) tgt_lang = next(iter([l for l in languages if l.code == target_lang]), None) if tgt_lang is None: abort(400, description="%s is not supported" % target_lang) if not text_format: text_format = "text" if text_format not in ["text", "html"]: abort(400, description="%s format is not supported" % text_format) try: if batch: results = [] for idx, text in enumerate(q): translator = src_langs[idx].get_translation(tgt_lang) if text_format == "html": translated_text = str(translate_html(translator, transliterate(text, target_lang=source_langs[idx]))) else: translated_text = translator.translate(transliterate(text, target_lang=source_langs[idx])) results.append(translated_text) return jsonify( { "translatedText": results } ) else: translator = src_langs[0].get_translation(tgt_lang) if text_format == "html": translated_text = str(translate_html(translator, transliterate(q, target_lang=source_langs[0]))) else: translated_text = translator.translate(transliterate(q, target_lang=source_langs[0])) return jsonify( { "translatedText": translated_text } ) except Exception as e: abort(500, description="Cannot translate text: %s" % str(e)) @app.route("/detect", methods=["POST"]) @access_check def detect(): """ Detect the language of a single text --- tags: - translate parameters: - in: formData name: q schema: type: string example: Hello world! required: true description: Text to detect - in: formData name: api_key schema: type: string example: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx required: false description: API key responses: 200: description: Detections schema: id: detections type: array items: type: object properties: confidence: type: number format: float minimum: 0 maximum: 1 description: Confidence value example: 0.6 language: type: string description: Language code example: en 400: description: Invalid request schema: id: error-response type: object properties: error: type: string description: Error message 500: description: Detection error schema: id: error-response type: object properties: error: type: string description: Error message 429: description: Slow down schema: id: error-slow-down type: object properties: error: type: string description: Reason for slow down 403: description: Banned schema: id: error-response type: object properties: error: type: string description: Error message """ if flood.is_banned(get_remote_address()): abort(403, description="Too many request limits violations") if request.is_json: json = get_json_dict(request) q = json.get("q") else: q = request.values.get("q") if not q: abort(400, description="Invalid request: missing q parameter") return jsonify(detect_languages(q)) @app.route("/frontend/settings") @limiter.exempt def frontend_settings(): """ Retrieve frontend specific settings --- tags: - frontend responses: 200: description: frontend settings schema: id: frontend-settings type: object properties: charLimit: type: integer description: Character input limit for this language (-1 indicates no limit) frontendTimeout: type: integer description: Frontend translation timeout language: type: object properties: source: type: object properties: code: type: string description: Language code name: type: string description: Human-readable language name (in English) target: type: object properties: code: type: string description: Language code name: type: string description: Human-readable language name (in English) """ return jsonify( { "charLimit": args.char_limit, "frontendTimeout": args.frontend_timeout, "language": { "source": { "code": frontend_argos_language_source.code, "name": frontend_argos_language_source.name, }, "target": { "code": frontend_argos_language_target.code, "name": frontend_argos_language_target.name, }, }, } ) swag = swagger(app) swag["info"]["version"] = "1.2" swag["info"]["title"] = "LibreTranslate" @app.route("/spec") @limiter.exempt def spec(): return jsonify(swag) SWAGGER_URL = "/docs" # URL for exposing Swagger UI (without trailing '/') API_URL = "/spec" # Call factory function to create our blueprint swaggerui_blueprint = get_swaggerui_blueprint(SWAGGER_URL, API_URL) app.register_blueprint(swaggerui_blueprint) return app