在Peewee中处理PostgreSQL记录重复：一站式数据摄取教程

import peewee

# 假设 BaseModel 包含了数据库连接配置
class BaseModel(peewee.Model):
    class Meta:
        database = peewee.PostgresDatabase('your_db', user='your_user', password='your_password', host='localhost', port=5432)

class Device(BaseModel):
    id = peewee.AutoField()
    md5 = peewee.FixedCharField(32) # 文件哈希，用于标识设备
    # ... 其他设备属性

class Message(BaseModel):
    id = peewee.AutoField()
    dev_ref = peewee.ForeignKeyField(Device, backref="messages") # 外键关联Device
    # ... 其他消息属性

# 连接数据库
BaseModel._meta.database.connect()

import pandas as pd
import hashlib

def calculate_file_hash(file_path: str) -> str:
    """计算文件的MD5哈希值"""
    with open(file_path, 'rb') as f:
        return hashlib.md5(f.read()).hexdigest()

def sheet_to_model(
    source_file_path: str,
    sheet_name: str,
    model: peewee.Model):

    df = pd.read_excel(source_file_path, sheet_name=sheet_name)
    file_hash = calculate_file_hash(source_file_path)

    # 潜在问题区域：Device 记录的“获取或创建”逻辑
    try:
        device = Device.select().where(Device.md5 == file_hash).get()
    except peewee.DoesNotExist: # 明确捕获 Peewee 的 DoesNotExist 异常
        device = Device(md5=file_hash)
        device.save()
    except Exception as e:
        # 记录其他潜在错误，并重新抛出或适当处理
        print(f"Error getting/creating device for hash {file_hash}: {e}")
        raise

    # 遍历数据框行，创建子表记录
    for index, row in df.iterrows():
        attrs = { 'column_name' : row['excel_column_name'] } # 示例：从Excel行映射属性
        # 创建子表记录，并关联到获取到的设备ID
        model.create(dev_ref=device.id, **attrs)

import glob
import openpyxl

# 示例：工作表名称到 Peewee 模型的映射
sheet_model_map = {
    "Messages" : Message,
    "Files" : File,
    "User Accounts": UserAccounts
}

# 遍历指定目录下的所有Excel文件
for file_path in glob.glob("file/location/whatever/*.xlsx"):
    # 考虑在此处使用数据库事务，以确保文件级别的数据一致性
    # with BaseModel._meta.database.atomic():
    xl_file = openpyxl.load_workbook(file_path, read_only=True)
    for sheet_name in filter(lambda k: k in sheet_model_map, xl_file.sheetnames):
        sheet_to_model(file_path, sheet_name, sheet_model_map[sheet_name])

# 关闭数据库连接
BaseModel._meta.database.close()

class Device(BaseModel):
    id = peewee.AutoField()
    md5 = peewee.FixedCharField(32, unique=True) # 强制 md5 字段唯一
    # ... 其他设备属性

# 在 sheet_to_model 函数中替换现有的 try-except 块
def sheet_to_model(
    source_file_path: str,
    sheet_name: str,
    model: peewee.Model):

    df = pd.read_excel(source_file_path, sheet_name=sheet_name)
    file_hash = calculate_file_hash(source_file_path)

    # 使用 Peewee 的 get_or_create 方法
    # 如果 md5 存在，则返回现有 device 对象；否则创建新 device
    device, created = Device.get_or_create(md5=file_hash)

    if created:
        print(f"Created new device with md5: {file_hash}")
    else:
        print(f"Found existing device with md5: