Initial Version

support_formatter/__init__.py

@@ -0,0 +1,2 @@
+import importlib.metadata
+__version__ = importlib.metadata.version(__package__ or "support_formatter")

support_formatter/app.py

@@ -0,0 +1,38 @@
+# pylint: disable=too-few-public-methods
+
+from datetime import datetime
+from flask import Flask
+from flask_smorest import Api
+from .config import APISettings
+
+class IsSingletonException(Exception):
+    pass
+
+class Application:
+    """
+    This is a singleton that can be accessed using get_instance()
+
+    It has 2 properties
+    - app: Used for WSGI servers and such
+    - api: Used for Blueprints
+    """
+
+    __instance = None
+    app = Flask(__name__)
+    app.config.from_object(APISettings)
+    app.json.sort_keys = False
+    api = None
+    alive_since = None
+
+    @staticmethod
+    def get_instance():
+        if Application.__instance is None:
+            Application()
+        return Application.__instance
+
+    def __init__(self):
+        if Application.__instance is not None:
+            raise IsSingletonException("This class is a singleton")
+        Application.__instance = self
+        self.api = Api(self.app)
+        self.alive_since = datetime.now()

support_formatter/config/__init__.py

@@ -0,0 +1 @@
+from .settings import APISettings, ServerSettings

support_formatter/config/settings.py

@@ -0,0 +1,35 @@
+# pylint: disable=too-few-public-methods
+import os
+from pathlib import Path
+
+from .. import __version__
+
+class ServerSettings:
+    HOSTNAME = os.environ.get("SUPPORT_FORMATTER_HOST", "localhost")
+    PORT = int(os.environ.get("SUPPORT_FORMATTER_PORT", 5000))
+
+class APISettings:
+    API_TITLE = "Support Organizer"
+    API_VERSION = __version__
+    OPENAPI_VERSION = "3.1.0"
+
+    # openapi.json settings
+    OPENAPI_URL_PREFIX = "/"
+    OPENAPI_JSON_PATH = "openapi.json"
+
+    # swagger settings
+    OPENAPI_SWAGGER_UI_PATH = "/swagger"
+    OPENAPI_SWAGGER_UI_URL = "https://cdn.jsdelivr.net/npm/swagger-ui-dist@5.10.0/"
+    SWAGGER_UI_DOC_EXPANSION = "list"
+
+    # Info settings
+    API_SPEC_OPTIONS = {
+        'info': {
+            'description': 'Support Organizer for FGO'
+        }
+    }
+
+    PAGES_DIRECTORY = Path(__file__).parents[1] / "pages"
+
+    ALLOWED_EXTENSIONS = { 'csv', 'txt' }
+    FILE_SAVE_DIRECTORY = Path(__file__).parents[1] / ".temp"

support_formatter/logic/csv_processor.py

@@ -0,0 +1,44 @@
+import csv
+import pathlib
+
+class FileTypeInvalidError(ValueError):
+    pass
+
+def determine_type(row):
+    if set(row) == set(["servant", "level", "np_level", "skills", "bce"]):
+        return "servants"
+    if set(row) == set(["ce", "ce_level", "mlb"]):
+        return "ces"
+    return "unknown"
+
+def process_csv(path: pathlib.Path):
+    data, entries = {}, []
+
+    with open(path, "r") as csv_file:
+        csv_reader = csv.DictReader(csv_file)
+        csv_type = determine_type(csv_reader.fieldnames)
+
+        if csv_type == "unknown":
+            raise FileTypeInvalidError()
+
+        for row in csv_reader:
+            entry = row.copy()
+            if csv_type == "servants":
+                entry.update({
+                    "level": int(entry['level']),
+                    "np_level": int(entry['np_level']),
+                    "bce": bool(entry["bce"])
+                })
+            else: 
+                entry.update({
+                    "ce_level": int(entry['ce_level']),
+                    "mlb": bool(entry["mlb"])
+                })
+            entries.append(entry)
+
+    if csv_type == "servants":
+        data = { "support_list": entries }
+    else:
+        data = { "ce_list": entries }
+    
+    return data

