Python脚本如何批量适配字段结构变更

wen python案例 2026-06-25 252

本文目录导读：

核心工具与思路
常见场景解决方案
高级方案：使用适配器模式
实战：数据库字段变更自动适配
最佳实践与注意事项
完整示例：通用适配器

针对Python批量适配字段结构变更，核心思路是抽象变更逻辑 + 自动化执行,以下是几种常见场景及对应的解决方案：

核心工具与思路

统一数据模型：定义标准的字段映射关系
适配器模式：为不同版本的数据结构编写适配器
自动化脚本：批量读取、转换、写入数据

常见场景解决方案

JSON/字典字段名变更

import json
# 字段映射字典（旧字段名 -> 新字段名）
FIELD_MAPPING = {
    "user_name": "username",
    "user_age": "age",
    "phone_number": "phone"
}
def adapt_dict_fields(data):
    """适配字典中的字段名"""
    new_data = {}
    for old_key, value in data.items():
        # 获取新字段名（如果没有映射则保留原字段名）
        new_key = FIELD_MAPPING.get(old_key, old_key)
        new_data[new_key] = value
    return new_data
def batch_adapt_json_files(file_list):
    """批量处理JSON文件"""
    for file_path in file_list:
        with open(file_path, 'r', encoding='utf-8') as f:
            data = json.load(f)
        # 处理单个对象或对象列表
        if isinstance(data, list):
            adapted_data = [adapt_dict_fields(item) for item in data]
        else:
            adapted_data = adapt_dict_fields(data)
        # 保存
        with open(file_path, 'w', encoding='utf-8') as f:
            json.dump(adapted_data, f, ensure_ascii=False, indent=2)
        print(f"已适配：{file_path}")

CSV/Excel表头变更

import pandas as pd
def adapt_csv_headers(file_path, field_mapping):
    """适配CSV文件表头"""
    df = pd.read_csv(file_path)
    df.rename(columns=field_mapping, inplace=True)
    df.to_csv(file_path, index=False)
    print(f"已适配：{file_path}")
# 批量处理
def batch_adapt_csv(file_list, field_mapping):
    for file_path in file_list:
        adapt_csv_headers(file_path, field_mapping)

数据库表结构变更

