Merge pull request #103 from rmobis/rmobis/clm-integration

CLM plugin integration

Author: mgerni

app_setup.py

@@ -0,0 +1,26 @@
+from flask import Flask
+import config
+from flask_recaptcha import ReCaptcha # type: ignore
+
+app = Flask(__name__)
+
+isProd = config.IS_PROD
+
+# Set secret key for Flask App.
+app.config['SECRET_KEY'] = config.SECRET_KEY
+
+if isProd:
+    # Keys for Google reCAPTCHA.
+    app.config['RECAPTCHA_SITE_KEY'] = config.RECAPTCHA_SITE_KEY
+    app.config['RECAPTCHA_SECRET_KEY'] = config.RECAPTCHA_SECRET_KEY
+    # initialize reCAPTCHA
+    recaptcha = ReCaptcha(app)
+else:
+    recaptcha = "Disabled for DEV"
+
+
+# email service account.
+taskapp_email = config.SECRET_KEY
+
+# specifies database to use.
+db = config.MONGO_CLIENT["TaskAppLoginDB"]

dev_env.ps1

@@ -0,0 +1,5 @@
+$env:FLASK_ENV='development'
+$env:MONGO_URI='mongodb://root:example@localhost:27017/'
+$env:SENDGRID_API_KEY='192b9bdd22ab9ed4d12e236c78afcb9a393ec15f71bbf5dc987d54727823bcbf'
+$env:SECRET_KEY='192b9bdd22ab9ed4d12e236c78afcb9a393ec15f71bbf5dc987d54727823bcbf'
+$env:SERVER_PORT='8080'

task_api.py

@@ -0,0 +1,112 @@
+import bcrypt
+import datetime
+import jwt
+
+from flask import Response, request
+from functools import wraps
+from http import HTTPStatus
+
+from task_database import complete_task, generate_task, get_user, manual_complete_tasks, manual_revert_tasks
+from app_setup import app, db
+from tasklists import get_task_tier, list_for_tier
+from user_dao import UserDatabaseObject
+
+
+def token_required_v2(f):
+    @wraps(f)
+    def decorated(*args, **kwargs):
+        token = None
+        if 'Authorization' in request.headers:
+            auth_header = request.headers['Authorization']
+            token = auth_header.split(" ")[1] if " " in auth_header else auth_header
+
+        if not token:
+            return { 'error': 'No token found' }, HTTPStatus.UNAUTHORIZED
+
+        try:
+            data = jwt.decode(token, app.config['SECRET_KEY'], algorithms=['HS256'])
+            user = get_user(data['sub'])
+        except:
+            return { 'error': 'Token is invalid' }, HTTPStatus.UNAUTHORIZED
+
+        return f(user, *args, **kwargs)
+    return decorated
+
+
+@app.route('/api/v2/login', methods=['POST'])
+def apiv2_login():
+    body = request.json
+    if not body or not body['username'] or not body['password']:
+        return { 'error': 'Missing username and/or password' }, HTTPStatus.UNAUTHORIZED
+
+    users = db['users']
+    user = users.find_one({ 'username': body['username'] }, { '_id': 0 })
+    if not user:
+        return { 'error': 'Invalid credentials' }, HTTPStatus.UNAUTHORIZED
+
+    if not bcrypt.checkpw(body['password'].encode('utf-8'), user['hashed_password']):
+        return { 'error': 'Invalid credentials' }, HTTPStatus.UNAUTHORIZED
+
+    now = datetime.datetime.now(datetime.timezone.utc)
+    token = jwt.encode({
+        'sub': user['username'],
+        'iat': now,
+        'exp': now + datetime.timedelta(hours=24)
+    }, app.config['SECRET_KEY'])
+
+    return { 'token': token }
+
+
+@app.route('/api/v2/task-list', methods=['GET'])
+def apiv2_get_task_list():
+    tiers = request.args.getlist('tier') or ['easy', 'medium', 'hard', 'elite', 'master']
+
+    return { tier: list_for_tier(tier) for tier in tiers }
+
+
+@app.route('/api/v2/user/profile', methods=['GET'])
+@token_required_v2
+def apiv2_get_user_profile(user: UserDatabaseObject):
+    return {
+        'username': user.username,
+        'is_official': user.is_official,
+        'is_lms_enabled': user.lms_enabled,
+        'active_task_id': user.current_task_id(),
+        'completed_tasks': [
+            *user.easy.completed_tasks,
+            *user.medium.completed_tasks,
+            *user.hard.completed_tasks,
+            *user.elite.completed_tasks,
+            *user.master.completed_tasks,
+        ]
+    }
+
+
+@app.route('/api/v2/user/tasks/<id>', methods=['PATCH'])
+@token_required_v2
+def apiv2_update_user_task(user: UserDatabaseObject, id: str) -> None:
+    tier = get_task_tier(id)
+    body = request.json
+
+    if body['completed'] == True:
+        if user.current_task_id() == id:
+            complete_task(user.username)
+        else:
+            manual_complete_tasks(user.username, tier, id)
+    elif body['completed'] == False:
+        manual_revert_tasks(user.username, tier, id)
+
+    return Response(status=HTTPStatus.NO_CONTENT)
+
+
+@app.route('/api/v2/user/generate-task', methods=['POST'])
+@token_required_v2
+def apiv2_generate_task(user: UserDatabaseObject):
+    if user.current_task():
+        return { 'error': 'User already has an active task' }, HTTPStatus.BAD_REQUEST
+
+    generated_task = generate_task(user.username)
+    if generated_task:
+        return { 'task_id': generated_task.id }
+
+    return { 'error': 'No available tasks to generate' }, HTTPStatus.BAD_REQUEST

