from flask import Flask, render_template, request, jsonify, send_file from Bio import Entrez, SeqIO import primer3 import ssl import pandas as pd import io import json from datetime import datetime import time app = Flask(__name__) ssl._create_default_https_context = ssl._create_unverified_context def get_ready_for_primers(gene_symbol, species="human"): """获取基因序列和外显子交界点信息""" Entrez.email = "your_email@example.com" # 1. 搜索基因并获取 NCBI 内部 ID search_term = f"{gene_symbol}[Gene Name] AND {species}[Organism]" handle = Entrez.esearch(db="gene", term=search_term) record = Entrez.read(handle) if not record["IdList"]: return {"error": "未找到该基因"} gene_id = record["IdList"][0] # 2. 获取该基因关联的 NM_ 编号 link_handle = Entrez.elink(dbfrom="gene", db="nucleotide", id=gene_id, term="srcdb_refseq[prop] AND mRNA[filter]") link_record = Entrez.read(link_handle) try: # 获取第一个关联的核苷酸 UID nucl_id = link_record[0]["LinkSetDb"][0]["Link"][0]["Id"] # 3. 下载完整的 GenBank 格式数据 handle = Entrez.efetch(db="nucleotide", id=nucl_id, rettype="gb", retmode="text") seq_record = SeqIO.read(handle, "genbank") # 提取外显子分界点 junctions = [] current_pos = 0 for feature in seq_record.features: if feature.type == "exon": start, end = feature.location.start, feature.location.end current_pos += (end - start) junctions.append(int(current_pos)) if junctions: junctions.pop() # 移除最后一个边界 return { "symbol": gene_symbol, "nm_id": seq_record.id, "sequence": str(seq_record.seq), "junctions": junctions } except Exception as e: return {"error": f"获取基因信息失败: {str(e)}"} def design_qpcr_primers(gene_data): """设计qPCR引物，包含失败重试机制""" if "error" in gene_data: return gene_data # 基础序列参数 seq_args = { 'SEQUENCE_ID': gene_data['nm_id'], 'SEQUENCE_TEMPLATE': gene_data['sequence'], } # 基础约束参数 global_args = { 'PRIMER_OPT_SIZE': 20, 'PRIMER_MIN_SIZE': 18, 'PRIMER_MAX_SIZE': 25, 'PRIMER_OPT_TM': 60.0, 'PRIMER_MIN_TM': 57.0, 'PRIMER_MAX_TM': 63.0, 'PRIMER_TM_MAX_DIFF': 2.0, # 放宽一点点 Tm 差限制 'PRIMER_MIN_GC': 30.0, # 放宽 GC 下限 'PRIMER_MAX_GC': 70.0, # 放宽 GC 上限 'PRIMER_PRODUCT_SIZE_RANGE': [80, 250], # qPCR 通常不需要太长的产物 'PRIMER_NUM_RETURN': 5, } try: # 尝试 1: 强制要求跨外显子交界 # 注意：这里我们只取前几个交界点，有时太多点会导致搜索空间太小 junctions = gene_data.get('junctions', []) if junctions: seq_args['SEQUENCE_OVERLAP_JUNCTION_LIST'] = junctions try: results = primer3.bindings.design_primers(seq_args, global_args) except Exception: # 尝试 2: 如果失败，去掉跨外显子限制，进行普通设计 if 'SEQUENCE_OVERLAP_JUNCTION_LIST' in seq_args: del seq_args['SEQUENCE_OVERLAP_JUNCTION_LIST'] results = primer3.bindings.design_primers(seq_args, global_args) else: raise primer_pairs = [] for i in range(global_args['PRIMER_NUM_RETURN']): try: # 检查这个引物是否真的跨越了交界点 # (在返回结果中查看是否有指定标志，或者统一标记) is_junction = "是" if 'SEQUENCE_OVERLAP_JUNCTION_LIST' in seq_args else "否（普通设计）" pair = { "id": i + 1, "forward": results[f'PRIMER_LEFT_{i}_SEQUENCE'], "reverse": results[f'PRIMER_RIGHT_{i}_SEQUENCE'], "f_tm": f"{results[f'PRIMER_LEFT_{i}_TM']:.2f}", "r_tm": f"{results[f'PRIMER_RIGHT_{i}_TM']:.2f}", "product_size": results[f'PRIMER_PAIR_{i}_PRODUCT_SIZE'], "junction_info": is_junction } primer_pairs.append(pair) except KeyError: break if not primer_pairs: return {"error": "无法找到符合条件的引物，请尝试放宽筛选标准"} return {"primers": primer_pairs, "gene_info": gene_data} except Exception as e: return {"error": f"引物设计深度失败: {str(e)}"} @app.route('/') def index(): return render_template('index.html') @app.route('/design_primers', methods=['POST']) def design_primers_api(): data = request.json gene_symbol = data.get('gene_symbol', '').strip() species = data.get('species', 'human') if not gene_symbol: return jsonify({"error": "请输入基因名称"}) # 获取基因信息 gene_data = get_ready_for_primers(gene_symbol, species) # 设计引物 result = design_qpcr_primers(gene_data) return jsonify(result) @app.route('/batch_design_primers', methods=['POST']) def batch_design_primers_api(): data = request.json gene_list = data.get('gene_list', []) species = data.get('species', 'human') if not gene_list: return jsonify({"error": "请输入基因列表"}) results = [] for gene_symbol in gene_list: gene_symbol = gene_symbol.strip() if not gene_symbol: continue time.sleep(1) # 获取基因信息 gene_data = get_ready_for_primers(gene_symbol, species) # 设计引物 result = design_qpcr_primers(gene_data) if "error" in result: results.append({ "gene": gene_symbol, "status": "failed", "error": result["error"] }) else: results.append({ "gene": gene_symbol, "status": "success", "data": result }) return jsonify({"results": results}) @app.route('/export_primers', methods=['POST']) def export_primers(): data = request.json export_format = data.get('format', 'excel') # excel, csv, json results_data = data.get('data', []) if export_format == 'excel': return export_to_excel(results_data) elif export_format == 'csv': return export_to_csv(results_data) elif export_format == 'json': return export_to_json(results_data) else: return jsonify({"error": "不支持的导出格式"}) def export_to_excel(results_data): """导出为Excel格式""" rows = [] for result in results_data: if result.get('status') == 'success': gene_info = result['data']['gene_info'] primers = result['data']['primers'] for primer in primers: rows.append({ '基因名称': gene_info['symbol'], 'RefSeq ID': gene_info['nm_id'], '引物对编号': primer['id'], '正向引物序列': primer['forward'], '反向引物序列': primer['reverse'], '正向引物Tm(°C)': primer['f_tm'], '反向引物Tm(°C)': primer['r_tm'], '产物长度(bp)': primer['product_size'], '外显子交界点': ', '.join(map(str, gene_info['junctions'])), '设计时间': datetime.now().strftime('%Y-%m-%d %H:%M:%S') }) else: rows.append({ '基因名称': result['gene'], 'RefSeq ID': 'N/A', '引物对编号': 'N/A', '正向引物序列': 'N/A', '反向引物序列': 'N/A', '正向引物Tm(°C)': 'N/A', '反向引物Tm(°C)': 'N/A', '产物长度(bp)': 'N/A', '外显子交界点': 'N/A', '设计时间': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), '错误信息': result.get('error', '未知错误') }) df = pd.DataFrame(rows) # 创建Excel文件 output = io.BytesIO() with pd.ExcelWriter(output, engine='openpyxl') as writer: df.to_excel(writer, sheet_name='引物设计结果', index=False) output.seek(0) filename = f"qPCR_primers_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx" return send_file( output, mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', as_attachment=True, download_name=filename ) def export_to_csv(results_data): """导出为CSV格式""" rows = [] for result in results_data: if result.get('status') == 'success': gene_info = result['data']['gene_info'] primers = result['data']['primers'] for primer in primers: rows.append({ '基因名称': gene_info['symbol'], 'RefSeq ID': gene_info['nm_id'], '引物对编号': primer['id'], '正向引物序列': primer['forward'], '反向引物序列': primer['reverse'], '正向引物Tm(°C)': primer['f_tm'], '反向引物Tm(°C)': primer['r_tm'], '产物长度(bp)': primer['product_size'], '外显子交界点': ', '.join(map(str, gene_info['junctions'])), '设计时间': datetime.now().strftime('%Y-%m-%d %H:%M:%S') }) df = pd.DataFrame(rows) output = io.StringIO() df.to_csv(output, index=False, encoding='utf-8-sig') filename = f"qPCR_primers_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv" return send_file( io.BytesIO(output.getvalue().encode('utf-8-sig')), mimetype='text/csv', as_attachment=True, download_name=filename ) def export_to_json(results_data): """导出为JSON格式""" export_data = { "export_time": datetime.now().strftime('%Y-%m-%d %H:%M:%S'), "total_genes": len(results_data), "results": results_data } filename = f"qPCR_primers_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" return send_file( io.BytesIO(json.dumps(export_data, ensure_ascii=False, indent=2).encode('utf-8')), mimetype='application/json', as_attachment=True, download_name=filename ) if __name__ == '__main__': app.run(debug=True, host='0.0.0.0', port=5000)