11.24 速率限制

flask_app/ConnectionLimiter.py

@@ -1,31 +1,21 @@
-from threading import Semaphore
+# flask_app/ConnectionLimiter.py
+import threading
 from functools import wraps
-from flask import jsonify, current_app
 
 
 class ConnectionLimiter:
-    _instance = None
-
-    def __new__(cls, *args, **kwargs):
-        if cls._instance is None:
-            cls._instance = super().__new__(cls)
-        return cls._instance
-
     def __init__(self, max_connections=10):
-        if not hasattr(self, 'semaphore'):
+        self.semaphore = threading.Semaphore(max_connections)
-            self.semaphore = Semaphore(max_connections)
 
     def limit_connections(self, f):
+        """装饰器：限制并发连接数"""
+
         @wraps(f)
-        def decorated_function(*args, **kwargs):
+        def wrapped(*args, **kwargs):
-            if not self.semaphore.acquire(blocking=False):
+            self.semaphore.acquire()
-                return jsonify({
-                    'error': 'Server is busy. Maximum number of concurrent connections reached.',
-                    'code': 429
-                }), 429
             try:
                 return f(*args, **kwargs)
             finally:
                 self.semaphore.release()
 
-        return decorated_function
+        return wrapped

flask_app/main/工程标解析main.py

