finance_data_process/car/car-wei-zhang/car_wei_zhang.py

"""车辆违章数据处理
"""

import re
import subprocess
from datetime import datetime
from dateutil.relativedelta import relativedelta
from loguru import logger
import pandas as pd
import psycopg
import paramiko

# 添加日志记录，将日志输出到文件 a.log
logger.add(sink='a.log')
ssh_hostname = '172.16.107.4'  # 定义远程主机地址
ssh_port = 22  # 定义SSH服务的端口号
ssh_username = 'app'  # 定义登录远程主机的用户名
ssh_password = '(l4w0ST_'  # 定义登录远程主机的密码
# 服务器文件夹路径
remote_dir_path = '/data/history/car/wei-zhang/'
# 数据库连接信息
db_host = "172.16.107.5"  # 数据库主机地址
db_port = 5432  # 数据库端口号
db_username = "finance"  # 数据库用户名
db_password = "Finance@unicom23"  # 数据库密码
dbname = "financialdb"  # 数据库名称
conn_info = f"host='{db_host}' port={db_port} user='{db_username}' password='{db_password}' dbname='{dbname}'"
# 获取当前日期，并计算上个月的第一天
today = datetime.today()
start_date = today - relativedelta(months=1, day=1)
year_month = start_date.strftime('%Y%m')
# 数据文件路径
input_path = 'data.xlsx'
# 输出文件路径
output_path = 'output.csv'


