# 數(shù)據(jù)質量檢查框架
class DataQualityFramework:
    """數(shù)據(jù)質量評估框架"""
    
    @staticmethod
    def identify_common_issues(data_source_type):
        """識別不同數(shù)據(jù)源的常見問題"""
        
        issues = {
            'csv': [
                '編碼問題（特別是中文）',
                '日期格式不一致',
                '缺失值表示方式多樣',
                '列分隔符不一致',
                '標題行位置不固定'
            ],
            'excel': [
                '合并單元格',
                '多工作表結構',
                '公式計算結果',
                '隱藏行列',
                '數(shù)據(jù)驗證規(guī)則'
            ],
            'sql': [
                '字符集不匹配',
                'NULL值處理',
                '外鍵約束違反',
                '數(shù)據(jù)類型轉換錯誤',
                '時區(qū)問題'
            ]
        }
        
        return issues.get(data_source_type, [])
    
    @staticmethod
    def calculate_data_quality_score(df, data_source_type):
        """計算數(shù)據(jù)質量分數(shù)"""
        
        quality_metrics = {
            'completeness': 1 - (df.isnull().sum().sum() / (df.shape[0] * df.shape[1])),
            'consistency': DataQualityFramework.check_consistency(df, data_source_type),
            'accuracy': DataQualityFramework.check_accuracy(df, data_source_type),
            'uniqueness': 1 - (df.duplicated().sum() / len(df))
        }
        
        # 加權平均計算總分
        weights = {'completeness': 0.3, 'consistency': 0.3, 'accuracy': 0.2, 'uniqueness': 0.2}
        total_score = sum(quality_metrics[metric] * weights[metric] for metric in quality_metrics)
        
        return total_score, quality_metrics

# 使用示例
quality_checker = DataQualityFramework()
csv_issues = quality_checker.identify_common_issues('csv')
print("CSV文件常見問題:", csv_issues)

# requirements.txt
"""
pandas>=1.5.0
numpy>=1.21.0
openpyxl>=3.0.0
xlrd>=2.0.0
sqlalchemy>=1.4.0
psycopg2-binary>=2.9.0
mysql-connector-python>=8.0.0
pyodbc>=4.0.0
python-dotenv>=0.19.0
loguru>=0.6.0
pydantic>=1.9.0
"""

# 環(huán)境配置和檢查腳本
import sys
import importlib
from typing import Dict, List, Tuple

class EnvironmentValidator:
    """環(huán)境驗證器：檢查所有必要的依賴包"""
    
    REQUIRED_PACKAGES = {
        'pandas': ('數(shù)據(jù)操作', '1.5.0'),
        'numpy': ('數(shù)值計算', '1.21.0'),
        'openpyxl': ('Excel文件處理', '3.0.0'),
        'sqlalchemy': ('數(shù)據(jù)庫ORM', '1.4.0'),
        'psycopg2': ('PostgreSQL連接', '2.9.0'),
        'mysql.connector': ('MySQL連接', '8.0.0'),
        'pyodbc': ('ODBC連接', '4.0.0'),
        'python-dotenv': ('環(huán)境變量管理', '0.19.0'),
        'loguru': ('日志記錄', '0.6.0')
    }
    
    OPTIONAL_PACKAGES = {
        'pydantic': ('數(shù)據(jù)驗證', '1.9.0'),
        'requests': ('HTTP請求', '2.27.0'),
        'boto3': ('AWS服務', '1.24.0')
    }
    
    @classmethod
    def validate_environment(cls) -> Tuple[bool, Dict[str, Tuple[bool, str]]]:
        """驗證環(huán)境依賴"""
        
        results = {}
        all_passed = True
        
        print("=" * 60)
        print("環(huán)境依賴檢查")
        print("=" * 60)
        
        # 檢查必需包
        print("\n必需依賴包檢查:")
        for package, (description, min_version) in cls.REQUIRED_PACKAGES.items():
            installed, version = cls._check_package(package, min_version)
            status = "?" if installed else "?"
            results[package] = (installed, version)
            
            if not installed:
                all_passed = False
                
            print(f"{status} {package:20} {description:15} 要求版本: {min_version:8} 安裝版本: {version}")
        
        # 檢查可選包
        print("\n可選依賴包檢查:")
        for package, (description, min_version) in cls.OPTIONAL_PACKAGES.items():
            installed, version = cls._check_package(package, min_version)
            status = "??" if not installed else "?"
            results[package] = (installed, version)
            print(f"{status} {package:20} {description:15} 要求版本: {min_version:8} 安裝版本: {version}")
        
        print(f"\n總體結果: {'所有依賴已滿足' if all_passed else '缺少必需依賴'}")
        return all_passed, results
    
    @staticmethod
    def _check_package(package_name: str, min_version: str) -> Tuple[bool, str]:
        """檢查單個包的安裝情況和版本"""
        try:
            module = importlib.import_module(package_name.replace('-', '_'))
            installed_version = getattr(module, '__version__', '未知')
            
            # 簡單的版本比較
            if installed_version != '未知':
                installed_parts = list(map(int, installed_version.split('.')[:3]))
                min_parts = list(map(int, min_version.split('.')[:3]))
                is_compatible = installed_parts >= min_parts
            else:
                is_compatible = True
                
            return is_compatible, installed_version
            
        except ImportError:
            return False, "未安裝"

# 運行環(huán)境檢查
if __name__ == "__main__":
    env_ok, package_status = EnvironmentValidator.validate_environment()
    
    if not env_ok:
        print("\n?? 請安裝缺失的依賴包:")
        for package, (installed, version) in package_status.items():
            if not installed and package in EnvironmentValidator.REQUIRED_PACKAGES:
                print(f"  pip install {package}>={EnvironmentValidator.REQUIRED_PACKAGES[package][1]}")

import os
from typing import Dict, Any, Optional
from dotenv import load_dotenv
import json

