import dmPython import openpyxl from openpyxl.styles import Font, PatternFill, Alignment, Border, Side # ==================== 数据库配置 ==================== DB_CONFIG = { "host": "192.168.1.250", "port": 5236, "user": "SYSDBA", "password": "Aa15790!123", } SCHEMA = "AYGC_JKYWK" TABLE_PREFIX = "QH_DAT_WATCH" OUTPUT_FILE = "export_qh_dat_watch.xlsx" # 排除不需要导出的表 TABLE_EXCLUDE = { "QH_DAT_WATCH_EME_MOBILE", "QH_DAT_WATCH_ERROR_LOG", "QH_DAT_WATCH_LOG", "QH_DAT_WATCH_MQTT_LOG", "QH_DAT_WATCH_NOTIFY", "QH_DAT_WATCH_NOTIFY_LOG", "QH_DAT_WATCH_PERIOD", "QH_DAT_WATCH_PERIOD_PRI", "QH_DAT_WATCH_STAT_ALL", "QH_DAT_WATCH_STAT_MAX_ID", "QH_DAT_WATCH_STAT_TODAY", "QH_DAT_WATCH_TASK", "QH_DAT_WATCH_TASKS", "QH_DAT_WATCH_TASKS_CYCLE", "QH_DAT_WATCH_TASKS_PRI", "QH_DAT_WATCH_WARN_FLAG", "QH_DAT_WATCH_WARN_PRI", } # 当前表名 → 原始表名映射（来自 xc_boot.sql 的 RENAME 记录） TABLE_NAME_MAPPING = { "QH_DAT_WATCH_SPO2": "DC_WATCH_DATA_COMMON", "QH_DAT_WATCH_ECG": "DC_WATCH_DATA_ECG", "QH_DAT_WATCH_ECG_ITEM": "DC_WATCH_DATA_ECG_ITEM", "QH_DAT_WATCH_EXERCISE": "DC_WATCH_DATA_EXERCISE", "QH_DAT_WATCH_GPS": "DC_WATCH_DATA_GPS", "QH_DAT_WATCH_HEART_RATE": "DC_WATCH_DATA_HEART_RATE", "QH_DAT_WATCH_LOG": "DC_WATCH_DATA_LOG", "QH_DAT_WATCH_MONITOR": "DC_WATCH_DATA_MONITOR", "QH_DAT_WATCH_MONITOR_HR": "DC_WATCH_DATA_MONITOR_HEART_RATE", "QH_DAT_WATCH_MONITOR_SPO2": "DC_WATCH_DATA_MONITOR_SPO2", "QH_DAT_WATCH_MONITOR_STRE": "DC_WATCH_DATA_MONITOR_STRESS", "QH_DAT_WATCH_MONITOR_TEMP": "DC_WATCH_DATA_MONITOR_TEMPERATURE", "QH_DAT_WATCH_SDC": "DC_WATCH_DATA_SDC", "QH_DAT_WATCH_SDC_ITEM": "DC_WATCH_DATA_SDC_ITEM", "QH_DAT_WATCH_SDC_JSON": "DC_WATCH_DATA_SDC_JSON", "QH_DAT_WATCH_SLEEP": "DC_WATCH_DATA_SLEEP", "QH_DAT_WATCH_SLEEP_NEW": "DC_WATCH_DATA_SLEEP_NEW", "QH_DAT_WATCH_SLEEP_NEWDAY": "DC_WATCH_DATA_SLEEP_NEW_DAY", "QH_DAT_WATCH_SLEEP_NEWERR": "DC_WATCH_DATA_SLEEP_NEW_ERR", "QH_DAT_WATCH_SLEEP_NEWMIN": "DC_WATCH_DATA_SLEEP_NEW_MINUTE", "QH_DAT_WATCH5_SLEEP": "DC_WATCH_DATA_SLEEP_WATCH5", "QH_DAT_WATCH5_SLEEP_DETA": "DC_WATCH_DATA_SLEEP_WATCH5_DETAIL", "QH_DAT_WATCH_STEP": "DC_WATCH_DATA_STEP", "QH_DAT_WATCH_STEP_NEW": "DC_WATCH_DATA_STEP_NEW", "QH_DAT_WATCH_STRESS": "DC_WATCH_DATA_STRESS", "QH_DAT_WATCH_TASK": "DC_WATCH_DATA_TASK", "QH_DAT_WATCH_TEMPERATURE": "DC_WATCH_DATA_TEMPERATURE", "QH_DAT_WATCH_WORKOUT": "DC_WATCH_DATA_WORKOUT", "QH_DAT_WATCH_WORKOUT_TRAC":"DC_WATCH_DATA_WORKOUT_TRACE", "QH_DAT_WATCH_DEVICE": "DC_WATCH_DEVICE", "QH_DAT_WATCH_DEVICE_SWIT": "DC_WATCH_DEVICE_SWITCH", "QH_DAT_WATCH_VERSION_HIS": "DC_WATCH_DEVICE_VERSION_HIS", "QH_DAT_WATCH_EME_MOBILE": "DC_WATCH_EME_MOBILE", "QH_DAT_WATCH_ERROR_LOG": "DC_WATCH_ERROR_LOG", "QH_DAT_WATCH_MQTT_LOG": "DC_WATCH_MQTT_LOG", "QH_DAT_WATCH_NET_TYPE_LOG":"DC_WATCH_NET_TYPE_LOG", "QH_DAT_WATCH_NOTIFY": "DC_WATCH_NOTIFY", "QH_DAT_WATCH_NOTIFY_LOG": "DC_WATCH_NOTIFY_LOG", "QH_DAT_WATCH_PERIOD": "DC_WATCH_PERIOD", "QH_DAT_WATCH_PERIOD_PRI": "DC_WATCH_PERIOD_PERSONAL", "QH_DAT_WATCH_SET_INFO": "DC_WATCH_SET_INFO", "QH_DAT_WATCH_STAT_ALL": "DC_WATCH_STAT_ALL", "QH_DAT_WATCH_STAT_MAX_ID": "DC_WATCH_STAT_MAX_ID", "QH_DAT_WATCH_STAT_TODAY": "DC_WATCH_STAT_TODAY", "QH_DAT_WATCH_TASKS": "DC_WATCH_TASK", "QH_DAT_WATCH_TASKS_CYCLE": "DC_WATCH_TASK_CYCLE", "QH_DAT_WATCH_TASKS_PRI": "DC_WATCH_TASK_PERSONAL", "QH_DAT_WATCH_WARN_FLAG": "DC_WATCH_WARN_FLAG", "QH_DAT_WATCH_WARN_PRI": "DC_WATCH_WARN_PERSONAL", } # ==================================================== def get_connection(): """建立达梦数据库连接""" return dmPython.connect( host=DB_CONFIG["host"], port=DB_CONFIG["port"], user=DB_CONFIG["user"], password=DB_CONFIG["password"], ) def get_all_tables(cursor): """获取指定模式下以 TABLE_PREFIX 开头的所有表名及表注释""" cursor.execute( """ SELECT t.TABLE_NAME, tc.COMMENTS FROM ALL_TABLES t LEFT JOIN ALL_TAB_COMMENTS tc ON tc.TABLE_NAME = t.TABLE_NAME AND tc.OWNER = t.OWNER WHERE t.OWNER = :1 AND t.TABLE_NAME LIKE :2 ORDER BY t.TABLE_NAME """, (SCHEMA, TABLE_PREFIX + "%") ) return [(name, comment) for name, comment in cursor.fetchall() if name not in TABLE_EXCLUDE] def get_columns_with_comments(cursor, table_name): """获取表的字段名和中文注释，按字段顺序排列""" cursor.execute( """ SELECT c.COLUMN_NAME, cc.COMMENTS FROM ALL_TAB_COLUMNS c LEFT JOIN ALL_COL_COMMENTS cc ON cc.TABLE_NAME = c.TABLE_NAME AND cc.COLUMN_NAME = c.COLUMN_NAME AND cc.SCHEMA_NAME = :1 WHERE c.OWNER = :2 AND c.TABLE_NAME = :3 ORDER BY c.COLUMN_ID """, (SCHEMA, SCHEMA, table_name) ) return cursor.fetchall() # [(column_name, comment), ...] def get_sample_rows(cursor, table_name, limit=3): """获取表的示例数据""" try: cursor.execute(f'SELECT * FROM "{SCHEMA}"."{table_name}" LIMIT {limit}') return cursor.fetchall() except Exception: return [] def make_header_style(): """生成表头样式""" fill = PatternFill(fill_type="solid", fgColor="4472C4") font = Font(bold=True, color="FFFFFF", size=11) alignment = Alignment(horizontal="center", vertical="center", wrap_text=True) thin = Side(style="thin", color="FFFFFF") border = Border(left=thin, right=thin, top=thin, bottom=thin) return fill, font, alignment, border def make_field_name_style(): """字段名列样式（浅蓝底）""" fill = PatternFill(fill_type="solid", fgColor="DCE6F1") font = Font(bold=True, size=10) alignment = Alignment(horizontal="left", vertical="center") return fill, font, alignment def make_table_title_style(): """表名标题行样式（深绿底白字）""" fill = PatternFill(fill_type="solid", fgColor="375623") font = Font(bold=True, color="FFFFFF", size=12) alignment = Alignment(horizontal="left", vertical="center") return fill, font, alignment def write_all_tables_to_sheet(ws, tables_data): """将所有表写入同一个 Sheet，每张表前有标题行""" header_fill, header_font, header_align, header_border = make_header_style() field_fill, field_font, field_align = make_field_name_style() title_fill, title_font, title_align = make_table_title_style() current_row = 1 for table_name, table_comment, columns, sample_rows in tables_data: # 优先使用原始表名显示，找不到映射则用当前名 display_name = TABLE_NAME_MAPPING.get(table_name.upper(), table_name) # ---------- 第1行：表名 + 表注释（合并A-E列） ---------- title_text = f"{display_name} {table_comment or ''}" ws.merge_cells( start_row=current_row, start_column=1, end_row=current_row, end_column=5 ) title_cell = ws.cell(row=current_row, column=1, value=title_text) title_cell.fill = title_fill title_cell.font = title_font title_cell.alignment = title_align ws.row_dimensions[current_row].height = 22 current_row += 1 # ---------- 第2行：列头 ---------- headers = ["数据名", "中文注释", "示例数据1", "示例数据2", "示例数据3"] for col_idx, header in enumerate(headers, start=1): cell = ws.cell(row=current_row, column=col_idx, value=header) cell.fill = header_fill cell.font = header_font cell.alignment = header_align cell.border = header_border ws.row_dimensions[current_row].height = 18 current_row += 1 # ---------- 字段数据行 ---------- for field_index, (col_name, comment) in enumerate(columns): cell_a = ws.cell(row=current_row, column=1, value=col_name) cell_a.fill = field_fill cell_a.font = field_font cell_a.alignment = field_align ws.cell(row=current_row, column=2, value=comment or "") for sample_col, sample_row in enumerate(sample_rows, start=3): val = sample_row[field_index] if field_index < len(sample_row) else "" ws.cell(row=current_row, column=sample_col, value=str(val) if val is not None else "NULL") current_row += 1 # 表与表之间空一行 current_row += 1 # ---------- 统一设置列宽 ---------- ws.column_dimensions["A"].width = 28 ws.column_dimensions["B"].width = 32 ws.column_dimensions["C"].width = 30 ws.column_dimensions["D"].width = 30 ws.column_dimensions["E"].width = 30 def main(): print(f"正在连接达梦数据库 {DB_CONFIG['host']}:{DB_CONFIG['port']} ...") conn = get_connection() cursor = conn.cursor() print(f"获取模式 {SCHEMA} 下以 '{TABLE_PREFIX}' 开头的表...") tables = get_all_tables(cursor) print(f"共找到 {len(tables)} 张表") # 预先收集所有表的数据 tables_data = [] for i, (table_name, table_comment) in enumerate(tables, start=1): print(f"[{i}/{len(tables)}] 处理表: {table_name}") columns = get_columns_with_comments(cursor, table_name) if not columns: print(f" ⚠ 字段为空: {table_name}") else: print(f" 字段数: {len(columns)}, 首字段: {columns[0]}") sample_rows = get_sample_rows(cursor, table_name) tables_data.append((table_name, table_comment, columns, sample_rows)) cursor.close() conn.close() wb = openpyxl.Workbook() ws = wb.active ws.title = "表结构数据" write_all_tables_to_sheet(ws, tables_data) wb.save(OUTPUT_FILE) print(f"\n导出完成 -> {OUTPUT_FILE}") if __name__ == "__main__": main()