finance_data_process/house/house-land/house_land.py

"""不动产土地数据处理
"""

# 导入必要的库
import re  # 用于正则表达式处理
from datetime import datetime  # 用于日期时间操作
from dateutil.relativedelta import relativedelta  # 用于相对日期计算
from loguru import logger  # 日志记录工具
import pandas as pd  # 数据处理库
import psycopg  # PostgreSQL数据库连接库
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/land/'
# 数据库连接信息
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():
    org_map = {}  # 组织机构ID到组织机构信息的映射
    third_org_list_map = {}  # 二级组织机构ID到其三级子组织列表的映射
    area_map = {}  # 区域ID到区域信息的映射
    districts_list_map = {}  # 城市ID到其区县列表的映射

    # 连接到PostgreSQL数据库
    with psycopg.connect(
            conninfo=conn_info,
            row_factory=psycopg.rows.dict_row  # 使用字典格式返回查询结果
    ) as conn:
        with conn.cursor() as curs:
            # 查询所有一级组织机构（grade=1）
            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}")
            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  # 将组织机构信息存入全局映射

            # 查询所有一级区域（area_grade=1）
            sql = """
            select * from common.area where area_grade = 1 order by area_id
            """
            logger.info(f"sql: {sql}")
            curs.execute(sql)
            cities = curs.fetchall()  # 获取所有一级区域（城市）
            for x in cities:
                districts_list_map[x['area_id']] = []  # 初始化城市的区县列表

            # 查询所有区域
            sql = """
            select * from common.area
            """
            logger.info(f"sql: {sql}")
            curs.execute(sql)
            areas = curs.fetchall()
            for x in areas:
                if x['parent_id'] in districts_list_map:
                    # 如果该区域的父级是城市，则将其加入对应城市的区县列表
                    districts_list_map[x['parent_id']].append(x)
                area_map[x['area_id']] = x  # 将区域信息存入全局映射

    # 读取Excel文件中的数据
    df = pd.read_excel(io=input_path)
    # 获取当前 DataFrame 的列名列表
    columns = df.columns.tolist()
    # 定义所需的字段列表
    required_columns = ['资产所属单位(一级)', '资产所属单位(二级)', '资产所属单位(三级)', '土地别名', '盘点情况',
                        '盘点状态', '是否修改', '是否存在待核对信息', '土地编号', '土地ID', '土地性质', '使用权类型',
                        '地类用途', '取得日期', '闲置开始时间', '局址别名', '局址ID', '标准地址', '投资主体',
                        '管理层级', '权属状态', '使用状态', '土地总面积(㎡)', '土地自用面积(㎡)', '土地闲置面积(㎡)',
                        '土地出租面积(㎡)', '土地不可使用面积(㎡)', '是否有土地证', '无土地证原因', '是否有保地风险',
                        '是否独立成宗土地', '是否空地', '是否有院落', '是否有资产卡片', '资产编号', '资产标签号',
                        '责任部门', '责任人', '经度', '纬度', '实际产权人', '特殊说明']
    # 检查是否有缺失的字段
    missing_columns = [col for col in required_columns if col not in columns]
    # 检查是否有多余的字段
    ex_columns = [col for col in columns if col not in required_columns]
    # 如果存在缺失字段，则抛出运行时错误并提示缺少哪些字段
    if missing_columns or ex_columns:
        raise RuntimeError(f"缺少以下字段: {missing_columns}；存在以下多余字段：{ex_columns}")
    # 删除字符串字段中的空白字符
    df = df.map(lambda x: re.sub(r'\s+', '', x) if type(x) is str else x)

    # 去重：根据“土地ID”列去重，保留最后一条记录
    df.drop_duplicates(subset=['土地ID'], keep='last', inplace=True)


    # 定义函数：获取二级组织机构编码
    def get_area_no(x):
        second_unit = x['资产所属单位(二级)']
        third_unit = x['资产所属单位(三级)']
        if '河北' == second_unit:
            return '-12'
        if '长途通信传输局' == 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
        raise RuntimeError(f'二级组织机构编码匹配失败: {second_unit}')


    # 应用函数，生成“二级组织机构编码”列
    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


    # 应用函数，生成“二级组织机构名称”列
    df['area_name'] = df.apply(get_area_name, axis=1)


    # 定义函数：获取三级组织机构编码
    def get_city_no(x):
        third_unit = x['资产所属单位(三级)']
        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'


    # 应用函数，生成“三级组织机构编码”列
    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


    # 应用函数，生成“三级组织机构名称”列
    df['city_name'] = df.apply(get_city_name, axis=1)


    # 定义函数：获取城市ID
    def get_city_id(x):
        address = x['标准地址']
        second_unit = x['资产所属单位(二级)']
        third_unit = x['资产所属单位(三级)']
        if '雄安' in address or ('保定' in address and ('雄县' in address or '容城' in address or '安新' in address)):
            return '133100'
        for city in cities:
            area_name = city['short_name']
            area_id = city['area_id']
            if area_name in second_unit:
                return area_id
            if area_name in third_unit:
                return area_id
            if area_name in address:
                return area_id
        return ''


    # 应用函数，生成“city_id”列
    df['city_id'] = df.apply(get_city_id, axis=1)


    # 定义函数：获取城市名称
    def get_city(x):
        city_id = x['city_id']
        area = area_map.get(city_id)
        if pd.notna(area):
            city = area['area_name']
            return city
        return ''


    # 应用函数，生成“city”列
    df['city'] = df.apply(get_city, axis=1)


    # 定义函数：获取区县ID
    def get_district_id(x):
        address = x['标准地址']
        city = x['city']
        city_id = x['city_id']
        if pd.isna(city) or pd.isna(address):
            return ''
        if city == '石家庄':
            if '矿区' in address:
                return '130107'
            if '井陉' in address:
                return '130121'
        if city == '唐山':
            if '滦县' in address:
                return '130284'
        if city == '邢台':
            if '内邱' in address:
                return '130523'
            if '任县' in address:
                return '130505'
        if city == '雄安':
            address = address.replace('雄安新区', '')
        districts = districts_list_map.get(city_id)
        if not districts:
            return ''
        for district in districts:
            district_name = district['short_name']
            if district_name in address:
                return district['area_id']
        return ''


    # 应用函数，生成“district_id”列
    df['district_id'] = df.apply(get_district_id, axis=1)


    # 定义函数：获取区县名称
    def get_district(x):
        district_id = x['district_id']
        area = area_map.get(district_id)
        if pd.notna(area):
            district = area['area_name']
            return district
        return ''


    # 应用函数，生成“district”列
    df['district'] = df.apply(get_district, axis=1)
    # 在DataFrame中插入“年月”列
    df.insert(0, 'year_month', year_month)
    df.rename(
        columns={'资产所属单位(一级)': 'first_unit', '资产所属单位(二级)': 'second_unit', '资产所属单位(三级)': 'third_unit',
                 '土地别名': 'land_name', '盘点情况': 'inventory_situation', '盘点状态': 'inventory_status',
                 '是否修改': 'modify', '是否存在待核对信息': 'to_be_verified', '土地编号': 'land_no', '土地ID': 'land_id',
                 '土地性质': 'land_ownership', '使用权类型': 'use_right_type', '地类用途': 'land_use',
                 '取得日期': 'acquisition_date', '闲置开始时间': 'idle_start_date', '局址别名': 'site_name',
                 '局址ID': 'site_id', '标准地址': 'address', '投资主体': 'investor', '管理层级': 'management_level',
                 '权属状态': 'ownership_status', '使用状态': 'usage_status', '土地总面积(㎡)': 'total_land_area',
                 '土地自用面积(㎡)': 'land_area_self_use', '土地闲置面积(㎡)': 'land_area_idle',
                 '土地出租面积(㎡)': 'land_area_rent', '土地不可使用面积(㎡)': 'land_area_unusable',
                 '是否有土地证': 'has_land_deed', '无土地证原因': 'no_land_deed_reason',
                 '是否有保地风险': 'land_preservation_risk', '是否独立成宗土地': 'independent_parcel_of_land',
                 '是否空地': 'open_space', '是否有院落': 'courtyard', '是否有资产卡片': 'has_asset_card',
                 '资产编号': 'assets_num', '资产标签号': 'assets_tag_num', '责任部门': 'responsible_department',
                 '责任人': 'person_in_charge', '经度': 'lng_jt', '纬度': 'lat_jt', '实际产权人': 'property_owner',
                 '特殊说明': 'special_specification'}, inplace=True)
    df = df[['year_month', 'first_unit', 'second_unit', 'third_unit', 'land_name', 'inventory_situation',
             'inventory_status', 'modify', 'to_be_verified', 'land_no', 'land_id', 'land_ownership',
             'use_right_type', 'land_use', 'acquisition_date', 'idle_start_date', 'site_name', 'site_id', 'address',
             'investor', 'management_level', 'ownership_status', 'usage_status', 'total_land_area',
             'land_area_self_use', 'land_area_idle', 'land_area_rent', 'land_area_unusable', 'has_land_deed',
             'no_land_deed_reason', 'land_preservation_risk', 'independent_parcel_of_land', 'open_space', 'courtyard',
             'has_asset_card', 'assets_num', 'assets_tag_num', 'responsible_department', 'person_in_charge', 'lng_jt',
             'lat_jt','property_owner', 'special_specification', 'area_no', 'area_name', 'city_no', 'city_name',
             'city_id', 'city', 'district_id', 'district']]
    # 打印DataFrame的基本信息
    df.info()

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


def data_import():
    # 定义 PowerShell 脚本的路径
    script_path = r"../../copy.ps1"
    # 目标表和文件信息
    table = "house.land_month"  # 数据库目标表名
    # 表字段列名，用于指定导入数据的列顺序
    columns = "year_month,first_unit,second_unit,third_unit,land_name,inventory_situation,inventory_status,modify,to_be_verified,land_no,land_id,land_ownership,use_right_type,land_use,acquisition_date,idle_start_date,site_name,site_id,address,investor,management_level,ownership_status,usage_status,total_land_area,land_area_self_use,land_area_idle,land_area_rent,land_area_unusable,has_land_deed,no_land_deed_reason,land_preservation_risk,independent_parcel_of_land,open_space,courtyard,has_asset_card,assets_num,assets_tag_num,responsible_department,person_in_charge,lng_jt,lat_jt,property_owner,special_specification,area_no,area_name,city_no,city_name,city_id,city,district_id,district"
    # 构造执行 PowerShell 脚本的命令
    command = f"powershell -NoProfile -NonInteractive -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"""
            update
                house.land_month a
            set
                site_num = b.site_num
            from
                house.site_month b
            where
                a.site_id = b.site_id
                and a.year_month = b.year_month
                and a.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()