class ConfigManager:
    """配置管理器：統(tǒng)一管理所有數(shù)據(jù)源配置"""
    
    def __init__(self, config_path: str = None):
        self.config_path = config_path or 'config.json'
        self._load_configuration()
    
    def _load_configuration(self) -> None:
        """加載配置文件和環(huán)境變量"""
        
        # 加載環(huán)境變量
        load_dotenv()
        
        # 加載JSON配置文件
        if os.path.exists(self.config_path):
            with open(self.config_path, 'r', encoding='utf-8') as f:
                self.config = json.load(f)
        else:
            self.config = self._create_default_config()
    
    def _create_default_config(self) -> Dict[str, Any]:
        """創(chuàng)建默認配置"""
        
        default_config = {
            "data_sources": {
                "csv": {
                    "default_encoding": "utf-8",
                    "fallback_encodings": ["gbk", "latin1"],
                    "chunk_size": 10000
                },
                "excel": {
                    "engine": "openpyxl",
                    "na_values": ["", "NULL", "N/A", "null"],
                    "keep_default_na": True
                },
                "database": {
                    "timeout": 30,
                    "pool_size": 5,
                    "max_overflow": 10
                }
            },
            "processing": {
                "max_workers": 4,
                "chunk_size": 10000,
                "temp_directory": "./temp"
            },
            "logging": {
                "level": "INFO",
                "format": "{time:YYYY-MM-DD HH:mm:ss} | {level} | {message}",
                "rotation": "10 MB"
            }
        }
        
        # 保存默認配置
        with open(self.config_path, 'w', encoding='utf-8') as f:
            json.dump(default_config, f, indent=2, ensure_ascii=False)
        
        print(f"已創(chuàng)建默認配置文件: {self.config_path}")
        return default_config
    
    def get_database_config(self, db_alias: str) -> Dict[str, str]:
        """獲取數(shù)據(jù)庫配置"""
        
        # 優(yōu)先從環(huán)境變量讀取
        db_config = {
            'host': os.getenv(f'{db_alias.upper()}_HOST'),
            'port': os.getenv(f'{db_alias.upper()}_PORT'),
            'database': os.getenv(f'{db_alias.upper()}_DATABASE'),
            'username': os.getenv(f'{db_alias.upper()}_USERNAME'),
            'password': os.getenv(f'{db_alias.upper()}_PASSWORD')
        }
        
        # 檢查配置完整性
        missing = [key for key, value in db_config.items() if not value]
        if missing:
            raise ValueError(f"數(shù)據(jù)庫 {db_alias} 配置不完整，缺少: {missing}")
        
        return db_config
    
    def get_file_config(self, file_type: str) -> Dict[str, Any]:
        """獲取文件處理配置"""
        return self.config['data_sources'].get(file_type, {})
    
    def update_config(self, section: str, updates: Dict[str, Any]) -> None:
        """更新配置"""
        if section in self.config:
            self.config[section].update(updates)
        else:
            self.config[section] = updates
        
        # 保存更新
        with open(self.config_path, 'w', encoding='utf-8') as f:
            json.dump(self.config, f, indent=2, ensure_ascii=False)

# 初始化配置管理器
config_manager = ConfigManager()

import pandas as pd
import numpy as np
from typing import Union, List, Dict, Any, Optional
from pathlib import Path
import logging
from loguru import logger
import chardet

class BaseDataReader:
    """基礎數(shù)據(jù)讀取器：提供通用數(shù)據(jù)讀取功能"""
    
    def __init__(self, config: ConfigManager):
        self.config = config
        self._setup_logging()
    
    def _setup_logging(self) -> None:
        """設置日志"""
        logging_config = self.config.get_file_config('logging')
        logger.remove()  # 移除默認處理器
        logger.add(
            "logs/data_integration_{time:YYYY-MM-DD}.log",
            level=logging_config.get('level', 'INFO'),
            format=logging_config.get('format', '{time:YYYY-MM-DD HH:mm:ss} | {level} | {message}'),
            rotation=logging_config.get('rotation', '10 MB'),
            retention=logging_config.get('retention', '30 days')
        )
    
    def detect_encoding(self, file_path: str) -> str:
        """檢測文件編碼"""
        try:
            with open(file_path, 'rb') as f:
                raw_data = f.read(10000)  # 讀取前10000字節(jié)進行檢測
                result = chardet.detect(raw_data)
                encoding = result['encoding']
                confidence = result['confidence']
                
                logger.info(f"檢測到文件編碼: {encoding} (置信度: {confidence:.2f})")
                return encoding if confidence > 0.7 else 'utf-8'
                
        except Exception as e:
            logger.warning(f"編碼檢測失敗: {e}，使用默認編碼")
            return 'utf-8'
    
    def validate_dataframe(self, df: pd.DataFrame, source_name: str) -> bool:
        """驗證DataFrame的基本完整性"""
        
        checks = {
            '空DataFrame': len(df) > 0,
            '包含數(shù)據(jù)': not df.empty,
            '列名唯一性': len(df.columns) == len(set(df.columns)),
            '無全空列': not df.isnull().all().any()
        }
        
        failed_checks = [check for check, passed in checks.items() if not passed]
        
        if failed_checks:
            logger.warning(f"數(shù)據(jù)驗證失敗 [{source_name}]: {failed_checks}")
            return False
        
        logger.info(f"數(shù)據(jù)驗證通過 [{source_name}]: 形狀 {df.shape}")
        return True
    
    def handle_read_error(self, error: Exception, source: str) -> pd.DataFrame:
        """統(tǒng)一錯誤處理"""
        logger.error(f"讀取數(shù)據(jù)源失敗 [{source}]: {error}")
        # 返回空的DataFrame而不是拋出異常
        return pd.DataFrame()