def data_process():
    # 正则表达式匹配车牌省份简称（如京、津、晋等）
    has_che_pai_province_pattern = re.compile(
        "[" + re.escape("京津晋冀蒙辽吉黑沪苏浙皖闽赣鲁豫鄂湘粤桂琼渝川贵云藏陕甘青宁国防") + "]")

    # 正则表达式匹配非车牌字符，排除车牌可能包含的字符（如字母、数字、特殊标志等）
    not_che_pai_pattern = re.compile(
        "[^京津晋冀蒙辽吉黑沪苏浙皖闽赣鲁豫鄂湘粤桂琼渝川贵云藏陕甘青宁新港澳学挂领试超练警国防A-Z\\d]")

    # 正则表达式匹配完整的车牌号格式
    che_pai_pattern = re.compile(
        r"([京津沪渝冀豫云辽黑湘皖鲁新苏浙赣鄂桂甘晋蒙陕吉闽贵粤青藏川宁琼使领A-Z][A-Z]"
        r"(([DF]((?![IO])[A-Z0-9](?![IO]))\d{4})|(\d{5}[DF]))|"
        r"[京津沪渝冀豫云辽黑湘皖鲁新苏浙赣鄂桂甘晋蒙陕吉闽贵粤青藏川宁琼使领A-Z][A-Z][A-Z0-9]{4}[A-Z0-9挂学警港澳])"
    )

    # 一级单位字典
    first_unit_map = {
        "保定市分公司": "保定",
        "沧州市分公司": "沧州",
        "承德市分公司": "承德",
        "邯郸市分公司": "邯郸",
        "河北省分公司线路维护中心": "机动局",
        "河北省机动通信局": "机动局",
        "衡水市分公司": "衡水",
        "华北基地建设部": "华北基地建设部",
        "廊坊市分公司": "廊坊",
        "秦皇岛市分公司": "秦皇岛",
        "省公司本部": "省公司本部",
        "石家庄市分公司": "石家庄",
        "唐山市分公司": "唐山",
        "邢台市分公司": "邢台",
        "雄安基地建设部": "雄安基地建设部",
        "雄安新区分公司": "雄安",
        "张家口市分公司": "张家口"
    }

    # 定义二级行政区划映射表（地级市及其下属区县）
    er_ji_map = {
        "石家庄": ["鹿泉", "藁城", "栾城", "井陉矿区", "井陉", "无极", "正定", "元氏", "新乐", "晋州", "平山", "灵寿",
                   "赞皇", "赵县", "行唐", "高邑", "辛集", "深泽"],
        "唐山": ["唐山高开区", "迁西", "海港", "开平", "丰南", "滦县", "乐亭", "丰润", "玉田", "古冶", "曹妃甸", "遵化",
                 "滦南", "迁安"],
        "秦皇岛": ["北戴河新区", "北戴河", "山海关", "昌黎", "卢龙", "青龙", "抚宁"],
        "邯郸": ["曲周", "魏县", "馆陶", "磁县", "大名", "鸡泽", "成安", "涉县", "永年", "武安", "峰峰", "广平", "临漳",
                 "邱县", "肥乡"],
        "邢台": ["新河", "南宫", "隆尧", "内邱", "平乡", "宁晋", "广宗", "清河", "临西", "任县", "巨鹿", "沙河", "威县",
                 "临城", "柏乡", "南和"],
        "保定": ["涞水", "蠡县", "顺平", "博野", "安国", "涞源", "唐县", "定州", "高阳", "曲阳", "阜平", "清苑",
                 "高碑店",
                 "满城", "涿州", "易县", "望都", "徐水", "定兴", "白沟"],
        "张家口": ["张北", "崇礼", "康保", "赤城", "阳原", "万全", "下花园", "尚义", "怀安", "怀来", "蔚县", "涿鹿",
                   "沽源",
                   "宣化"],
        "承德": ["承德县", "兴隆", "宽城", "平泉", "营子", "隆化", "滦平", "围场", "丰宁", "双滦"],
        "廊坊": ["文安", "霸州", "大城", "廊坊开发区", "三河", "香河", "永清", "胜芳", "燕郊", "固安", "大厂"],
        "沧州": ["东光", "吴桥", "黄骅", "盐山", "孟村", "泊头", "献县", "南皮", "渤海新区", "海兴", "沧县", "河间",
                 "青县",
                 "任丘", "肃宁"],
        "衡水": ["景县", "阜城", "枣强", "深州", "饶阳", "故城", "武强", "武邑", "冀州", "安平"],
        "雄安": ["容城", "雄县", "安新"]
    }

    # 初始化组织结构映射表
    org_map = {}
    third_org_map = {}
    third_org_list_map = {}
    area_map = {}
    district_list_map = {}

    # 连接PostgreSQL数据库
    with psycopg.connect(
            conninfo=conn_info,
            row_factory=psycopg.rows.dict_row
    ) as conn:
        with conn.cursor() as curs:
            # 查询一级组织数据，并按order_num排序
            sql = """
                select * from common.organization where grade = 1 order by order_num
            """
            logger.info(f"sql: {sql}")
            curs.execute(sql)
            second_orgs = curs.fetchall()

            # 遍历一级组织数据，构建org_map和third_org_list_map
            for x in second_orgs:
                org_map[x['id']] = x
                third_org_list_map[x['id']] = []

            # 查询二级组织数据，并按parent_id和order_num排序
            sql = """
                select * from common.organization where grade = 2 order by parent_id, order_num
            """
            logger.info(f"sql: {sql}")
            curs.execute(sql)
            third_orgs = curs.fetchall()

            # 遍历二级组织数据，构建org_map、third_org_list_map和third_org_map
            for x in third_orgs:
                org_map[x['id']] = x
                third_org_list_map[x['parent_id']].append(x)
                third_org_map[x['id']] = x

            # 查询一级行政区划数据，并按area_id排序
            sql = """
                select * from common.area where area_grade = 1 order by area_id
            """
            logger.info(f"sql: {sql}")
            curs.execute(sql)
            cities = curs.fetchall()

            # 遍历一级行政区划数据，构建area_map
            for city in cities:
                area_map[city['area_id']] = city

            # 查询二级行政区划数据，并按parent_id和area_id排序
            sql = """
                select * from common.area where area_grade = 2 order by parent_id, area_id
            """
            logger.info(f"sql: {sql}")
            curs.execute(sql)
            districts = curs.fetchall()

            # 遍历二级行政区划数据，构建area_map和district_list_map
            for district in districts:
                area_map[district['area_id']] = district

            # 构建城市与区县的映射关系
            for city in cities:
                district_list_map[city['area_id']] = []
                for district in districts:
                    if city['area_id'] == district['parent_id']:
                        district_list_map[city['area_id']].append(district)

    # 读取 Excel 文件中的数据
    df = pd.read_excel(io=input_path)
    # 获取需要清理的列名列表，排除 "违章时间" 和 "处理时间" 列
    columns_to_clean = list(filter(lambda x: x not in ('违章时间', '处理时间'), df.columns))
    # 对需要清理的列进行字符串清理，移除多余的空白字符
    df[columns_to_clean] = df[columns_to_clean].map(lambda x: re.sub(r'\s+', '', x) if type(x) is str else x)
    df['账期'] = year_month
    # 保存原始单位和车牌号信息到新的列中
    df['原始一级单位'] = df['一级单位']
    df['原始二级单位'] = df['二级单位']
    df['原始三级单位'] = df['三级单位']
    df['原始车牌号'] = df['车牌号']

    # 定义函数，用于提取并标准化车牌号
    def get_che_pai(che_pai):
        # 如果车牌号为空或无效，则返回空字符串
        if pd.isna(che_pai) or not che_pai or not che_pai.strip():
            return ""
        # 将车牌号转换为大写
        upper_case = che_pai.upper()
        # 移除车牌号中不符合规则的字符
        s = not_che_pai_pattern.sub("", upper_case)
        # 使用正则表达式匹配合法的车牌号
        m = che_pai_pattern.search(s)
        if m:
            return m.group(0)

        # 如果车牌号包含省份简称但未匹配成功，记录警告日志
        if has_che_pai_province_pattern.search(che_pai):
            logger.warning(f"车牌匹配失败: {che_pai} -> {s}")
            return s

        # 如果完全无法匹配，记录警告日志并返回原车牌号
        logger.warning(f"车牌匹配失败: {che_pai} -> {upper_case}")
        return upper_case

    # 应用 get_che_pai 函数处理车牌号列
    df['车牌号'] = df['车牌号'].apply(get_che_pai)

    # 定义函数，用于标记车牌号是否匹配失败
    def che_pai_fail(che_pai):
        # 如果车牌号为空或无效，则标记为失败
        if pd.isna(che_pai) or not che_pai or not che_pai.strip():
            return "1"

        # 移除车牌号中不符合规则的字符
        s = not_che_pai_pattern.sub("", che_pai.upper())
        # 使用正则表达式匹配合法的车牌号
        m = che_pai_pattern.search(s)
        if m:
            return "0"  # 匹配成功

        return "1"  # 匹配失败

    # 应用 che_pai_fail 函数生成车牌匹配失败标记列
    df['车牌匹配失败'] = df['车牌号'].apply(che_pai_fail)

    # 获取一级单位
    def get_first_unit(x):
        raw_che_pai = x['原始车牌号']
        raw_first_unit = str(x['原始一级单位']) if pd.notna(x['原始一级单位']) else ""
        if not raw_first_unit or not raw_first_unit.strip():
            raise RuntimeError(f"一级单位为空：{raw_che_pai}")
        if raw_first_unit in first_unit_map:
            return first_unit_map.get(raw_first_unit)
        raise RuntimeError(f"一级单位匹配失败：{raw_che_pai} {raw_first_unit}")

    # 应用 get_first_unit 函数生成一级单位列
    df['一级单位'] = df.apply(get_first_unit, axis=1)

    # 获取二级单位
    def get_second_unit(x):
        first_unit = str(x['一级单位']) if pd.notna(x['一级单位']) else ""
        raw_second_unit = str(x['原始二级单位']) if pd.notna(x['原始二级单位']) else ""
        if first_unit in ["华北基地建设部", "雄安基地建设部", "省公司本部"]:
            return first_unit
        if not raw_second_unit or not raw_second_unit.strip():
            return f"{first_unit}本部"
        if first_unit == "机动局":
            for yj in er_ji_map.keys():
                if yj in raw_second_unit:
                    return f"机动局{yj}"
            return "机动局本部"
        if first_unit == "石家庄":
            if "开发区" in raw_second_unit:
                return "石家庄开发区"
        if first_unit == "廊坊":
            if "开发区" in raw_second_unit:
                return "廊坊开发区"
        if first_unit == "邢台":
            if "内丘" in raw_second_unit:
                return "内邱"
            if "任泽" in raw_second_unit:
                return "任县"
        if first_unit == "唐山":
            if "高开区" in raw_second_unit:
                return "唐山高开区"
            if "滦州" in raw_second_unit:
                return "滦县"
        ejs = er_ji_map.get(first_unit, [])
        if first_unit == "雄安":
            raw_second_unit = raw_second_unit.replace("雄安新区", "")
        for ej in ejs:
            if ej in raw_second_unit:
                return ej
        return f"{first_unit}本部"

    # 应用 get_second_unit 函数生成二级单位列
    df['二级单位'] = df.apply(get_second_unit, axis=1)

    # 获取三级单位
    def get_third_unit(x):
        second_unit = str(x['二级单位']) if pd.notna(x['二级单位']) else ""
        raw_third_unit = str(x['原始三级单位']) if pd.notna(x['原始三级单位']) else ""
        return raw_third_unit if raw_third_unit and raw_third_unit.strip() else second_unit

    # 应用 get_third_unit 函数生成三级单位列
    df['三级单位'] = df.apply(get_third_unit, axis=1)

    # 定义一个函数，用于根据单位名称获取二级组织机构编码
    def get_area_no(first_unit):
        if first_unit == "机动局":
            return "-11"
        if first_unit in ["省公司本部", "雄安基地建设部", "华北基地建设部"]:
            return "-12"
        for second_org in second_orgs:
            if second_org.get('name') in first_unit:
                return second_org.get('id')
        return ''

    df['二级组织机构编码'] = df['一级单位'].apply(get_area_no)

    # 用于根据组织机构编码获取组织机构名称
    def get_org_name(x):
        org_no = str(x) if pd.notna(x) else ''
        # 如果编码为空或无效，则返回空字符串
        if not org_no or not org_no.strip():
            return ''
        # 在org_map中查找对应编码的组织机构信息，并返回其名称
        po = org_map.get(org_no)
        if po is not None:
            return po.get('name')
        return ''

    # 将get_org_name函数应用到'二级组织机构编码'列，生成'二级组织机构名称'列
    df['二级组织机构名称'] = df['二级组织机构编码'].apply(get_org_name)

    # 定义一个函数，用于根据行数据获取三级组织机构编码
    def get_city_no(x):
        # 获取相关字段值，如果为空则设置为""
        area_no = str(x['二级组织机构编码']) if pd.notna(x['二级组织机构编码']) else ""
        area_name = str(x['二级组织机构名称']) if pd.notna(x['二级组织机构名称']) else ""
        first_unit = str(x['一级单位']) if pd.notna(x['一级单位']) else ""
        second_unit = str(x['原始二级单位']) if pd.notna(x['原始二级单位']) else ""
        if not area_no or not area_no.strip() or not area_name or not area_name.strip():
            return ''
        if '华北基地建设部' == first_unit:
            return 'HE018'
        if '雄安基地建设部' == first_unit:
            return 'HE019'
        if second_unit and second_unit.strip():
            if area_name == "石家庄":
                if "井陉矿区" in second_unit:
                    return "D0130185"
                if "井陉" in second_unit:
                    return "D0130121"
            if area_name == "秦皇岛":
                if "北戴河新区" in second_unit:
                    return "D0130325"
                if "北戴河" in second_unit:
                    return "D0130304"
            if area_name == "邯郸":
                if "峰峰" in second_unit:
                    return "D0130406"
            if area_name == "邢台":
                if "内丘" in second_unit:
                    return "D0130523"
                if "任泽" in second_unit:
                    return "D0130526"
            if area_name == "省机动局":
                if "沧州" in second_unit:
                    return "HECS180"
                if "唐山" in second_unit:
                    return "HECS181"
                if "秦皇岛" in second_unit:
                    return "HECS182"
                if "廊坊" in second_unit:
                    return "HECS183"
                if "张家口" in second_unit:
                    return "HECS184"
                if "邢台" in second_unit:
                    return "HECS185"
                if "邯郸" in second_unit:
                    return "HECS186"
                if "保定" in second_unit:
                    return "HECS187"
                if "石家庄" in second_unit:
                    return "HECS188"
                if "承德" in second_unit:
                    return "HECS189"
                if "衡水" in second_unit:
                    return "HECS720"
                if "雄安" in second_unit:
                    return "HECS728"
                return "HECS018"
            if area_name == "雄安":
                second_unit = second_unit.replace("雄安新区", "")
            l3 = third_org_list_map.get(area_no, [])
            for organization_po in l3:
                if organization_po.get('name') in second_unit:
                    return organization_po.get('id')
        if area_name == '省本部':
            return 'HE001'
        if area_name == "省机动局":
            return "HECS018"
        if area_name == "沧州":
            return "D0130911"
        if area_name == "唐山":
            return "D0130202"
        if area_name == "秦皇岛":
            return "D0130302"
        if area_name == "廊坊":
            return "D0131000"
        if area_name == "张家口":
            return "D0130701"
        if area_name == "邢台":
            return "D0130502"
        if area_name == "邯郸":
            return "D0130402"
        if area_name == "保定":
            return "D0130601"
        if area_name == "石家庄":
            return "D0130186"
        if area_name == "承德":
            return "D0130801"
        if area_name == "衡水":
            return "D0133001"
        if area_name == "雄安":
            return "D0130830"
        return ''

    # 将get_city_no函数应用到DataFrame的每一行，生成'三级组织机构编码'列
    df['三级组织机构编码'] = df.apply(get_city_no, axis=1)
    # 将get_org_name函数应用到'三级组织机构编码'列，生成'三级组织机构名称'列
    df['三级组织机构名称'] = df['三级组织机构编码'].apply(get_org_name)

    # 定义一个函数，用于根据行数据获取二级组织机构编码2
    def get_area_no2(x):
        # 获取相关字段值，如果为空则设置为""
        area_name = str(x['二级组织机构名称']) if pd.notna(x['二级组织机构名称']) else ""
        city_name = str(x['三级组织机构名称']) if pd.notna(x['三级组织机构名称']) else ""
        if not area_name or not area_name.strip() or '省本部' == area_name:
            return ''
        # 根据二级组织机构名称和三级组织机构名称的内容，返回对应的编码
        if area_name == "省机动局" and city_name and city_name.strip():
            if "沧州" in city_name:
                return "180"
            if "唐山" in city_name:
                return "181"
            if "秦皇岛" in city_name:
                return "182"
            if "廊坊" in city_name:
                return "183"
            if "张家口" in city_name:
                return "184"
            if "邢台" in city_name:
                return "185"
            if "邯郸" in city_name:
                return "186"
            if "保定" in city_name:
                return "187"
            if "石家庄" in city_name:
                return "188"
            if "承德" in city_name:
                return "189"
            if "衡水" in city_name:
                return "720"
            if "雄安" in city_name:
                return "782"
            return ''
        if "沧州" in area_name:
            return "180"
        if "唐山" in area_name:
            return "181"
        if "秦皇岛" in area_name:
            return "182"
        if "廊坊" in area_name:
            return "183"
        if "张家口" in area_name:
            return "184"
        if "邢台" in area_name:
            return "185"
        if "邯郸" in area_name:
            return "186"
        if "保定" in area_name:
            return "187"
        if "石家庄" in area_name:
            return "188"
        if "承德" in area_name:
            return "189"
        if "衡水" in area_name:
            return "720"
        if "雄安" in area_name:
            return "782"
        return ""

    # 将get_area_no2函数应用到DataFrame的每一行，生成'二级组织机构编码2'列
    df['二级组织机构编码2'] = df.apply(get_area_no2, axis=1)
    # 将get_org_name函数应用到'二级组织机构编码2'列，生成'二级组织机构名称2'列
    df['二级组织机构名称2'] = df['二级组织机构编码2'].apply(get_org_name)

    # 获取城市ID
    def get_city_id(x):
        raw_first_unit = str(x['原始一级单位']) if pd.notna(x['原始一级单位']) else ""
        raw_second_unit = str(x['原始二级单位']) if pd.notna(x['原始二级单位']) else ""
        raw_third_unit = str(x['原始三级单位']) if pd.notna(x['原始三级单位']) else ""
        unit = f"{raw_first_unit}_{raw_second_unit}_{raw_third_unit}"
        if not unit or not unit.strip():
            return ''
        # 遍历cities列表，匹配单位名称并返回对应的城市ID
        for city in cities:
            if city.get('short_name') in unit:
                return city.get('area_id', '')
        return ''

    df['city_id'] = df.apply(get_city_id, axis=1)

    # 定义一个函数，用于根据ID获取区域名称
    def get_area_name(x):
        id = str(x) if pd.notna(x) else ""
        if not id or not id.strip():
            return ""
        # 在area_map中查找对应ID的区域信息，并返回其名称
        area_po = area_map.get(id)
        if area_po is not None:
            return area_po.get("area_name", "")
        return ""

    # 将get_area_name函数应用到'city_id'列，生成'city'列
    df['city'] = df['city_id'].apply(get_area_name)

    # 定义一个函数，用于根据行数据获取区县ID
    def get_district_id(x):
        # 获取相关字段值，如果为空则设置为""
        city_id = str(x['city_id']) if pd.notna(x['city_id']) else ""
        city = str(x['city']) if pd.notna(x['city']) else ""
        raw_first_unit = str(x['原始一级单位']) if pd.notna(x['原始一级单位']) else ""
        raw_second_unit = str(x['原始二级单位']) if pd.notna(x['原始二级单位']) else ""
        raw_third_unit = str(x['原始三级单位']) if pd.notna(x['原始三级单位']) else ""
        unit = f"{raw_first_unit}_{raw_second_unit}_{raw_third_unit}"
        # 如果城市ID、城市名称或单位为空，则返回""
        if not city_id or not city_id.strip() or not city or not city.strip() or not unit or not unit.strip():
            return ""
        # 根据城市名称和单位内容，返回对应的区县ID
        if city == "石家庄":
            if "井陉矿区" in unit:
                return "130107"
            if "井陉" in unit:
                return "130121"
        if city == "雄安":
            unit = unit.replace("雄安新区", "")
        districts = district_list_map.get(city_id, [])
        for district in districts:
            if district.get('short_name') in unit:
                return district.get('area_id')
        return ""

    df['district_id'] = df.apply(get_district_id, axis=1)
    # 将get_area_name函数应用到'district_id'列，生成'district'列
    df['district'] = df['district_id'].apply(get_area_name)

    # 提取账期年份和月份信息
    df['year_no'] = df['账期'].apply(lambda x: None if pd.isna(x) else str(x)[:4])
    df['month_no'] = df['账期'].apply(lambda x: None if pd.isna(x) else str(x)[-2:])
    # 格式化违章时间和处理时间为年月格式
    df['违章年月'] = df['违章时间'].apply(lambda x: None if pd.isna(x) else pd.to_datetime(x).strftime('%Y%m'))
    df['处理年月'] = df['处理时间'].apply(lambda x: None if pd.isna(x) else pd.to_datetime(x).strftime('%Y%m'))

    # 打印DataFrame的信息
    print(df.info())

    # 将处理后的数据保存到CSV文件中
    df.to_csv(path_or_buf=output_path,
              header=['year_month', 'che_pai_hao', 'first_unit', 'second_unit', 'third_unit', 'che_jia_hao',
                      'wei_zhang_shi_jian', 'wei_zhang_di_dian', 'wei_zhang_xiang_qing', 'kou_fen', 'fa_kuan',
                      'chu_li_zhuang_tai', 'chu_li_shi_jian', 'wei_zhang_wei_chu_li_shi_chang', 'raw_yi_ji',
                      'raw_er_ji',
                      'raw_san_ji', 'raw_che_pai_hao', 'che_pai_fail', 'area_no', 'area_name', 'city_no', 'city_name',
                      'area_no2', 'area_name2', 'city_id', 'city', 'district_id', 'district', 'year_no', 'month_no',
                      'wei_zhang_nian_yue', 'chu_li_nian_yue'],
              index=False,
              encoding='utf-8-sig')


