from flask import Flask, request, render_template, redirect, url_for import pandas as pd from sklearn.metrics import roc_auc_score, average_precision_score import os app = Flask(__name__) @app.route('/') def index(): return render_template('index.html') @app.route('/upload', methods=['POST']) def upload_file(): # 定义保存结果的文件路径 results_file = 'results.xlsx' # 如果结果文件不存在，创建文件并写入表头 if not os.path.exists(results_file): with open(results_file, 'w') as f: f.write('Name,clinvar_test_auroc,clinvar_test_auprc,cancer_auroc,cancer_auprc,TP53_auroc,TP53_auprc,goflof_HGMD_auroc,goflof_HGMD_auprc,PPARG_auroc,PPARG_auprc\n') #接受从前段传来的参数 if 'file' not in request.files: return redirect(request.url) file = request.files['file'] user_name = request.form['name'] #判断有没有当前用户 df = pd.read_excel(results_file) if user_name not in df['Name'].values: # 如果名字不存在，添加新行 new_row = { 'Name': user_name, 'clinvar_test_auroc': '', 'clinvar_test_auprc': '', 'cancer_auroc': '', 'cancer_auprc': '', 'TP53_auroc': '', 'TP53_auprc': '', 'goflof_HGMD_auroc': '', 'goflof_HGMD_auprc': '', 'PPARG_auroc': '', 'PPARG_auprc': '' } #df = df.append(new_row, ignore_index=True) df = pd.concat([df, pd.DataFrame([new_row])], ignore_index=True) # 保存更新后的DataFrame回excel文件 df.to_excel(results_file, index=False) #判断用户有没有上传文件 if file.filename == '': return redirect(request.url) if file: #user_df = pd.read_excel(file) user_df = pd.read_excel(file, dtype={'CHROM': str}) #pd.read_excel(file, dtype={'CHROM': str}) # 判断用户输入的数据集是哪个数据集 excel_labels = [ pd.read_excel('cancer_discover_test.xlsx', dtype={'CHROM': str}), # 将 'path_to_excel_label1.xlsx' 替换为实际文件路径 pd.read_excel('clinvar_20230326_test.xlsx', dtype={'CHROM': str}), # 将 'path_to_excel_label2.xlsx' 替换为实际文件路径 pd.read_excel('goflof_HGMD2019_with_gnomAD_test.xlsx', dtype={'CHROM': str}), # 将 'path_to_excel_label3.xlsx' 替换为实际文件路径 pd.read_excel('PPARG_test.xlsx', dtype={'CHROM': str}), # 将 'path_to_excel_label4.xlsx' 替换为实际文件路径 pd.read_excel('TP53_test.xlsx', dtype={'CHROM': str}) # 将 'path_to_excel_label5.xlsx' 替换为实际文件路径 ] # 定义标识符列 identifier_columns = ['CHROM', 'Nuc-Pos', 'REF-Nuc', 'ALT-Nuc'] #print(user_df) samples_test_set = set(tuple(row) for row in user_df[identifier_columns].to_numpy()) # 初始化比较结果列表 samples_identical = [] # 遍历每个标签数据集，进行比较 for excel_label in excel_labels: samples_label_set = set(tuple(row) for row in excel_label[identifier_columns].to_numpy()) samples_identical.append(samples_test_set == samples_label_set) # 检查是否有完全相同的数据集 any_identical = any(samples_identical) print(samples_identical) # 找到对应的测试集 if any_identical: for i, identical in enumerate(samples_identical): if identical: if i==0: result_test_name ='cancer' true_labels_df = pd.read_excel('cancer_discover_test.xlsx', dtype={'CHROM': str}) #print("aaaaaaaa11111111111111111111111") elif i==1: result_test_name ='clinvar_test' #print("2222222222222222222222222") true_labels_df = pd.read_excel('clinvar_20230326_test.xlsx', dtype={'CHROM': str}) elif i==2: result_test_name ='goflof_HGMD' true_labels_df = pd.read_excel('goflof_HGMD2019_with_gnomAD_test.xlsx', dtype={'CHROM': str}) #print("33333333333333333333") elif i==3: result_test_name ='PPARG' true_labels_df = pd.read_excel('PPARG_test.xlsx', dtype={'CHROM': str}) #print("444444444444444444444444444") elif i==4: result_test_name ='TP53' true_labels_df = pd.read_excel('TP53_test.xlsx', dtype={'CHROM': str}) #print("5555555555555555555555555") else: no_test = 1 print("数据集没有匹配成功") return render_template('index.html', no_test=no_test) user_df['predicted_score'] = user_df['predicted_score'].fillna(0) # 合并数据集，使用 CHROM、Nuc-Pos、REF-Nuc、ALT-Nuc 作为合并键 merged_df = pd.merge(user_df, true_labels_df, on=['CHROM', 'Nuc-Pos', 'REF-Nuc', 'ALT-Nuc'], how='left') #print("合并成功") #计算并更新值 if 'True Label' in merged_df.columns and 'predicted_score' in merged_df.columns: true_labels = merged_df['True Label'] predicted_scores = merged_df['predicted_score'] auroc = roc_auc_score(true_labels, predicted_scores) auprc = average_precision_score(true_labels, predicted_scores) # 读取已有的exvel文件 df = pd.read_excel('results.xlsx') # 找到需要更新的行 name_to_update = user_name auroc_to_update = result_test_name+'_'+'auroc' auprc_to_update = result_test_name+'_'+'auprc' new_auroc = auroc # 更新后的cancer_auroc值 new_auprc = auprc # 更新后的cancer_auprc值 # 更新指定的值 df.loc[df['Name'] == name_to_update, auroc_to_update] = new_auroc df.loc[df['Name'] == name_to_update, auprc_to_update] = new_auprc # 保存更新后的DataFrame回excel文件 df.to_excel('results.xlsx', index=False) return render_template('index.html', auroc=auroc, auprc=auprc) else: error_message = "The uploaded excel must contain 'True Label' and 'predicted_score' columns" return render_template('index.html', error=error_message) return redirect(url_for('index')) # 处理查询结果的路由和函数 @app.route('/query_results', methods=['GET', 'POST']) #查询计算结果 def query_results(): user_name = request.form['query_name'] if not user_name: return jsonify({"error": "No name provided"}) # 定义保存结果的文件路径 results_file = 'results.xlsx' # 读取结果文件 results_df = pd.read_excel(results_file) # 查询用户的计算结果 user_results = results_df[results_df['Name'] == user_name] if user_results.empty: no_score = 1 return render_template('index.html', no_score = 1) # 获取最新的结果 latest_result = user_results.iloc[0] yes_score = 1 #return jsonify({"auroc": latest_result['auroc'], "auprc": latest_result['auprc']}) #return jsonify({"clinvar_test_auroc": latest_result['clinvar_test_auroc'], "clinvar_test_auprc": latest_result['clinvar_test_auprc'], "cancer_auroc": latest_result['cancer_auroc'], "cancer_auprc": latest_result['cancer_auprc'], "goflof_HGMD_auroc": latest_result['goflof_HGMD_auroc'], "goflof_HGMD_auprc": latest_result['goflof_HGMD_auprc'], "PPARG_auroc": latest_result['PPARG_auroc'], "PPARG_auprc": latest_result['PPARG_auprc'], "TP53_auroc": latest_result['TP53_auroc'], "TP53_auprc": latest_result['TP53_auprc']}) return render_template('index.html', yes_score = 1, clinvar_test_auroc=latest_result['clinvar_test_auroc'], clinvar_test_auprc=latest_result['clinvar_test_auprc'], cancer_auroc=latest_result['cancer_auroc'], cancer_auprc=latest_result['cancer_auprc'], goflof_HGMD_auroc=latest_result['goflof_HGMD_auroc'], goflof_HGMD_auprc=latest_result['goflof_HGMD_auprc'], PPARG_auroc=latest_result['PPARG_auroc'], PPARG_auprc=latest_result['PPARG_auprc'], TP53_auroc=latest_result['TP53_auroc'], TP53_auprc=latest_result['TP53_auprc']) @app.errorhandler(Exception) def handle_exception(e): # 记录错误详细信息（可选） print(f"Error: {e}") # 返回自定义错误信息 return render_template('index.html', error="上传的文件存在格式问题。") @app.errorhandler(404) def page_not_found(e): return render_template('index.html', error="页面未找到。") @app.errorhandler(500) def internal_server_error(e): return render_template('index.html', error="服务器内部错误。") if __name__ == '__main__': app.run(host='0.0.0.0', port=8000, debug=False)