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