#!/usr/bin/env python3 """ Excel数据导入脚本 从docs/example.xls导入工程项目数据到数据库 """ import sys import xlrd import mysql.connector from mysql.connector import Error from datetime import datetime import os # 配置 EXCEL_FILE = '../docs/example.xls' DB_CONFIG = { 'host': 'localhost', 'user': 'root', 'password': 'rootpassword', 'database': 'project_manager', 'charset': 'utf8mb4' } def get_database_connection(): """获取数据库连接""" try: connection = mysql.connector.connect(**DB_CONFIG) return connection except Error as e: print(f'数据库连接错误: {e}', file=sys.stderr) sys.exit(1) def parse_date_value(cell_value, workbook): """解析Excel日期值""" if cell_value == '': return None # 检查是否是日期类型 try: date_tuple = xlrd.xldate_as_tuple(cell_value, workbook.datemode) return f'{date_tuple[0]}-{date_tuple[1]:02d}-{date_tuple[2]:02d}' except: pass # 检查特殊值 if cell_value in ['未到期', '可退质保金', '未开工']: return None # 尝试解析字符串日期 try: date_obj = datetime.strptime(str(cell_value), '%Y-%m-%d') return date_obj.strftime('%Y-%m-%d') except: pass return None def parse_number_value(cell_value): """解析数字值""" if cell_value == '': return None try: return float(cell_value) except: return None def parse_int_value(cell_value): """解析整数值""" num = parse_number_value(cell_value) if num is not None: return int(num) return None def parse_enum_value(cell_value, allowed_values): """解析枚举值""" if cell_value == '': return None if cell_value in allowed_values: return cell_value return None def clean_text_value(cell_value): """清理文本值""" if cell_value == '': return None return str(cell_value).strip() def import_projects_from_sheet(sheet, connection, cursor, sheet_name): """从工作表导入项目数据""" print(f'\n正在导入工作表: {sheet_name}') print(f'总行数: {sheet.nrows}') # 获取管理员用户ID（用于created_by字段） cursor.execute("SELECT id FROM users WHERE username='admin' LIMIT 1") result = cursor.fetchone() if not result: print('错误: 未找到管理员用户', file=sys.stderr) return False admin_id = result[0] success_count = 0 skip_count = 0 error_count = 0 # 从第6行开始（前5行是表头） for row_idx in range(5, sheet.nrows): try: # 读取数据 row_data = [sheet.cell_value(row_idx, col_idx) for col_idx in range(min(61, sheet.ncols))] # 检查是否是合计行或空行 if row_data[0] == '' or str(row_data[0]).strip() == '': skip_count += 1 continue # 准备SQL插入语句 sql = """ INSERT INTO projects ( project_no, power_contract_no, name, subitem_count, subitem_code, total_investment, contract_amount, warranty_ratio, settlement_amount, total_cost_estimated, voltage_level, engineering_type, owner_unit, owner_contact, bidding_type, signing_date, start_date, planned_end_date, actual_end_date, warranty_amount, warranty_expiry_date, actual_warranty_refund_date, project_department, project_leader, payment_method, total_cost_control, is_adjusted, labor_cost_control, labor_cost_planned, labor_cost_paid, material_cost_control, material_cost_payable, material_cost_actual, material_cost_paid, other_cost_control, other_cost_payable, other_cost_actual, tax_amount, profit, actual_profit, cost_settlement_amount, cumulative_progress, receivable_amount, invoice_amount, actual_receipt_amount, receipt_completion_rate, payable_amount, actual_payment_amount, unpaid_amount, payment_completion_rate, labor_debt_amount, settlement_cost_amount, settlement_labor_cost, settlement_material_cost, settlement_other_cost, due_settlement_count, unsettlement_count, problems, suggestions, remarks, status, created_by ) 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 ) """ # 准备参数 params = ( clean_text_value(row_data[1]), # project_no clean_text_value(row_data[2]), # power_contract_no clean_text_value(row_data[3]), # name parse_int_value(row_data[4]), # subitem_count clean_text_value(row_data[5]), # subitem_code parse_number_value(row_data[6]), # total_investment parse_number_value(row_data[7]), # contract_amount parse_number_value(row_data[8]), # warranty_ratio parse_number_value(row_data[9]), # settlement_amount parse_number_value(row_data[10]), # total_cost_estimated clean_text_value(row_data[11]), # voltage_level clean_text_value(row_data[12]), # engineering_type clean_text_value(row_data[13]), # owner_unit clean_text_value(row_data[14]), # owner_contact clean_text_value(row_data[15]), # bidding_type parse_date_value(row_data[16], connection._connection.workbook), # signing_date parse_date_value(row_data[17], connection._connection.workbook), # start_date parse_date_value(row_data[18], connection._connection.workbook), # planned_end_date parse_date_value(row_data[19], connection._connection.workbook), # actual_end_date parse_number_value(row_data[20]), # warranty_amount parse_date_value(row_data[21], connection._connection.workbook), # warranty_expiry_date parse_date_value(row_data[22], connection._connection.workbook), # actual_warranty_refund_date clean_text_value(row_data[23]), # project_department clean_text_value(row_data[24]), # project_leader clean_text_value(row_data[25]), # payment_method parse_number_value(row_data[26]), # total_cost_control parse_enum_value(row_data[27], ['是', '否']), # is_adjusted parse_number_value(row_data[28]), # labor_cost_control parse_number_value(row_data[29]), # labor_cost_planned parse_number_value(row_data[30]), # labor_cost_paid parse_number_value(row_data[31]), # material_cost_control parse_number_value(row_data[32]), # material_cost_payable parse_number_value(row_data[33]), # material_cost_actual parse_number_value(row_data[34]), # material_cost_paid parse_number_value(row_data[35]), # other_cost_control parse_number_value(row_data[36]), # other_cost_payable parse_number_value(row_data[37]), # other_cost_actual parse_number_value(row_data[38]), # tax_amount parse_number_value(row_data[39]), # profit parse_number_value(row_data[40]), # actual_profit parse_number_value(row_data[41]), # cost_settlement_amount parse_number_value(row_data[42]), # cumulative_progress parse_number_value(row_data[43]), # receivable_amount parse_number_value(row_data[44]), # invoice_amount parse_number_value(row_data[45]), # actual_receipt_amount parse_number_value(row_data[46]), # receipt_completion_rate parse_number_value(row_data[47]), # payable_amount parse_number_value(row_data[48]), # actual_payment_amount parse_number_value(row_data[49]), # unpaid_amount parse_number_value(row_data[50]), # payment_completion_rate parse_number_value(row_data[51]), # labor_debt_amount parse_number_value(row_data[52]), # settlement_cost_amount parse_number_value(row_data[53]), # settlement_labor_cost parse_number_value(row_data[54]), # settlement_material_cost parse_number_value(row_data[55]), # settlement_other_cost parse_int_value(row_data[56]), # due_settlement_count parse_int_value(row_data[57]), # unsettlement_count clean_text_value(row_data[58]), # problems clean_text_value(row_data[59]), # suggestions clean_text_value(row_data[60]), # remarks '新建', # status (默认新建) admin_id # created_by ) # 跳过没有项目名称的行 if not params[2]: skip_count += 1 continue cursor.execute(sql, params) success_count += 1 except Error as e: print(f'第{row_idx+1}行导入错误: {e}', file=sys.stderr) error_count += 1 continue print(f'导入完成: 成功 {success_count} 条, 跳过 {skip_count} 条, 错误 {error_count} 条') return True def main(): """主函数""" print('='*60) print('工程项目数据导入脚本') print('='*60) # 检查Excel文件是否存在 if not os.path.exists(EXCEL_FILE): print(f'错误: Excel文件不存在: {EXCEL_FILE}', file=sys.stderr) sys.exit(1) # 打开Excel文件 try: workbook = xlrd.open_workbook(EXCEL_FILE) print(f'已打开Excel文件: {EXCEL_FILE}') print(f'工作表数量: {len(workbook.sheet_names())}') except Exception as e: print(f'错误: 无法打开Excel文件: {e}', file=sys.stderr) sys.exit(1) # 获取数据库连接 connection = get_database_connection() cursor = connection.cursor() try: # 导入主要工作表 main_sheets = [ '1-1基建', '1-2业扩', '1-3客户', '1-4营销', '2检修、技改、应急抢修项目' ] for sheet_name in main_sheets: if sheet_name in workbook.sheet_names(): sheet = workbook.sheet_by_name(sheet_name) import_projects_from_sheet(sheet, connection, cursor, sheet_name) else: print(f'警告: 工作表不存在: {sheet_name}') # 提交事务 connection.commit() # 显示统计信息 cursor.execute("SELECT COUNT(*) FROM projects") total_count = cursor.fetchone()[0] print(f'\n数据库中共有 {total_count} 个项目') cursor.execute("SELECT engineering_type, COUNT(*) as count FROM projects GROUP BY engineering_type") print('\n按工程类别统计:') for row in cursor.fetchall(): print(f' {row[0]}: {row[1]} 个') print('\n' + '='*60) print('数据导入完成！') print('='*60) except Exception as e: print(f'错误: {e}', file=sys.stderr) connection.rollback() sys.exit(1) finally: cursor.close() connection.close() if __name__ == '__main__': main()