如何拆解并重构Python代码修复DataFrame.append弃用问题

1. 替换已弃用的DataFrame.append方法 #

Pandas 2.0+ 彻底移除了append方法，这就是报错的直接原因。最优替代方案是先把所有数据字典收集到列表，最后一次性转换为DataFrame——这种方式比多次调用append效率高得多（因为append每次都会创建新的DataFrame并复制全部数据，而列表收集是分摊O(1)的轻量操作）。

错误写法（旧代码） #

df = pd.DataFrame()
for _ in range(100):
    contact_dict = generate_contact()
    df = df.append(contact_dict, ignore_index=True)

正确写法（重构后） #

# 先收集所有数据到列表
contacts_list = []
for _ in range(100):
    contacts_list.append(generate_contact())

# 一次性转换为DataFrame
df_contacts = pd.DataFrame(contacts_list)

如果是合并多个已存在的DataFrame，可用pd.concat([df1, df2, df3], ignore_index=True)，但生成虚拟数据的场景优先用列表收集后转DataFrame的方式。

2. 代码拆解为单一职责模块 #

把152行的大代码拆分成多个小函数，每个函数只负责1-2个核心功能，让代码更易维护、测试和扩展。以下是具体的拆分结构：

拆分思路 #

• 工具函数：生成基础原子数据（邮箱、电话、公司名等）
• 单实体生成函数：组装单个联系人/公司/交易的完整数据字典
• 批量生成函数：循环调用单实体函数，生成指定数量的数据集列表
• 整合关联函数：处理实体间的关联（比如联系人绑定公司ID），并转换为DataFrame
• 输出函数（可选）：将DataFrame保存为CSV/Excel等格式

重构后的完整代码示例 #

import pandas as pd
import random
import string
from uuid import uuid4

# ------------------------------
# 工具函数：生成基础原子数据
# ------------------------------
def generate_random_email(domain: str = "example.com") -> str:
    username = ''.join(random.choices(string.ascii_lowercase + string.digits, k=8))
    return f"{username}@{domain}"

def generate_random_phone() -> str:
    area_code = random.randint(100, 999)
    prefix = random.randint(100, 999)
    line_number = random.randint(1000, 9999)
    return f"({area_code}) {prefix}-{line_number}"

def generate_random_company_name() -> str:
    prefixes = ["Tech", "Global", "Innovative", "Prime", "Alpha"]
    suffixes = ["Solutions", "Corp", "Group", "Labs", "Enterprises"]
    return f"{random.choice(prefixes)} {random.choice(suffixes)}"

# ------------------------------
# 单实体生成函数：组装单个数据字典
# ------------------------------
def generate_single_contact() -> dict:
    return {
        "contact_id": str(uuid4()),
        "first_name": random.choice(["John", "Jane", "Mike", "Sarah", "David"]),
        "last_name": random.choice(["Doe", "Smith", "Johnson", "Williams", "Brown"]),
        "email": generate_random_email(),
        "phone": generate_random_phone(),
        "company_id": None  # 后续关联公司时填充
    }

def generate_single_company() -> dict:
    return {
        "company_id": str(uuid4()),
        "name": generate_random_company_name(),
        "industry": random.choice(["Technology", "Healthcare", "Finance", "Retail", "Manufacturing"]),
        "employee_count": random.randint(10, 1000)
    }

def generate_single_deal(contact_ids: list[str], company_ids: list[str]) -> dict:
    return {
        "deal_id": str(uuid4()),
        "contact_id": random.choice(contact_ids),
        "company_id": random.choice(company_ids),
        "deal_name": f"Deal {random.randint(1000, 9999)}",
        "amount": round(random.uniform(1000, 100000), 2),
        "stage": random.choice(["Appointment Scheduled", "Qualified to Buy", "Proposal Sent", "Closed Won", "Closed Lost"])
    }

# ------------------------------
# 批量生成函数：生成指定数量的数据集列表
# ------------------------------
def generate_contacts_batch(num_records: int) -> list[dict]:
    return [generate_single_contact() for _ in range(num_records)]

def generate_companies_batch(num_records: int) -> list[dict]:
    return [generate_single_company() for _ in range(num_records)]

def generate_deals_batch(num_records: int, contact_ids: list[str], company_ids: list[str]) -> list[dict]:
    return [generate_single_deal(contact_ids, company_ids) for _ in range(num_records)]

# ------------------------------
# 整合关联函数：处理关联并转换为DataFrame
# ------------------------------
def create_hubspot_datasets(num_contacts: int = 100, num_companies: int = 50, num_deals: int = 75) -> dict[str, pd.DataFrame]:
    # 生成公司数据集
    companies = generate_companies_batch(num_companies)
    df_companies = pd.DataFrame(companies)
    
    # 生成联系人并关联公司ID
    contacts = generate_contacts_batch(num_contacts)
    company_ids = df_companies["company_id"].tolist()
    for contact in contacts:
        contact["company_id"] = random.choice(company_ids)
    df_contacts = pd.DataFrame(contacts)
    
    # 生成交易并关联联系人和公司ID
    contact_ids = df_contacts["contact_id"].tolist()
    deals = generate_deals_batch(num_deals, contact_ids, company_ids)
    df_deals = pd.DataFrame(deals)
    
    return {
        "contacts": df_contacts,
        "companies": df_companies,
        "deals": df_deals
    }

# ------------------------------
# 使用示例
# ------------------------------
if __name__ == "__main__":
    # 生成自定义数量的HubSpot虚拟数据
    hubspot_data = create_hubspot_datasets(num_contacts=200, num_companies=100, num_deals=150)
    
    # 查看联系人数据前5行
    print("Contacts Sample:")
    print(hubspot_data["contacts"].head())
    
    # 保存到CSV（可选）
    # hubspot_data["contacts"].to_csv("hubspot_contacts.csv", index=False)

拆分优势 #

• 每个函数职责单一，比如修改邮箱生成规则时，只需改动generate_random_email，不影响其他代码
• 函数名直接体现功能，代码可读性大幅提升
• 便于单元测试，比如单独验证generate_single_contact是否生成符合要求的字典
• 扩展方便，新增“任务”“票据”等实体时，只需添加对应的单实体和批量生成函数

内容的提问来源于stack exchange，提问作者Seattle Python Noobie