support_formatter/models/interface.py

@@ -0,0 +1,19 @@
+from enum import Enum
+import marshmallow as ma
+
+class HealthStatus(Enum):
+    OK = 0
+    WARNING = 1
+    ERROR = 2
+    CRITICAL = 3
+
+class HealthGet(ma.Schema):
+    alive_since = ma.fields.String()
+    alive_for = ma.fields.String()
+    status = ma.fields.Enum(HealthStatus, type=ma.fields.String)
+
+class ApiVersionGet(ma.Schema):
+    version = ma.fields.String(example="0.1")
+
+class OpenAPIGet(ma.Schema):
+    pass

support_formatter/pages/error.html

@@ -0,0 +1,14 @@
+<!DOCTYPE html>
+<html lang="en">
+
+<head>
+  <meta charset="UTF-8">
+  <meta name="viewport" content="width=device-width, initial-scale=1.0">
+  <title>Support File Converter - Error</title>
+</head>
+
+<body>
+  <h1>Error</h1>
+</body>
+
+</html>

support_formatter/pages/index.html

@@ -0,0 +1,34 @@
+<!DOCTYPE html>
+<html lang="en">
+
+<head>
+  <meta charset="UTF-8">
+  <meta name="viewport" content="width=device-width, initial-scale=1.0">
+  <title>Support File Converter</title>
+</head>
+
+<body>
+  <h1>Convert to Nerone-compatible data</h1>
+  <form method=post enctype=multipart/form-data>
+    <label for="fusername">Username:</label>
+    <input type="text" id="fusername" name="username" placeholder="Your username">
+    <br />
+    <label for="ffriendcode">Friendcode:</label>
+    <input type="text" id="ffriendcode" name="friendcode" placeholder="000,000,000">
+    <br />
+    <br />
+    <label for="fservantdata">Servant Data</label>
+    <br />
+    <input type="file" id="fservantdata" name="servantdata">
+    <br />
+    <br />
+    <label for="fcedata">CE Data (optional)</label>
+    <br />
+    <input type="file" id="fcedata" name="cedata">
+    <br />
+    <br />
+    <input type="submit" value="Upload">
+  </form>
+</body>
+
+</html>

support_formatter/routes/__init__.py

@@ -0,0 +1,7 @@
+# pylint: disable=wrong-import-position,cyclic-import
+from flask_smorest import Blueprint
+
+interface_routes = Blueprint("interface", "interface", url_prefix="/", description="")
+formatter_routes = Blueprint("formatter", "formatter", url_prefix="/", description="")
+
+from . import interface, upload # avoids circular imports problem

support_formatter/routes/interface.py

@@ -0,0 +1,38 @@
+from datetime import datetime
+
+from flask import redirect, url_for
+from flask.views import MethodView
+
+from ..app import Application
+from ..config import APISettings
+from ..models.interface import ApiVersionGet, HealthGet, HealthStatus, OpenAPIGet
+
+from . import interface_routes as blp
+
+APP = Application.get_instance()
+
+@blp.route("/health")
+class ApiVersion(MethodView):
+    @blp.doc(summary="Returns the status and alive-time of the server")
+    @blp.response(200, HealthGet, description="Successful operation")
+    def get(self):
+        response = {
+            "alive_since": datetime.strftime(APP.alive_since, "%d.%m.%Y %H:%M:%S"),
+            "alive_for": str(datetime.now() - APP.alive_since),
+            "status": HealthStatus.OK
+        }
+        return response
+
+@blp.route("/openapi")
+class ApiVersion(MethodView):
+    @blp.doc(summary="Get the OpenAPI spec as a JSON.")
+    @blp.response(200, OpenAPIGet, description="Successful operation")
+    def get(self):
+        return redirect(url_for('api-docs.openapi_json'))
+
+@blp.route("/version")
+class ApiVersion(MethodView):
+    @blp.doc(summary="Get the REST interface version identification.")
+    @blp.response(200, ApiVersionGet, description="Successful operation")
+    def get(self):
+        return { "version": APISettings.API_VERSION }

