finance_data_process/house/house-abnormal-data/house_abnormal_data.py

"""不动产异常数据稽核数据处理
"""

import os  # 导入os模块，用于处理文件和目录操作
import re  # 导入re模块，用于正则表达式操作
from loguru import logger  # 导入loguru模块，用于日志记录
import pandas as pd  # 导入pandas模块，用于数据处理和分析
import psycopg  # 导入psycopg模块，用于连接PostgreSQL数据库
from datetime import datetime  # 导入datetime模块，用于日期和时间操作
from dateutil.relativedelta import relativedelta  # 导入relativedelta模块，用于日期的相对计算
import xlwings as xw  # 导入xlwings模块，用于操作Excel文件
import subprocess
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/house/abnormal-data/'
# 数据库连接信息
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}'"
# 创建一个不可见的Excel应用程序实例
app = xw.App(visible=False)
# 获取当前日期和时间
today = datetime.today()
# 计算上个月的第一天作为起始日期
start_date = today - relativedelta(months=1, day=1)
# 将起始日期格式化为YYYYMM的形式
year_month = start_date.strftime('%Y%m')
# 定义数据输入和输出目录
input_dir = 'data/'
output_dir = 'output/'
# 输出文件路径
output_path = 'output.csv'
remote_dir_path = f"{remote_dir_path}{year_month}/"


def xls_to_xlsx():
    # 如果输出目录不存在，则创建该目录
    if not os.path.exists(output_dir):
        os.makedirs(output_dir)

    # 遍历数据目录中的所有文件
    for filename in os.listdir(input_dir):
        # 只处理以.xls结尾的文件
        if filename.endswith(".xls"):
            file_path = os.path.join(input_dir, filename)  # 构造文件路径
            workbook = app.books.open(file_path)  # 打开Excel文件
            pattern = r'\D+(\d+)'  # 定义正则表达式，用于匹配非数字部分
            # 使用正则表达式替换文件名，并将扩展名改为.xlsx
            new_file_name = re.sub(pattern, year_month + '_', filename).replace(".xls", ".xlsx")
            # 保存修改后的文件到输出目录
            workbook.save(os.path.join(output_dir, new_file_name))
            logger.info(f"{filename} -> {new_file_name}")  # 记录日志
            workbook.close()  # 关闭工作簿


