zbparse/flask_app/routes/utils.py

#flask_app/routes/utils.py
import json
from functools import wraps

from flask import request, jsonify, current_app, g

from flask_app.general.清除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', 1)
    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 = {
                'get_deviation': 'output3',
                'little_zbparse': 'output2',
                'upload': 'output1',
                'test_zbparse': 'test_output'
            }
            subfolder = subfolder_map.get(blueprint, 'output1')

            # 创建 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)}")