Python脚本如何适配各类业务数据场景

wen python案例 2026-06-24 29

本文目录导读：

核心架构设计原则
配置驱动设计
插件化架构
数据转换管道
业务场景适配示例
统一错误处理
性能优化策略
测试和监控
最佳实践总结
关键原则

我来系统地介绍Python脚本适配各类业务数据场景的方法和最佳实践。

核心架构设计原则

1 分层设计

# 数据接入层 → 处理逻辑层 → 输出适配层
from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional
class DataProcessor(ABC):
    """数据处理器抽象基类"""
    @abstractmethod
    def load_data(self, source: Any) -> Any:
        """加载数据"""
        pass
    @abstractmethod
    def transform(self, data: Any) -> Any:
        """转换数据"""
        pass
    @abstractmethod
    def save_data(self, data: Any, target: Any) -> None:
        """保存数据"""
        pass
class BusinessProcessor(DataProcessor):
    """业务处理器实现"""
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.plugins = []

配置驱动设计

1 灵活的配置系统

import json
import yaml
from typing import Dict, Any
class ConfigManager:
    """配置管理器"""
    def __init__(self, config_path: str):
        self.config = self._load_config(config_path)
    def _load_config(self, path: str) -> Dict[str, Any]:
        """加载配置，支持多种格式"""
        if path.endswith(('.yaml', '.yml')):
            with open(path, 'r') as f:
                return yaml.safe_load(f)
        elif path.endswith('.json'):
            with open(path, 'r') as f:
                return json.load(f)
        else:
            raise ValueError(f"Unsupported config format: {path}")
    def get_pipeline_config(self, pipeline_name: str) -> Dict:
        """获取流水线配置"""
        return self.config.get('pipelines', {}).get(pipeline_name, {})

2 配置示例

# config.yaml
databases:
  mysql:
    host: localhost
    port: 3306
  mongodb:
    host: localhost
    port: 27017
pipelines:
  ecommerce:
    input: 
      type: api
      endpoint: /api/orders
      format: json
    processing:
      steps:
        - name: validate
          module: validators.order_validator
        - name: enrich
          module: enrichers.customer_enricher
    output:
      database: postgresql
      table: processed_orders

插件化架构

1 插件管理器

import importlib
from typing import Dict, Type
class PluginManager:
    """插件管理器"""
    def __init__(self):
        self.plugins: Dict[str, Type] = {}
    def register_plugin(self, name: str, module_path: str, class_name: str):
        """注册插件"""
        module = importlib.import_module(module_path)
        plugin_class = getattr(module, class_name)
        self.plugins[name] = plugin_class
    def get_plugin(self, name: str):
        """获取插件"""
        if name not in self.plugins:
            raise ValueError(f"Plugin '{name}' not registered")
        return self.plugins[name]

2 适配器模式示例

from abc import ABC, abstractmethod
class DataSourceAdapter(ABC):
    """数据源适配器基类"""
    @abstractmethod
    def connect(self):
        pass
    @abstractmethod
    def read(self, query: str) -> List[Dict]:
        pass
    @abstractmethod
    def write(self, data: List[Dict]) -> bool:
        pass
class MySQLAdapter(DataSourceAdapter):
    def __init__(self, config: Dict):
        self.config = config
        self.connection = None
    def connect(self):
        import pymysql
        self.connection = pymysql.connect(**self.config)
    def read(self, query: str) -> List[Dict]:
        with self.connection.cursor() as cursor:
            cursor.execute(query)
            return cursor.fetchall()
    def write(self, data: List[Dict]) -> bool:
        # 实现MySQL写入逻辑
        pass
class MongoDBAdapter(DataSourceAdapter):
    def __init__(self, config: Dict):
        self.config = config
        self.client = None
    def connect(self):
        from pymongo import MongoClient
        self.client = MongoClient(self.config['uri'])
    def read(self, query: str) -> List[Dict]:
        collection = self.client[self.config['db']][self.config['collection']]
        return list(collection.find(json.loads(query)))
    def write(self, data: List[Dict]) -> bool:
        # 实现MongoDB写入逻辑
        pass

