yuxin_pms/etl_utility.py

"""
寓信水电费收入ETL
水电费收入报表
"""

import yaml
import pandas as pd
from sqlalchemy import create_engine, text
from sqlalchemy.types import NVARCHAR, Date, DECIMAL, INTEGER
from datetime import datetime, timedelta
from tqdm import tqdm
import json
from dateutil.relativedelta import relativedelta
import calendar
import taosrest as taos
from taosrest import connect, TaosRestConnection, TaosRestCursor
from urllib.parse import quote_plus
from utils import load_config, truncate_target_db


debug = False
if debug:
    # debug_condition = " and crc.id='1634463949758156801'"  # debug charge amount
    # debug_condition = " and crc.id='1635124251453575170'"  # debug renew contract
    # debug_condition = " and crc.id='1649702247584620545'"  # debug 期末读数
    debug_condition = " and crc.id='1674347182259716097'"  # debug 退租日期
else:
    debug_condition = ''



def query_total(conn) -> int:
    """ Queries the total number of records to be processed
        合同状态不为无效4，签约类型不为续签2的合同
    """
    query = """
        select count(1) from yuxin_contract.cont_renter_contract crc 
        where crc.is_delete=0 and crc.contract_status<>4 and crc.sign_type<>2 {debug_condition}
        """.format(debug_condition=debug_condition)
    return conn.execute(text(query)).scalar()


def extract(conn, td_cursor, batch_size, i) -> pd.DataFrame:
    """ This API extracts data from
    """
    query = """
        select hhb.dept_id '房源所属门店ID', house_sd.name '房源所属门店名称', hhr.address '房源地址', 
            crc_sd.id '合同所属门店/部门ID', crc_sd.name '合同所属门店/部门',
            crc.id '合同ID', crc_ei.name '合同维护人', cri.name '租客姓名', crc.terminate_type '退租类型', 
            crc.begin_time, crc.end_time, crc.quite_date, crc.house_id
        from yuxin_contract.cont_renter_contract crc 
        left join yuxin_house.hse_house_room hhr on hhr.is_delete=0 and hhr.id=crc.house_id
        left join yuxin_house.hse_house_base hhb on hhb.is_delete=0 and hhb.id=crc.house_id
        left join yuxin_setting.setting_department house_sd on house_sd.id=hhb.dept_id and house_sd.is_delete=0
        left join yuxin_setting.setting_employee_dept crc_ed on crc_ed.emp_id=crc.maintainer_id and crc_ed.is_delete=0
        left join yuxin_setting.setting_department crc_sd on crc_sd.id=crc_ed.dept_id and crc_sd.is_delete=0
        left join yuxin_setting.setting_employee_info crc_ei on crc_ei.is_delete=0 and crc_ei.id=crc.maintainer_id
        left join yuxin_contract.cont_renter_info cri on cri.is_delete=0 and cri.customer_type=1 and cri.contract_id=crc.id
        where crc.is_delete=0 and crc.contract_status<>4 and crc.sign_type<>2 {debug_condition}
        limit {batch_size} offset {offset}
        """.format(batch_size=batch_size, offset=i, debug_condition=debug_condition)
    source_data = pd.read_sql(query, con=conn)
    return source_data


def find_renewal_contract(conn, contract_id):
    query = """
        select crc1.id, crc1.begin_time, crc1.end_time, crc1.quite_date, 
               crc2.id, crc2.begin_time, crc2.end_time, crc2.quite_date, 
               crc3.id, crc3.begin_time, crc3.end_time, crc3.quite_date
        from yuxin_contract.cont_renter_contract crc1
        left join yuxin_contract.cont_renter_contract crc2 on crc2.contract_pid=crc1.id and crc2.is_delete=0 and crc2.contract_status<>4 and crc2.sign_type=2
        left join yuxin_contract.cont_renter_contract crc3 on crc3.contract_pid=crc2.id and crc3.is_delete=0 and crc3.contract_status<>4 and crc3.sign_type=2
        where crc1.is_delete=0 and crc1.contract_status<>4 and crc1.sign_type=2
            and crc1.contract_pid='{contract_id}'
        """.format(contract_id=contract_id)
    
    result = conn.execute(text(query)).fetchone()
    if result is None:
        return None

    if result[11] is not None:
        return result[11]
    elif result[10] is not None:
        return result[10]
    elif result[7] is not None:
        return result[7]
    elif result[6] is not None:
        return result[6]
    elif result[3] is not None:
        return result[3]
    elif result[2] is not None:
        return result[2]
    return None


def query_charge_data(conn, contract_id, room_id):
    query = """
        select DATE_FORMAT(charge_create_time,'%%Y-%%m-%%d') as 'time', sum(total_price) as 'charge_amount', sum(clear_amount) as 'clear_amount', device_id
        from yuxin_finance.fin_finance_charge_record
        where status=1 and charge_status=2 and contract_id='{contract_id}' and room_id='{room_id}'
        group by DATE_FORMAT(charge_create_time,'%%Y-%%m-%%d'), device_id
        """.format(contract_id=contract_id, room_id=room_id)
    data = pd.read_sql(query, con=conn)
    return data