class CSVDataReader(BaseDataReader):
    """CSV文件讀取器"""
    
    def read(self, file_path: str, **kwargs) -> pd.DataFrame:
        """讀取CSV文件"""
        
        try:
            csv_config = self.config.get_file_config('csv')
            
            # 自動檢測編碼
            encoding = kwargs.pop('encoding', None)
            if not encoding:
                encoding = self.detect_encoding(file_path)
            
            # 嘗試多種編碼
            encodings_to_try = [encoding] + csv_config.get('fallback_encodings', [])
            
            for enc in encodings_to_try:
                try:
                    logger.info(f"嘗試使用編碼 {enc} 讀取CSV文件: {file_path}")
                    
                    df = pd.read_csv(
                        file_path,
                        encoding=enc,
                        na_values=csv_config.get('na_values', ['', 'NULL']),
                        keep_default_na=True,
                        **kwargs
                    )
                    
                    if self.validate_dataframe(df, f"CSV: {file_path}"):
                        return df
                        
                except UnicodeDecodeError:
                    logger.warning(f"編碼 {enc} 失敗，嘗試下一個")
                    continue
                except Exception as e:
                    logger.error(f"讀取CSV文件失敗: {e}")
                    break
            
            return self.handle_read_error(Exception("所有編碼嘗試都失敗"), f"CSV: {file_path}")
            
        except Exception as e:
            return self.handle_read_error(e, f"CSV: {file_path}")
    
    def read_chunked(self, file_path: str, chunk_size: int = None, **kwargs) -> pd.DataFrame:
        """分塊讀取大型CSV文件"""
        
        csv_config = self.config.get_file_config('csv')
        chunk_size = chunk_size or csv_config.get('chunk_size', 10000)
        
        chunks = []
        try:
            for i, chunk in enumerate(pd.read_csv(file_path, chunksize=chunk_size, **kwargs)):
                chunks.append(chunk)
                logger.info(f"讀取第 {i + 1} 塊數(shù)據(jù)，形狀: {chunk.shape}")
                
                # 每10塊輸出一次進度
                if (i + 1) % 10 == 0:
                    total_rows = sum(len(c) for c in chunks)
                    logger.info(f"已讀取 {total_rows} 行數(shù)據(jù)")
            
            if chunks:
                result_df = pd.concat(chunks, ignore_index=True)
                logger.info(f"CSV文件讀取完成，總行數(shù): {len(result_df)}")
                return result_df
            else:
                return pd.DataFrame()
                
        except Exception as e:
            return self.handle_read_error(e, f"CSV(chunked): {file_path}")

class ExcelDataReader(BaseDataReader):
    """Excel文件讀取器"""
    
    def read(self, file_path: str, sheet_name: Union[str, int, List] = 0, **kwargs) -> pd.DataFrame:
        """讀取Excel文件"""
        
        try:
            excel_config = self.config.get_file_config('excel')
            
            # 獲取所有工作表名稱
            excel_file = pd.ExcelFile(file_path)
            sheet_names = excel_file.sheet_names
            
            logger.info(f"Excel文件包含工作表: {sheet_names}")
            
            if sheet_name is None:
                # 讀取所有工作表
                dfs = {}
                for sheet in sheet_names:
                    df = self._read_single_sheet(file_path, sheet, excel_config, **kwargs)
                    if not df.empty:
                        dfs[sheet] = df
                return dfs
            else:
                # 讀取指定工作表
                return self._read_single_sheet(file_path, sheet_name, excel_config, **kwargs)
                
        except Exception as e:
            return self.handle_read_error(e, f"Excel: {file_path}")
    
    def _read_single_sheet(self, file_path: str, sheet_name: str, config: Dict, **kwargs) -> pd.DataFrame:
        """讀取單個工作表"""
        
        try:
            df = pd.read_excel(
                file_path,
                sheet_name=sheet_name,
                engine=config.get('engine', 'openpyxl'),
                na_values=config.get('na_values', ['', 'NULL', 'N/A']),
                keep_default_na=config.get('keep_default_na', True),
                **kwargs
            )
            
            if self.validate_dataframe(df, f"Excel[{sheet_name}]: {file_path}"):
                # 清理Excel特有的問題
                df = self._clean_excel_data(df)
                return df
            else:
                return pd.DataFrame()
                
        except Exception as e:
            logger.error(f"讀取Excel工作表失敗 [{sheet_name}]: {e}")
            return pd.DataFrame()
    
    def _clean_excel_data(self, df: pd.DataFrame) -> pd.DataFrame:
        """清理Excel數(shù)據(jù)特有的問題"""
        
        # 移除完全空的行和列
        df = df.dropna(how='all').dropna(axis=1, how='all')
        
        # 重置索引
        df = df.reset_index(drop=True)
        
        # 清理列名中的特殊字符
        df.columns = [str(col).strip().replace('\n', ' ') for col in df.columns]
        
        return df

class DatabaseReader(BaseDataReader):
    """數(shù)據(jù)庫讀取器"""
    
    def __init__(self, config: ConfigManager):
        super().__init__(config)
        self.connections = {}
    
    def get_connection(self, db_alias: str):
        """獲取數(shù)據(jù)庫連接"""
        
        if db_alias in self.connections:
            return self.connections[db_alias]
        
        try:
            db_config = self.config.get_database_config(db_alias)
            db_type = db_alias.lower()
            
            if db_type in ['postgresql', 'postgres']:
                import psycopg2
                conn = psycopg2.connect(
                    host=db_config['host'],
                    port=db_config['port'],
                    database=db_config['database'],
                    user=db_config['username'],
                    password=db_config['password']
                )
            elif db_type == 'mysql':
                import mysql.connector
                conn = mysql.connector.connect(
                    host=db_config['host'],
                    port=db_config['port'],
                    database=db_config['database'],
                    user=db_config['username'],
                    password=db_config['password']
                )
            else:
                # 使用SQLAlchemy作為通用連接
                from sqlalchemy import create_engine
                connection_string = self._build_connection_string(db_alias, db_config)
                engine = create_engine(connection_string)
                conn = engine.connect()
            
            self.connections[db_alias] = conn
            logger.info(f"數(shù)據(jù)庫連接已建立: {db_alias}")
            return conn
            
        except Exception as e:
            logger.error(f"數(shù)據(jù)庫連接失敗 [{db_alias}]: {e}")
            raise
    
    def _build_connection_string(self, db_alias: str, db_config: Dict) -> str:
        """構建數(shù)據(jù)庫連接字符串"""
        
        db_type = db_alias.lower()
        
        if db_type in ['postgresql', 'postgres']:
            return f"postgresql://{db_config['username']}:{db_config['password']}@{db_config['host']}:{db_config['port']}/{db_config['database']}"
        elif db_type == 'mysql':
            return f"mysql+mysqlconnector://{db_config['username']}:{db_config['password']}@{db_config['host']}:{db_config['port']}/{db_config['database']}"
        else:
            raise ValueError(f"不支持的數(shù)據(jù)庫類型: {db_type}")
    
    def read_query(self, db_alias: str, query: str, params: Dict = None) -> pd.DataFrame:
        """執(zhí)行SQL查詢"""
        
        try:
            conn = self.get_connection(db_alias)
            
            logger.info(f"執(zhí)行SQL查詢 [{db_alias}]: {query[:100]}...")
            
            df = pd.read_sql_query(query, conn, params=params)
            
            if self.validate_dataframe(df, f"DB[{db_alias}]"):
                return df
            else:
                return pd.DataFrame()
                
        except Exception as e:
            return self.handle_read_error(e, f"DB[{db_alias}]")
    
    def read_table(self, db_alias: str, table_name: str, columns: List[str] = None, 
                   where_clause: str = None, limit: int = None) -> pd.DataFrame:
        """讀取整個表或表的部分數(shù)據(jù)"""
        
        try:
            # 構建查詢
            column_list = '*' if not columns else ', '.join(columns)
            query = f"SELECT {column_list} FROM {table_name}"
            
            if where_clause:
                query += f" WHERE {where_clause}"
            
            if limit:
                query += f" LIMIT {limit}"
            
            return self.read_query(db_alias, query)
            
        except Exception as e:
            return self.handle_read_error(e, f"DB[{db_alias}.{table_name}]")
    
    def close_connections(self):
        """關閉所有數(shù)據(jù)庫連接"""
        for alias, conn in self.connections.items():
            try:
                conn.close()
                logger.info(f"數(shù)據(jù)庫連接已關閉: {alias}")
            except Exception as e:
                logger.error(f"關閉數(shù)據(jù)庫連接失敗 [{alias}]: {e}")
        
        self.connections.clear()

from typing import Dict, List, Any, Callable
import hashlib
from datetime import datetime

class DataIntegrationEngine:
    """數(shù)據(jù)整合引擎：統(tǒng)一處理多源數(shù)據(jù)"""
    
    def __init__(self, config: ConfigManager):
        self.config = config
        self.readers = {
            'csv': CSVDataReader(config),
            'excel': ExcelDataReader(config),
            'database': DatabaseReader(config)
        }
        
        # 數(shù)據(jù)緩存
        self.data_cache = {}
        self.schema_registry = {}
    
    def load_data(self, data_sources: List[Dict[str, Any]]) -> Dict[str, pd.DataFrame]:
        """加載多個數(shù)據(jù)源"""
        
        loaded_data = {}
        
        for source in data_sources:
            source_type = source.get('type')
            source_name = source.get('name', f"source_{len(loaded_data)}")
            
            if source_type not in self.readers:
                logger.warning(f"不支持的數(shù)據(jù)源類型: {source_type}，跳過 {source_name}")
                continue
            
            try:
                # 生成緩存鍵
                cache_key = self._generate_cache_key(source)
                
                # 檢查緩存
                if cache_key in self.data_cache:
                    logger.info(f"使用緩存數(shù)據(jù): {source_name}")
                    loaded_data[source_name] = self.data_cache[cache_key]
                    continue
                
                # 讀取數(shù)據(jù)
                reader = self.readers[source_type]
                df = self._read_with_reader(reader, source)
                
                if not df.empty:
                    # 數(shù)據(jù)預處理
                    df = self._preprocess_data(df, source)
                    
                    # 注冊schema
                    self._register_schema(source_name, df)
                    
                    # 緩存數(shù)據(jù)
                    self.data_cache[cache_key] = df
                    loaded_data[source_name] = df
                    
                    logger.info(f"成功加載數(shù)據(jù)源: {source_name}，形狀: {df.shape}")
                else:
                    logger.warning(f"數(shù)據(jù)源為空: {source_name}")
                    
            except Exception as e:
                logger.error(f"加載數(shù)據(jù)源失敗 [{source_name}]: {e}")
                continue
        
        return loaded_data
    
    def _read_with_reader(self, reader: BaseDataReader, source: Dict) -> pd.DataFrame:
        """使用對應的讀取器讀取數(shù)據(jù)"""
        
        source_type = source.get('type')
        parameters = source.get('parameters', {})
        
        if source_type == 'csv':
            file_path = source['path']
            return reader.read(file_path, **parameters)
        
        elif source_type == 'excel':
            file_path = source['path']
            sheet_name = parameters.pop('sheet_name', 0)
            return reader.read(file_path, sheet_name=sheet_name, **parameters)
        
        elif source_type == 'database':
            db_alias = source['connection']
            query = source.get('query')
            table_name = source.get('table')
            
            if query:
                return reader.read_query(db_alias, query, parameters.get('params'))
            elif table_name:
                return reader.read_table(
                    db_alias, 
                    table_name,
                    columns=parameters.get('columns'),
                    where_clause=parameters.get('where'),
                    limit=parameters.get('limit')
                )
            else:
                raise ValueError("數(shù)據(jù)庫源必須提供query或table參數(shù)")
        
        else:
            raise ValueError(f"未知的數(shù)據(jù)源類型: {source_type}")
    
    def _generate_cache_key(self, source: Dict) -> str:
        """生成緩存鍵"""
        
        source_str = json.dumps(source, sort_keys=True)
        return hashlib.md5(source_str.encode()).hexdigest()
    
    def _preprocess_data(self, df: pd.DataFrame, source: Dict) -> pd.DataFrame:
        """數(shù)據(jù)預處理"""
        
        # 應用自定義預處理函數(shù)
        preprocess_func = source.get('preprocess')
        if preprocess_func and callable(preprocess_func):
            df = preprocess_func(df)
        
        # 標準預處理步驟
        df = self._standard_preprocessing(df, source)
        
        return df
    
    def _standard_preprocessing(self, df: pd.DataFrame, source: Dict) -> pd.DataFrame:
        """標準預處理流程"""
        
        # 1. 清理列名
        df.columns = [self._clean_column_name(col) for col in df.columns]
        
        # 2. 處理數(shù)據(jù)類型
        type_mapping = source.get('type_mapping', {})
        for col, dtype in type_mapping.items():
            if col in df.columns:
                try:
                    df[col] = self._convert_column_type(df[col], dtype)
                except Exception as e:
                    logger.warning(f"列類型轉換失敗 [{col} -> {dtype}]: {e}")
        
        # 3. 添加數(shù)據(jù)源標識
        df['_data_source'] = source.get('name', 'unknown')
        df['_load_timestamp'] = datetime.now()
        
        return df
    
    def _clean_column_name(self, column_name: str) -> str:
        """清理列名"""
        
        # 移除特殊字符，保留字母、數(shù)字、下劃線
        import re
        cleaned = re.sub(r'[^\w]', '_', str(column_name))
        # 移除連續(xù)的下劃線
        cleaned = re.sub(r'_+', '_', cleaned)
        # 移除首尾的下劃線
        cleaned = cleaned.strip('_')
        # 轉換為小寫
        return cleaned.lower()
    
    def _convert_column_type(self, series: pd.Series, target_type: str) -> pd.Series:
        """轉換列數(shù)據(jù)類型"""
        
        type_handlers = {
            'string': lambda s: s.astype(str),
            'integer': lambda s: pd.to_numeric(s, errors='coerce').fillna(0).astype(int),
            'float': lambda s: pd.to_numeric(s, errors='coerce'),
            'datetime': lambda s: pd.to_datetime(s, errors='coerce'),
            'boolean': lambda s: s.astype(str).str.lower().isin(['true', '1', 'yes', 'y']).astype(bool)
        }
        
        handler = type_handlers.get(target_type.lower())
        if handler:
            return handler(series)
        else:
            return series
    
    def _register_schema(self, source_name: str, df: pd.DataFrame):
        """注冊數(shù)據(jù)schema"""
        
        schema = {
            'columns': list(df.columns),
            'dtypes': {col: str(dtype) for col, dtype in df.dtypes.items()},
            'row_count': len(df),
            'null_counts': df.isnull().sum().to_dict(),
            'sample_data': df.head(3).to_dict('records')
        }
        
        self.schema_registry[source_name] = schema
        logger.info(f"Schema已注冊: {source_name}")
    
    def merge_data(self, data_dict: Dict[str, pd.DataFrame], 
                   merge_strategy: Dict[str, Any]) -> pd.DataFrame:
        """合并多個數(shù)據(jù)源"""
        
        if not data_dict:
            return pd.DataFrame()
        
        data_frames = list(data_dict.values())
        merge_type = merge_strategy.get('type', 'concat')
        
        try:
            if merge_type == 'concat':
                result = self._concat_dataframes(data_frames, merge_strategy)
            elif merge_type == 'join':
                result = self._join_dataframes(data_dict, merge_strategy)
            elif merge_type == 'union':
                result = self._union_dataframes(data_frames, merge_strategy)
            else:
                raise ValueError(f"不支持的合并類型: {merge_type}")
            
            logger.info(f"數(shù)據(jù)合并完成，最終形狀: {result.shape}")
            return result
            
        except Exception as e:
            logger.error(f"數(shù)據(jù)合并失敗: {e}")
            return pd.DataFrame()
    
    def _concat_dataframes(self, data_frames: List[pd.DataFrame], strategy: Dict) -> pd.DataFrame:
        """垂直合并數(shù)據(jù)框"""
        
        # 對齊列
        if strategy.get('align_columns', True):
            all_columns = set()
            for df in data_frames:
                all_columns.update(df.columns)
            
            aligned_dfs = []
            for df in data_frames:
                # 添加缺失的列
                for col in all_columns:
                    if col not in df.columns:
                        df[col] = None
                # 按統(tǒng)一順序排列列
                df = df[list(all_columns)]
                aligned_dfs.append(df)
            
            data_frames = aligned_dfs
        
        return pd.concat(data_frames, ignore_index=True, sort=False)
    
    def _join_dataframes(self, data_dict: Dict[str, pd.DataFrame], strategy: Dict) -> pd.DataFrame:
        """連接數(shù)據(jù)框"""
        
        join_keys = strategy.get('keys', [])
        join_type = strategy.get('join_type', 'inner')
        
        if not data_dict:
            return pd.DataFrame()
        
        data_frames = list(data_dict.values())
        result = data_frames[0]
        
        for i in range(1, len(data_frames)):
            result = result.merge(
                data_frames[i],
                on=join_keys,
                how=join_type,
                suffixes=(f'_{i-1}', f'_{i}')
            )
        
        return result
    
    def _union_dataframes(self, data_frames: List[pd.DataFrame], strategy: Dict) -> pd.DataFrame:
        """求并集合并數(shù)據(jù)框"""
        
        common_columns = set.intersection(*[set(df.columns) for df in data_frames])
        
        if not common_columns:
            logger.warning("沒有共同列，無法執(zhí)行union操作")
            return pd.DataFrame()
        
        # 只保留共同列
        union_dfs = [df[list(common_columns)] for df in data_frames]
        
        return pd.concat(union_dfs, ignore_index=True).drop_duplicates()
    
    def clear_cache(self):
        """清理緩存"""
        self.data_cache.clear()
        logger.info("數(shù)據(jù)緩存已清理")

