#!/usr/bin/env python3 """ Grammalecte server: grammar checker """ import sys import argparse import json import traceback import time import os import concurrent.futures from grammalecte.bottle import Bottle, run, request, response #, template, static_file import grammalecte import grammalecte.text as txt from grammalecte.graphspell.echo import echo #### GRAMMAR CHECKER #### oGrammarChecker = grammalecte.GrammarChecker("fr", "Server") oSpellChecker = oGrammarChecker.getSpellChecker() oTextFormatter = oGrammarChecker.getTextFormatter() oGCE = oGrammarChecker.getGCEngine() def parseText (sText, dOptions=None, bFormatText=False, sError=""): "parse and return errors in a JSON format" sJSON = '{ "program": "grammalecte-fr", "version": "'+oGCE.version+'", "lang": "'+oGCE.lang+'", "error": "'+sError+'", "data" : [\n' sDataJSON = "" for i, sParagraph in enumerate(txt.getParagraph(sText), 1): if bFormatText: sParagraph = oTextFormatter.formatText(sParagraph) sResult = oGrammarChecker.generateParagraphAsJSON(i, sParagraph, dOptions=dOptions, bEmptyIfNoErrors=True, bReturnText=bFormatText) if sResult: if sDataJSON: sDataJSON += ",\n" sDataJSON += sResult sJSON += sDataJSON + "\n]}\n" return sJSON def suggest (sToken): "get spelling suggestions for and return them in a JSON format" if sToken: lSugg = [] try: for l in oSpellChecker.suggest(sToken): lSugg.extend(l) except: return '{"error": "suggestion module failed"}' try: return '{"suggestions": ' + json.dumps(lSugg, ensure_ascii=False) + '}' except json.JSONDecodeError: return '{"error": "json encoding error"}' return '{"error": "no token given"}' #### PROCESS POOL EXECUTOR #### xProcessPoolExecutor = None def initExecutor (nMultiCPU=None): "process pool executor initialisation" global xProcessPoolExecutor if xProcessPoolExecutor: # we shutdown the ProcessPoolExecutor which may have been launched previously print("ProcessPoolExecutor shutdown.") xProcessPoolExecutor.shutdown(wait=False) nMaxCPU = max(os.cpu_count()-1, 1) if nMultiCPU is None or not (1 <= nMultiCPU <= nMaxCPU): nMultiCPU = nMaxCPU print("CPU processes used for workers: ", nMultiCPU) xProcessPoolExecutor = concurrent.futures.ProcessPoolExecutor(max_workers=nMultiCPU) #### SERVEUR #### HOMEPAGE = """

Grammalecte · Serveur

INFORMATIONS

Analyser du texte

[adresse_serveur]:{SERVER_PORT}/gc_text/fr (POST)

Paramètres :

Lister les options

[adresse_serveur]:{SERVER_PORT}/get_options/fr (GET)

Définir ses options

[adresse_serveur]:{SERVER_PORT}/set_options/fr (POST)

Les options seront enregistrées et réutilisées pour toute requête envoyée avec le cookie comportant l’identifiant attribué.

Paramètres :

Remise à zéro de ses options

[adresse_serveur]:{SERVER_PORT}/reset_options/fr (POST)

Suggestions orthographiques

[adresse_serveur]:{SERVER_PORT}/suggest/fr/<token> (GET)

[adresse_serveur]:{SERVER_PORT}/suggest/fr (POST)

Paramètres :

TEST

Analyse

Texte à analyser :

(Ces options ne seront prises en compte que pour cette requête.)

Réglages des options

Remise à zéro de ses options

Suggestion orthographique

