doris-sync/jobs/patient_exam/ods_sync.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
ODS 层原始表同步任务
说明：轻量级增量同步，只同步昨天的数据，对 MySQL 友好
"""

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

logger = get_logger("job.ods_patient_exam_sync")


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

    # 定义需要同步的表
    tables = [
        {
            "name": "ods_patient_custome",
            "source_table": "patient_custome",
            "time_field": "update_time",
            "sync_mode": "increment",  # 增量
            "columns": """
                id,
                patientId,
                name,
                sex,
                birthday,
                identity,
                mobile,
                source,
                inviter,
                inviter_phone,
                customer_optometrist_staff_id,
                create_time,
                update_time
            """
        },
        {
            "name": "ods_clinic_clinic",
            "source_table": "clinic_clinic",
            "time_field": "create_time",
            "sync_mode": "increment",
            "columns": """
                id,
                patient_id,
                org_id,
                type,
                channel,
                add_user,
                appoint_date,
                AppointmentTime_start,
                matter,
                accept_doctor_id,
                accept_optometrist_id,
                accept_assistant_id,
                come_time,
                come_date,
                reg_status,
                pno,
                create_time,
                update_time
            """
        },
        {
            "name": "ods_exam_examresult",
            "source_table": "exam_examresult",
            "time_field": "created",
            "sync_mode": "increment",
            "columns": """
                id,
                clinic_id,
                org_id,
                result,
                tab,
                visual,
                exam_id,
                created,
                create_time,
                update_time
            """
        },
        {
            "name": "ods_treat_sellgoods",
            "source_table": "treat_sellgoods",
            "time_field": "create_time",
            "sync_mode": "increment",
            "columns": """
                id,
                sell_goods_list_id,
                goods_id,
                out_time,
                out_date,
                remark,
                create_time,
                update_time
            """
        },
        {
            "name": "ods_material_goods",
            "source_table": "material_goods",
            "time_field": None,
            "sync_mode": "full",  # 全量（数据量小）
            "columns": """
                id,
                name,
                uuid_no,
                fee_class,
                group_id,
                create_time,
                update_time
            """
        },
        {
            "name": "ods_fee_order",
            "source_table": "fee_order",
            "time_field": "create_time",
            "sync_mode": "increment",
            "columns": """
                id,
                patientId,
                org_id,
                pay_status,
                pay_time,
                pay_date,
                total_amt,
                create_time,
                update_time
            """
        },
        {
            "name": "ods_fee_orderdetail",
            "source_table": "fee_orderdetail",
            "time_field": "create_time",
            "sync_mode": "increment",
            "columns": """
                id,
                order_id,
                goods_id,
                patient_id,
                name,
                pay_time,
                pay_date,
                price,
                quantity,
                total_amt,
                create_time,
                update_time
            """
        },
        {
            "name": "ods_staff_staff",
            "source_table": "staff_staff",
            "time_field": None,
            "sync_mode": "full",  # 全量（数据量小）
            "columns": """
                id,
                user_id,
                main_org_id,
                name,
                mobile,
                qywx_userid,
                status,
                create_time,
                update_time
            """
        },
        {
            "name": "ods_org_org",
            "source_table": "org_org",
            "time_field": None,
            "sync_mode": "full",  # 全量（数据量小）
            "columns": """
                id,
                name,
                group_id,
                is_effective,
                create_time,
                update_time
            """
        },
    ]

    mysql_conn = get_mysql_conn()
    doris_conn = get_doris_conn()

    try:
        success_count = 0
        for table_config in tables:
            try:
                count = sync_table(
                    mysql_conn,
                    doris_conn,
                    table_config,
                    start_date,
                    end_date
                )
                if count > 0:
                    success_count += 1
            except Exception as e:
                logger.error(f"同步 {table_config['name']} 失败: {e}")
                import traceback
                traceback.print_exc()

        logger.info("=" * 60)
        logger.info(f"[ODS同步] 同步完成，成功 {success_count}/{len(tables)} 个表")
        logger.info("=" * 60)
        return success_count

    finally:
        mysql_conn.close()
        doris_conn.close()


def sync_table(mysql_conn, doris_conn, table_config, start_date, end_date):
    """同步单个表"""
    table_name = table_config["name"]
    source_table = table_config["source_table"]
    time_field = table_config.get("time_field")
    sync_mode = table_config["sync_mode"]
    columns = table_config["columns"]

    logger.info(f"\n开始同步: {table_name} ({sync_mode})")

    mysql_cursor = mysql_conn.cursor()
    doris_cursor = doris_conn.cursor()

    try:
        # 1. 从 MySQL 查询数据
        if sync_mode == "increment" and time_field:
            # 增量同步：只查询昨天的数据
            sql = f"""
                SELECT {columns}
                FROM {source_table}
                WHERE {time_field} >= %s
                  AND {time_field} < %s
            """
            params = [f"{start_date} 00:00:00", f"{end_date} 23:59:59"]
        else:
            # 全量同步
            sql = f"SELECT {columns} FROM {source_table}"
            params = []

        logger.info(f"执行SQL: {sql}")
        logger.info(f"参数: {params}")

        mysql_cursor.execute(sql, params)
        data_list = mysql_cursor.fetchall()
        row_count = len(data_list)

        logger.info(f"查询到 {row_count} 条记录")

        if row_count == 0:
            logger.info(f"无新数据，跳过")
            return 0

        # 2. 构建插入 SQL
        # 获取列名
        columns_list = [col.strip() for col in columns.split(',')]

        # 添加源表时间字段
        if sync_mode == "increment" and time_field:
            columns_list.extend(["source_create_time", "source_update_time"])

        placeholders = ', '.join(['%s'] * len(columns_list))
        insert_sql = f"INSERT INTO {table_name} ({', '.join(columns_list)}) VALUES ({placeholders})"

        # 3. 准备数据
        values_list = []
        for row in data_list:
            row_dict = dict(row)

            # 转换数据
            values = []
            for col in columns_list:
                if col in ["source_create_time", "source_update_time"]:
                    # 源表时间字段
                    val = row_dict.get(time_field)
                else:
                    # 普通字段
                    val = row_dict.get(col)

                # 处理日期时间
                if val and col.endswith("_time"):
                    if isinstance(val, str):
                        pass  # 已经是字符串
                    elif hasattr(val, 'strftime'):
                        val = val.strftime('%Y-%m-%d %H:%M:%S')
                    else:
                        val = None
                elif val and col.endswith("_date"):
                    if isinstance(val, str):
                        pass  # 已经是字符串
                    elif hasattr(val, 'strftime'):
                        val = val.strftime('%Y-%m-%d')
                    else:
                        val = None

                values.append(val)

            values_list.append(tuple(values))

        # 4. 批量插入 Doris
        batch_size = 100
        total_inserted = 0

        for i in range(0, len(values_list), batch_size):
            batch = values_list[i:i+batch_size]
            doris_cursor.executemany(insert_sql, batch)
            doris_conn.commit()
            total_inserted += len(batch)
            logger.info(f"已写入 {total_inserted}/{row_count}")

        logger.info(f"✓ {table_name} 同步完成: {total_inserted} 条")
        return total_inserted

    except Exception as e:
        logger.error(f"同步 {table_name} 失败: {e}")
        raise
    finally:
        mysql_cursor.close()
        doris_cursor.close()