class DataQualityMonitor:
    """數(shù)據(jù)質量監(jiān)控器"""
    
    def __init__(self):
        self.quality_metrics = {}
        self.validation_rules = {}
    
    def add_validation_rule(self, rule_name: str, rule_func: Callable, 
                           description: str = "") -> None:
        """添加數(shù)據(jù)驗證規(guī)則"""
        self.validation_rules[rule_name] = {
            'function': rule_func,
            'description': description
        }
        logger.info(f"驗證規(guī)則已添加: {rule_name}")
    
    def validate_dataset(self, df: pd.DataFrame, dataset_name: str) -> Dict[str, Any]:
        """驗證數(shù)據(jù)集質量"""
        
        validation_results = {
            'dataset_name': dataset_name,
            'timestamp': datetime.now(),
            'basic_stats': self._get_basic_stats(df),
            'quality_metrics': self._calculate_quality_metrics(df),
            'rule_violations': self._check_validation_rules(df),
            'data_issues': self._detect_data_issues(df)
        }
        
        # 計算總體質量分數(shù)
        validation_results['quality_score'] = self._calculate_overall_score(
            validation_results['quality_metrics'],
            validation_results['rule_violations']
        )
        
        self.quality_metrics[dataset_name] = validation_results
        return validation_results
    
    def _get_basic_stats(self, df: pd.DataFrame) -> Dict[str, Any]:
        """獲取基本統(tǒng)計信息"""
        
        return {
            'row_count': len(df),
            'column_count': len(df.columns),
            'memory_usage_mb': df.memory_usage(deep=True).sum() / 1024**2,
            'data_types': {col: str(dtype) for col, dtype in df.dtypes.items()}
        }
    
    def _calculate_quality_metrics(self, df: pd.DataFrame) -> Dict[str, float]:
        """計算數(shù)據(jù)質量指標"""
        
        total_cells = df.shape[0] * df.shape[1]
        
        if total_cells == 0:
            return {
                'completeness': 0.0,
                'consistency': 0.0,
                'accuracy': 0.0,
                'uniqueness': 0.0,
                'timeliness': 1.0
            }
        
        # 完整性：非空值比例
        completeness = 1 - (df.isnull().sum().sum() / total_cells)
        
        # 一致性：數(shù)據(jù)類型一致性
        type_consistency = self._check_type_consistency(df)
        
        # 準確性：基于業(yè)務規(guī)則（需要自定義）
        accuracy = self._estimate_accuracy(df)
        
        # 唯一性：重復行比例
        uniqueness = 1 - (df.duplicated().sum() / len(df))
        
        # 時效性：基于時間戳（如果存在）
        timeliness = self._check_timeliness(df)
        
        return {
            'completeness': completeness,
            'consistency': type_consistency,
            'accuracy': accuracy,
            'uniqueness': uniqueness,
            'timeliness': timeliness
        }
    
    def _check_type_consistency(self, df: pd.DataFrame) -> float:
        """檢查數(shù)據(jù)類型一致性"""
        
        consistent_columns = 0
        for col in df.columns:
            try:
                # 嘗試轉換為數(shù)值類型
                pd.to_numeric(df[col], errors='raise')
                consistent_columns += 1
            except:
                try:
                    # 嘗試轉換為日期類型
                    pd.to_datetime(df[col], errors='raise')
                    consistent_columns += 1
                except:
                    # 保持為字符串類型
                    consistent_columns += 1
        
        return consistent_columns / len(df.columns) if df.columns.any() else 1.0
    
    def _estimate_accuracy(self, df: pd.DataFrame) -> float:
        """估計數(shù)據(jù)準確性（基于簡單啟發(fā)式規(guī)則）"""
        
        accuracy_indicators = []
        
        # 檢查數(shù)值列的合理性
        numeric_cols = df.select_dtypes(include=[np.number]).columns
        for col in numeric_cols:
            if df[col].notna().any():
                # 檢查異常值（超出3個標準差）
                z_scores = np.abs((df[col] - df[col].mean()) / df[col].std())
                outlier_ratio = (z_scores > 3).sum() / len(df)
                accuracy_indicators.append(1 - outlier_ratio)
        
        # 檢查分類列的合理性（值在合理范圍內）
        categorical_cols = df.select_dtypes(include=['object']).columns
        for col in categorical_cols[:5]:  # 只檢查前5個分類列
            if df[col].notna().any():
                # 簡單檢查：最常見的值應該占一定比例
                value_counts = df[col].value_counts()
                if len(value_counts) > 0:
                    top_ratio = value_counts.iloc[0] / len(df)
                    accuracy_indicators.append(min(top_ratio * 2, 1.0))
        
        return np.mean(accuracy_indicators) if accuracy_indicators else 0.8
    
    def _check_timeliness(self, df: pd.DataFrame) -> float:
        """檢查數(shù)據(jù)時效性"""
        
        # 查找可能的時間戳列
        time_columns = []
        for col in df.columns:
            col_lower = col.lower()
            if any(time_keyword in col_lower for time_keyword in 
                  ['time', 'date', 'timestamp', 'created', 'updated']):
                time_columns.append(col)
        
        if not time_columns:
            return 1.0  # 沒有時間列，默認時效性為1
        
        # 檢查最近的數(shù)據(jù)時間
        latest_times = []
        for col in time_columns:
            try:
                time_series = pd.to_datetime(df[col], errors='coerce')
                if time_series.notna().any():
                    latest_time = time_series.max()
                    if pd.notna(latest_time):
                        days_ago = (datetime.now() - latest_time).days
                        # 數(shù)據(jù)在30天內為新鮮
                        timeliness = max(0, 1 - (days_ago / 30))
                        latest_times.append(timeliness)
            except:
                continue
        
        return np.mean(latest_times) if latest_times else 0.5
    
    def _check_validation_rules(self, df: pd.DataFrame) -> Dict[str, List[str]]:
        """檢查驗證規(guī)則"""
        
        violations = {}
        for rule_name, rule_info in self.validation_rules.items():
            try:
                rule_func = rule_info['function']
                result = rule_func(df)
                
                if result is not True and result is not None:
                    violations[rule_name] = result
                    
            except Exception as e:
                violations[rule_name] = f"規(guī)則執(zhí)行錯誤: {e}"
        
        return violations
    
    def _detect_data_issues(self, df: pd.DataFrame) -> List[str]:
        """檢測數(shù)據(jù)問題"""
        
        issues = []
        
        # 檢查完全空的列
        empty_cols = df.columns[df.isnull().all()].tolist()
        if empty_cols:
            issues.append(f"完全空的列: {empty_cols}")
        
        # 檢查常數(shù)列
        constant_cols = []
        for col in df.columns:
            if df[col].nunique() == 1:
                constant_cols.append(col)
        if constant_cols:
            issues.append(f"常數(shù)列: {constant_cols}")
        
        # 檢查高基數(shù)分類列
        object_cols = df.select_dtypes(include=['object']).columns
        high_cardinality_cols = []
        for col in object_cols:
            if df[col].nunique() > len(df) * 0.5:  # 唯一值超過50%
                high_cardinality_cols.append(col)
        if high_cardinality_cols:
            issues.append(f"高基數(shù)分類列: {high_cardinality_cols}")
        
        return issues
    
    def _calculate_overall_score(self, quality_metrics: Dict, 
                               rule_violations: Dict) -> float:
        """計算總體質量分數(shù)"""
        
        # 基礎質量指標權重
        weights = {
            'completeness': 0.3,
            'consistency': 0.2,
            'accuracy': 0.25,
            'uniqueness': 0.15,
            'timeliness': 0.1
        }
        
        # 計算加權平均
        base_score = sum(quality_metrics[metric] * weight 
                        for metric, weight in weights.items())
        
        # 規(guī)則違反懲罰
        violation_penalty = min(len(rule_violations) * 0.1, 0.3)
        
        return max(0, base_score - violation_penalty)
    
    def generate_quality_report(self, dataset_name: str) -> str:
        """生成質量報告"""
        
        if dataset_name not in self.quality_metrics:
            return f"未找到數(shù)據(jù)集: {dataset_name}"
        
        metrics = self.quality_metrics[dataset_name]
        report = []
        
        report.append("=" * 60)
        report.append(f"數(shù)據(jù)質量報告 - {dataset_name}")
        report.append("=" * 60)
        report.append(f"生成時間: {metrics['timestamp']}")
        report.append(f"總體質量分數(shù): {metrics['quality_score']:.2%}")
        report.append("")
        
        # 基本統(tǒng)計
        report.append("基本統(tǒng)計:")
        stats = metrics['basic_stats']
        report.append(f"  行數(shù): {stats['row_count']:,}")
        report.append(f"  列數(shù): {stats['column_count']}")
        report.append(f"  內存使用: {stats['memory_usage_mb']:.2f} MB")
        report.append("")
        
        # 質量指標
        report.append("質量指標:")
        for metric, value in metrics['quality_metrics'].items():
            report.append(f"  {metric:15}: {value:.2%}")
        report.append("")
        
        # 規(guī)則違反
        if metrics['rule_violations']:
            report.append("規(guī)則違反:")
            for rule, violation in metrics['rule_violations'].items():
                report.append(f"  {rule}: {violation}")
            report.append("")
        
        # 數(shù)據(jù)問題
        if metrics['data_issues']:
            report.append("檢測到的問題:")
            for issue in metrics['data_issues']:
                report.append(f"  ? {issue}")
        
        return "\n".join(report)