@@ -98,35 +98,48 @@ def preprocess_files(output_folder, downloaded_file_path, file_type,unique_id,lo
 
 # 基本信息
 
-def fetch_project_basic_info(invalid_path, merged_baseinfo_path, merged_baseinfo_path_more,tobidders_notice, clause_path, logger):
+def fetch_project_basic_info(invalid_path, merged_baseinfo_path, merged_baseinfo_path_more, tobidders_notice, clause_path, logger):
     logger.info("starting 基础信息...")
     start_time = time.time()
+    try:
         if not merged_baseinfo_path:
             merged_baseinfo_path = invalid_path
         if not merged_baseinfo_path_more:
-        merged_baseinfo_path_more=invalid_path
+            merged_baseinfo_path_more = invalid_path
         if not tobidders_notice:
             tobidders_notice = invalid_path
-    basic_res = combine_basic_info(merged_baseinfo_path,merged_baseinfo_path_more,tobidders_notice, clause_path)
+        basic_res = combine_basic_info(merged_baseinfo_path, merged_baseinfo_path_more, tobidders_notice, clause_path)
+        result = basic_res
+    except Exception as exc:
+        logger.error(f"Error in 基础信息: {exc}")
+        # 返回默认值
+        result = {"基础信息": {}}
     end_time = time.time()
     logger.info(f"基础信息 done，耗时：{end_time - start_time:.2f} 秒")
-    return basic_res
+    return result
 
 
-# 形式、响应、资格评审
+def fetch_qualification_review(evaluation_method, qualification, output_folder, tobidders_notice_table, clause_path, invalid_path, merged_baseinfo_path, notice_path, logger):
-def fetch_qualification_review(evaluation_method, qualification, output_folder, tobidders_notice_table, clause_path, invalid_path, merged_baseinfo_path,notice_path,logger):
     logger.info("starting 资格审查...")
     start_time = time.time()
+    try:
         if not notice_path:
-        notice_path=invalid_path
+            notice_path = invalid_path
         if not evaluation_method:
             evaluation_method = invalid_path
         if not merged_baseinfo_path:
             merged_baseinfo_path = invalid_path
-    review_standards_res = combine_review_standards(evaluation_method, qualification, output_folder, tobidders_notice_table, clause_path, invalid_path, merged_baseinfo_path,notice_path)
+        review_standards_res = combine_review_standards(
+            evaluation_method, qualification, output_folder, tobidders_notice_table, clause_path, invalid_path, merged_baseinfo_path, notice_path
+        )
+        result = {"资格审查": review_standards_res}
+    except Exception as exc:
+        logger.error(f"Error in 资格审查: {exc}")
+        # 返回默认值
+        result = {"资格审查": {}}
     end_time = time.time()
     logger.info(f"资格审查 done，耗时：{end_time - start_time:.2f} 秒")
-    return review_standards_res
+    return result
 
 
 # 评分细则 流式
@@ -147,38 +160,56 @@ def fetch_evaluation_standards(invalid_path, evaluation_method,logger):
 
 
 # 无效、废标项解析
-def fetch_invalid_requirements(invalid_docpath, output_folder,logger):
+def fetch_invalid_requirements(invalid_docpath, output_folder, logger):
     logger.info("starting 无效标与废标...")
     start_time = time.time()
+    try:
         find_invalid_res = combine_find_invalid(invalid_docpath, output_folder)
+        result = find_invalid_res
+    except Exception as exc:
+        logger.error(f"Error in 无效标与废标: {exc}")
+        # 返回默认值
+        result = {"无效标与废标": {}}
     end_time = time.time()
     logger.info(f"无效标与废标 done，耗时：{end_time - start_time:.2f} 秒")
-    return find_invalid_res
+    return result
 
 
 # 投标文件要求
-def fetch_bidding_documents_requirements(invalid_path, merged_baseinfo_path_more,clause_path,logger):
+def fetch_bidding_documents_requirements(invalid_path, merged_baseinfo_path_more, clause_path, logger):
     logger.info("starting 投标文件要求...")
     start_time = time.time()
+    try:
         if not merged_baseinfo_path_more:
-        merged_baseinfo_path_more=invalid_path
+            merged_baseinfo_path_more = invalid_path
         selection = 1
         fetch_bidding_documents_requirements_json = extract_from_notice(merged_baseinfo_path_more, clause_path, selection)
+        result = {"投标文件要求": fetch_bidding_documents_requirements_json}
+    except Exception as exc:
+        logger.error(f"Error in 投标文件要求: {exc}")
+        # 返回默认值
+        result = {"投标文件要求": {}}
     end_time = time.time()
     logger.info(f"投标文件要求 done，耗时：{end_time - start_time:.2f} 秒")
-    return {"投标文件要求": fetch_bidding_documents_requirements_json}
+    return result
 
 # 开评定标流程
-def fetch_bid_opening(invalid_path, merged_baseinfo_path_more,clause_path,logger):
+def fetch_bid_opening(invalid_path, merged_baseinfo_path_more, clause_path, logger):
     logger.info("starting 开评定标流程...")
     start_time = time.time()
+    try:
         if not merged_baseinfo_path_more:
-        merged_baseinfo_path_more=invalid_path
+            merged_baseinfo_path_more = invalid_path
         selection = 2
         fetch_bid_opening_json = extract_from_notice(merged_baseinfo_path_more, clause_path, selection)
+        result = {"开评定标流程": fetch_bid_opening_json}
+    except Exception as exc:
+        logger.error(f"Error in 开评定标流程: {exc}")
+        # 返回默认值
+        result = {"开评定标流程": {}}
     end_time = time.time()
     logger.info(f"开评定标流程 done，耗时：{end_time - start_time:.2f} 秒")
-    return {"开评定标流程": fetch_bid_opening_json}
+    return result
 
 #分段返回
 
@@ -226,7 +257,14 @@ def engineering_bid_main(output_folder, downloaded_file_path, file_type, unique_
 
             except Exception as exc:
                 logger.error(f"Error processing {key}: {exc}")
-                yield json.dumps({'error': f'Error processing {key}: {str(exc)}'}, ensure_ascii=False)
+                if key == 'evaluation_standards':
+                    # 返回默认的商务评分和技术评分
+                    default_evaluation = {
+                        'technical_standards': {"技术评分": ""},
+                        'commercial_standards': {"商务评分": ""}
+                    }
+                    yield json.dumps(default_evaluation, ensure_ascii=False)
+                # yield json.dumps({'error': f'Error processing {key}: {str(exc)}'}, ensure_ascii=False)
 
 #TODO:基本信息，判断是否这里，打勾逻辑取消了。
 if __name__ == "__main__":

flask_app/routes/upload.py

@@ -22,20 +22,16 @@ def require_connection_limit():
         @wraps(f)
         def wrapped(*args, **kwargs):
             limiter = current_app.connection_limiter
-            return limiter.limit_connections(f)(*args, **kwargs)
+            limiter.semaphore.acquire()  # 阻塞式获取信号量
+            try:
+                return f(*args, **kwargs)
+            finally:
+                limiter.semaphore.release()
         return wrapped
     return decorator
 @upload_bp.route('/upload', methods=['POST'])
+@require_connection_limit()
 def zbparse():
-    # 获取当前应用的 connection_limiter
-    limiter = current_app.connection_limiter
-
-    if not limiter.semaphore.acquire(blocking=False):
-        return jsonify({
-            'error': 'Server is busy. Maximum number of concurrent connections reached.',
-            'code': 429
-        }), 429
-
     try:
         logger = g.logger
         logger.info("zbparse start!!!")
@@ -55,8 +51,9 @@ def zbparse():
         except Exception as e:
             logger.error('Exception occurred: ' + str(e))
             return jsonify({'error': str(e)}), 500
-    finally:
+    except Exception as e:
-        limiter.semaphore.release()
+        logger.error('Unexpected exception: ' + str(e))
+        return jsonify({'error': 'Internal server error'}), 500
 
 def process_and_stream(file_url, zb_type):
     """

flask_app/货物标/货物标解析main.py

@@ -78,28 +78,39 @@ def preprocess_files(output_folder, file_path, file_type,logger):
         'merged_baseinfo_path': merged_baseinfo_path
     }
 
-def fetch_project_basic_info(invalid_path,merged_baseinfo_path, procurement_path, clause_path,logger):
+def fetch_project_basic_info(invalid_path, merged_baseinfo_path, procurement_path, clause_path, logger):
     logger.info("starting 基础信息...")
     start_time = time.time()
+    try:
         if not merged_baseinfo_path:
             merged_baseinfo_path = invalid_path
         if not procurement_path:
-        procurement_path=invalid_path
+            procurement_path = invalid_path
-    basic_res = combine_basic_info(merged_baseinfo_path, procurement_path,clause_path,invalid_path)
+        basic_res = combine_basic_info(merged_baseinfo_path, procurement_path, clause_path, invalid_path)
-    base_info, good_list = post_process_baseinfo(basic_res,logger)
+        base_info, good_list = post_process_baseinfo(basic_res, logger)
+        result = base_info, good_list
+    except Exception as exc:
+        logger.error(f"Error in 基础信息: {exc}")
+        # 返回默认值
+        result = {"基础信息": {}}, []
     end_time = time.time()
     logger.info(f"基础信息 done，耗时：{end_time - start_time:.2f} 秒")
-    return base_info, good_list
+    return result
 
 
-def fetch_qualification_review(invalid_path,qualification_path, notice_path,logger):
+def fetch_qualification_review(invalid_path, qualification_path, notice_path, logger):
     logger.info("starting 资格审查...")
     start_time = time.time()
-    review_standards_res = combine_qualification_review(invalid_path,qualification_path, notice_path)
+    try:
+        review_standards_res = combine_qualification_review(invalid_path, qualification_path, notice_path)
+        result = {"资格审查": review_standards_res}
+    except Exception as exc:
+        logger.error(f"Error in 资格审查: {exc}")
+        # 返回默认值
+        result = {"资格审查": {}}
     end_time = time.time()
     logger.info(f"资格审查 done，耗时：{end_time - start_time:.2f} 秒")
-    return review_standards_res
+    return result
-
 
 def fetch_evaluation_standards(invalid_path, evaluation_method_path,logger):
     logger.info("starting 商务评分和技术评分...")
@@ -116,37 +127,55 @@ def fetch_evaluation_standards(invalid_path, evaluation_method_path,logger):
         "commercial_standards": commercial_standards
     }
 
-def fetch_invalid_requirements(invalid_docpath, output_folder,logger):
+def fetch_invalid_requirements(invalid_docpath, output_folder, logger):
     logger.info("starting 无效标与废标...")
     start_time = time.time()
+    try:
         find_invalid_res = combine_find_invalid(invalid_docpath, output_folder)
+        result = find_invalid_res
+    except Exception as exc:
+        logger.error(f"Error in 无效标与废标: {exc}")
+        # 返回默认值
+        result = {"无效标与废标": {}}
     end_time = time.time()
     logger.info(f"无效标与废标 done，耗时：{end_time - start_time:.2f} 秒")
-    return find_invalid_res
+    return result
 
-def fetch_bidding_documents_requirements(invalid_path,merged_baseinfo_path,clause_path,logger):
+def fetch_bidding_documents_requirements(invalid_path, merged_baseinfo_path, clause_path, logger):
     logger.info("starting 投标文件要求...")
     if not merged_baseinfo_path:
-        merged_baseinfo_path=invalid_path
+        merged_baseinfo_path = invalid_path
     start_time = time.time()
-    selection=1
+    selection = 1
-    fetch_bidding_documents_requirements_json = extract_from_notice(merged_baseinfo_path,clause_path, selection)
+    try:
+        fetch_bidding_documents_requirements_json = extract_from_notice(merged_baseinfo_path, clause_path, selection)
+        result = {"投标文件要求": fetch_bidding_documents_requirements_json}
+    except Exception as exc:
+        logger.error(f"Error in 投标文件要求: {exc}")
+        # 返回默认值，假设默认值为一个空字典
+        result = {"投标文件要求": {}}
     end_time = time.time()
     logger.info(f"投标文件要求 done，耗时：{end_time - start_time:.2f} 秒")
-    return {"投标文件要求": fetch_bidding_documents_requirements_json}
+    return result
 
 
 # 开评定标流程
-def fetch_bid_opening(invalid_path,merged_baseinfo_path,clause_path,logger):
+def fetch_bid_opening(invalid_path, merged_baseinfo_path, clause_path, logger):
     logger.info("starting 开评定标流程...")
     if not merged_baseinfo_path:
-        merged_baseinfo_path=invalid_path
+        merged_baseinfo_path = invalid_path
     start_time = time.time()
-    selection=2
+    selection = 2
-    fetch_bid_opening_json = extract_from_notice(merged_baseinfo_path,clause_path, selection)
+    try:
+        fetch_bid_opening_json = extract_from_notice(merged_baseinfo_path, clause_path, selection)
+        result = {"开评定标流程": fetch_bid_opening_json}
+    except Exception as exc:
+        logger.error(f"Error in 开评定标流程: {exc}")
+        # 返回默认值，假设默认值为一个空字典
+        result = {"开评定标流程": {}}
     end_time = time.time()
     logger.info(f"开评定标流程 done，耗时：{end_time - start_time:.2f} 秒")
-    return {"开评定标流程": fetch_bid_opening_json}
+    return result
 
 
 def post_process_baseinfo(base_info,logger):
@@ -230,7 +259,14 @@ def goods_bid_main(output_folder, file_path, file_type, unique_id):
                     yield json.dumps({key: transform_json_values(result)}, ensure_ascii=False)
             except Exception as exc:
                 logger.error(f"Error processing {key}: {exc}")
-                yield json.dumps({'error': f'Error processing {key}: {str(exc)}'}, ensure_ascii=False)
+                if key == 'evaluation_standards':
+                    # 返回默认的商务评分和技术评分
+                    default_evaluation = {
+                        'technical_standards': {"技术评分": ""},
+                        'commercial_standards': {"商务评分": ""}
+                    }
+                    yield json.dumps(default_evaluation, ensure_ascii=False)
+                # yield json.dumps({'error': f'Error processing {key}: {str(exc)}'}, ensure_ascii=False)
         if collected_good_list is not None:
             yield json.dumps({'good_list': transform_json_values(collected_good_list)}, ensure_ascii=False)