291 lines
12 KiB
Python
291 lines
12 KiB
Python
#!/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()
|