task_database.py

@@ -432,7 +432,7 @@ def get_roll_candidates_for_tier(username: str, tier: str, min_candidates: int =
 '''
 
 
-def generate_task(username: str) -> TaskData or None: # type: ignore
+def generate_task(username: str) -> TaskData | None:
     user = get_user(username)
     if user.current_task() is not None:
         return

task_types.py

@@ -20,6 +20,7 @@ class TaskData:
     tip: str
     wiki_link: str
     image_link: str
+    display_item_id: int
     tags: list["TaskTag"]
     verification: VerificationData
 

taskapp.py

@@ -1,10 +1,9 @@
-from flask import Flask, render_template, request, redirect, flash, url_for, session, jsonify, make_response # type: ignore
-from flask_recaptcha import ReCaptcha # type: ignore
+from app_setup import app, isProd, db, taskapp_email, recaptcha
+from flask import render_template, request, redirect, flash, url_for, session, jsonify, make_response # type: ignore
 import jwt
 import datetime
 import json
 import bcrypt # type: ignore
-import config
 from functools import wraps
 import task_login
 import tasklists
@@ -17,29 +16,7 @@
 import send_grid_email
 from templesync import check_logs, temple_player_data, import_logs
 from task_types import CollectionLogVerificationData
-
-app = Flask(__name__)
-
-isProd = config.IS_PROD
-
-# Set secret key for Flask App.
-app.config['SECRET_KEY'] = config.SECRET_KEY
-
-if isProd:
-    # Keys for Google reCAPTCHA.
-    app.config['RECAPTCHA_SITE_KEY'] = config.RECAPTCHA_SITE_KEY
-    app.config['RECAPTCHA_SECRET_KEY'] = config.RECAPTCHA_SECRET_KEY
-    # initialize reCAPTCHA
-    recaptcha = ReCaptcha(app)
-else:
-    recaptcha = "Disabled for DEV"
-
-
-# email service account.
-taskapp_email = config.SECRET_KEY
-
-# specifies database to use.
-db = config.MONGO_CLIENT["TaskAppLoginDB"]
+import task_api
 
 
 '''

tasklists.py

@@ -19,6 +19,7 @@ def to_task_class(data: dict) -> TaskData:
                     tip=data.get('tip'),
                     wiki_link=data['wikiLink'],
                     image_link=data['imageLink'],
+                    display_item_id=data['displayItemId'],
                     tags=data.get('tags', []),
                     verification=to_verification_data(data.get('verification')))
 
@@ -27,6 +28,16 @@ def read_tasks(filename: str) -> list[TaskData]:
         json_list = json.load(f)
         return list(map(to_task_class, json_list.get('tasks')))
 
+def get_task_tier(task_id: str) -> str:
+    tiers = ['easy', 'medium', 'hard', 'elite', 'master', 'passive', 'extra', 'pets']
+    for tier in tiers:
+        tier_tasks = list_for_tier(tier)
+        is_task_in_tier = next((True for task in tier_tasks if task.id == task_id), False)
+        if (is_task_in_tier):
+            return tier
+
+    return None
+
 
 easy = read_tasks('easy')
 medium = read_tasks('medium')

user_dao.py

@@ -64,15 +64,15 @@ def get_task_list(self, tier: str) -> UserTaskList:
         }[tier]
 
 
-    def current_task_for_tier(self, tier: str) -> tuple or None: # type: ignore
+    def current_task_for_tier(self, tier: str) -> tuple | None:
         user_task_list = self.get_task_list(tier)
         if user_task_list.current_task is None:
             return None
         task = task_info_for_id(tasklists.list_for_tier(tier), user_task_list.current_task.id)
         # TODO Fix this format
         return task.name, task.image_link, tier, task.id, task.tip, task.wiki_link, task.image_link
 
-    def current_task(self) -> tuple or None: # type: ignore
+    def current_task(self) -> tuple | None:
         if self.easy.current_task is not None:
             return self.current_task_for_tier('easyTasks')
         elif self.medium.current_task is not None:
@@ -86,6 +86,15 @@ def current_task(self) -> tuple or None: # type: ignore
         else:
             return None
 
+    def current_task_id(self) -> str | None:
+        current_task = self.current_task()
+
+        if not current_task:
+            return None
+
+        return current_task[3]
+
+
     def get_tier_progress(self, tier: str) -> TierProgress:
         # remove instance of duplicate uuid in completed tasks
         def clean_tasklists(tier: str):