lianlonggang/rkutil
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

添加数据导出脚本：导出达梦数据库和MySQL数据到Excel

Browse Source 
- export_dm_to_excel.py: 达梦数据库数据导出到Excel
- export_mysql_to_excel.py: MySQL数据导出到Excel

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
lianlonggang
2026-06-03 09:47:07 +08:00
co-authored by Claude Opus 4.7
commit cceaa62d7c
2 changed files with 538 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
export_dm_to_excel.py
+269
View File
@@ -0,0 +1,269 @@
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()
export_mysql_to_excel.py
+269
View File
@@ -0,0 +1,269 @@
import pymysql
import openpyxl
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
# ==================== 数据库配置 ====================
DB_CONFIG = {
"host": "192.168.10.100",
"port": 3308,
"user": "root",
"password": "Aa135790123",
"database": "cqyt_new_prod",
"charset": "utf8mb4",
}
OUTPUT_FILE = "export_medical_to_excel.xlsx"
# 需要导出的表名列表（来自 mysql表名.txt）
TARGET_TABLES = [
"medical_announcement_info",
"medical_banner",
"medical_check_package",
"medical_check_package_item",
"medical_conclusion",
"medical_defend_cancer_code",
"medical_depart_hospital",
"medical_diseases_dic",
"medical_diseases_dic_rel",
"medical_doctor_check_item",
"medical_doctor_sign",
"medical_emp_defend_cancer",
"medical_emp_occup_import",
"medical_emp_occupational",
"medical_health_check_statistics",
"medical_health_check_statistics_info",
"medical_hospital",
"medical_hospital_area",
"medical_hospital_doctor_check",
"medical_hospital_emp",
"medical_hospital_except_count",
"medical_hospital_except_date",
"medical_hospital_item",
"medical_hospital_item_rel",
"medical_hospital_user",
"medical_hospital_virtual",
"medical_info",
"medical_item",
"medical_item_cost",
"medical_item_mutex",
"medical_item_package",
"medical_item_package_model",
"medical_item_package_model_append",
"medical_item_package_model_append_sub",
"medical_item_package_model_sub",
"medical_item_package_sub",
"medical_item_syn",
"medical_limit_float",
"medical_limit_model",
"medical_model_float",
"medical_order_info",
"medical_order_queue",
"medical_package_check",
"medical_paper_question",
"medical_paper_question_item",
"medical_plan",
"medical_plan_user",
"medical_precheck_paper",
"medical_precheck_paper_questions",
"medical_record_model",
"medical_record_model_base",
"medical_record_model_config",
"medical_serious_result_item",
"medical_stat_year",
"medical_uni_item_class",
"medical_uni_item_class_office",
"medical_uni_item_class_sicks",
"medical_uni_item_hospital",
"medical_uni_item_info",
"medical_uni_item_risk_rel",
"medical_uni_item_unusual_rel",
"medical_user_defend_cancer_result",
"medical_user_diseases",
"medical_user_diseases_item",
"medical_user_exception",
"medical_user_exception_item",
"medical_user_form",
"medical_user_form_sub",
"medical_user_ill",
"medical_user_ill_sub",
"medical_user_image",
"medical_user_image_item",
"medical_user_limit",
"medical_user_paper",
"medical_user_paper_answer",
"medical_user_result",
"medical_user_result_ext",
"medical_user_result_item",
"medical_user_result_item_tmp_unmatch",
"medical_user_sign",
]
# ====================================================
def get_connection():
"""建立 MySQL 数据库连接"""
return pymysql.connect(
host=DB_CONFIG["host"],
port=DB_CONFIG["port"],
user=DB_CONFIG["user"],
password=DB_CONFIG["password"],
database=DB_CONFIG["database"],
charset=DB_CONFIG["charset"],
)
def get_table_comment(cursor, table_name):
"""获取表的中文注释"""
cursor.execute(
"""
SELECT TABLE_COMMENT
FROM information_schema.TABLES
WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s
""",
(DB_CONFIG["database"], table_name)
)
row = cursor.fetchone()
return row[0] if row else ""
def get_columns_with_comments(cursor, table_name):
"""获取表的字段名和中文注释，按字段顺序排列"""
cursor.execute(
"""
SELECT COLUMN_NAME, COLUMN_COMMENT
FROM information_schema.COLUMNS
WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s
ORDER BY ORDINAL_POSITION
""",
(DB_CONFIG["database"], table_name)
)
return cursor.fetchall() # [(column_name, comment), ...]
def get_sample_rows(cursor, table_name, limit=3):
"""获取表的示例数据"""
try:
cursor.execute(f"SELECT * FROM `{table_name}` LIMIT %s", (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:
# ---------- 第1行：表名 + 表注释（合并A-E列） ----------
title_text = f"{table_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"正在连接 MySQL {DB_CONFIG['host']}:{DB_CONFIG['port']} ...")
conn = get_connection()
cursor = conn.cursor()
print(f"共 {len(TARGET_TABLES)} 张表待处理")
tables_data = []
for i, table_name in enumerate(TARGET_TABLES, start=1):
print(f"[{i}/{len(TARGET_TABLES)}] 处理表: {table_name}")
table_comment = get_table_comment(cursor, table_name)
columns = get_columns_with_comments(cursor, table_name)
if not columns:
print(f" ⚠ 字段为空或表不存在: {table_name}")
else:
print(f" 字段数: {len(columns)}, 表注释: {table_comment or '（无）'}")
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()