1

Author: LAPTOP-7V78BBO2\ydf19

.qoder/rules/r1.md

@@ -0,0 +1,7 @@
+---
+trigger: always_on
+alwaysApply: true
+---
+
+1. 我的操作系统是win,别用linux命令执行
+2. 如果你要生成临时的测试验证代码,请生成到 tests/ai_gen 这个目录下

.qoderignore

@@ -0,0 +1 @@
+# Specify files or folders to ignore during indexing. Use commas to separate entries. Glob patterns like *.log，my-security/ are supported.

README.md

@@ -491,14 +491,14 @@ def _user_convert_msg_before_run(self, msg: typing.Union[str, dict]) -> dict:
         return msg
 
     def _submit_task(self, kw):
-
         kw['body'] = self._convert_msg_before_run(kw['body'])
         self._print_message_get_from_broker(kw['body'])
         if self._judge_is_daylight():
             self._requeue(kw)
             time.sleep(self.time_interval_for_check_do_not_run_time)
             return
         function_only_params = delete_keys_and_return_new_dict(kw['body'], )
+        kw['function_only_params'] = function_only_params
         if self._get_priority_conf(kw, 'do_task_filtering') and self._redis_filter.check_value_exists(
                 function_only_params,self._get_priority_conf(kw, 'filter_str')):  # 对函数的参数进行检查，过滤已经执行过并且成功的任务。
             self.logger.warning(f'redis的 [{self._redis_filter_key_name}] 键 中 过滤任务 {kw["body"]}')
@@ -688,7 +688,7 @@ def _run(self, kw: dict, ):
             max_retry_times = self._get_priority_conf(kw, 'max_retry_times')
             current_function_result_status = FunctionResultStatus(self.queue_name, self.consuming_function.__name__, kw['body'], )
             current_retry_times = 0
-            function_only_params = delete_keys_and_return_new_dict(kw['body'])
+            function_only_params = kw['function_only_params']
             for current_retry_times in range(max_retry_times + 1):
                 current_function_result_status.run_times = current_retry_times + 1
                 current_function_result_status.run_status = RunStatus.running
@@ -747,10 +747,10 @@ def _run(self, kw: dict, ):
     # noinspection PyProtectedMember
     def _run_consuming_function_with_confirm_and_retry(self, kw: dict, current_retry_times,
                                                        function_result_status: FunctionResultStatus, ):
-        function_only_params = delete_keys_and_return_new_dict(kw['body']) if self._do_not_delete_extra_from_msg is False else kw['body']
+        function_only_params = kw['function_only_params'] if self._do_not_delete_extra_from_msg is False else kw['body']
         task_id = kw['body']['extra']['task_id']
         t_start = time.time()
