doris-sync/jobs/grossprofit.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
毛利分析数据同步任务
功能：将MySQL中的fee_incomedetail数据同步到Doris DWD层宽表
"""

import datetime
from db.pool import get_mysql_conn, get_doris_conn
from logger.handler import get_logger

logger = get_logger("job.grossprofit")


def run(start_date=None, end_date=None, group_id=None):
    if start_date is None or end_date is None:
        yesterday = datetime.date.today() - datetime.timedelta(days=1)
        start_date = end_date = yesterday.strftime("%Y-%m-%d")

    logger.info("=" * 50)
    logger.info(f"[毛利分析] 开始同步: {start_date} ~ {end_date}")
    logger.info("=" * 50)

    mysql_conn = get_mysql_conn()
    doris_conn = get_doris_conn()
    mysql_cursor = mysql_conn.cursor()
    doris_cursor = doris_conn.cursor()

    if group_id:
        mysql_cursor.execute("SELECT id, name FROM org_group WHERE id = %s", (int(group_id),))
    else:
        mysql_cursor.execute("SELECT id, name FROM org_group")
    groups = mysql_cursor.fetchall()
    logger.info(f"发现 {len(groups)} 个集团")

    if not groups:
        logger.info(f"未找到集团: {group_id}")
        mysql_cursor.close()
        doris_cursor.close()
        mysql_conn.close()
        doris_conn.close()
        return 0

    total_count = 0
    sdate = start_date + " 00:00:00"
    # 结束时间用下一天0点，SQL用<，刚好覆盖整天
    edate_dt = datetime.datetime.strptime(end_date, "%Y-%m-%d") + datetime.timedelta(days=1)
    edate = edate_dt.strftime("%Y-%m-%d") + " 00:00:00"

    logger.info(f"删除 {start_date} ~ {end_date} 的旧数据...")
    if group_id:
        doris_cursor.execute(
            "DELETE FROM fact_fee_grossprofit WHERE pay_time >= %s AND pay_time < %s AND group_id = %s",
            (sdate, edate, int(group_id))
        )
    else:
        doris_cursor.execute(
            "DELETE FROM fact_fee_grossprofit WHERE pay_time >= %s AND pay_time < %s",
            (sdate, edate)
        )
    doris_conn.commit()
    logger.info(f"旧数据已删除")

    for group in groups:
        group_id = group["id"]
        group_name = group["name"]
        logger.info(f"开始同步集团: {group_name} (id={group_id})")
        count = sync_group_data(mysql_cursor, doris_cursor, doris_conn, group_id, group_name, sdate, edate)
        total_count += count
        logger.info(f"集团 {group_name} 同步完成: {count} 条")

    logger.info(f"[毛利分析] 全部同步完成: 共 {total_count} 条")

    mysql_cursor.close()
    doris_cursor.close()
    mysql_conn.close()
    doris_conn.close()


def sync_group_data(mysql_cursor, doris_cursor, doris_conn, group_id, group_name, sdate, edate):
    try:
        params = {"group_id": group_id, "sdate": sdate, "edate": edate}

        sql = """
        SELECT
            x.id, x.created, x.modified, x.patient_name, x.patient_identity,
            x.patient_mobile, x.patientId, x.org_name, x.order_no, x.fee_class,
            x.class_name, x.combo_name, x.per_id, x.name, x.unit,
            x.price, x.num, x.sum_price, x.should_amt, x.discount,
            x.discount_memo, x.coupon_name, x.flag, x.type_flag, x.order_status,
            x.created AS pay_time, x.exec_time, x.exec_status, x.cost_price, x.mcost_price,
            x.jcost_price, x.cost_flag, x.cost_flag2, x.cost_update_time, x.cost_update_time2,
            x.profit_amt, x.profit_rate, x.mprofit_amt, x.mprofit_rate,
            x.jprofit_amt, x.jprofit_rate, x.blq_amt, x.swap_card_amt, x.eye_card_amt,
            x.hx_amt, x.hx_cost_price, x.hx_sum_price, x.hx_num,
            x.pay_amt, x.pay_cost_price, x.pay_sum_price, x.pay_num,
            x.pay_profit_amt, x.pay_profit_rate, x.by_user_name, x.doctor_name,
            x.acp_doctor_name, x.acp_opt_name, x.acp_assistant_name,
            x.clinic_id, x.org_id, x.order_id,
            x.pk_code, x.uuid_no, x.birthday, x.first_accept_date,
            x.first_distance_day, x.age, x.source1, x.source2,
            x.inviter_name, x.inviter_phone, x.line_name, x.flag_memo,
            x.brands, x.factory, x.variety, x.visual, x.exclude_flag, x.pay_status,
            c.come_time, x.pay_time AS sell_time,
            CASE WHEN x.flag IN (9, 11) THEN f.pay_time ELSE NULL END AS refund_pay_time,
            CASE WHEN x.type_flag = 2 AND x.flag NOT IN (9, 11) THEN fo.pay_time ELSE NULL END AS refund_order_pay_time
        FROM fee_incomedetail x
        JOIN clinic_clinic c ON c.id = x.clinic_id
        LEFT JOIN finance_bill f ON x.type_flag = 2 AND x.flag IN (9, 11) AND f.order_no = x.order_no
        LEFT JOIN fee_order fo ON x.type_flag = 2 AND x.flag NOT IN (9, 11) AND fo.id = x.order_id
        WHERE x.created >= %(sdate)s AND x.created < %(edate)s
          AND x.group_id = %(group_id)s
          AND x.visual = 1 AND x.org_id != 16
          AND (x.flag IN (4, 5) OR (x.flag IN (1, 2, 6, 7, 8, 9, 10, 11, 12) AND x.should_amt != 0))
        ORDER BY x.id
        """

        logger.info(f"查询MySQL数据... (created: {sdate} ~ {edate})")
        mysql_cursor.execute(sql, params)
        rows = mysql_cursor.fetchall()
        logger.info(f"获取到 {len(rows)} 条数据")

        if not rows:
            logger.info("没有数据需要同步")
            return 0

        doris_insert_sql = """
        INSERT INTO fact_fee_grossprofit (
            fee_incomedetail_id, created, modified, month, org_name, group_id, group_name,
            pay_time, fee_time, clinic_id, come_time, patientId,
            patient_name, patient_mobile, patient_identity, birthday, age,
            source1, source2, inviter_name, inviter_phone,
            first_accept_date, first_distance_day, by_user_name,
            doctor_name, opt_name, assistant_name,
            pay_status, order_no, pk_code, uuid_no,
            class_name, line_name, fee_class, name, combo_name, flag_memo,
            price, num, unit, sum_price, should_amt,
            cost_amt, profit_amt, profit_rate, discount,
            discount_memo, coupon_name, blq_amt,
            hx_sum_price, hx_num, hx_amt, hx_cost_price,
            pay_sum_price, pay_num, pay_amt, pay_cost_price,
            pay_profit_amt, pay_profit_rate,
            factory_name, brands_name, variety_name,
            exec_time, exec_status_name, order_status_name,
            cost_flag, update_time, cost_flag2, update_time2,
            exec_status, flag, per_id, order_id, type_flag, order_status, sell_time, sync_date
        ) VALUES (
            %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
            %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
            %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
            %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s
        )
        """

        def format_datetime(v):
            if isinstance(v, datetime.datetime):
                return v.strftime("%Y-%m-%d %H:%M:%S") if v else None
            elif isinstance(v, datetime.date):
                return v.strftime("%Y-%m-%d") if v else None
            return v

        def format_profit_rate(v):
            if v is None or v == 0:
                return ""
            return f"{float(v) * 100:.2f}%"

        def format_discount(v):
            if v is None or v == 0:
                return ""
            return f"{float(v) * 100:.2f}%"

        batch_size = 1000
        for i in range(0, len(rows), batch_size):
            batch = rows[i:i + batch_size]
            values = []
            for row in batch:
                r = row  # DictCursor 已经返回字典
                created = format_datetime(r.get("created"))
                month = created[:7] if created else ""
                pay_time = format_datetime(r.get("pay_time")) # 即 x.created
                sell_time = format_datetime(r.get("sell_time")) # 即 x.pay_time

                # fee_time 计算逻辑完全对齐 API
                if r.get("name") != "系统逾期项目":
                    fee_time = created
                else:
                    fee_time = sell_time

                type_flag = r.get("type_flag", 1)
                if type_flag == 2:
                    if r.get("flag") in [9, 11]:
                        fee_time = format_datetime(r.get("refund_pay_time"))
                    else:
                        if r.get("name") != "系统逾期项目":
                            fee_time = format_datetime(r.get("refund_order_pay_time"))
                        else:
                            fee_time = sell_time

                exec_status = r.get("exec_status", 0)
                exec_status_name = "已交付" if exec_status == 1 else "未交付"
                order_status = r.get("order_status", 1)
                order_status_name = "正常"
                # 完全对齐 API 中的 Bug: 若 order_status == 2，把 exec_status_name 设为 "作废"
                if order_status == 2:
                    exec_status_name = "作废"

                pay_status = "已退费" if type_flag == 2 else "已收费"

                cost_amt = r.get("mcost_price", 0) or 0
                profit_amt = r.get("mprofit_amt", 0) or 0
                
                sum_price = float(r.get("sum_price", 0) or 0)
                should_amt = float(r.get("should_amt", 0) or 0)
                num = r.get("num", 0) or 0
                
                if type_flag == 2:
                    cost_amt = -(cost_amt)
                    profit_amt = -(profit_amt)
                    sum_price = -sum_price
                    should_amt = -should_amt
                    num = -num

                profit_rate = format_profit_rate(r.get("mprofit_rate"))
                discount = format_discount(r.get("discount"))

                come_time = format_datetime(r.get("come_time"))
                exec_time = format_datetime(r.get("exec_time"))
                update_time = format_datetime(r.get("cost_update_time"))
                update_time2 = format_datetime(r.get("cost_update_time2"))
                sync_date = pay_time.split(" ")[0] if pay_time else None

                blq_amt = float(r.get("blq_amt", 0) or 0) + float(r.get("swap_card_amt", 0) or 0) + float(r.get("eye_card_amt", 0) or 0)

                hx_sum_price = float(r.get("hx_sum_price", 0) or 0)
                hx_num = r.get("hx_num", 0) or 0
                hx_amt = float(r.get("hx_amt", 0) or 0)
                hx_cost_price = float(r.get("hx_cost_price", 0) or 0)
                pay_sum_price = float(r.get("pay_sum_price", 0) or 0)
                pay_num = r.get("pay_num", 0) or 0
                pay_amt = float(r.get("pay_amt", 0) or 0)
                pay_cost_price = float(r.get("pay_cost_price", 0) or 0)
                pay_profit_amt = float(r.get("pay_profit_amt", 0) or 0)
                pay_profit_rate = r.get("pay_profit_rate", 0) or 0

                if type_flag == 2:
                    hx_sum_price = -hx_sum_price
                    hx_num = -hx_num
                    hx_amt = -hx_amt
                    hx_cost_price = -hx_cost_price
                    pay_sum_price = -pay_sum_price
                    pay_num = -pay_num
                    pay_amt = -pay_amt
                    pay_cost_price = -pay_cost_price
                    pay_profit_amt = -pay_profit_amt
                    pay_profit_rate = -pay_profit_rate

                values.append((
                    r.get("id"), created, format_datetime(r.get("modified")), month,
                    r.get("org_name"), group_id, group_name,
                    pay_time, fee_time, r.get("clinic_id"), come_time, r.get("patientId"),
                    r.get("patient_name"), r.get("patient_mobile"), r.get("patient_identity"),
                    format_datetime(r.get("birthday")), r.get("age", 0),
                    r.get("source1"), r.get("source2"),
                    r.get("inviter_name"), r.get("inviter_phone"),
                    format_datetime(r.get("first_accept_date")), r.get("first_distance_day", 0),
                    r.get("by_user_name"),
                    r.get("doctor_name"), r.get("acp_opt_name"), r.get("acp_assistant_name"),
                    pay_status, r.get("order_no"), r.get("pk_code"), r.get("uuid_no"),
                    r.get("class_name"), r.get("line_name"), r.get("fee_class"),
                    r.get("name"), r.get("combo_name"), r.get("flag_memo"),
                    r.get("price"), num, r.get("unit"), sum_price, should_amt,
                    cost_amt, profit_amt, profit_rate, discount,
                    r.get("discount_memo"), r.get("coupon_name"), blq_amt,
                    hx_sum_price, hx_num, hx_amt, hx_cost_price,
                    pay_sum_price, pay_num, pay_amt, pay_cost_price,
                    pay_profit_amt, pay_profit_rate,
                    r.get("factory"), r.get("brands"), r.get("variety"),
                    exec_time, exec_status_name, order_status_name,
                    r.get("cost_flag", 0), update_time, r.get("cost_flag2", 0), update_time2,
                    exec_status, r.get("flag"), r.get("per_id"), r.get("order_id"),
                    type_flag, order_status, sell_time, sync_date
                ))

            doris_cursor.executemany(doris_insert_sql, values)
            doris_conn.commit()
            logger.info(f"已插入 {i + len(batch)} / {len(rows)} 条")

        logger.info(f"[毛利分析] 同步完成: {len(rows)} 条")
        return len(rows)

    except Exception as e:
        logger.error(f"同步失败: {e}")
        doris_conn.rollback()
        raise


if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--start", type=str, help="开始日期 YYYY-MM-DD")
    parser.add_argument("--end", type=str, help="结束日期 YYYY-MM-DD")
    parser.add_argument("--group-id", type=str, help="集团ID，不传则同步全部集团")
    args = parser.parse_args()
    run(args.start, args.end, args.group_id)