data-clean/data_clean/handle/company_court_open_announcement.py

# -*- coding: utf-8 -*-
# @Time : 2023/7/20 16:27
# @Author : XuJiakai
# @File : company_court_open_announcement
# @Software: PyCharm
import os

from data_clean.dim_handle_registry import get_dim_handle
from data_clean.exception.fetch_exception import FetchException
from data_clean.exception.ruler_validation_exception import RulerValidationException
from data_clean.utils.date_utils import *
from data_clean.utils.party_name_verify_utils import person_name_verify
from data_clean.utils.str_utils import json_str_2_list

# 必须命名为dim_handle
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("前置程序错误")

    pass


@dim_handle.registry_postfix_func()
async def post_func(dim_data: list):
    # print("后置程序：", dim_data)

    for r in dim_data:
        r['update_time'] = get_update_time()
        pass

    pass


@dim_handle.registry_row_func
async def party_intersect(row_data: dict) -> dict:
    # 判断当事人有交叉
    plaintiff_info = json_str_2_list(row_data['plaintiff_info'], "name")
    defendant_info = json_str_2_list(row_data['defendant_info'], "name")

    inter = list(set(plaintiff_info).intersection(set(defendant_info)))

    if len(inter) == 0:
        return row_data
    else:
        raise RulerValidationException("ccoa_001", "当事人有交叉：%s" % inter)

    pass


@dim_handle.registry_row_func
async def open_ann_date(row_data: dict) -> dict:
    # 过滤开庭时间早于建国时间问题
    if 'start_date' in row_data:
        if not row_data['start_date']:
            raise RulerValidationException("ccoa_002", "开庭时间为空")

        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'])

    return row_data
    pass


@dim_handle.registry_row_func
async def party_unknown(row_data: dict) -> dict:
    # 过滤当事人名字异常，Z某某、xxx
    plaintiff_info = json_str_2_list(row_data['plaintiff_info'], "name")
    defendant_info = json_str_2_list(row_data['defendant_info'], "name")
    li = plaintiff_info + defendant_info
    for i in li:
        if not person_name_verify(i):
            raise RulerValidationException("ccoa_003", "人名不符合规范：%s" % i)
    return row_data


if __name__ == '__main__':
    pass