def data_process():
    # 初始化组织映射和三级组织列表映射
    org_map = {}
    third_org_list_map = {}

    # 连接PostgreSQL数据库
    with psycopg.connect(
            conninfo=conn_info,
            row_factory=psycopg.rows.dict_row  # 使用字典格式返回查询结果
    ) as conn:
        with conn.cursor() as curs:
            # 查询一级组织信息
            sql = """
            select * from common.organization where grade = 1
            """
            logger.info(f"sql: {sql}")  # 记录SQL语句
            curs.execute(sql)
            second_orgs = curs.fetchall()
            # 初始化三级组织列表映射
            for x in second_orgs:
                third_org_list_map[x['id']] = []
            # 查询所有组织信息
            sql = """
            select * from common.organization
            """
            logger.info(f"sql: {sql}")  # 记录SQL语句
            curs.execute(sql)
            orgs = curs.fetchall()
            # 构建组织映射和三级组织列表映射
            for x in orgs:
                if x['parent_id'] in third_org_list_map:
                    third_org_list_map[x['parent_id']].append(x)
                org_map[x['id']] = x

    files = os.listdir(output_dir)
    file_list = list(filter(lambda x: x.endswith('.xlsx'), files))  # 筛选出以.xlsx结尾的文件
    logger.info('file_list: {}', file_list)  # 记录文件列表

    # 初始化数据列表
    data_list = []

    # 遍历文件列表，读取Excel文件内容并进行处理
    for t in file_list:
        logger.info(f'{t}')  # 记录当前文件名
        ny = t.replace('.xlsx', '').split('_')[0]  # 提取年月信息
        ds = t.replace('.xlsx', '').split('_')[1]  # 提取地市信息
        tmp = pd.read_excel(output_dir + t, skiprows=8, header=None)  # 读取Excel文件内容
        if '省本部' in ds or '省公司' in ds:
            tmp = pd.read_excel(output_dir + t, skiprows=8, header=None, nrows=1)  # 特殊处理省本部文件
        tmp['年月'] = ny  # 添加年月列
        tmp['地市'] = ds  # 添加地市列
        tmp['source'] = t  # 添加来源文件名列
        data_list.append(tmp)  # 将数据添加到数据列表中

    # 合并所有数据到一个DataFrame中
    df = pd.concat(data_list)

    # 去除字符串中的多余空格
    df = df.map(lambda x: re.sub(r'\s+', '', x) if type(x) is str else x)

    # 定义DataFrame列名
    df.columns = ['third_unit', 'wei_guan_lian_tu_di_zheng', 'wei_guan_lian_fang_chan_zheng', 'wei_guan_lian_ju_zhi',
                  'jian_zhu_tu_di_ju_zhi_bu_dui_ying', 'tu_di_ju_zhi_bian_hao_bu_cun_zai',
                  'jian_zhu_ju_zhi_bian_hao_bu_cun_zai', 'jian_zhu_tu_di_biao_qian_hao_bu_cun_zai',
                  'dai_guan_lian_tu_di', 'dai_guan_lian_jian_zhu', 'ju_zhi_gte_three_tu_di', 'ju_zhi_gt_ten_tu_di',
                  'ju_zhi_gte_five_jian_zhu', 'ju_zhi_gte_ten_jian_zhu', 'tu_di_gte_seven_jian_zhu',
                  'tu_di_gte_ten_jian_zhu', 'tu_di_zheng_xia_ci_kong_bai', 'fang_chan_zheng_xia_ci_kong_bai',
                  'ju_zhi_di_duan_kong_bai', 'xian_zhi_jian_zhu_mian_ji', 'xian_zhi_tu_di_mian_ji',
                  'wu_jian_zhu_fei_xian_zhi_tu_di_mian_ji', 'tu_di_shi_yong_quan_qi_ta', 'year_month', 'second_unit',
                  'source']

    # 定义函数，根据二级单位和三级单位获取区域编号
    def get_area_no(x):
        second_unit = x['second_unit']
        third_unit = x['third_unit']
        if '长途局' in second_unit or '长途通信传输局' in second_unit or '机动局' in second_unit or '传输局' in second_unit:
            return '-11'
        if '保定' in second_unit and ('雄县' in third_unit or '容城' in third_unit or '安新' in third_unit):
            return '782'
        for second_org in second_orgs:
            area_name = second_org['name']
            area_no = second_org['id']
            if area_name in second_unit:
                return area_no
        return '-12'

    # 应用get_area_no函数，生成area_no列
    df['area_no'] = df.apply(get_area_no, axis=1)

    # 定义函数，根据区域编号获取区域名称
    def get_area_name(x):
        area_no = x['area_no']
        second_org = org_map[area_no]
        area_name = second_org['name']
        return area_name

    # 应用get_area_name函数，生成area_name列
    df['area_name'] = df.apply(get_area_name, axis=1)

    # 定义函数，根据三级单位、区域名称和区域编号获取城市编号
    def get_city_no(x):
        third_unit = x['third_unit']
        area_name = x['area_name']
        area_no = x['area_no']
        if area_name == '石家庄':
            if '矿区' in third_unit:
                return 'D0130185'
            if '井陉' in third_unit:
                return 'D0130121'
        if area_name == '秦皇岛':
            if '北戴河新区' in third_unit:
                return 'D0130185'
            if '北戴河' in third_unit:
                return 'D0130304'
        if area_name == '唐山':
            if '滦县' in third_unit:
                return 'D0130223'
            if '高新技术开发区' in third_unit:
                return 'D0130205'
        if area_name == '邢台':
            if '内丘' in third_unit:
                return 'D0130523'
            if '任泽' in third_unit:
                return 'D0130526'
        if area_name == '邯郸':
            if '峰峰' in third_unit:
                return 'D0130406'
        if area_name == '省机动局':
            if '沧州' in third_unit:
                return 'HECS180'
            if '唐山' in third_unit:
                return 'HECS181'
            if '秦皇岛' in third_unit:
                return 'HECS182'
            if '廊坊' in third_unit:
                return 'HECS183'
            if '张家口' in third_unit:
                return 'HECS184'
            if '邢台' in third_unit:
                return 'HECS185'
            if '邯郸' in third_unit:
                return 'HECS186'
            if '保定' in third_unit:
                return 'HECS187'
            if '石家庄' in third_unit:
                return 'HECS188'
            if '承德' in third_unit:
                return 'HECS189'
            if '衡水' in third_unit:
                return 'HECS720'
            if '雄安' in third_unit:
                return 'HECS728'
            return 'HECS018'
        if '雄安' == area_name:
            third_unit = third_unit.replace('雄安新区', '')
        third_org_list = third_org_list_map[area_no]
        for third_org in third_org_list:
            city_name = third_org['name']
            if city_name in third_unit:
                return third_org['id']
        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 'HE001'

    # 应用get_city_no函数，生成city_no列
    df['city_no'] = df.apply(get_city_no, axis=1)

    # 定义函数，根据城市编号获取城市名称
    def get_city_name(x):
        city_no = x['city_no']
        third_org = org_map[city_no]
        city_name = third_org['name']
        return city_name

    # 应用get_city_name函数，生成city_name列
    df['city_name'] = df.apply(get_city_name, axis=1)

    # 输出DataFrame的基本信息
    print(df.info())

    # 将处理后的数据保存为CSV文件
    df.to_csv(path_or_buf=output_path,
              index=False,
              encoding='utf-8-sig')