数据转换管道

1 管道模式

class TransformPipeline:
    """数据转换管道"""
    def __init__(self):
        self.steps = []
    def add_step(self, func, **kwargs):
        """添加处理步骤"""
        self.steps.append({
            'func': func,
            'params': kwargs
        })
    def execute(self, data: Any) -> Any:
        """执行管道"""
        result = data
        for step in self.steps:
            try:
                result = step['func'](result, **step['params'])
            except Exception as e:
                raise PipelineError(f"Step failed: {step['func'].__name__}: {e}")
        return result
class PipelineError(Exception):
    pass

2 通用数据清洗器

import pandas as pd
import numpy as np
class DataCleaner:
    """数据清洗类"""
    @staticmethod
    def handle_missing_values(df: pd.DataFrame, strategy: str = 'drop') -> pd.DataFrame:
        """处理缺失值"""
        if strategy == 'drop':
            return df.dropna()
        elif strategy == 'fill_mean':
            return df.fillna(df.mean())
        elif strategy == 'fill_median':
            return df.fillna(df.median())
        elif strategy == 'forward_fill':
            return df.ffill()
        return df
    @staticmethod
    def standardize_formats(df: pd.DataFrame, column_mappings: Dict) -> pd.DataFrame:
        """标准化格式"""
        result = df.copy()
        for col, config in column_mappings.items():
            if col in result.columns:
                if config.get('type') == 'date':
                    result[col] = pd.to_datetime(result[col], 
                                                  format=config.get('format'))
                elif config.get('type') == 'number':
                    result[col] = pd.to_numeric(result[col], errors='coerce')
        return result

业务场景适配示例

1 电商数据处理

class ECommerceProcessor:
    """电商数据处理"""
    def process_orders(self, raw_orders: List[Dict]) -> Dict:
        """处理订单数据"""
        return {
            'total_revenue': sum(order['amount'] for order in raw_orders),
            'order_count': len(raw_orders),
            'top_products': self._get_top_products(raw_orders),
            'customer_metrics': self._calculate_customer_metrics(raw_orders)
        }
    def _get_top_products(self, orders: List[Dict]) -> List[Dict]:
        """获取热销产品"""
        from collections import Counter
        product_counter = Counter()
        for order in orders:
            for item in order['items']:
                product_counter[item['product_id']] += item['quantity']
        return product_counter.most_common(10)
    def _calculate_customer_metrics(self, orders: List[Dict]) -> Dict:
        """计算客户指标"""
        customer_data = {}
        for order in orders:
            customer_id = order['customer_id']
            if customer_id not in customer_data:
                customer_data[customer_id] = {
                    'total_spent': 0,
                    'order_count': 0,
                    'last_order_date': None
                }
            customer_data[customer_id]['total_spent'] += order['amount']
            customer_data[customer_id]['order_count'] += 1
        return customer_data

2 金融数据处理

class FinancialDataProcessor:
    """金融数据处理"""
    def __init__(self, risk_config: Dict):
        self.risk_thresholds = risk_config
    def validate_transaction(self, transaction: Dict) -> bool:
        """交易验证"""
        amount = transaction['amount']
        # 反欺诈检查
        if amount > self.risk_thresholds['max_amount']:
            return False
        if transaction['frequency'] > self.risk_thresholds['max_frequency']:
            return False
        return True
    def calculate_risk_score(self, customer_data: Dict) -> float:
        """计算风险得分"""
        score = 0.0
        factors = {
            'age': 0.1,
            'credit_score': 0.3,
            'transaction_history': 0.3,
            'income_level': 0.3
        }
        for factor, weight in factors.items():
            if factor in customer_data:
                score += self._evaluate_factor(factor, customer_data[factor]) * weight
        return score

统一错误处理

1 错误处理系统

class BusinessError(Exception):
    """业务错误基类"""
    def __init__(self, message: str, error_code: str, details: Any = None):
        self.message = message
        self.error_code = error_code
        self.details = details
        super().__init__(self.message)
