fix: 调整sls日志输出的编码问题

JobMain.py

@@ -5,6 +5,7 @@
 # @Software: PyCharm
 import asyncio
 import json
+from functools import partial
 
 import aio_pika
 from aio_pika import IncomingMessage
@@ -14,6 +15,8 @@ from data_clean.task_distributor import task_distribute
 from data_clean.utils.async_client import get_aio_kafka_consumer, get_aio_kafka_producer, get_rabbitmq_connection
 from data_clean.utils.asyncio_pool import AsyncPool
 
+json.dumps = partial(json.dumps, ensure_ascii=False)
+
 env = Env()
 
 base_topic = env.str("base_topic", "rt_other_dim")

data_clean/exception/exception_handle.py

@@ -22,6 +22,7 @@ async def ruler_valid_exception_sink(ex: RulerValidationException, tn: str, data
     :param ex:
     :param tn:
     :param data:
+    :param session_id:
     :return:
     """
     col = "a_data_clean_ruler_valid_error"
@@ -43,6 +44,7 @@ async def fetch_exception_sink(ex: FetchException, tn: str, data, session_id):
     :param ex:
     :param tn:
     :param data:
+    :param session_id:
     :return:
     """
     col_pre = "a_data_clean_fetch_error"
@@ -64,6 +66,7 @@ async def error_sink(ex: Exception, tn: str, data, session_id):
     :param ex:
     :param tn:
     :param data:
+    :param session_id:
     :return:
     """
     col_pre = f"a_data_clean_error"

data_clean/handle/company_court_open_announcement.py

@@ -19,7 +19,7 @@ dim_handle = get_dim_handle(os.path.basename(__file__))
 # @dim_handle.registry_prefix_func
 async def prefix_func(dim_data: list):
     print("前置程序：", dim_data)
-    # raise ValueError("前置程序错误")
+    raise ValueError("前置程序错误")
 
     pass
 
@@ -55,6 +55,11 @@ async def party_intersect(row_data: dict) -> dict:
 async def open_ann_date(row_data: dict) -> dict:
     # 过滤开庭时间早于建国时间问题
     if 'start_date' in row_data:
+
+        if row_data['start_date']:
+            raise RulerValidationException("ccoa_002", "开庭时间为空")
+            pass
+
         this_date = str_2_date_time(row_data['start_date'])
         if this_date < establish_state_time:
             raise RulerValidationException("ccoa_002", "开庭时间早于建国时间：%s" % row_data['start_date'])