#!/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() """