def data_import():
    # 定义 PowerShell 脚本的路径
    script_path = r"../../copy.ps1"
    # 目标表和文件信息
    table = "car.car_wei_zhang"  # 数据库目标表名
    # 表字段列名，用于指定导入数据的列顺序
    columns = "year_month,che_pai_hao,first_unit,second_unit,third_unit,che_jia_hao,wei_zhang_shi_jian,wei_zhang_di_dian,wei_zhang_xiang_qing,kou_fen,fa_kuan,chu_li_zhuang_tai,chu_li_shi_jian,wei_zhang_wei_chu_li_shi_chang,raw_yi_ji,raw_er_ji,raw_san_ji,raw_che_pai_hao,che_pai_fail,area_no,area_name,city_no,city_name,area_no2,area_name2,city_id,city,district_id,district,year_no,month_no,wei_zhang_nian_yue,chu_li_nian_yue"
    # 构造执行 PowerShell 脚本的命令
    command = f"powershell -File {script_path} -db_host {db_host} -db_port {db_port} -db_username {db_username} -db_password {db_password} -dbname {dbname} -table {table} -filename {output_path} -columns {columns}"
    # 打印生成的命令，方便调试和日志记录
    logger.info("command: {}", command)
    # 使用 subprocess 模块运行 PowerShell 命令，并捕获输出
    completed_process = subprocess.run(
        command,  # 执行的命令
        check=False,  # 如果命令执行失败，不抛出异常
        text=True,  # 将输出作为字符串处理
        capture_output=True,  # 捕获标准输出和标准错误
    )
    # 打印命令执行的结果，包括返回码、标准输出和标准错误
    logger.info("导入结果：\n{}\n{}\n{}", completed_process.returncode, completed_process.stdout,
                completed_process.stderr)
    # 定义正则表达式，用于匹配标准输出中的 COPY 结果
    p = re.compile(r"^(COPY) (\d+)$")
    count = None  # 初始化计数变量
    matcher = p.match(completed_process.stdout)  # 匹配标准输出中的 COPY 结果
    if matcher:
        count = int(matcher.group(2))  # 提取导入的数据行数

    # 如果没有成功提取到导入数据的行数，抛出运行时异常
    if count is None:
        raise RuntimeError("导入数据失败")


