Evaluation-Backend/exp/knn/main.py

import importlib.util
import numpy as np
import requests
import time
import traceback

EXP_URL = "http://10.2.1.195:880"


def load_user_function(file_path, function_name):
    """
    动态加载函数。
    """
    spec = importlib.util.spec_from_file_location("user_code", file_path)
    user_code = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(user_code)
    return getattr(user_code, function_name)


def evaluate_knn(
    user_func, train_data, train_labels, test_data, test_labels, log_details
):
    """
    使用学生编写的 KNN 函数对测试集进行分类，并计算准确率。

    Args:
    - user_func (function): 学生编写的 KNN 函数。
    - train_data (numpy array): 训练数据集。
    - train_labels (numpy array): 训练标签集。
    - test_data (numpy array): 测试数据集。
    - test_labels (numpy array): 测试标签集。
    - log_details (list): 日志记录列表。

    Returns:
    - float: 分类准确率。
    """
    correct = 0
    total = len(test_data)

    try:
        for i in range(total):
            # 获取测试数据中的每个样本
            result = user_func(test_data[i], train_data, train_labels)
            if result == test_labels[i]:
                correct += 1
            else:
                log_details.append(
                    f"第 {i+1} 次测试错误，数据 {test_data[i]} 被分类到 {result} 实际是 {test_labels[i]}"
                )
                print(
                    f"第 {i+1} 次测试错误，数据 {test_data[i]} 被分类到 {result} 实际是 {test_labels[i]}"
                )

        accuracy = correct / total
        log_details.append(f"分类准确率: {accuracy * 100:.2f}%")
        return accuracy
    except Exception as e:
        log_details.append(f"评测期间发生错误: {e}")
        print(f"评测期间发生错误: {e}")
        log_details.append(traceback.format_exc())
        print(traceback.format_exc())
        log_details.append("成绩为零！")
        print("成绩为零！")
        return 0.0


def load_data(file_path):
    """
    加载 TSV 格式数据文件。
    """
    features = []
    labels = []

    with open(file_path, "r") as file:
        for line in file:
            values = line.strip().split("\t")
            features.append([float(x) for x in values[:-1]])
            labels.append(int(values[-1]))

    return np.array(features), np.array(labels)


def upload_score(name, id, score_data):
    """
    上传成绩到服务器。
    """
    url = f"{EXP_URL}/upload_score"
    try:
        print("上传成绩中...")
        # print(score_data)
        response = requests.post(
            url,
            json={
                "id": id,
                "name": name,
                "score": score_data["score"],
                "running_time": score_data["running_time"],
                "time": time.strftime(
                    "%Y-%m-%d %H:%M:%S", time.gmtime(time.time() + 8 * 3600)
                ),
                "details": score_data["details"],
            },
        )
        if response.status_code == 200:
            print("上传成绩成功!")
        elif response.status_code == 400:
            print("上传成绩失败!\n原因:", response.json()["message"])
    except Exception as e:
        print(f"上传成绩出错: {e}")


def main(name, id):
    user_code_file = "function.py"  # 代码文件
    user_function_name = "classify0"  # KNN 函数名

    # 加载训练集和测试集
    train_file = "datingTrainSet2.tsv"
    test_file = "datingTestSet2.tsv"

    train_data, train_labels = load_data(train_file)
    test_data, test_labels = load_data(test_file)

    # 加载自定义函数
    log_details = []
    try:
        user_func = load_user_function(user_code_file, user_function_name)
    except Exception as e:
        log_details.append(f"Error loading user function: {e}")
        print("\n".join(log_details))
        return

    # 记录评估时间
    start_time = time.time()

    # 评估分类准确率
    accuracy = evaluate_knn(
        user_func, train_data, train_labels, test_data, test_labels, log_details
    )

    # 计算运行时间
    running_time = time.time() - start_time

    # 输出结果
    print("")
    log_details.append(f"得分: {accuracy * 100:.2f}")
    print(f"得分: {accuracy * 100:.2f}")
    log_details.append(f"运行时间: {running_time * 1000:.10f}ms")
    print(f"运行时间: {running_time * 1000:.10f}ms")

    # 上传成绩
    score_data = {
        "score": accuracy * 100,
        "running_time": f"{running_time * 1000:.10f}ms",
        "details": "<br>".join(log_details),
    }

    upload_score(name, id, score_data)


if __name__ == "__main__":
    try:
        from function import name, id

        print(f"正在评测 {name}@{id} 的代码...\n")
        main(name, id)
    except Exception as e:
        log_details = [
            f"评测出错: {e}",
            traceback.format_exc(),
            "请检查代码和返回值类型是否正确！",
            "本次评测无效！",
        ]
        print("\n".join(log_details))