def complete_data_integration_example():
    """完整的數(shù)據(jù)整合示例"""
    
    # 1. 初始化配置
    config = ConfigManager()
    
    # 2. 初始化數(shù)據(jù)整合引擎
    integration_engine = DataIntegrationEngine(config)
    quality_monitor = DataQualityMonitor()
    
    # 3. 定義數(shù)據(jù)源配置
    data_sources = [
        {
            'name': 'sales_data_csv',
            'type': 'csv',
            'path': 'data/sales_data.csv',
            'parameters': {
                'encoding': 'utf-8',
                'sep': ','
            }
        },
        {
            'name': 'customer_data_excel',
            'type': 'excel',
            'path': 'data/customer_data.xlsx',
            'parameters': {
                'sheet_name': 'Customers',
                'header': 0
            }
        },
        {
            'name': 'product_data_db',
            'type': 'database',
            'connection': 'postgresql',
            'table': 'products',
            'parameters': {
                'columns': ['product_id', 'product_name', 'category', 'price'],
                'where': 'active = true'
            }
        }
    ]
    
    # 4. 添加數(shù)據(jù)驗證規(guī)則
    def validate_sales_data(df):
        """驗證銷售數(shù)據(jù)規(guī)則"""
        issues = []
        
        # 檢查銷售額非負
        if 'sales_amount' in df.columns:
            negative_sales = (df['sales_amount'] < 0).sum()
            if negative_sales > 0:
                issues.append(f"發(fā)現(xiàn) {negative_sales} 條負銷售額記錄")
        
        # 檢查日期合理性
        if 'sale_date' in df.columns:
            try:
                sale_dates = pd.to_datetime(df['sale_date'])
                future_sales = (sale_dates > datetime.now()).sum()
                if future_sales > 0:
                    issues.append(f"發(fā)現(xiàn) {future_sales} 條未來日期銷售記錄")
            except:
                issues.append("銷售日期格式異常")
        
        return issues if issues else True
    
    quality_monitor.add_validation_rule(
        'sales_data_validation', 
        validate_sales_data,
        '驗證銷售數(shù)據(jù)的基本業(yè)務規(guī)則'
    )
    
    # 5. 加載數(shù)據(jù)
    print("開始加載數(shù)據(jù)源...")
    loaded_data = integration_engine.load_data(data_sources)
    
    # 6. 數(shù)據(jù)質量檢查
    print("\n進行數(shù)據(jù)質量檢查...")
    quality_reports = {}
    for name, df in loaded_data.items():
        quality_report = quality_monitor.validate_dataset(df, name)
        quality_reports[name] = quality_report
        
        # 打印質量報告
        print(quality_monitor.generate_quality_report(name))
        print()
    
    # 7. 數(shù)據(jù)整合
    print("開始數(shù)據(jù)整合...")
    merge_strategy = {
        'type': 'concat',
        'align_columns': True
    }
    
    integrated_data = integration_engine.merge_data(loaded_data, merge_strategy)
    
    # 8. 最終質量檢查
    print("最終整合數(shù)據(jù)質量檢查...")
    final_quality = quality_monitor.validate_dataset(integrated_data, 'integrated_dataset')
    print(quality_monitor.generate_quality_report('integrated_dataset'))
    
    # 9. 保存整合結果
    output_path = 'output/integrated_data.csv'
    integrated_data.to_csv(output_path, index=False, encoding='utf-8')
    print(f"\n整合數(shù)據(jù)已保存至: {output_path}")
    print(f"最終數(shù)據(jù)集形狀: {integrated_data.shape}")
    
    # 10. 清理資源
    integration_engine.clear_cache()
    if hasattr(integration_engine.readers['database'], 'close_connections'):
        integration_engine.readers['database'].close_connections()
    
    return integrated_data, quality_reports