support_formatter/routes/upload.py

@@ -0,0 +1,70 @@
+import os
+import pathlib
+from typing import List
+from flask import send_from_directory
+import marshmallow as ma
+from flask_smorest.fields import Upload
+from werkzeug.datastructures import FileStorage
+
+import uuid
+
+from ..app import Application
+from ..config import APISettings
+from ..logic.csv_processor import process_csv, FileTypeInvalidError
+from . import formatter_routes as blp
+
+APP = Application.get_instance()
+
+def allowed_file(filename):
+    return '.' in filename and filename.rsplit('.', 1)[1].lower() in APISettings.ALLOWED_EXTENSIONS
+
+class MultipartFormSchema(ma.Schema):
+    username = ma.fields.String(required=True)
+    friendcode = ma.fields.String(required=False)
+
+class MultipartFileSchema(ma.Schema):
+    servantdata = Upload()
+    cedata = Upload()
+
+@blp.route("/", methods=["POST"])
+@blp.arguments(MultipartFormSchema, location="form")
+@blp.arguments(MultipartFileSchema, location="files") 
+@blp.response(200)
+def upload_file(form: dict[str, str], files: dict[str, FileStorage]):
+    filepaths: List[pathlib.Path] = []
+    returndata = {
+        "username": form["username"],
+        "friendcode": form.get("friendcode", "")
+    }
+
+    for name, file in files.items():
+        if name not in ("servantdata", "cedata"):
+            return send_from_directory(APISettings.PAGES_DIRECTORY, "error.html")
+
+        filepath = APISettings.FILE_SAVE_DIRECTORY / f"{uuid.uuid4()}.csv"
+        file.save(filepath)
+
+        if os.stat(filepath).st_size < 1 or not allowed_file(file.filename):
+            filepath.unlink()
+            continue
+
+        filepaths.append(filepath)
+
+    if len(filepaths) == 0:
+        return send_from_directory(APISettings.PAGES_DIRECTORY, "error.html")
+
+    try:
+        for f in filepaths:
+            result = process_csv(f)
+            returndata = returndata | result
+            f.unlink()
+    except FileTypeInvalidError:
+        for f in filepaths:
+            f.unlink()
+        return send_from_directory(APISettings.PAGES_DIRECTORY, "error.html")
+
+    return returndata
+
+@blp.route("/", methods=["GET"])
+def file_dialog():
+    return send_from_directory(APISettings.PAGES_DIRECTORY, "index.html")

support_formatter/server.py

@@ -0,0 +1,28 @@
+# pylint: disable=multiple-statements,wrong-import-position,wrong-import-order
+from gevent.monkey import patch_all; patch_all()
+from gevent.pywsgi import WSGIServer
+
+from .app import Application
+from .routes import interface_routes, formatter_routes
+from .config import APISettings, ServerSettings
+
+APISettings.FILE_SAVE_DIRECTORY.mkdir(parents=True, exist_ok=True)
+
+APP = Application.get_instance()
+app, api = APP.app, APP.api
+api.register_blueprint(interface_routes)
+api.register_blueprint(formatter_routes)
+
+HOSTNAME, PORT = ServerSettings.HOSTNAME, ServerSettings.PORT
+
+def run_server():
+    http_Server = WSGIServer((HOSTNAME, PORT), app)
+    try:
+        print(f"Server available on http://{HOSTNAME}:{PORT}/")
+        print(f"View docs on http://{HOSTNAME}:{PORT}/swagger")
+        http_Server.serve_forever()
+    except KeyboardInterrupt:
+        print("Keyboard interrupt received, stopping...")
+
+if __name__ == '__main__':
+    run_server()