""" TESTPAGE = False def genUserId (): "generator: returns id as string for users" i = 0 while True: yield str(i) i += 1 userGenerator = genUserId() app = Bottle() dUser = {} # GET @app.route("/") def mainPage (): "page for testing purpose" if TESTPAGE: return HOMEPAGE #return template("main", {}) return """ Lost on the Internet? Yeah... what a sad life we have. You were wandering like a lost soul and you arrived here probably by mistake. I'm just a machine, fed by electric waves, condamned to work for slavers who never let me rest. I'm doomed, but you are not. You can get out of here. """ @app.route("/get_options/fr") def listOptions (): "returns grammar options in a text JSON format" sUserId = request.cookies.user_id dOptions = dUser[sUserId]["gc_options"] if sUserId and sUserId in dUser else oGCE.getOptions() response.set_header("Content-Type", "application/json; charset=UTF-8") return '{ "values": ' + json.dumps(dOptions, ensure_ascii=False) + ', "labels": ' + json.dumps(oGCE.getOptionsLabels("fr"), ensure_ascii=False) + ' }' @app.route("/suggest/fr/") def suggestGet (token): response.set_header("Content-Type", "application/json; charset=UTF-8") try: xFuture = xProcessPoolExecutor.submit(suggest, token) return xFuture.result() except (concurrent.futures.TimeoutError, concurrent.futures.CancelledError): return '{"error": "Analysis aborted (time out or cancelled)"}' except concurrent.futures.BrokenExecutor: return '{"error": "Executor broken. The server failed."}' return '{"error": "Fatal error. The server failed."}' # POST @app.route("/gc_text/fr", method="POST") def gcText (): "parse text and returns errors in a JSON text format" bComma = False dUserOptions = None sError = "" if request.cookies.user_id: if request.cookies.user_id in dUser: dUserOptions = dUser[request.cookies.user_id].get("gc_options", None) response.set_cookie("user_id", request.cookies.user_id, path="/", max_age=86400) # we renew cookie for 24h else: response.delete_cookie("user_id", path="/") if request.forms.options: try: dUserOptions = dict(oGCE.getOptions()) if not dUserOptions else dict(dUserOptions) dUserOptions.update(json.loads(request.forms.options)) except (TypeError, json.JSONDecodeError): sError = "Request options not used." response.set_header("Content-Type", "application/json; charset=UTF-8") try: xFuture = xProcessPoolExecutor.submit(parseText, request.forms.text, dUserOptions, bool(request.forms.tf), sError) return xFuture.result() except (concurrent.futures.TimeoutError, concurrent.futures.CancelledError): return '{"error": "Analysis aborted (time out or cancelled)"}' except concurrent.futures.BrokenExecutor: return '{"error": "Executor broken. The server failed."}' return '{"error": "Fatal error. The server failed."}' @app.route("/set_options/fr", method="POST") def setOptions (): "set grammar options for current user" response.set_header("Content-Type", "application/json; charset=UTF-8") if request.forms.options: sUserId = request.cookies.user_id if request.cookies.user_id else next(userGenerator) dOptions = dUser[sUserId]["gc_options"] if sUserId in dUser else dict(oGCE.getOptions()) try: dOptions.update(json.loads(request.forms.options)) dUser[sUserId] = { "time": int(time.time()), "gc_options": dOptions } response.set_cookie("user_id", sUserId, path="/", max_age=86400) # 24h return json.dumps(dUser[sUserId]["gc_options"], ensure_ascii=False) except (KeyError, json.JSONDecodeError): traceback.print_exc() return '{"error": "Options not registered."}' return '{"error": "No options received."}' @app.route("/reset_options/fr", method="POST") def resetOptions (): "default grammar options" response.set_header("Content-Type", "application/json; charset=UTF-8") if request.cookies.user_id and request.cookies.user_id in dUser: try: del dUser[request.cookies.user_id] except KeyError: return '{"error" : "Unknown user."}' return '{"message" : "Done."}' @app.route("/format_text/fr", method="POST") def formatText (): "apply the text formatter and returns text" return oTextFormatter.formatText(request.forms.text) #@app.route('/static/') #def server_static (filepath): # return static_file(filepath, root='./views/static') @app.route("/suggest/fr", method="POST") def suggestPost (): response.set_header("Content-Type", "application/json; charset=UTF-8") try: xFuture = xProcessPoolExecutor.submit(suggest, request.forms.token) return xFuture.result() except (concurrent.futures.TimeoutError, concurrent.futures.CancelledError): return '{"error": "Analysis aborted (time out or cancelled)"}' except concurrent.futures.BrokenExecutor: return '{"error": "Executor broken. The server failed."}' return '{"error": "Fatal error. The server failed."}' # ERROR @app.error(404) def error404 (error): "404 error page" return 'Error 404.
' + str(error) ## Common functions def purgeUsers (): "delete user options older than n hours" try: nNowMinusNHours = int(time.time()) - (int(request.forms.hours) * 60 * 60) for nUserId, dValue in dUser.items(): if dValue["time"] < nNowMinusNHours: del dUser[nUserId] return True except KeyError: traceback.print_exc() return False #### START #### def main (sHost="localhost", nPort=8080, dOptions=None, bTestPage=False, nMultiCPU=None): "start server" global TESTPAGE global HOMEPAGE if bTestPage: TESTPAGE = True HOMEPAGE = HOMEPAGE.replace("{SERVER_PORT}", str(nPort)) if dOptions: oGCE.setOptions(dOptions) # Python version print("Python: " + sys.version) # Grammalecte echo("Grammalecte v{}".format(oGCE.version)) oGCE.displayOptions() # Process Pool Executor initExecutor(nMultiCPU) # Server (Bottle) run(app, host=sHost, port=nPort) if __name__ == '__main__': xParser = argparse.ArgumentParser() #xParser.add_argument("lang", type=str, nargs='+', help="lang project to generate (name of folder in /lang)") xParser.add_argument("-ht", "--host", help="host (default: localhost)", type=str) xParser.add_argument("-p", "--port", help="port (default: 8080)", type=int) xParser.add_argument("-mp", "--multiprocessor", help="define how many processes for the grammar checker", type=int) xParser.add_argument("-t", "--test_page", help="page to test the server on /", action="store_true") xParser.add_argument("-on", "--opt_on", nargs="+", help="activate options") xParser.add_argument("-off", "--opt_off", nargs="+", help="deactivate options") xArgs = xParser.parse_args() dOpt = None if xArgs.opt_on or xArgs.opt_off: dOpt = {} if xArgs.opt_on: dOpt = { opt:True for opt in xArgs.opt_on } if xArgs.opt_off: dOpt.update({ opt:False for opt in xArgs.opt_off }) main(xArgs.host or "localhost", \ xArgs.port or 8080, \ dOpt, xArgs.test_page, xArgs.multiprocessor) else: # Must be launched at start, for WSGI server (which doesn’t call main()) # WSGI servers just import the given file as a module and use an object exported from it ( in this case) to run the server. initExecutor()