#!/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