import sqlite3
def adapt_database_table(db_path, table_name, column_mapping):
    """适配数据库表字段变更"""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    # 1. 创建新表
    # 假设从旧表 user_info 迁移到新表 users
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY,
            name TEXT NOT NULL,
            email TEXT
        )
    """)
    # 2. 迁移数据（如果有旧表）
    try:
        cursor.execute("""
            INSERT INTO users (id, name, email)
            SELECT id, user_name, email_address
            FROM user_info
        """)
        conn.commit()
        print(f"数据迁移完成：{table_name}")
    except Exception as e:
        print(f"迁移失败：{e}")
        conn.rollback()
    conn.close()

复杂嵌套结构变更

def adapt_nested_structure(data):
    """适配嵌套的复杂数据结构"""
    if isinstance(data, dict):
        new_data = {}
        for key, value in data.items():
            # 处理字段拆分
            if key == "full_name":
                parts = value.split(" ", 1)
                new_data["first_name"] = parts[0]
                new_data["last_name"] = parts[1] if len(parts) > 1 else ""
            # 处理字段合并
            elif key in ["city", "province"]:
                if "address" not in new_data:
                    new_data["address"] = {}
                new_data["address"][key] = value
            # 递归处理
            else:
                new_data[key] = adapt_nested_structure(value)
        return new_data
    elif isinstance(data, list):
        return [adapt_nested_structure(item) for item in data]
    else:
        return data

高级方案：使用适配器模式

from abc import ABC, abstractmethod
class DataAdapter(ABC):
    """数据适配器基类"""
    @abstractmethod
    def adapt(self, data):
        pass
class UserDataAdapter(DataAdapter):
    """用户数据适配器"""
    def __init__(self, version="v2"):
        self.version = version
    def adapt(self, data):
        if self.version == "v2":
            return self._adapt_v1_to_v2(data)
        elif self.version == "v3":
            return self._adapt_v2_to_v3(data)
    def _adapt_v1_to_v2(self, data):
        # v1: {user_name, user_age, phone}
        # v2: {name, age, phone}
        return {
            "name": data.get("user_name"),
            "age": data.get("user_age"),
            "phone": data.get("phone"),
            "email": data.get("email", "")  # v2新增字段
        }
    def _adapt_v2_to_v3(self, data):
        # v2: {name, age, phone, email}
        # v3: {full_name, age, contact_info}
        return {
            "full_name": f"{data['name']}",
            "age": data["age"],
            "contact_info": {
                "phone": data["phone"],
                "email": data["email"]
            }
        }
# 使用示例
def batch_adapt_with_version(adapter, data_list):
    return [adapter.adapt(item) for item in data_list]

实战：数据库字段变更自动适配

import sqlalchemy
from sqlalchemy import inspect, text
class DBFieldAdapter:
    """数据库字段变更适配器"""
    def __init__(self, db_url):
        self.engine = sqlalchemy.create_engine(db_url)
        self.inspector = inspect(self.engine)
    def get_field_changes(self, table_name):
        """获取字段变更"""
        old_columns = self.inspector.get_columns(table_name + "_v1")
        new_columns = self.inspector.get_columns(table_name + "_v2")
        return self._compare_columns(old_columns, new_columns)
    def _compare_columns(self, old_cols, new_cols):
        """比较新旧字段"""
        old_col_names = {col['name'] for col in old_cols}
        new_col_names = {col['name'] for col in new_cols}
        return {
            "added": new_col_names - old_col_names,
            "removed": old_col_names - new_col_names,
            "modified": self._get_modified(old_cols, new_cols)
        }
    def auto_migrate(self, old_table, new_table, field_mapping):
        """自动迁移数据"""
        with self.engine.connect() as conn:
            # 生成迁移SQL
            old_fields = ", ".join(field_mapping.keys())
            new_fields = ", ".join(field_mapping.values())
            sql = f"""
                INSERT INTO {new_table} ({new_fields})
                SELECT {old_fields} FROM {old_table}
            """
            conn.execute(text(sql))
            conn.commit()

最佳实践与注意事项

✅ 推荐的做法：

版本控制：为数据添加版本号，便于区分
备份原始数据：适配前先备份
日志记录：记录每次变更的详细信息
增量处理：支持断点续传
验证机制：适配后验证数据完整性

⚠️ 需要避免的：

直接修改生产数据
忽略字段类型转换
没有异常处理机制
硬编码字段名
一次性处理大量数据（建议分批）

完整示例：通用适配器

import json
import os
from typing import Dict, List, Any
import logging
logging.basicConfig(level=logging.INFO)
class UniversalFieldAdapter:
    """通用字段适配器"""
    def __init__(self, mapping_file: str = None):
        self.mappings = {}
        if mapping_file:
            self.load_mappings(mapping_file)
    def load_mappings(self, file_path: str):
        """从JSON文件加载字段映射"""
        with open(file_path, 'r', encoding='utf-8') as f:
            self.mappings = json.load(f)
    def add_mapping(self, data_type: str, field_map: Dict[str, str]):
        """添加字段映射"""
        self.mappings[data_type] = field_map
    def adapt(self, data: Any, data_type: str = None) -> Any:
        """适配数据"""
        if data_type not in self.mappings:
            return data
        field_map = self.mappings[data_type]
        if isinstance(data, dict):
            return self._adapt_dict(data, field_map)
        elif isinstance(data, list):
            return [self._adapt_dict(item, field_map) for item in data]
        else:
            logging.warning(f"不支持的数据类型: {type(data)}")
            return data
    def _adapt_dict(self, data: Dict, field_map: Dict[str, str]) -> Dict:
        """适配字典"""
        adapted = {}
        for key, value in data.items():
            new_key = field_map.get(key, key)
            adapted[new_key] = value
        return adapted
    def batch_adapt_files(self, directory: str, pattern: str = "*.json"):
        """批量适配文件"""
        import glob
        files = glob.glob(os.path.join(directory, pattern))
        for file_path in files:
            try:
                with open(file_path, 'r', encoding='utf-8') as f:
                    data = json.load(f)
                # 自动检测数据类型（基于文件名或内容）
                data_type = self._detect_data_type(file_path, data)
                if data_type:
                    adapted_data = self.adapt(data, data_type)
                    # 备份原始文件
                    backup_file = file_path + ".bak"
                    os.rename(file_path, backup_file)
                    # 保存适配后的数据
                    with open(file_path, 'w', encoding='utf-8') as f:
                        json.dump(adapted_data, f, ensure_ascii=False, indent=2)
                    logging.info(f"适配完成: {file_path}")
                else:
                    logging.warning(f"未找到映射规则: {file_path}")
            except Exception as e:
                logging.error(f"适配失败 {file_path}: {str(e)}")
# 使用示例
if __name__ == "__main__":
    adapter = UniversalFieldAdapter()
    # 添加字段映射
    adapter.add_mapping("user_data", {
        "user_name": "name",
        "user_age": "age",
        "phone_number": "phone"
    })
    # 批量适配JSON文件
    adapter.batch_adapt_files("./data", "*.json")

批量适配字段结构变更的关键在于：