def query_meter_read_record(conn, td_cursor, contract_id, room_id, start_date, end_date):

    end_date_obj = end_date if isinstance(end_date, datetime) else datetime.strptime(end_date, '%Y-%m-%d')
    end_date_obj = end_date_obj + timedelta(days=1)
    start_date_obj = start_date if isinstance(start_date, datetime) else datetime.strptime(start_date, '%Y-%m-%d')
    start_date_obj = start_date_obj - timedelta(days=1)

    # 获取meter信息
    # 忽略合同ID，续约情况下合同ID会有多个
    query = """
        SELECT device_type, first(ts), first(amount), last(amount), sum(cost), last(`balance`), device_id
        FROM device.meter_read_record 
        WHERE room_id='{room_id}' and ts >= '{start_date}' AND ts < '{end_date}' and pay_method=2
        PARTITION BY device_type, device_id
        INTERVAL(1d)
        """.format(contract_id=contract_id, room_id=room_id, 
                   start_date=start_date_obj.strftime('%Y-%m-%d'), end_date=end_date)
    td_cursor.execute(query)

    # get total rows
    if td_cursor.rowcount == 0:
        return None

    # get rows
    rows = td_cursor.fetchall()

    # 获取充值信息
    charge_data = query_charge_data(conn, contract_id, room_id)

    data = {}
    for idx, row in enumerate(rows):
        if idx == 0:
            prev_row = None
        else:
            prev_row = rows[idx - 1]
        day = row[1].strftime('%Y-%m-%d')           # ts
        device_type = row[0]                        # 设备类型
        begin_amount = prev_row[3] if prev_row else row[2]   # 期初读数, 取上个时间窗口的最后一个读数，否则取当前时间窗口的最早一个读数
        end_amount = row[3]                         # 期末读数
        amount = end_amount - begin_amount          # 使用量
        cost = row[4]                               # 使用金额
        balance = row[5]                            # 期末余额
        device_id = row[6]

        charge_row = charge_data.loc[(charge_data['time'] == day) & (charge_data['device_id'] == device_id)]
        if charge_row.empty:
            charge_amount = 0
            clear_amount = 0
        else:
            charge_row = charge_row.iloc[0]
            charge_amount = charge_row['charge_amount']
            clear_amount = charge_row['clear_amount']

        # device type
        # WATER_MATER(1, "冷水表",DictConstant.FeeItemEnum.WATER_FEE),
        # RWATER_MATER(2, "中水表",DictConstant.FeeItemEnum.RWATER_FEE),
        # HWATER_MATER(3, "热水表",DictConstant.FeeItemEnum.HWATER_FEE),
        # ELE_MATER(4, "电表",DictConstant.FeeItemEnum.ELECTRIC_FEE),
        if device_type == 1:
            device_type = '冷水费'
        elif device_type == 2:
            device_type = '中水费'
        elif device_type == 3:
            device_type = '热水费'
        elif device_type == 4:
            device_type = '电费'
        else:
            device_type = '?'
        # print('device_type', device_type)
        # print('begin_amount', begin_amount)
        # print('end_amount', end_amount)
        # print('amount', amount)
        # print('balance', balance)
        # print('cost', cost)

        # ensure key exists
        if device_type not in data:
            data[device_type] = {}

        data[device_type][day] = {
            'device_type': device_type,
            'begin_amount': begin_amount,
            'end_amount': end_amount,
            'amount': amount,
            'balance': balance,
            'cost': cost,
            'charge_amount': charge_amount,
            'clear_amount': clear_amount,
            'device_id': device_id
        }

    return data


