add api/v2 routes for clm integration

Author: rmobis

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

taskapp.py

@@ -16,6 +16,7 @@
 import send_grid_email
 from templesync import check_logs, temple_player_data, import_logs
 from task_types import CollectionLogVerificationData
+import task_api
 
 
 '''

tasklists.py

@@ -28,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

@@ -86,6 +86,15 @@ def current_task(self) -> tuple | None:
         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):