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

[CLM integration] add sync endpoint

Author: mgerni

task_api.py

@@ -9,6 +9,7 @@
 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 templesync import sync_user_tasks
 from user_dao import UserDatabaseObject
 
 
@@ -110,3 +111,20 @@ def apiv2_generate_task(user: UserDatabaseObject):
         return { 'task_id': generated_task.id }
 
     return { 'error': 'No available tasks to generate' }, HTTPStatus.BAD_REQUEST
+
+
+@app.route('/api/v2/user/sync', methods=['POST'])
+@token_required_v2
+def apiv2_sync(user: UserDatabaseObject):
+    body = request.json
+
+    collection_log = set(body['collection_log'])
+    diaries = body['diaries']
+    skills = body['skills']
+
+    changed_tasks = sync_user_tasks(user.username, collection_log, diaries, skills)
+
+    return {
+        'completed': list(changed_tasks[0]),
+        'uncompleted': list(changed_tasks[1])
+    }

task_types.py

@@ -12,6 +12,16 @@ class CollectionLogVerificationData(VerificationData):
     item_ids: list[int]
     count: int
 
+@dataclass
+class AchievementDiaryVerificationData(VerificationData):
+    region: str
+    difficulty: str
+
+@dataclass
+class SkillVerificationData(VerificationData):
+    experience: dict
+    count: int
+
 
 @dataclass
 class TaskData:

tasklists.py

@@ -1,17 +1,31 @@
 import json
-from task_types import TaskData, TaskTag, VerificationData, CollectionLogVerificationData
+from task_types import AchievementDiaryVerificationData, SkillVerificationData, TaskData, TaskTag, VerificationData, CollectionLogVerificationData
 
-def to_verification_data(data: dict) -> VerificationData or None: # type: ignore
+def to_verification_data(data: dict) -> VerificationData | None:
     if data is None:
         return None
 
     method: str = data['method']
-    if (method != "collection-log"):
-        return None
+    if method == "collection-log":
+        return CollectionLogVerificationData(
+            method=data['method'],
+            item_ids=data['itemIds'],
+            count=data['count']
+        )
+    elif method == "achievement-diary":
+        return AchievementDiaryVerificationData(
+            method=data['method'],
+            region=data['region'],
+            difficulty=data['difficulty']
+        )
+    elif method == "skill":
+        return SkillVerificationData(
+            method=data['method'],
+            experience=data['experience'],
+            count=data['count']
+        )
 
-    return CollectionLogVerificationData(method=data['method'],
-                                         item_ids=data['itemIds'],
-                                         count=data['count'])
+    return None
 
 def to_task_class(data: dict) -> TaskData:
     return TaskData(id=data['id'],

templesync.py

@@ -1,7 +1,8 @@
 import requests
+from task_database import __set_current_task, __set_task_complete, get_user
 import tasklists
 from datetime import datetime, timezone
-from task_types import CollectionLogVerificationData, TaskData
+from task_types import AchievementDiaryVerificationData, CollectionLogVerificationData, SkillVerificationData, TaskData
 
 
 def temple_player_data(username: str):
@@ -172,5 +173,82 @@ def format_completed_tasks(completed_tasks: set):
         # print(sorted_completed_tasks)
         return format_completed_tasks(sorted_completed_tasks)
 
+
+def sync_user_tasks(username: str, collection_log: set[int], diaries: dict, skills: dict) -> tuple[set[int], set[int]]:
+    tiers = ['easy', 'medium', 'hard', 'elite', 'master']
+    tasks = [ task for tier in tiers for task in tasklists.list_for_tier(tier) ]
+
+    completed_tasks: set[int] = set()
+    uncompleted_tasks: set[int] = set()
+
+    for task in tasks:
+        verification_data = task.verification
+        task_completed = False
+
+        if isinstance(verification_data, CollectionLogVerificationData):
+            task_completed = verify_collection_log(collection_log, verification_data)
+        elif isinstance(verification_data, AchievementDiaryVerificationData):
+            task_completed = verify_achievement_diary(diaries, verification_data)
+        elif isinstance(verification_data, SkillVerificationData):
+            task_completed = verify_skill(skills, verification_data)
+
+        if task_completed:
+            completed_tasks.add(task.id)
+        else:
+            uncompleted_tasks.add(task.id)
+
+    user = get_user(username)
+    old_completed_tasks = set([task.id for task in [
+        *user.easy.completed_tasks,
+        *user.medium.completed_tasks,
+        *user.hard.completed_tasks,
+        *user.elite.completed_tasks,
+        *user.master.completed_tasks,
+    ]])
+
+    new_completed_tasks: set[int] = set()
+    new_uncompleted_tasks: set[int] = set()
+
+    for task_id in completed_tasks:
+        if task_id not in old_completed_tasks:
+            __set_task_complete(username, None, task_id, True)
+            new_completed_tasks.add(task_id)
+
+    for task_id in uncompleted_tasks:
+        if task_id in old_completed_tasks:
+            __set_task_complete(username, None, task_id, False)
+            new_uncompleted_tasks.add(task_id)
+
+    current_task_id = user.current_task_id()
+    if current_task_id in new_completed_tasks:
+        __set_current_task(username, None, None, False)
+
+    return new_completed_tasks, new_uncompleted_tasks
+
+
+def verify_collection_log(collection_log: set[int], verification_data: CollectionLogVerificationData) -> bool:
+    log_count = 0
+    for item_id in verification_data.item_ids:
+        if item_id in collection_log:
+            log_count += 1
+
+    return log_count >= verification_data.count
+
+
+def verify_achievement_diary(diaries: dict, verification_data: AchievementDiaryVerificationData) -> bool:
+    region = verification_data.region
+    difficulty = verification_data.difficulty
+
+    return diaries[region][difficulty]
+
+
+def verify_skill(skills: dict, verification_data: SkillVerificationData) -> bool:
+    skill_count = 0
+    for skill, exp in verification_data.experience.items():
+        if skills[skill] >= exp:
+            skill_count += 1
+
+    return skill_count >= verification_data.count
+
 if __name__ == "__main__":
     print(check_logs('Gerni Task', tasklists.list_for_tier('elite'), 'check'))