management-llm/task-server/app.py

from flask import Flask, request, jsonify
import pandas as pd
import matplotlib.pyplot as plt
import os
import openai
import joblib
import random

app = Flask(__name__)
openai.api_key = os.getenv("OPENAI_API_KEY")

# Sample data to store tasks and users
tasks_data = {}
users_data = {}

# Loads the SVM model, vectorizer, and encoder
model_filename = "svm_model.sav"
svm_model = joblib.load(model_filename)

vectorizer_filename = 'fitted_vectorizer.joblib'
word_vectorizer = joblib.load(vectorizer_filename)

encoder_filename = 'fitted_encoder.joblib'
encoder = joblib.load(encoder_filename)

def generate_technical_tags(task_form):
    # Vectorize 
    vectorized_text = word_vectorizer.transform([task_form])   
    decision_function_scores = svm_model.decision_function(vectorized_text)

    # Get the top 3 predicted labels based on highest decision function scores
    top_4_indices = decision_function_scores.argsort()[0][-3:][::-1]
    top_4_labels = encoder.inverse_transform(top_4_indices)

    tags = top_4_labels.tolist()
    return tags

def match_tasks_to_users(tasks, users, user_max_tasks):
    task_ids = list(tasks.keys())

    # Create a dictionary to store the matched tasks and users
    task_to_user_matches = {}
    user_to_task_matches = {user_id: None for user_id in users}

    # Helper function to calculate the preference score between a task and a user
    def calculate_preference(task_id, user_id):
        task_tags = set(tasks[task_id]['tags'])
        user_strengths = set(users[user_id]['strengths'])
        user_current_tasks = users[user_id]['current_tasks']

        # Calculate the preference score based on matching tags and strengths
        tag_score = len(task_tags.intersection(user_strengths))

        # Calculate the total preference score, considering currentTasks as a tiebreaker
        preference_score = tag_score - user_current_tasks * 0.1

        # Makes the preference score negative if the user is already at max tasks
        if user_current_tasks == user_max_tasks:
            preference_score -= user_max_tasks

        return preference_score

    # Assign tasks to users based on preferences
    for task_id in task_ids:
        task_info = tasks[task_id]

        # Sort the users based on their preference for this task
        sorted_users = sorted(users.keys(), key=lambda user_id: calculate_preference(task_id, user_id), reverse=True)
        # Assign the task to the first user in the sorted list who is not already matched
        for user_id in sorted_users:
            if user_to_task_matches[user_id] is None:
                task_to_user_matches[task_id] = user_id
                user_to_task_matches[user_id] = task_id
                break
    print(task_to_user_matches)
    return task_to_user_matches

@app.route('/label_tasks', methods=['POST'])
def get_task_tags():
    try:
        data = request.get_json()
        tasks = data[0]['tasks']
        result = {}
        for task, task_description in tasks.items():
            task_id = task_description["id"]
            task_content = task_description["content"]
            task_complexity = task_description["complexityScore"]
            
            tags = generate_technical_tags(task_content) # where the function that gets the tags is placed
            result[task_id] = tags
        return jsonify(result)
    except Exception as e:
        return jsonify({"error": str(e)}), 500


@app.route('/assign_tasks', methods=['POST'])
def assign_tasks_to_users():
    # try:
    data = request.get_json()
    tasks = data[0]['tasks']
    users = data[0]['users']
    result = {}

    tasks_dict = {}
    users_dict = {}

    for task, task_description in tasks.items():
        tasks_dict[task_description["id"]] = {"tags" : generate_technical_tags(task_description["content"]), "complexity" : task_description["complexityScore"]}

    for user, user_description in users.items():
        users_dict[user_description["id"]] = {"strengths" : user_description["strengths"], "current_tasks" : user_description["currentTasks"]}
    
    
    matching = match_tasks_to_users(tasks_dict, users_dict, 3)
    # result = {}
    # for task, task_description in tasks.items():
    #     result[task_description["content"]] = {"assignedTo" : matchings[task_description["id"]]}

    # print(result)
    result = matching
    return jsonify(result)
    # return result
    # except Exception as e:
    #     return jsonify({"error": str(e)}), 500


if __name__ == '__main__':
    app.run(debug=True)