#!/usr/bin/env python # -*- coding: utf-8 -*- """ 毛利分析数据同步任务 功能：将MySQL中的fee_incomedetail数据同步到Doris DWD层宽表 """ import datetime from db.pool import get_mysql_conn, get_doris_conn from logger.handler import get_logger logger = get_logger("job.grossprofit") def run(start_date=None, end_date=None, group_id=None): 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}") logger.info("=" * 50) mysql_conn = get_mysql_conn() doris_conn = get_doris_conn() mysql_cursor = mysql_conn.cursor() doris_cursor = doris_conn.cursor() if group_id: mysql_cursor.execute("SELECT id, name FROM org_group WHERE id = %s", (int(group_id),)) else: mysql_cursor.execute("SELECT id, name FROM org_group") groups = mysql_cursor.fetchall() logger.info(f"发现 {len(groups)} 个集团") if not groups: logger.info(f"未找到集团: {group_id}") mysql_cursor.close() doris_cursor.close() mysql_conn.close() doris_conn.close() return 0 total_count = 0 sdate = start_date + " 00:00:00" # 结束时间用下一天0点，SQL用<，刚好覆盖整天 edate_dt = datetime.datetime.strptime(end_date, "%Y-%m-%d") + datetime.timedelta(days=1) edate = edate_dt.strftime("%Y-%m-%d") + " 00:00:00" logger.info(f"删除 {start_date} ~ {end_date} 的旧数据...") if group_id: doris_cursor.execute( "DELETE FROM fact_fee_grossprofit WHERE pay_time >= %s AND pay_time < %s AND group_id = %s", (sdate, edate, int(group_id)) ) else: doris_cursor.execute( "DELETE FROM fact_fee_grossprofit WHERE pay_time >= %s AND pay_time < %s", (sdate, edate) ) doris_conn.commit() logger.info(f"旧数据已删除") for group in groups: group_id = group["id"] group_name = group["name"] logger.info(f"开始同步集团: {group_name} (id={group_id})") count = sync_group_data(mysql_cursor, doris_cursor, doris_conn, group_id, group_name, sdate, edate) total_count += count logger.info(f"集团 {group_name} 同步完成: {count} 条") logger.info(f"[毛利分析] 全部同步完成: 共 {total_count} 条") mysql_cursor.close() doris_cursor.close() mysql_conn.close() doris_conn.close() def sync_group_data(mysql_cursor, doris_cursor, doris_conn, group_id, group_name, sdate, edate): try: params = {"group_id": group_id, "sdate": sdate, "edate": edate} sql = """ SELECT x.id, x.created, x.modified, x.patient_name, x.patient_identity, x.patient_mobile, x.patientId, x.org_name, x.order_no, x.fee_class, x.class_name, x.combo_name, x.per_id, x.name, x.unit, x.price, x.num, x.sum_price, x.should_amt, x.discount, x.discount_memo, x.coupon_name, x.flag, x.type_flag, x.order_status, x.created AS pay_time, x.exec_time, x.exec_status, x.cost_price, x.mcost_price, x.jcost_price, x.cost_flag, x.cost_flag2, x.cost_update_time, x.cost_update_time2, x.profit_amt, x.profit_rate, x.mprofit_amt, x.mprofit_rate, x.jprofit_amt, x.jprofit_rate, x.blq_amt, x.swap_card_amt, x.eye_card_amt, x.hx_amt, x.hx_cost_price, x.hx_sum_price, x.hx_num, x.pay_amt, x.pay_cost_price, x.pay_sum_price, x.pay_num, x.pay_profit_amt, x.pay_profit_rate, x.by_user_name, x.doctor_name, x.acp_doctor_name, x.acp_opt_name, x.acp_assistant_name, x.clinic_id, x.org_id, x.order_id, x.pk_code, x.uuid_no, x.birthday, x.first_accept_date, x.first_distance_day, x.age, x.source1, x.source2, x.inviter_name, x.inviter_phone, x.line_name, x.flag_memo, x.brands, x.factory, x.variety, x.visual, x.exclude_flag, x.pay_status, c.come_time, x.pay_time AS sell_time, CASE WHEN x.flag IN (9, 11) THEN f.pay_time ELSE NULL END AS refund_pay_time, CASE WHEN x.type_flag = 2 AND x.flag NOT IN (9, 11) THEN fo.pay_time ELSE NULL END AS refund_order_pay_time FROM fee_incomedetail x JOIN clinic_clinic c ON c.id = x.clinic_id LEFT JOIN finance_bill f ON x.type_flag = 2 AND x.flag IN (9, 11) AND f.order_no = x.order_no LEFT JOIN fee_order fo ON x.type_flag = 2 AND x.flag NOT IN (9, 11) AND fo.id = x.order_id WHERE x.created >= %(sdate)s AND x.created < %(edate)s AND x.group_id = %(group_id)s AND x.visual = 1 AND x.org_id != 16 AND (x.flag IN (4, 5) OR (x.flag IN (1, 2, 6, 7, 8, 9, 10, 11, 12) AND x.should_amt != 0)) ORDER BY x.id """ logger.info(f"查询MySQL数据... (created: {sdate} ~ {edate})") mysql_cursor.execute(sql, params) rows = mysql_cursor.fetchall() logger.info(f"获取到 {len(rows)} 条数据") if not rows: logger.info("没有数据需要同步") return 0 doris_insert_sql = """ INSERT INTO fact_fee_grossprofit ( fee_incomedetail_id, created, modified, month, org_name, group_id, group_name, pay_time, fee_time, clinic_id, come_time, patientId, patient_name, patient_mobile, patient_identity, birthday, age, source1, source2, inviter_name, inviter_phone, first_accept_date, first_distance_day, by_user_name, doctor_name, opt_name, assistant_name, pay_status, order_no, pk_code, uuid_no, class_name, line_name, fee_class, name, combo_name, flag_memo, price, num, unit, sum_price, should_amt, cost_amt, profit_amt, profit_rate, discount, discount_memo, coupon_name, blq_amt, hx_sum_price, hx_num, hx_amt, hx_cost_price, pay_sum_price, pay_num, pay_amt, pay_cost_price, pay_profit_amt, pay_profit_rate, factory_name, brands_name, variety_name, exec_time, exec_status_name, order_status_name, cost_flag, update_time, cost_flag2, update_time2, exec_status, flag, per_id, order_id, type_flag, order_status, sell_time, sync_date ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s ) """ def format_datetime(v): if isinstance(v, datetime.datetime): return v.strftime("%Y-%m-%d %H:%M:%S") if v else None elif isinstance(v, datetime.date): return v.strftime("%Y-%m-%d") if v else None return v def format_profit_rate(v): if v is None or v == 0: return "" return f"{float(v) * 100:.2f}%" def format_discount(v): if v is None or v == 0: return "" return f"{float(v) * 100:.2f}%" batch_size = 1000 for i in range(0, len(rows), batch_size): batch = rows[i:i + batch_size] values = [] for row in batch: r = row # DictCursor 已经返回字典 created = format_datetime(r.get("created")) month = created[:7] if created else "" pay_time = format_datetime(r.get("pay_time")) # 即 x.created sell_time = format_datetime(r.get("sell_time")) # 即 x.pay_time # fee_time 计算逻辑完全对齐 API if r.get("name") != "系统逾期项目": fee_time = created else: fee_time = sell_time type_flag = r.get("type_flag", 1) if type_flag == 2: if r.get("flag") in [9, 11]: fee_time = format_datetime(r.get("refund_pay_time")) else: if r.get("name") != "系统逾期项目": fee_time = format_datetime(r.get("refund_order_pay_time")) else: fee_time = sell_time exec_status = r.get("exec_status", 0) exec_status_name = "已交付" if exec_status == 1 else "未交付" order_status = r.get("order_status", 1) order_status_name = "正常" # 完全对齐 API 中的 Bug: 若 order_status == 2，把 exec_status_name 设为 "作废" if order_status == 2: exec_status_name = "作废" pay_status = "已退费" if type_flag == 2 else "已收费" cost_amt = r.get("mcost_price", 0) or 0 profit_amt = r.get("mprofit_amt", 0) or 0 sum_price = float(r.get("sum_price", 0) or 0) should_amt = float(r.get("should_amt", 0) or 0) num = r.get("num", 0) or 0 if type_flag == 2: cost_amt = -(cost_amt) profit_amt = -(profit_amt) sum_price = -sum_price should_amt = -should_amt num = -num profit_rate = format_profit_rate(r.get("mprofit_rate")) discount = format_discount(r.get("discount")) come_time = format_datetime(r.get("come_time")) exec_time = format_datetime(r.get("exec_time")) update_time = format_datetime(r.get("cost_update_time")) update_time2 = format_datetime(r.get("cost_update_time2")) sync_date = pay_time.split(" ")[0] if pay_time else None blq_amt = float(r.get("blq_amt", 0) or 0) + float(r.get("swap_card_amt", 0) or 0) + float(r.get("eye_card_amt", 0) or 0) hx_sum_price = float(r.get("hx_sum_price", 0) or 0) hx_num = r.get("hx_num", 0) or 0 hx_amt = float(r.get("hx_amt", 0) or 0) hx_cost_price = float(r.get("hx_cost_price", 0) or 0) pay_sum_price = float(r.get("pay_sum_price", 0) or 0) pay_num = r.get("pay_num", 0) or 0 pay_amt = float(r.get("pay_amt", 0) or 0) pay_cost_price = float(r.get("pay_cost_price", 0) or 0) pay_profit_amt = float(r.get("pay_profit_amt", 0) or 0) pay_profit_rate = r.get("pay_profit_rate", 0) or 0 if type_flag == 2: hx_sum_price = -hx_sum_price hx_num = -hx_num hx_amt = -hx_amt hx_cost_price = -hx_cost_price pay_sum_price = -pay_sum_price pay_num = -pay_num pay_amt = -pay_amt pay_cost_price = -pay_cost_price pay_profit_amt = -pay_profit_amt pay_profit_rate = -pay_profit_rate values.append(( r.get("id"), created, format_datetime(r.get("modified")), month, r.get("org_name"), group_id, group_name, pay_time, fee_time, r.get("clinic_id"), come_time, r.get("patientId"), r.get("patient_name"), r.get("patient_mobile"), r.get("patient_identity"), format_datetime(r.get("birthday")), r.get("age", 0), r.get("source1"), r.get("source2"), r.get("inviter_name"), r.get("inviter_phone"), format_datetime(r.get("first_accept_date")), r.get("first_distance_day", 0), r.get("by_user_name"), r.get("doctor_name"), r.get("acp_opt_name"), r.get("acp_assistant_name"), pay_status, r.get("order_no"), r.get("pk_code"), r.get("uuid_no"), r.get("class_name"), r.get("line_name"), r.get("fee_class"), r.get("name"), r.get("combo_name"), r.get("flag_memo"), r.get("price"), num, r.get("unit"), sum_price, should_amt, cost_amt, profit_amt, profit_rate, discount, r.get("discount_memo"), r.get("coupon_name"), blq_amt, hx_sum_price, hx_num, hx_amt, hx_cost_price, pay_sum_price, pay_num, pay_amt, pay_cost_price, pay_profit_amt, pay_profit_rate, r.get("factory"), r.get("brands"), r.get("variety"), exec_time, exec_status_name, order_status_name, r.get("cost_flag", 0), update_time, r.get("cost_flag2", 0), update_time2, exec_status, r.get("flag"), r.get("per_id"), r.get("order_id"), type_flag, order_status, sell_time, sync_date )) doris_cursor.executemany(doris_insert_sql, values) doris_conn.commit() logger.info(f"已插入 {i + len(batch)} / {len(rows)} 条") logger.info(f"[毛利分析] 同步完成: {len(rows)} 条") return len(rows) except Exception as e: logger.error(f"同步失败: {e}") doris_conn.rollback() raise if __name__ == "__main__": import argparse parser = argparse.ArgumentParser() parser.add_argument("--start", type=str, help="开始日期 YYYY-MM-DD") parser.add_argument("--end", type=str, help="结束日期 YYYY-MM-DD") parser.add_argument("--group-id", type=str, help="集团ID，不传则同步全部集团") args = parser.parse_args() run(args.start, args.end, args.group_id)