doris-sync/jobs/patient_exam/dwd_build.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
DWD 层明细宽表计算任务
说明：在 Doris 内部关联 ODS 表生成就诊明细事实表
"""

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

logger = get_logger("job.dwd_patient_exam")


def run(start_date=None, end_date=None):
    """执行 DWD 层计算"""
    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("=" * 60)
    logger.info(f"[DWD计算] 开始计算: {start_date}")
    logger.info("=" * 60)

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

    try:
        # 先清空当天的数据（幂等性）
        delete_sql = """
            DELETE FROM fact_patient_exam_detail
            WHERE data_date = %s
        """
        cursor.execute(delete_sql, [start_date])
        affected_rows = cursor.rowcount
        if affected_rows > 0:
            logger.info(f"清理旧数据: {affected_rows} 条")
            conn.commit()

        # 执行计算 SQL
        build_sql = get_build_sql(start_date)
        logger.info("开始执行计算...")

        cursor.execute(build_sql)
        affected_rows = cursor.rowcount
        conn.commit()

        logger.info("=" * 60)
        logger.info(f"[DWD计算] 计算完成: {affected_rows} 条")
        logger.info("=" * 60)
        return affected_rows

    except Exception as e:
        logger.error(f"[DWD计算] 计算失败: {e}")
        conn.rollback()
        raise
    finally:
        cursor.close()
        conn.close()


def get_build_sql(data_date):
    """生成计算 SQL"""
    return f"""
    INSERT INTO fact_patient_exam_detail (
        id,
        patient_id,
        patient_id_str,
        patient_name,
        sex,
        age,
        birthday,
        mobile,
        identity,
        clinic_id,
        come_time,
        come_date,
        visit_type,
        channel,
        goods_id,
        goods_name,
        goods_uuid,
        fee_class,
        pickup_id,
        pickup_time,
        pickup_date,
        order_id,
        orderdetail_id,
        pay_time,
        pay_date,
        pay_amt,
        refund_time,
        org_id,
        org_name,
        group_id,
        group_name,
        customer_optometrist_staff_id,
        customer_optometrist_name,
        customer_optometrist_org_name,
        accept_doctor_id,
        accept_doctor_name,
        accept_optometrist_id,
        accept_optometrist_name,
        exam_result_json,
        exam_tab,
        exam_id,
        right_se_val,
        left_se_val,
        right_sph_val,
        left_sph_val,
        right_cyl_val,
        left_cyl_val,
        al_od_val,
        al_os_val,
        data_date,
        create_time,
        update_time
    )
    SELECT
        c.id,
        p.id as patient_id,
        p.patientId as patient_id_str,
        p.name as patient_name,
        CASE WHEN p.sex = '男' THEN 1 ELSE 2 END as sex,
        TIMESTAMPDIFF(YEAR, p.birthday, CURDATE()) as age,
        p.birthday,
        p.mobile,
        p.identity,
        c.id as clinic_id,
        c.come_time,
        DATE(c.come_time) as come_date,
        c.type as visit_type,
        c.channel,
        g.id as goods_id,
        g.name as goods_name,
        g.uuid_no as goods_uuid,
        g.fee_class,
        sg.id as pickup_id,
        sg.out_time as pickup_time,
        DATE(sg.out_time) as pickup_date,
        fo.id as order_id,
        fod.id as orderdetail_id,
        fo.pay_time,
        DATE(fo.pay_time) as pay_date,
        fod.quantity * fod.price as pay_amt,
        r.refund_time,
        c.org_id,
        org.name as org_name,
        org.group_id,
        grp.name as group_name,
        p.customer_optometrist_staff_id,
        s.name as customer_optometrist_name,
        s_org.name as customer_optometrist_org_name,
        c.accept_doctor_id,
        d.name as accept_doctor_name,
        c.accept_optometrist_id,
        o.name as accept_optometrist_name,
        er.result as exam_result_json,
        er.tab as exam_tab,
        er.exam_id,
        NULL as right_se_val,
        NULL as left_se_val,
        NULL as right_sph_val,
        NULL as left_sph_val,
        NULL as right_cyl_val,
        NULL as left_cyl_val,
        NULL as al_od_val,
        NULL as al_os_val,
        DATE(c.come_time) as data_date,
        NOW() as create_time,
        NOW() as update_time
    FROM ods_clinic_clinic c
    JOIN ods_patient_custome p ON c.patient_id = p.id
    LEFT JOIN ods_org_org org ON c.org_id = org.id
    LEFT JOIN ods_fee_order fo ON fo.patientId = p.patientId
        AND DATE(fo.pay_time) = '{data_date}'
    LEFT JOIN ods_fee_orderdetail fod ON fod.order_id = fo.id
    LEFT JOIN ods_material_goods g ON g.id = fod.goods_id
    LEFT JOIN ods_treat_sellgoods sg ON sg.sell_goods_list_id = fod.sell_goods_list_id
    LEFT JOIN ods_fee_refunddetail r ON r.orderdetail_id = fod.id
    LEFT JOIN ods_staff_staff s ON s.id = p.customer_optometrist_staff_id
    LEFT JOIN ods_org_org s_org ON s.main_org_id = s_org.id
    LEFT JOIN ods_staff_staff d ON d.id = c.accept_doctor_id
    LEFT JOIN ods_staff_staff o ON o.id = c.accept_optometrist_id
    LEFT JOIN ods_exam_examresult er ON er.clinic_id = c.id
        AND er.tab IN ('yj_dnyg', 'tj_gxsw')
    LEFT JOIN ods_org_group grp ON grp.id = org.group_id
    WHERE c.come_time >= '{data_date} 00:00:00'
      AND c.come_time < '{data_date} 23:59:59'
      AND c.reg_status IN (1, 2, 3, 4)
      AND c.pno = ''
      AND org.is_effective = 1
      AND g.fee_class IN ('长期治疗类', '离焦镜片')
    """