def upload_file():
    remote_path = f'{remote_dir_path}{year_month}.xlsx'  # 定义远程主机的目标文件路径
    # 使用paramiko.SSHClient创建一个SSH客户端对象，并通过with语句管理其上下文
    with paramiko.SSHClient() as ssh:
        # 设置自动添加主机密钥策略，避免因未知主机密钥导致连接失败
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        # 连接到远程主机，传入主机地址、端口、用户名和密码
        ssh.connect(ssh_hostname, port=ssh_port, username=ssh_username, password=ssh_password)
        # 执行远程命令，创建远程目录（如果不存在）
        ssh.exec_command(f'mkdir -p {remote_dir_path}')
        # 打开SFTP会话，用于文件传输，并通过with语句管理其上下文
        with ssh.open_sftp() as sftp:
            # 记录日志，提示即将上传的本地文件和远程目标路径
            logger.info("upload {} to {}", input_path, remote_path)
            # 使用SFTP的put方法将本地文件上传到远程主机
            sftp.put(input_path, remote_path)
            # 记录日志，提示文件已成功上传
            logger.info("uploaded {}", input_path)


def data_update():
    with psycopg.connect(
            conninfo=conn_info,
    ) as conn:
        with conn.cursor() as curs:
            # 插入违章长期未处理
            sql = f"""
                insert
                    into
                    car.car_wei_zhang_chang_qi
                (
                    year_month,
                    che_pai_hao,
                    raw_yi_ji,
                    raw_er_ji,
                    raw_san_ji,
                    wei_zhang_shi_jian,
                    wei_zhang_di_dian,
                    wei_zhang_xiang_qing,
                    kou_fen,
                    fa_kuan,
                    wei_zhang_wei_chu_li_shi_chang,
                    chu_li_zhuang_tai,
                    first_unit,
                    second_unit,
                    third_unit,
                    area_no,
                    area_name,
                    city_no,
                    city_name,
                    area_name2,
                    area_no2,
                    city_id,
                    city,
                    district_id,
                    district,
                    raw_che_pai_hao,
                    che_pai_fail,
                    wei_zhang_nian_yue,
                    year_no,
                    month_no,
                    source
                )
                select
                    year_month,
                    che_pai_hao,
                    raw_yi_ji,
                    raw_er_ji,
                    raw_san_ji,
                    wei_zhang_shi_jian,
                    wei_zhang_di_dian,
                    wei_zhang_xiang_qing,
                    kou_fen,
                    fa_kuan,
                    wei_zhang_wei_chu_li_shi_chang,
                    chu_li_zhuang_tai,
                    first_unit,
                    second_unit,
                    third_unit,
                    area_no,
                    area_name,
                    city_no,
                    city_name,
                    area_name2,
                    area_no2,
                    city_id,
                    city,
                    district_id,
                    district,
                    raw_che_pai_hao,
                    che_pai_fail,
                    wei_zhang_nian_yue,
                    year_no,
                    month_no,
                    source
                from
                    car.car_wei_zhang
                where
                    chu_li_zhuang_tai = '未处理'
                    and wei_zhang_wei_chu_li_shi_chang > 150
                    and year_month = {year_month}
                            """
            logger.info(f"sql: {sql}")
            curs.execute(sql)
            logger.info(f"update {curs.rowcount}")
            # 插入违章
            sql = f"""
            insert
                into
                car_theme.wz_f_violation_details
            (
                statistical_month,
                card_num,
                city,
                dpt_sec,
                grid,
                violation_time,
                violation_location,
                violation_details,
                deduction_points,
                fine,
                processing_time,
                unprocessed_duration_of_violation,
                offline_actual_processing_status
            )
            select
                year_month,
                che_pai_hao,
                first_unit,
                second_unit,
                third_unit,
                wei_zhang_shi_jian,
                wei_zhang_di_dian,
                wei_zhang_xiang_qing,
                kou_fen,
                fa_kuan,
                chu_li_shi_jian,
                wei_zhang_wei_chu_li_shi_chang,
                chu_li_zhuang_tai
            from
                car.car_wei_zhang
            where
                year_month = {year_month}
            """
            logger.info(f"sql: {sql}")
            curs.execute(sql)
            logger.info(f"update {curs.rowcount}")


data_process()
data_import()
upload_file()
data_update()