doris-sync/jobs/patient_exam/ads_build.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
ADS 层宽表计算任务
说明：在 Doris 内部生成患者检查周期宽表（基于末次取镜时间）
"""

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

logger = get_logger("job.ads_patient_exam")


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

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

    try:
        # 1. 查询昨天有取镜记录的患者
        logger.info("步骤1: 查询取镜患者...")
        patient_sql = get_patient_sql(start_date)
        cursor.execute(patient_sql)
        patients = cursor.fetchall()
        logger.info(f"找到 {len(patients)} 个患者")

        if not patients:
            logger.warning("没有取镜患者，跳过")
            return 0

        # 2. 为每个患者生成宽表数据
        logger.info("步骤2: 生成宽表数据...")
        all_rows = []
        for patient in patients:
            patient_id = patient['patient_id']
            patient_id_str = patient['patient_id_str']

            # 生成单条宽表记录
            row = build_wide_row(cursor, patient_id, patient_id_str)
            if row:
                all_rows.append(row)

        # 3. 写入宽表
        logger.info("步骤3: 写入宽表...")
        insert_sql = get_insert_sql()
        placeholders = ', '.join(['%s'] * len(all_rows[0]))
        final_sql = insert_sql.replace('{placeholders}', placeholders)

        batch_size = 100
        total_inserted = 0

        for i in range(0, len(all_rows), batch_size):
            batch = all_rows[i:i+batch_size]
            cursor.executemany(final_sql, batch)
            conn.commit()
            total_inserted += len(batch)
            logger.info(f"已写入 {total_inserted}/{len(all_rows)}")

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

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


def get_patient_sql(pickup_date):
    """查询有取镜记录的患者"""
    return f"""
    SELECT DISTINCT
        p.id as patient_id,
        p.patientId as patient_id_str
    FROM ods_treat_sellgoods sg
    JOIN ods_material_goods g ON g.id = sg.goods_id
    JOIN ods_treat_sellgoodslist sgl ON sgl.id = sg.sell_goods_list_id
    JOIN ods_clinic_clinic c ON c.id = sgl.clinic_id
    JOIN ods_patient_custome p ON p.id = c.patient_id
    WHERE sg.out_time >= '{pickup_date} 00:00:00'
      AND sg.out_time < '{pickup_date} 23:59:59'
      AND sg.out_time IS NOT NULL
      AND g.fee_class IN ('长期治疗类', '离焦镜片')
    """


def build_wide_row(cursor, patient_id, patient_id_str):
    """为单个患者构建宽表数据"""
    # 查询患者基本信息
    info_sql = f"""
    SELECT
        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,
        s.name as customer_optometrist_name,
        org_s.name as customer_optometrist_org_name,
        g.name as last_pickup_goods,
        g.id as last_pickup_goods_id,
        g.fee_class
    FROM ods_patient_custome p
    LEFT JOIN ods_staff_staff s ON s.id = p.customer_optometrist_staff_id
    LEFT JOIN ods_org_org org_s ON org_s.id = s.main_org_id
    JOIN (
        SELECT sg2.out_time, g2.name, g2.id, g2.fee_class
        FROM ods_treat_sellgoods sg2
        JOIN ods_material_goods g2 ON g2.id = sg2.goods_id
        JOIN ods_treat_sellgoodslist sgl2 ON sgl2.id = sg2.sell_goods_list_id
        JOIN ods_clinic_clinic c2 ON c2.id = sgl2.clinic_id
        WHERE c2.patient_id = {patient_id}
        ORDER BY sg2.out_time DESC
        LIMIT 1
    ) g ON 1=1
    WHERE p.id = {patient_id}
    """

    cursor.execute(info_sql)
    patient_info = cursor.fetchone()

    if not patient_info:
        return None

    # 查询购买统计
    buy_sql = f"""
    SELECT
        COUNT(CASE WHEN r.id IS NULL THEN 1 END) as buy_cnt,
        COUNT(CASE WHEN r.id IS NOT NULL THEN 1 END) as refund_cnt,
        MAX(CASE WHEN g.name LIKE '%铂视控%' THEN 1 END) as has_bs,
        MAX(CASE WHEN g.name LIKE '%星趣控%' THEN 1 END) as has_xq
    FROM ods_clinic_clinic c
    JOIN ods_fee_order fo ON fo.patientId IN (SELECT patientId FROM ods_patient_custome WHERE id = {patient_id})
    JOIN ods_fee_orderdetail fod ON fod.order_id = fo.id
    JOIN ods_material_goods g ON g.id = fod.goods_id
    LEFT JOIN ods_fee_refunddetail r ON r.orderdetail_id = fod.id
    WHERE c.patient_id = {patient_id}
      AND g.fee_class = '{patient_info['fee_class']}'
    """

    cursor.execute(buy_sql)
    buy_info = cursor.fetchone()

    # 查询末次取镜时间
    pickup_sql = f"""
    SELECT MAX(sg.out_time) as last_pickup_time
    FROM ods_treat_sellgoods sg
    JOIN ods_material_goods g ON g.id = sg.goods_id
    JOIN ods_treat_sellgoodslist sgl ON sgl.id = sg.sell_goods_list_id
    JOIN ods_clinic_clinic c ON c.id = sgl.clinic_id
    WHERE c.patient_id = {patient_id}
      AND sg.out_time IS NOT NULL
      AND g.fee_class IN ('长期治疗类', '离焦镜片')
    """

    cursor.execute(pickup_sql)
    pickup_info = cursor.fetchone()

    last_pickup_time = pickup_info['last_pickup_time']
    last_pickup_date = last_pickup_time.date() if last_pickup_time else None

    # 构建宽表记录
    row = {
        'patient_id': patient_id,
        'patient_id_str': patient_id_str,
        'patient_name': patient_info['patient_name'],
        'sex': patient_info['sex'],
        'age': patient_info['age'],
        'birthday': patient_info['birthday'],
        'mobile': patient_info['mobile'],
        'identity': patient_info['identity'],
        'fee_class': patient_info['fee_class'],
        'customer_optometrist_name': patient_info['customer_optometrist_name'] or '',
        'customer_optometrist_org_name': patient_info['customer_optometrist_org_name'] or '',
        'last_pickup_time': last_pickup_time,
        'last_pickup_date': last_pickup_date,
        'last_pickup_goods': patient_info['last_pickup_goods'],
        'last_pickup_goods_id': patient_info['last_pickup_goods_id'],
        'buy_cnt': buy_info['buy_cnt'] if buy_info else 0,
        'refund_cnt': buy_info['refund_cnt'] if buy_info else 0,
        'is_both_bs_xq': 1 if (buy_info['has_bs'] and buy_info['has_xq']) else 0,
        'data_date': last_pickup_date,
    }

    # TODO: 这里需要实现周期数据的计算逻辑
    # 可以参考 export_goods.py 中的 parse_exam_data 和 fill_cycle_exam_data 函数
    # 由于逻辑复杂，建议用存储过程或单独的 Python 函数实现

    return row


def get_insert_sql():
    """获取插入 SQL"""
    # 这里简化了字段，实际需要添加所有周期字段
    return """
    INSERT INTO fact_patient_exam_wide (
        patient_id,
        patient_id_str,
        patient_name,
        sex,
        age,
        birthday,
        mobile,
        identity,
        fee_class,
        customer_optometrist_name,
        customer_optometrist_org_name,
        last_pickup_time,
        last_pickup_date,
        last_pickup_goods,
        last_pickup_goods_id,
        buy_cnt,
        refund_cnt,
        is_both_bs_xq,
        baseline_date,
        baseline_right_se_val,
        baseline_left_se_val,
        baseline_al_od_val,
        baseline_al_os_val,
        m3_right_se_val,
        m3_left_se_val,
        m3_al_od_val,
        m3_al_os_val,
        p3_right_se_val,
        p3_left_se_val,
        p3_al_od_val,
        p3_al_os_val,
        p6_right_se_val,
        p6_left_se_val,
        p6_al_od_val,
        p6_al_os_val,
        data_date,
        create_time,
        update_time
    ) VALUES ({placeholders})
    ON DUPLICATE KEY UPDATE
        patient_name = VALUES(patient_name),
        update_time = NOW()
    """