# 運行完整示例
if __name__ == "__main__":
    try:
        final_data, reports = complete_data_integration_example()
        print("數(shù)據(jù)整合流程完成!")
    except Exception as e:
        print(f"數(shù)據(jù)整合流程失敗: {e}")
        import traceback
        traceback.print_exc()

class IncrementalDataLoader:
    """增量數(shù)據(jù)加載器"""
    
    def __init__(self, integration_engine: DataIntegrationEngine):
        self.engine = integration_engine
        self.last_load_info = self._load_last_run_info()
    
    def _load_last_run_info(self) -> Dict:
        """加載上次運行信息"""
        
        info_file = 'metadata/last_run_info.json'
        if os.path.exists(info_file):
            with open(info_file, 'r') as f:
                return json.load(f)
        return {}
    
    def _save_last_run_info(self, info: Dict) -> None:
        """保存運行信息"""
        
        os.makedirs('metadata', exist_ok=True)
        with open('metadata/last_run_info.json', 'w') as f:
            json.dump(info, f, indent=2)
    
    def incremental_load(self, data_sources: List[Dict]) -> Dict[str, pd.DataFrame]:
        """增量加載數(shù)據(jù)"""
        
        incremental_data = {}
        current_run_info = {'timestamp': datetime.now().isoformat()}
        
        for source in data_sources:
            source_name = source['name']
            source_type = source['type']
            
            # 獲取增量條件
            incremental_condition = self._get_incremental_condition(
                source_name, source_type, source
            )
            
            if incremental_condition:
                # 修改數(shù)據(jù)源配置以包含增量條件
                modified_source = self._apply_incremental_filter(
                    source, incremental_condition
                )
                
                logger.info(f"執(zhí)行增量加載: {source_name}，條件: {incremental_condition}")
                data = self.engine.load_data([modified_source])
                
                if data and source_name in data:
                    incremental_data[source_name] = data[source_name]
            
            # 記錄本次運行信息
            current_run_info[source_name] = {
                'loaded_at': datetime.now().isoformat(),
                'row_count': len(incremental_data.get(source_name, pd.DataFrame()))
            }
        
        # 保存運行信息
        self._save_last_run_info(current_run_info)
        self.last_load_info = current_run_info
        
        return incremental_data
    
    def _get_incremental_condition(self, source_name: str, source_type: str, 
                                 source: Dict) -> Optional[str]:
        """獲取增量加載條件"""
        
        if source_type == 'database':
            # 數(shù)據(jù)庫增量加載
            incremental_field = source.get('incremental_field', 'updated_at')
            last_value = self.last_load_info.get(source_name, {}).get('last_value')
            
            if last_value:
                return f"{incremental_field} > '{last_value}'"
            else:
                # 首次運行，加載所有數(shù)據(jù)
                return None
        
        elif source_type in ['csv', 'excel']:
            # 文件增量加載 - 基于文件修改時間
            file_path = source['path']
            if os.path.exists(file_path):
                current_mtime = os.path.getmtime(file_path)
                last_mtime = self.last_load_info.get(source_name, {}).get('file_mtime')
                
                if last_mtime and current_mtime > last_mtime:
                    # 文件已修改，需要重新加載
                    return "file_modified"
                elif not last_mtime:
                    # 首次運行
                    return None
        
        return None
    
    def _apply_incremental_filter(self, source: Dict, condition: str) -> Dict:
        """應用增量過濾條件"""
        
        source_type = source['type']
        modified_source = source.copy()
        
        if source_type == 'database':
            if 'where' in modified_source.get('parameters', {}):
                # 合并現(xiàn)有條件和增量條件
                modified_source['parameters']['where'] = \
                    f"({modified_source['parameters']['where']}) AND ({condition})"
            else:
                modified_source['parameters']['where'] = condition
        
        return modified_source

