doris-sync/jobs/clinic_log.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.clinic_log")


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. 获取来源字典
    cursor.execute("SELECT id, name FROM orgdict_origin")
    source_dict = {s["id"]: s["name"] for s in cursor.fetchall()}

    # 2. 查询主数据
    params = {
        "start_date": start_date + " 00:00:00",
        "end_date": end_date + " 23:59:59",
        "group_id": int(group_id),
    }

    sql = """
        SELECT
            c.id as clinic_id,
            p.patientId,
            p.name as patient_name,
            p.sex,
            p.birthday,
            p.identity,
            p.mobile,
            p.source,
            p.inviter,
            p.inviter_phone,
            c.type,
            c.channel,
            c.add_user,
            c.appoint_date,
            c.AppointmentTime_start,
            c.matter,
            c.accept_doctor_id,
            c.accept_optometrist_id,
            c.accept_assistant_id,
            c.customer_manager_staff_id,
            c.customer_optometrist_staff_id,
            c.come_time,
            c.patient_id,
            o.id as org_id,
            o.name as org_name,
            o.group_id,
            g.name as group_name
        FROM clinic_clinic c
        JOIN patient_custome p ON c.patient_id = p.id
        JOIN org_org o ON c.org_id = o.id
        LEFT JOIN org_group g ON o.group_id = g.id
        WHERE c.reg_status IN (2, 3, 4)
            AND c.pno = ''
            AND o.is_effective = 1
            AND o.group_id = %(group_id)s
            AND c.come_time >= %(start_date)s
            AND c.come_time <= %(end_date)s
        GROUP BY c.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

    # 3. 收集ID列表
    clinic_id_list = [d["clinic_id"] for d in data_list]
    patient_id_list = [d["patient_id"] for d in data_list]
    staff_id_list = []

    for d in data_list:
        if d.get("accept_doctor_id"):
            staff_id_list.append(d["accept_doctor_id"])
        if d.get("accept_optometrist_id"):
            staff_id_list.append(d["accept_optometrist_id"])
        if d.get("accept_assistant_id"):
            staff_id_list.append(d["accept_assistant_id"])
        if d.get("customer_manager_staff_id"):
            staff_id_list.append(d["customer_manager_staff_id"])
        if d.get("customer_optometrist_staff_id"):
            staff_id_list.append(d["customer_optometrist_staff_id"])

    staff_id_list = list(set(staff_id_list))

    # 4. 查询员工信息
    staff_dict = {}
    if staff_id_list:
        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_id_list})
        staff_dict = {s["id"]: s for s in cursor.fetchall()}

    # 5. 查询咨询信息
    recode_dict = {}
    user_dict = {}
    if clinic_id_list:
        sql = """
            SELECT r.clinic_id, r.collect_date, r.activity_location,
                   r.extension_user_id, r.created_user_id
            FROM returnvisit_customerconsultationrecord r
            WHERE r.clinic_id IN %(ids)s
        """
        cursor.execute(sql, {"ids": clinic_id_list})
        recode_list = cursor.fetchall()
        for r in recode_list:
            recode_dict[r["clinic_id"]] = dict(r)

        # 获取用户ID列表查询用户信息
        user_id_list = []
        for r in recode_list:
            if r.get("extension_user_id"):
                user_id_list.append(r["extension_user_id"])
            if r.get("created_user_id"):
                user_id_list.append(r["created_user_id"])

        if user_id_list:
            sql = "SELECT id, name, mobile FROM users_userprofile WHERE id IN %(ids)s"
            cursor.execute(sql, {"ids": list(set(user_id_list))})
            user_list = cursor.fetchall()
            user_dict = {u["id"]: u for u in user_list}

    # 6. 查询视力信息
    exam_dict = {}
    if clinic_id_list:
        sql = """
            SELECT clinic_id, result
            FROM exam_examresult
            WHERE clinic_id IN %(ids)s
            AND visual = 1
            AND (tab = 'yj_dnyg' OR tab = 'tj_zhygy')
        """
        cursor.execute(sql, {"ids": clinic_id_list})
        exam_list = cursor.fetchall()
        for e in exam_list:
            clinic_id = e["clinic_id"]
            if clinic_id not in exam_dict:
                exam_dict[clinic_id] = {"r_sph": "", "l_sph": "", "r_nakesight": "", "l_nakesight": ""}
            try:
                res_dict = eval(e["result"]) if e.get("result") else {}
                if res_dict.get("r_sph"):
                    exam_dict[clinic_id]["r_sph"] = res_dict.get("r_sph")
                if res_dict.get("l_sph"):
                    exam_dict[clinic_id]["l_sph"] = res_dict.get("l_sph")
                if res_dict.get("r_nakesight"):
                    exam_dict[clinic_id]["r_nakesight"] = res_dict.get("r_nakesight")
                if res_dict.get("l_nakesight"):
                    exam_dict[clinic_id]["l_nakesight"] = res_dict.get("l_nakesight")
            except:
                pass

    # 7. 查询标签
    patient_tag_dict = {}
    if patient_id_list:
        sql = """
            SELECT t.patient_id, u.name
            FROM patient_uspatienttotag t
            JOIN patient_ustag u ON t.tag_id = u.id
            JOIN patient_ustagparent p ON u.parent_id = p.id
            WHERE t.patient_id IN %(ids)s AND p.enabled = 1
        """
        cursor.execute(sql, {"ids": patient_id_list})
        tag_list = cursor.fetchall()
        for t in tag_list:
            patient_tag_dict.setdefault(t["patient_id"], []).append(t["name"])

    # 7.1 按原系统逻辑判断新老客户和当前就诊次数
    clinic_new_old_dict = {}
    if patient_id_list:
        sql = """
            SELECT patient_id, id
            FROM clinic_clinic
            WHERE patient_id IN %(ids)s
              AND reg_status IN (2, 3, 4)
              AND (pno IS NULL OR pno = '')
            ORDER BY come_time, id
        """
        cursor.execute(sql, {"ids": list(set(patient_id_list))})
        all_clinic_list = cursor.fetchall()

        patient_clinic_dict = {}
        for clinic in all_clinic_list:
            patient_clinic_dict.setdefault(clinic["patient_id"], []).append(str(clinic["id"]))

        for clinic_id, patient_id in zip(clinic_id_list, patient_id_list):
            clinic_ids = patient_clinic_dict.get(patient_id, [])
            clinic_new_old_dict[clinic_id] = {
                "patient_type": "新客户",
                "this_come_number": 0,
            }

            if clinic_ids:
                if str(clinic_id) != clinic_ids[0]:
                    clinic_new_old_dict[clinic_id]["patient_type"] = "老客户"
                try:
                    clinic_new_old_dict[clinic_id]["this_come_number"] = clinic_ids.index(str(clinic_id)) + 1
                except ValueError:
                    clinic_new_old_dict[clinic_id]["this_come_number"] = 0

    cursor.close()
    conn.close()

    # 8. 字段列表
    fields = [
        "clinic_id", "patientId", "patient_name", "sex", "birthday", "identity", "mobile",
        "source1", "source2", "tag_list", "patient_type", "visit_type", "r_sph", "l_sph",
        "r_nakesight", "l_nakesight", "org_id", "org_name", "appoint_time", "add_user",
        "group_id", "group_name", "matter", "accept_doctor_id", "accept_doctor_name", "accept_optometrist_id",
        "accept_optometrist_name", "accept_assistant_id", "accept_assistant_name",
        "manager_id", "manager_name", "manager_qywx_id", "manager_mobile",
        "optometrist_id", "optometrist_name", "optometrist_qywx_id", "optometrist_mobile",
        "come_time", "this_come_number", "inviter", "inviter_phone", "channel",
        "collect_date", "activity_location", "extension_user", "extension_user_mobile",
    ]

    # 9. 处理数据
    result = []
    for data in data_list:
        clinic_id = data["clinic_id"]
        patient_id = data["patient_id"]

        # 处理来源
        source = data.get("source", "")
        pat_source = source.split(",") if source else []
        source1 = source_dict.get(int(pat_source[0])) if len(pat_source) > 0 else ""
        source2 = source_dict.get(int(pat_source[1])) if len(pat_source) > 1 else ""

        # 处理渠道
        channel = data.get("channel", "")
        if channel == "wechat":
            channel = "微信小程序"
        elif channel == "crm":
            channel = "客服"
        elif channel == "doctor_wechat":
            channel = "渠道医生"
        elif channel == "eye_wechat":
            channel = "互联网医院"
        elif channel == "ileads":
            channel = "ileads"
        else:
            channel = "门店"

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

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

        # 处理日期
        birthday = data.get("birthday")
        if birthday:
            if isinstance(birthday, str):
                birthday = birthday.split(" ")[0]
            else:
                birthday = str(birthday).split(" ")[0]
        else:
            birthday = None

        # 辅助函数：处理日期时间格式
        def format_datetime(dt):
            """将日期时间转换为 Doris 兼容的格式"""
            if dt is None:
                return None
            if isinstance(dt, str):
                # 如果是字符串，检查是否有时分秒
                if " " in dt:
                    return dt  # 已包含时分秒
                else:
                    return f"{dt} 00:00:00"  # 只有日期，添加默认时间
            else:
                # datetime 对象
                return dt.strftime("%Y-%m-%d %H:%M:%S")

        come_time = format_datetime(data.get("come_time"))
        appoint_time = format_datetime(data.get("AppointmentTime_start"))

        # 获取其他信息
        exam_info = exam_dict.get(clinic_id, {})
        recode_info = recode_dict.get(clinic_id, {})
        new_old_info = clinic_new_old_dict.get(clinic_id, {})

        # 处理收集日期
        collect_date = recode_info.get("collect_date")
        if collect_date:
            if isinstance(collect_date, str):
                collect_date = collect_date.split(" ")[0]
            elif hasattr(collect_date, 'strftime'):
                collect_date = collect_date.strftime("%Y-%m-%d")
            else:
                collect_date = str(collect_date).split(" ")[0]

        # 确保整数字段类型正确
        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 = (
            safe_int(clinic_id),
            safe_str(data.get("patientId")),
            safe_str(data.get("patient_name")),
            safe_int(sex) if sex else None,
            birthday,
            safe_str(data.get("identity")),
            safe_str(data.get("mobile")),
            safe_str(source1),
            safe_str(source2),
            safe_str(",".join(patient_tag_dict.get(patient_id, []))),
            safe_str(new_old_info.get("patient_type", "")),
            safe_int(visit_type) if visit_type else None,
            safe_str(exam_info.get("r_sph")),
            safe_str(exam_info.get("l_sph")),
            safe_str(exam_info.get("r_nakesight")),
            safe_str(exam_info.get("l_nakesight")),
            safe_int(data.get("org_id")),
            safe_str(data.get("org_name")),
            appoint_time,
            safe_str(data.get("add_user")),
            safe_int(data.get("group_id")),
            safe_str(data.get("group_name")),
            safe_str(data.get("matter")),
            safe_int(data.get("accept_doctor_id")),
            safe_str(staff_dict.get(data.get("accept_doctor_id"), {}).get("name", "")),
            safe_int(data.get("accept_optometrist_id")),
            safe_str(staff_dict.get(data.get("accept_optometrist_id"), {}).get("name", "")),
            safe_int(data.get("accept_assistant_id")),
            safe_str(staff_dict.get(data.get("accept_assistant_id"), {}).get("name", "")),
            safe_int(data.get("customer_manager_staff_id")),
            safe_str(staff_dict.get(data.get("customer_manager_staff_id"), {}).get("name", "")),
            safe_str(staff_dict.get(data.get("customer_manager_staff_id"), {}).get("qywx_userid", "")),
            safe_str(staff_dict.get(data.get("customer_manager_staff_id"), {}).get("mobile", "")),
            safe_int(data.get("customer_optometrist_staff_id")),
            safe_str(staff_dict.get(data.get("customer_optometrist_staff_id"), {}).get("name", "")),
            safe_str(staff_dict.get(data.get("customer_optometrist_staff_id"), {}).get("qywx_userid", "")),
            safe_str(staff_dict.get(data.get("customer_optometrist_staff_id"), {}).get("mobile", "")),
            come_time,
            safe_int(new_old_info.get("this_come_number")) if new_old_info.get("this_come_number") else 1,
            safe_str(data.get("inviter")),
            safe_str(data.get("inviter_phone")),
            safe_str(channel),
            collect_date,
            safe_str(recode_info.get("activity_location")),
            safe_str(user_dict.get(recode_info.get("extension_user_id", 0), {}).get("name", "")),
            safe_str(user_dict.get(recode_info.get("extension_user_id", 0), {}).get("mobile", "")),
        )

        result.append(record)

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

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

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

    # 10.2 插入新数据
    sql = f"INSERT INTO fact_cliniclog ({', '.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