def data_import():
    # 定义 PowerShell 脚本的路径
    script_path = r"../../copy.ps1"
    # 目标表和文件信息
    table = "house.abnormal_data"  # 数据库目标表名
    # 表字段列名，用于指定导入数据的列顺序
    columns = "third_unit,wei_guan_lian_tu_di_zheng,wei_guan_lian_fang_chan_zheng,wei_guan_lian_ju_zhi,jian_zhu_tu_di_ju_zhi_bu_dui_ying,tu_di_ju_zhi_bian_hao_bu_cun_zai,jian_zhu_ju_zhi_bian_hao_bu_cun_zai,jian_zhu_tu_di_biao_qian_hao_bu_cun_zai,dai_guan_lian_tu_di,dai_guan_lian_jian_zhu,ju_zhi_gte_three_tu_di,ju_zhi_gt_ten_tu_di,ju_zhi_gte_five_jian_zhu,ju_zhi_gte_ten_jian_zhu,tu_di_gte_seven_jian_zhu,tu_di_gte_ten_jian_zhu,tu_di_zheng_xia_ci_kong_bai,fang_chan_zheng_xia_ci_kong_bai,ju_zhi_di_duan_kong_bai,xian_zhi_jian_zhu_mian_ji,xian_zhi_tu_di_mian_ji,wu_jian_zhu_fei_xian_zhi_tu_di_mian_ji,tu_di_shi_yong_quan_qi_ta,year_month,second_unit,source,area_no,area_name,city_no,city_name"
    # 构造执行 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():
    # 使用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:
            for filename in os.listdir(input_dir):
                local_path = os.path.join(input_dir, filename)
                remote_path = f'{remote_dir_path}{filename}'
                if os.path.isfile(local_path):
                    # 记录日志，提示即将上传的本地文件和远程目标路径
                    logger.info("upload {} to {}", local_path, remote_path)
                    # 使用SFTP的put方法将本地文件上传到远程主机
                    sftp.put(local_path, remote_path)
                    # 记录日志，提示文件已成功上传
                    logger.info("uploaded {}", local_path)


xls_to_xlsx()
data_process()
data_import()
upload_file()