# 使用增量加載的示例
def incremental_loading_example():
    """增量加載示例"""
    
    config = ConfigManager()
    engine = DataIntegrationEngine(config)
    incremental_loader = IncrementalDataLoader(engine)
    
    data_sources = [
        {
            'name': 'daily_sales',
            'type': 'database',
            'connection': 'postgresql',
            'table': 'sales',
            'incremental_field': 'sale_date',
            'parameters': {
                'columns': ['sale_id', 'sale_date', 'amount', 'customer_id']
            }
        }
    ]
    
    print("執(zhí)行增量數(shù)據(jù)加載...")
    new_data = incremental_loader.incremental_load(data_sources)
    
    for name, df in new_data.items():
        print(f"增量加載 {name}: {len(df)} 行新數(shù)據(jù)")
    
    return new_data

def pre_deployment_validation():
    """部署前驗證"""
    
    validation_steps = [
        ("環(huán)境依賴檢查", validate_environment_dependencies),
        ("配置文件驗證", validate_configuration_files),
        ("數(shù)據(jù)源連通性", validate_data_source_connectivity),
        ("功能完整性", validate_functional_completeness),
        ("性能基準測試", run_performance_benchmarks)
    ]
    
    print("開始部署前驗證...")
    results = {}
    
    for step_name, validation_func in validation_steps:
        print(f"\n驗證: {step_name}...")
        try:
            result = validation_func()
            results[step_name] = ("? 通過", result)
            print("? 通過")
        except Exception as e:
            results[step_name] = ("? 失敗", str(e))
            print(f"? 失敗: {e}")
    
    # 生成驗證報告
    print("\n" + "="*60)
    print("部署前驗證報告")
    print("="*60)
    
    all_passed = all("通過" in result[0] for result in results.values())
    
    for step_name, (status, details) in results.items():
        print(f"{step_name:20} {status}")
        if details and "失敗" in status:
            print(f"  詳細信息: {details}")
    
    print(f"\n總體結果: {'? 所有檢查通過，可以部署' if all_passed else '? 存在未通過檢查項'}")
    return all_passed, results

def validate_environment_dependencies():
    """驗證環(huán)境依賴"""
    return EnvironmentValidator.validate_environment()[0]

def validate_configuration_files():
    """驗證配置文件"""
    config = ConfigManager()
    required_sections = ['data_sources', 'processing', 'logging']
    
    for section in required_sections:
        if section not in config.config:
            raise ValueError(f"缺少配置段: {section}")
    
    return True

def validate_data_source_connectivity():
    """驗證數(shù)據(jù)源連通性"""
    # 這里實現(xiàn)具體的數(shù)據(jù)源連通性測試
    return True

def validate_functional_completeness():
    """驗證功能完整性"""
    # 這里實現(xiàn)核心功能的完整性測試
    return True

def run_performance_benchmarks():
    """運行性能基準測試"""
    # 這里實現(xiàn)性能基準測試
    return "性能測試通過"

# 運行部署前驗證
if __name__ == "__main__":
    deployment_ready, validation_report = pre_deployment_validation()

# 生產(chǎn)環(huán)境配置示例
PRODUCTION_CONFIG = {
    "data_sources": {
        "database": {
            "timeout": 60,
            "pool_size": 10,
            "max_overflow": 20,
            "pool_recycle": 3600
        }
    },
    "processing": {
        "max_workers": 8,
        "chunk_size": 50000,
        "temp_directory": "/data/temp"
    },
    "monitoring": {
        "enable_metrics": True,
        "alert_threshold": 0.85,  # 質量分數(shù)告警閾值
        "slack_webhook": "https://hooks.slack.com/..."  # 告警通知
    }
}