doris-sync/jobs/consumption.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
患者消费统计宽表同步任务
功能：将MySQL中的患者消费数据同步到Doris宽表
"""

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

logger = get_logger("job.consumption")


def run(start_date=None, end_date=None, group_id=1):
    """执行任务"""
    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}, group_id={group_id}")
    logger.info("=" * 50)

    conn = get_mysql_conn()
    cursor = conn.cursor()

    # 1. 查询主要的消费数据
    params = {
        "group_id": int(group_id),
        "fsdate": start_date + " 00:00:00",
        "fedate": end_date + " 23:59:59",
    }

    sql = """
        SELECT
            x.clinic_id,
            x.patientId,
            x.patient_name,
            x.inviter_name,
            x.sex,
            x.patient_mobile as mobile,
            x.patient_birthday as birthday,
            x.patient_identity as identity,
            x.come_number,
            x.apt_doctor_name,
            x.apt_opt_name as apt_optometrist_name,
            x.act_doctor_name,
            x.act_assistant_name,
            x.act_opt_name as act_optometrist_name,
            x.matter,
            x.type,
            x.come_time,
            x.org_name,
            x.source1,
            x.source2,
            x.manager_name,
            x.first_opt_name,
            x.first_accept_date,
            x.created as pay_time,
            x.type_flag,
            x.fee_class,
            x.by_user_name,
            x.coupon_name,
            x.sign_wxapp,
            x.add_manager,
            x.not_transaction_reason,
            c.org_id,
            c.customer_manager_staff_id,
            c.customer_optometrist_staff_id,
            SUM(x.pay_amt) as total_amt
        FROM fee_dealdetail x
        JOIN clinic_clinic c ON c.id = x.clinic_id
        WHERE x.group_id = %(group_id)s
            AND x.org_id != 16
            AND x.created >= %(fsdate)s
            AND x.created <= %(fedate)s
        GROUP BY x.clinic_id, x.patientId, x.patient_name, x.inviter_name, x.sex,
            x.patient_mobile, x.patient_birthday, x.patient_identity, x.come_number,
            x.apt_doctor_name, x.apt_opt_name, x.act_doctor_name, x.act_assistant_name,
            x.act_opt_name, x.matter, x.type, x.come_time, x.org_name, x.source1, x.source2,
            x.manager_name, x.first_opt_name, x.first_accept_date, x.created,
            x.type_flag, x.fee_class, x.by_user_name, x.coupon_name, x.sign_wxapp,
            x.add_manager, x.not_transaction_reason, c.org_id, c.customer_manager_staff_id, c.customer_optometrist_staff_id
    """

    logger.info("正在查询主数据...")
    cursor.execute(sql, params)
    data_list = cursor.fetchall()  # DictCursor 直接返回字典列表
    logger.info(f"查询到 {len(data_list)} 条记录")

    if not data_list:
        cursor.close()
        conn.close()
        logger.info("[患者消费] 无数据")
        return 0

    # 2. 获取 clinic_id 列表
    clinic_ids = [data["clinic_id"] for data in data_list]

    # 3. 查询诊所和分组信息
    org_ids = list(set([d.get("org_id") for d in data_list if d.get("org_id")]))
    org_info_dict = {}
    if org_ids:
        sql = """
            SELECT o.id as org_id, o.name as org_name, o.group_id, g.name as group_name
            FROM org_org o
            LEFT JOIN org_group g ON o.group_id = g.id
            WHERE o.id IN %(ids)s
        """
        cursor.execute(sql, {"ids": org_ids})
        org_list = cursor.fetchall()
        org_info_dict = {o["org_id"]: o for o in org_list}

    # 4. 查询员工信息
    staff_ids = []
    for data in data_list:
        if data.get("customer_manager_staff_id"):
            staff_ids.append(data["customer_manager_staff_id"])
        if data.get("customer_optometrist_staff_id"):
            staff_ids.append(data["customer_optometrist_staff_id"])

    staff_ids = list(set(staff_ids))
    staff_dict = {}
    if staff_ids:
        sql = """
            SELECT s.id, u.name, s.qywx_userid, u.mobile
            FROM staff_staff s
            JOIN users_userprofile u ON s.user_id = u.id
            WHERE s.id IN %(ids)s
        """
        cursor.execute(sql, {"ids": staff_ids})
        staff_list = cursor.fetchall()
        staff_dict = {s["id"]: s for s in staff_list}

    # 5. 查询推广信息
    clinic_info_dict = {}
    if clinic_ids:
        sql = """
            SELECT cc.collect_date, cc.activity_location,
                   u.name as extension_user_name, cc.clinic_id
            FROM returnvisit_customerconsultationrecord cc
            LEFT JOIN users_userprofile u ON cc.extension_user_id = u.id
            WHERE cc.clinic_id IN %(ids)s
        """
        cursor.execute(sql, {"ids": clinic_ids})
        clinic_info_list = cursor.fetchall()
        for c in clinic_info_list:
            clinic_info_dict[c["clinic_id"]] = c

    # 5.1 查询患者检查信息
    patient_ids = list(set([d.get("patientId") for d in data_list if d.get("patientId")]))
    patient_exam_dict = {}
    if patient_ids:
        sql = """
            SELECT patientId, r_sph, l_sph, r_nakeye, l_nakeye, al_od, al_os
            FROM patient_custome
            WHERE patientId IN %(ids)s
        """
        cursor.execute(sql, {"ids": patient_ids})
        exam_list = cursor.fetchall()
        patient_exam_dict = {e["patientId"]: e for e in exam_list}

    cursor.close()
    conn.close()

    # 6. 字段列表
    fields = [
        "clinic_id", "pay_date", "patientId", "patient_name", "mobile", "sex", "birthday",
        "identity", "come_number", "inviter_name", "come_time", "come_date", "type", "matter",
        "org_id", "org_name", "group_id", "group_name", "apt_doctor_id", "apt_doctor_name",
        "apt_optometrist_id", "apt_optometrist_name", "act_doctor_id", "act_doctor_name",
        "act_optometrist_id", "act_optometrist_name", "act_assistant_name",
        "source1", "source2", "manager_id", "manager_name", "manager_qywx_id", "manager_mobile",
        "optometrist_id", "optometrist_name", "optometrist_qywx_id", "optometrist_mobile",
        "first_opt_name", "first_accept_date", 
        "r_sph", "l_sph", "r_nakeye", "l_nakeye", "al_od", "al_os", "not_transaction_reason",
        "total_amt", "pay_time", "by_user_id",
        "by_user_name", "coupon_name", "add_manager", "sign_wxapp", "collect_date",
        "activity_location", "extension_user_id", "extension_user_name", "create_time",
    ]

    # 7. 处理数据
    result_dict = {}
    for data in data_list:
        clinic_id = data["clinic_id"]

        # 处理金额（根据type_flag正负）
        total_amt = data.get("total_amt", 0)
        try:
            total_amt = float(total_amt) if total_amt is not None else 0.0
        except (TypeError, ValueError):
            total_amt = 0.0

        if str(data.get("type_flag")) != "1":
            total_amt = -total_amt if total_amt else 0

        # 处理日期
        pay_time = data.get("pay_time")
        if pay_time:
            if isinstance(pay_time, str):
                pay_date = pay_time.split(" ")[0]
                pay_time_str = pay_time if " " in pay_time else f"{pay_time} 00:00:00"
            else:
                pay_date = pay_time.strftime("%Y-%m-%d")
                pay_time_str = pay_time.strftime("%Y-%m-%d %H:%M:%S")
        else:
            pay_date = None
            pay_time_str = None
            
        if not pay_date:
            continue
            
        key = f"{clinic_id}_{pay_date}"
        
        if key in result_dict:
            result_dict[key]["total_amt"] += total_amt
            continue

        come_time = data.get("come_time")
        if come_time:
            if isinstance(come_time, str):
                come_date = come_time.split(" ")[0]
                come_time_str = come_time if " " in come_time else f"{come_time} 00:00:00"
            else:
                come_date = come_time.strftime("%Y-%m-%d")
                come_time_str = come_time.strftime("%Y-%m-%d %H:%M:%S")
        else:
            come_date = None
            come_time_str = None

        # 获取诊所信息
        org_id = data.get("org_id")
        org_info = org_info_dict.get(org_id, {})

        # 获取员工信息
        manager_staff = staff_dict.get(data.get("customer_manager_staff_id"), {})
        optometrist_staff = staff_dict.get(data.get("customer_optometrist_staff_id"), {})

        # 获取推广信息
        clinic_info = clinic_info_dict.get(clinic_id, {})

        # 获取患者检查信息
        patient_exam = patient_exam_dict.get(data.get("patientId"), {})

        # 处理性别
        sex_val = data.get("sex")
        if sex_val == "男":
            sex_val = 1
        elif sex_val == "女":
            sex_val = 2

        # 处理就诊类型
        type_val = data.get("type")
        if type_val == "初诊":
            type_val = 1
        elif type_val == "复诊":
            type_val = 2
        elif type_val == "预约":
            type_val = 3
        elif type_val == "复查":
            type_val = 4

        # 确保整数字段类型正确
        def safe_int(v):
            if v is None or v == "":
                return None
            try:
                return int(v)
            except (TypeError, ValueError):
                return None

        # 确保字符串字段去除两边空格
        def safe_str(v):
            if v is None:
                return ""
            return str(v).strip()

        record_dict = {
            "clinic_id": safe_int(clinic_id),
            "pay_date": pay_date,
            "patientId": safe_str(data.get("patientId")),
            "patient_name": safe_str(data.get("patient_name")),
            "mobile": safe_str(data.get("mobile")),
            "sex": safe_int(sex_val) if sex_val else None,
            "birthday": data.get("birthday"),
            "identity": safe_str(data.get("identity")),
            "come_number": safe_int(data.get("come_number")),
            "inviter_name": safe_str(data.get("inviter_name")),
            "come_time": come_time_str,
            "come_date": come_date,
            "type": safe_int(type_val) if type_val else None,
            "matter": safe_str(data.get("matter")),
            "org_id": safe_int(org_id),
            "org_name": safe_str(org_info.get("org_name")),
            "group_id": safe_int(org_info.get("group_id")),
            "group_name": safe_str(org_info.get("group_name")),
            "apt_doctor_id": None,
            "apt_doctor_name": safe_str(data.get("apt_doctor_name")),
            "apt_optometrist_id": None,
            "apt_optometrist_name": safe_str(data.get("apt_optometrist_name")),
            "act_doctor_id": None,
            "act_doctor_name": safe_str(data.get("act_doctor_name")),
            "act_optometrist_id": None,
            "act_optometrist_name": safe_str(data.get("act_optometrist_name")),
            "act_assistant_name": safe_str(data.get("act_assistant_name")),
            "source1": safe_str(data.get("source1")),
            "source2": safe_str(data.get("source2")),
            "manager_id": safe_int(data.get("customer_manager_staff_id")),
            "manager_name": safe_str(manager_staff.get("name")),
            "manager_qywx_id": safe_str(manager_staff.get("qywx_userid")),
            "manager_mobile": safe_str(manager_staff.get("mobile")),
            "optometrist_id": safe_int(data.get("customer_optometrist_staff_id")),
            "optometrist_name": safe_str(optometrist_staff.get("name")),
            "optometrist_qywx_id": safe_str(optometrist_staff.get("qywx_userid")),
            "optometrist_mobile": safe_str(optometrist_staff.get("mobile")),
            "first_opt_name": safe_str(data.get("first_opt_name")),
            "first_accept_date": data.get("first_accept_date"),
            "r_sph": safe_str(patient_exam.get("r_sph")),
            "l_sph": safe_str(patient_exam.get("l_sph")),
            "r_nakeye": safe_str(patient_exam.get("r_nakeye")),
            "l_nakeye": safe_str(patient_exam.get("l_nakeye")),
            "al_od": safe_str(patient_exam.get("al_od")),
            "al_os": safe_str(patient_exam.get("al_os")),
            "not_transaction_reason": safe_str(data.get("not_transaction_reason")),
            "total_amt": total_amt,
            "pay_time": pay_time_str,
            "by_user_id": None,
            "by_user_name": safe_str(data.get("by_user_name")),
            "coupon_name": safe_str(data.get("coupon_name")),
            "add_manager": "已添加" if data.get("add_manager") == 1 else "未添加",
            "sign_wxapp": "已注册" if data.get("sign_wxapp") == 1 else "未注册",
            "collect_date": clinic_info.get("collect_date"),
            "activity_location": safe_str(clinic_info.get("activity_location")),
            "extension_user_id": None,
            "extension_user_name": safe_str(clinic_info.get("extension_user_name")),
            "create_time": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        }
        result_dict[key] = record_dict

    result = []
    for key, record_dict in result_dict.items():
        record = (
            record_dict["clinic_id"], record_dict["pay_date"], record_dict["patientId"],
            record_dict["patient_name"], record_dict["mobile"], record_dict["sex"],
            record_dict["birthday"], record_dict["identity"], record_dict["come_number"],
            record_dict["inviter_name"], record_dict["come_time"], record_dict["come_date"],
            record_dict["type"], record_dict["matter"], record_dict["org_id"],
            record_dict["org_name"], record_dict["group_id"], record_dict["group_name"],
            record_dict["apt_doctor_id"], record_dict["apt_doctor_name"],
            record_dict["apt_optometrist_id"], record_dict["apt_optometrist_name"],
            record_dict["act_doctor_id"], record_dict["act_doctor_name"],
            record_dict["act_optometrist_id"], record_dict["act_optometrist_name"],
            record_dict["act_assistant_name"], record_dict["source1"], record_dict["source2"],
            record_dict["manager_id"], record_dict["manager_name"],
            record_dict["manager_qywx_id"], record_dict["manager_mobile"],
            record_dict["optometrist_id"], record_dict["optometrist_name"],
            record_dict["optometrist_qywx_id"], record_dict["optometrist_mobile"],
            record_dict["first_opt_name"], record_dict["first_accept_date"],
            record_dict["r_sph"], record_dict["l_sph"], record_dict["r_nakeye"],
            record_dict["l_nakeye"], record_dict["al_od"], record_dict["al_os"],
            record_dict["not_transaction_reason"],
            float(record_dict["total_amt"]), record_dict["pay_time"],
            record_dict["by_user_id"], record_dict["by_user_name"], record_dict["coupon_name"],
            record_dict["add_manager"], record_dict["sign_wxapp"], record_dict["collect_date"],
            record_dict["activity_location"], record_dict["extension_user_id"],
            record_dict["extension_user_name"], record_dict["create_time"]
        )
        result.append(record)

    # 8. 写入Doris
    logger.info(f"数据处理完成，共 {len(result)} 条记录")

    conn = get_doris_conn()
    cursor = conn.cursor()

    # 8.1 先删除该时间段的数据（确保幂等性）
    logger.info(f"清理该时间段的历史数据: {start_date} ~ {end_date}")
    delete_sql = """
        DELETE FROM fact_patientconsumption
        WHERE pay_date >= %s AND pay_date <= %s
            AND group_id = %s
    """
    cursor.execute(delete_sql, (start_date, end_date, int(group_id)))
    deleted_count = cursor.rowcount
    logger.info(f"已删除 {deleted_count} 条历史数据")
    conn.commit()

    # 8.2 插入新数据
    sql = f"INSERT INTO fact_patientconsumption ({', '.join(fields)}) VALUES ({', '.join(['%s'] * len(fields))})"

    total = len(result)
    batch_size = 100
    success_count = 0

    for i in range(0, total, batch_size):
        batch = result[i:i + batch_size]
        try:
            cursor.executemany(sql, batch)
            conn.commit()
            success_count += len(batch)
            logger.info(f"[患者消费] 已写入 {min(i + batch_size, total)}/{total}")
        except Exception as e:
            logger.error(f"写入失败: {e}")
            conn.rollback()
            raise

    cursor.close()
    conn.close()

    logger.info("=" * 50)
    logger.info(f"[患者消费] 同步完成，共 {success_count} 条记录")
    logger.info("=" * 50)
    return success_count