#flask_app/routes/utils.py import json import os from functools import wraps from flask import request, jsonify, current_app, g from flask_app.general.llm.清除file_id import read_file_ids, delete_file_by_ids from flask_app.logger_setup import create_logger def validate_request(): """ 验证请求中的JSON数据。 """ if not request.is_json: return jsonify({'error': 'Missing JSON in request'}), 400 file_url = request.json.get('file_url') zb_type = request.json.get('zb_type', 2) #zb_type:默认按货物标解析 if not file_url: return jsonify({'error': 'No file URL provided'}), 400 try: zb_type = int(zb_type) except (ValueError, TypeError): return jsonify({'error': 'Invalid zb_type provided'}), 400 return file_url, zb_type def generate_deviation_response(tech_deviation, tech_star_deviation, business_deviation, business_star_deviation, zigefuhe_deviation,proof_materials, logger): logger.info(f"技术偏离表: {json.dumps(tech_deviation, ensure_ascii=False, indent=4)}") logger.info(f"技术偏离表带星: {json.dumps(tech_star_deviation, ensure_ascii=False, indent=4)}") logger.info(f"商务偏离表: {json.dumps(business_deviation, ensure_ascii=False, indent=4)}") logger.info(f"商务偏离表带星: {json.dumps(business_star_deviation, ensure_ascii=False, indent=4)}") logger.info(f"资格检查偏离表: {json.dumps(zigefuhe_deviation, ensure_ascii=False, indent=4)}") logger.info(f"所需提交的材料: {json.dumps(proof_materials, ensure_ascii=False, indent=4)}") tech_deviation_response = { 'message': 'procurement_reqs', 'status': 'success', 'data': json.dumps(tech_deviation, ensure_ascii=False) } tech_deviation_star_response = { 'message': 'jishu_star_deviation', 'status': 'success', 'data': json.dumps(tech_star_deviation, ensure_ascii=False) } zigefuhe_deviation_response = { 'message': 'zigefuhe_deviation', 'status': 'success', 'data': json.dumps(zigefuhe_deviation, ensure_ascii=False) } shangwu_deviation_response = { 'message': 'shangwu_deviation', 'status': 'success', 'data': json.dumps(business_deviation, ensure_ascii=False) } shangwu_star_deviation_response = { 'message': 'shangwu_star_deviation', 'status': 'success', 'data': json.dumps(business_star_deviation, ensure_ascii=False) } proof_materials_response = { 'message': 'proof_materials', 'status': 'success', 'data': json.dumps(proof_materials, ensure_ascii=False) } return ( tech_deviation_response, tech_deviation_star_response, zigefuhe_deviation_response, shangwu_deviation_response, shangwu_star_deviation_response, proof_materials_response ) def require_connection_limit(): """装饰器：确保路由使用连接限制，并正确处理生成器函数""" def decorator(f): @wraps(f) def wrapped(*args, **kwargs): limiter = getattr(current_app, 'connection_limiter', None) if limiter is None: current_app.logger.error("ConnectionLimiter 未初始化") return jsonify({'error': 'Server configuration error'}), 500 acquired = limiter.semaphore.acquire(blocking=True) if not acquired: return jsonify({ 'error': 'Server is busy. Please try again later.', 'code': 503 }), 503 generator = f(*args, **kwargs) try: for item in generator: yield item finally: limiter.semaphore.release() return wrapped return decorator #装饰器来简化验证和日志初始化 def validate_and_setup_logger(f): @wraps(f) def decorated_function(*args, **kwargs): # 进行请求验证 validation_result = validate_request() if isinstance(validation_result, tuple): file_url, zb_type = validation_result # 根据蓝图确定子文件夹 blueprint = request.blueprint subfolder_map = { 'judge_zbfile': 'output4', 'get_deviation': 'output3', 'little_zbparse': 'output2', 'upload': 'output1', 'test_zbparse': 'test_output', 'test_preprocess':'test_output', 'test_readpdf':'test_output' } subfolder = subfolder_map.get(blueprint, 'test_output') # 创建 logger 和 output_folder create_logger(current_app, subfolder) # 将验证后的数据存储在 g 对象中 g.file_url = file_url g.zb_type = zb_type return f(*args, **kwargs) else: # 验证失败，返回错误响应 return validation_result return decorated_function def perform_cleanup(output_folder, logger): """ 清理逻辑：删除临时文件或其他资源。 """ try: if output_folder: logger.info(f"正在清理输出文件夹: {output_folder}") file_ids = read_file_ids(output_folder) failed_file_ids = delete_file_by_ids(file_ids) if failed_file_ids: logger.error(f"以下文件删除失败: {failed_file_ids}") else: logger.info("清理完毕！") except Exception as e: logger.error(f"清理过程中发生异常: {str(e)}") def create_response_normal(message, status, data=''): """ 创建统一格式的 JSON 响应。 参数: message (str): 响应消息。 status (str): 状态标记，'success' 或 'error'。 data (str, optional): 响应数据。默认为空字符串。 status_code (int, optional): HTTP 状态码。默认为 200。 返回: Response: Flask 响应对象。 """ response = jsonify({ 'message': message, 'status': status, 'data': data }) return response def create_response(message, status, data): """ 创建一个统一格式的响应字典。 :param message: 响应消息 :param status: 响应状态（如 'success', 'error', 'processing'） :param data: 具体的数据内容 :return: 字典格式的响应 """ return { 'message': message, 'status': status, 'data': data } def sse_format(response): """ 将响应格式化为 Server-Sent Events (SSE) 的格式。 """ return f"data: {json.dumps(response, ensure_ascii=False)}\n\n" def log_error_unique_id(unique_id, selection=6): """ 记录失败调用的 unique_id 到对应的错误日志文件中。 参数： unique_id (str): 唯一标识符 selection (int): 接口选择编号，用于确定记录到哪个错误文件 """ error_dir = os.path.join("flask_app", "static", "output", 'error_record') os.makedirs(error_dir, exist_ok=True) # 根据 selection 映射到不同的错误文件 error_file_map = { 1: 'upload_error.txt', 2: 'little_zbparse_error.txt', 3: 'get_deviation_error.txt', 4: 'judge_zbfile_error.txt', 5: 'test_zbparse_error.txt' # 如果有更多接口，可以在这里继续添加 } # 获取对应的错误文件名，如果 selection 不在映射中，则使用默认文件名 error_file_name = error_file_map.get(selection, 'general_error.txt') error_file_path = os.path.join(error_dir, error_file_name) # 将 unique_id 写入错误文件 with open(error_file_path, 'a', encoding='utf-8') as f: f.write(f"{unique_id}\n")