-        # function_result_status.run_times = current_retry_times + 1
+      
         fct = funboost_current_task()
         fct_context = FctContext(function_params=function_only_params,
                                  full_msg=kw['body'],
@@ -872,7 +872,7 @@ def _unified_run(self, kw, is_async):
             max_retry_times = self._get_priority_conf(kw, 'max_retry_times')
             current_function_result_status = FunctionResultStatus(self.queue_name, self.consuming_function.__name__, kw['body'], )
             current_retry_times = 0
-            function_only_params = delete_keys_and_return_new_dict(kw['body'])
+            function_only_params = kw['function_only_params']
             for current_retry_times in range(max_retry_times + 1):
                 current_function_result_status.run_times = current_retry_times + 1
                 current_function_result_status.run_status = RunStatus.running
@@ -936,8 +936,8 @@ async def _async_run_consuming_function_with_confirm_and_retry(self, kw: dict, c
                                                                    function_result_status: FunctionResultStatus, ):
         """虽然和上面有点大面积重复相似，这个是为了asyncio模式的，asyncio模式真的和普通同步模式的代码思维和形式区别太大，
         框架实现兼容async的消费函数很麻烦复杂，连并发池都要单独写"""
-        function_only_params = delete_keys_and_return_new_dict(kw['body']) if self._do_not_delete_extra_from_msg is False else kw['body']
-        function_result_status.run_times = current_retry_times + 1
+        function_only_params = kw['function_only_params'] if self._do_not_delete_extra_from_msg is False else kw['body']
+      
         # noinspection PyBroadException
         t_start = time.time()
         fct = funboost_current_task()

funboost/consumers/base_consumer.py

@@ -1,6 +1,6 @@
 import pytz
 import time
-
+import sys
 import datetime
 
 import typing
@@ -28,6 +28,46 @@ def get_str_fast(self):
         return t_str
 
 
+
+
+# 缓存时区对象，提升性能（避免重复解析）
+_tz_cache = {}
+
+def get_now_time_str_by_tz(tz_name: str=None) -> str:                   
+    """
+    根据时区名（如 'Asia/Shanghai'）返回当前时间字符串，格式：'%Y-%m-%d %H:%M:%S'
+    
+    兼容 Python 3.6+，优先使用 zoneinfo（3.9+），否则尝试 pytz
+    
+    :param tz_name: IANA 时区名称，如 'Asia/Shanghai', 'America/New_York'
+    :return: 格式化时间字符串
+    """
+    # 检查缓存
+    tz_name = tz_name or FunboostCommonConfig.TIMEZONE
+    if tz_name not in _tz_cache:
+        if sys.version_info >= (3, 9):
+            from zoneinfo import ZoneInfo
+            _tz_cache[tz_name] = ZoneInfo(tz_name)
+        else:
+            # Python < 3.9，使用 pytz
+            try:
+                import pytz
+                _tz_cache[tz_name] = pytz.timezone(tz_name)
+            except ImportError:
+                raise RuntimeError(
+                    f"Python < 3.9 requires 'pytz' to handle timezones. "
+                    f"Install it with: pip install pytz"
+                ) from None
+            except pytz.UnknownTimeZoneError:
+                raise pytz.UnknownTimeZoneError(tz_name)
+
+    tz = _tz_cache[tz_name]
+    
+    # 获取当前时间并格式化（注意：datetime.now(tz) 是最高效的方式）
+    now = datetime.datetime.now(tz)
+    return f'{now.year:04d}-{now.month:02d}-{now.day:02d} {now.hour:02d}:{now.minute:02d}:{now.second:02d}'
+    # return now.strftime("%Y-%m-%d %H:%M:%S")
+
 if __name__ == '__main__':
     print(FunboostTime().get_str())
     tz=pytz.timezone(FunboostCommonConfig.TIMEZONE)

funboost/core/funboost_time.py

@@ -14,7 +14,7 @@
 from pymongo import IndexModel, ReplaceOne
 
 from funboost.core.func_params_model import FunctionResultStatusPersistanceConfig
-from funboost.core.helper_funs import get_publish_time, delete_keys_and_return_new_dict
+from funboost.core.helper_funs import get_publish_time, delete_keys_and_return_new_dict, get_publish_time_format
 from funboost.core.serialization import Serialization
 from funboost.utils import time_util, decorators
 from funboost.utils.mongo_util import MongoMixin
@@ -42,9 +42,9 @@ def __init__(self, queue_name: str, fucntion_name: str, msg_dict: dict):
         self.task_id = self.msg_dict.get('extra', {}).get('task_id', '')
         self.process_id = os.getpid()
         self.thread_id = threading.get_ident()
-        self.publish_time = publish_time = get_publish_time(msg_dict)
-        if publish_time:
-            self.publish_time_str = time_util.DatetimeConverter(publish_time).datetime_str
+        self.publish_time  = get_publish_time(msg_dict)
+        self.publish_time_format = get_publish_time_format(msg_dict)
+        # print(self.publish_time_format)
         function_params = delete_keys_and_return_new_dict(msg_dict, )
         self.params = function_params
         self.params_str = Serialization.to_json_str(function_params)

funboost/core/function_result_status_saver.py

@@ -3,7 +3,7 @@
 import time
 import uuid
 import datetime
-from funboost.core.funboost_time import FunboostTime
+from funboost.core.funboost_time import FunboostTime, get_now_time_str_by_tz
 
 
 def get_publish_time(paramsx: dict):
@@ -14,6 +14,14 @@ def get_publish_time(paramsx: dict):
     return paramsx.get('extra', {}).get('publish_time', None)
 
 
+def get_publish_time_format(paramsx: dict):
+    """
+    :param paramsx:
+    :return:
+    """
+    return paramsx.get('extra', {}).get('publish_time_format', None)
+
+
 def delete_keys_and_return_new_dict(dictx: dict, keys: list = None):
     dict_new = copy.deepcopy(dictx)  # 主要是去掉一级键 publish_time，浅拷贝即可。新的消息已经不是这样了。
     keys = ['publish_time', 'publish_time_format', 'extra'] if keys is None else keys
@@ -49,13 +57,18 @@ def generate_task_id(queue_name:str) -> str:
     def generate_publish_time() -> float:
         return round(time.time(),4)
 
+    # @staticmethod  # 性能不好
+    # def generate_publish_time_format() -> str:
+    #     return FunboostTime().get_str()
+    
     @staticmethod
     def generate_publish_time_format() -> str:
-        return FunboostTime().get_str()
+        return get_now_time_str_by_tz()
 
     @classmethod
     def generate_pulish_time_and_task_id(cls,queue_name:str,task_id=None):
-        extra_params = {'task_id': task_id or cls.generate_task_id(queue_name), 'publish_time': cls.generate_publish_time(),
+        extra_params = {'task_id': task_id or cls.generate_task_id(queue_name), 
+                        'publish_time': cls.generate_publish_time(),
                         'publish_time_format': cls.generate_publish_time_format()}
         return extra_params