class ErrorHandler:
    """统一错误处理器"""
    @staticmethod
    def handle_error(error: Exception, context: Dict = None):
        """处理错误"""
        if isinstance(error, BusinessError):
            # 业务错误日志
            logging.error(f"Business Error [{error.error_code}]: {error.message}")
            return {
                'status': 'error',
                'error_code': error.error_code,
                'message': error.message,
                'details': error.details
            }
        elif isinstance(error, DataValidationError):
            # 数据验证错误
            return {
                'status': 'validation_error',
                'message': str(error),
                'field': error.field
            }
        else:
            # 未知错误
            logging.exception("Unexpected error")
            return {
                'status': 'internal_error',
                'message': 'An unexpected error occurred'
            }

性能优化策略

1 批量处理

class BatchProcessor:
    """批量处理器"""
    def __init__(self, batch_size: int = 1000):
        self.batch_size = batch_size
    def process_in_batches(self, data_generator, process_func):
        """批量处理数据"""
        batch = []
        results = []
        for item in data_generator:
            batch.append(item)
            if len(batch) >= self.batch_size:
                results.extend(process_func(batch))
                batch = []
        # 处理剩余数据
        if batch:
            results.extend(process_func(batch))
        return results

2 缓存机制

from functools import lru_cache
import time
class CacheManager:
    """缓存管理器"""
    def __init__(self, ttl: int = 3600):
        self.cache = {}
        self.ttl = ttl
    def get_or_compute(self, key: str, compute_func):
        """获取或计算缓存"""
        if key in self.cache:
            value, timestamp = self.cache[key]
            if time.time() - timestamp < self.ttl:
                return value
        value = compute_func()
        self.cache[key] = (value, time.time())
        return value
    @lru_cache(maxsize=128)
    def get_cached_prediction(self, features: tuple):
        """缓存的预测函数"""
        return self.predict(features)

测试和监控

1 单元测试示例

import unittest
from unittest.mock import Mock, patch
class TestDataProcessor(unittest.TestCase):
    def setUp(self):
        self.processor = DataProcessor()
    def test_csv_loading(self):
        # 测试CSV加载
        test_data = "name,age\nAlice,30\nBob,25"
        result = self.processor.load_csv(test_data)
        self.assertEqual(len(result), 2)
    def test_json_to_dataframe(self):
        # 测试JSON转换
        test_json = [
            {"id": 1, "value": 100},
            {"id": 2, "value": 200}
        ]
        df = self.processor.json_to_dataframe(test_json)
        self.assertEqual(df.shape[0], 2)

最佳实践总结

1 适配策略选择

class AdapterFactory:
    """适配器工厂"""
    @staticmethod
    def create_adapter(business_type: str, config: Dict) -> DataProcessor:
        """根据业务类型创建适配器"""
        adapters = {
            'ecommerce': ECommerceProcessor,
            'finance': FinancialDataProcessor,
            'healthcare': HealthcareProcessor,
            'logistics': LogisticsProcessor
        }
        adapter_class = adapters.get(business_type)
        if not adapter_class:
            raise ValueError(f"Unsupported business type: {business_type}")
        return adapter_class(config)

2 使用示例

def main():
    """主函数"""
    # 加载配置
    config = ConfigManager('config.yaml')
    # 创建适配器
    processor = AdapterFactory.create_adapter(
        business_type='ecommerce',
        config=config.get_pipeline_config('ecommerce')
    )
    # 执行数据管道
    pipeline = TransformPipeline()
    pipeline.add_step(processor.validate)
    pipeline.add_step(processor.transform)
    pipeline.add_step(processor.enrich)
    # 处理数据
    raw_data = load_source_data()
    try:
        result = pipeline.execute(raw_data)
        save_processed_data(result)
    except Exception as e:
        error_response = ErrorHandler.handle_error(e)
        log_error(error_response)

关键原则

配置驱动：将可变参数外部化
插件化：支持功能扩展
接口统一：定义清晰的抽象接口
错误隔离：完善的异常处理机制
性能优化：批处理、缓存策略
可观测性：完善的日志和监控
灵活扩展：支持新业务快速接入

这种架构设计能最大化代码复用,适应不同业务场景的差异需求。