def transform(conn, td_cursor, data) -> pd.DataFrame:
    # 字段命名	        功能说明			                举例	
    # 房源所属门店	     显示当前房源维护人所属的门店			如木繁公寓南山店、寓信公寓	house_base里的dept_id
    # 房源地址	        调取合同关联房源的完整地址			   深圳市福田区香蜜湖街道乐山集团1栋1单元-203	
    # 合同所属门店/部门	  显示当前合同维护人所属的门店或部门	  如木繁公寓南山店、寓信公寓	
    # 合同ID	        系统给合同分配的唯一ID			      2467347234871234772	
    # 合同维护人	     显示合同和子合同管理的合同的合同维护人	  刘飞	
    # 租客姓名	        显示合同和子合同关联的租客信息			租客合同显示租客；包租子合同显示承租人	
    # 合同开始日期	    显示合同和子合同关联的合约开始日期		  -	
    # 合同结束日期	    显示合同和子合同关联的合约结束日期；	  -	
    # 退租类型	        若已退租，则显示退租类型			    违约退；正常退，若未退租则显示“-”	
    # 退租日期	        若已退租，则显示退租日期			    -	
    # 费用类型	        显示所选能耗费账单费用类型			    电费、热水费、冷水费、中水费	

    # target columns
    columns = ['theday', 'house_dept_id', 'house_dept_name', 'address', 'contract_dept_id', 'contract_dept_name',
               'contract_id', 'contract_maintainer', 'renter_name', 'contract_terminate_type',
               'begin_time', 'end_time', 'quit_time', 'kind', 'begin_amount', 'end_amount',
               'amount', 'balance', 'cost', 'charge_amount', 'clear_amount']
    # target data
    target_data = []

    for index, row in data.iterrows():
        # 查找续租合同
        renewed_end_time = find_renewal_contract(conn, row['合同ID'])

        begin_time = row['begin_time']
        if renewed_end_time is None:
            end_time = row['quite_date'] if row['quite_date'] is not None and not pd.isnull(
                row['quite_date']) else row['end_time']
        else:
            end_time = renewed_end_time
        
        meter_data = query_meter_read_record(conn, td_cursor,
                                             row['合同ID'], row['house_id'], begin_time, end_time)

        if meter_data is None:
            continue

        the_time = begin_time
        while the_time < end_time:
            # 查询电表读数

            for device_type in meter_data.keys():
                key = the_time.strftime('%Y-%m-%d')
                if key not in meter_data[device_type]:
                    continue

                meter_record = meter_data[device_type][key]
                if meter_record is None:
                    continue

                target_data.append({
                    'theday': the_time,
                    'house_dept_id': row['房源所属门店ID'],
                    'house_dept_name': row['房源所属门店名称'],
                    'address': row['房源地址'],
                    'contract_dept_id': row['合同所属门店/部门ID'],
                    'contract_dept_name': row['合同所属门店/部门'],
                    'contract_id': row['合同ID'],
                    'contract_maintainer': row['合同维护人'],
                    'renter_name': row['租客姓名'],
                    'contract_terminate_type': row['退租类型'],
                    'begin_time': begin_time,
                    'end_time': end_time,
                    'quit_time': row['quite_date'],
                    'kind': meter_record['device_type'],
                    'begin_amount': meter_record['begin_amount'],
                    'end_amount': meter_record['end_amount'],
                    'amount': meter_record['amount'],
                    'balance': meter_record['balance'],
                    'cost': meter_record['cost'],
                    'charge_amount': meter_record['charge_amount'],
                    'clear_amount': meter_record['clear_amount']
                })

            the_time = the_time + relativedelta(days=1)

        pass

    df = pd.DataFrame(columns=columns, data=target_data)
    return df


def load(conn, df: pd.DataFrame, target_db) -> None:
    """ Loads data into a MySQL database
    """

    # Define the column types for the table
    dtype = {
        'theday': Date,
        'house_dept_id': NVARCHAR(32),
        'house_dept_name': NVARCHAR(32),
        'address': NVARCHAR(255),
        'contract_dept_id': NVARCHAR(32),
        'contract_dept_name': NVARCHAR(32),
        'contract_id': NVARCHAR(36),
        'contract_maintainer': NVARCHAR(32),
        'renter_name': NVARCHAR(50),
        'contract_terminate_type': INTEGER(),
        'begin_time': Date,
        'end_time': Date,
        'quit_time': Date,
        'kind': NVARCHAR(32),
        'begin_amount': DECIMAL(10, 2),
        'end_amount': DECIMAL(10, 2),
        'amount': DECIMAL(10, 2),
        'balance': DECIMAL(10, 2),
        'cost': DECIMAL(10, 2),
        'charge_amount': DECIMAL(10, 2),
        'clear_amount': DECIMAL(10, 2)
    }
    # create target table with df.dtypes
    df.to_sql(target_db, con=conn, if_exists='append',
              index=False, dtype=dtype)

    pass


def etl():
    config = load_config('production')

    target_db = 'bi_utility_'

    # connect TDEngine
    td_conn = taos.connect(url=config['tdengine']['url'],
                        user=config['tdengine']['username'],
                        password=config['tdengine']['password'],
                        timeout=30)
    td_cursor = td_conn.cursor()


    connection_string = 'mysql+pymysql://{username}:{pwd}@{host}:{port}/{db}?charset=utf8'\
        .format(username=config['mysql']['username'],
                pwd=quote_plus(config['mysql']['password']),
                host=config['mysql']['host'],
                port=config['mysql']['port'],
                db=config['mysql']['db'])

    engine = create_engine(connection_string)

    with engine.begin() as conn:
        # Get the total number of rows
        total = query_total(conn)
        # Define the batch size
        batch_size = 100

        truncate_target_db(conn, target_db)

        if debug:
            total = 1000
            batch_size = 100
        print('total', total)

        # Write the data to the table in batches
        for i in tqdm(range(0, total, batch_size)):
            data = extract(conn, td_cursor, batch_size, i)
            data = transform(conn, td_cursor, data)

            # if debug:
            #     print(data.head())
            # else:
            load(conn, data, target_db)

    td_conn.close()
    pass


etl()