class Calculator:
    def add(self, a, b):
        return a + b
    
    def subtract(self, a, b):
        return a - b
    
    def multiply(self, a, b):
        return a * b
    
    def divide(self, a, b):
        if b != 0:
            return a / b
        raise ValueError("除數(shù)不能為零")

# 創(chuàng)建計(jì)算器實(shí)例
calc = Calculator()

# 動態(tài)調(diào)用方法
method_name = "add"
result = getattr(calc, method_name)(5, 3)
print(f"5 + 3 = {result}")  # 輸出: 5 + 3 = 8

method_name = "multiply"
result = getattr(calc, method_name)(4, 6)
print(f"4 × 6 = {result}")  # 輸出: 4 × 6 = 24

def safe_dynamic_call(obj, method_name, *args, **kwargs):
    """安全地動態(tài)調(diào)用方法"""
    if hasattr(obj, method_name):
        method = getattr(obj, method_name)
        if callable(method):
            return method(*args, **kwargs)
        else:
            raise AttributeError(f"'{method_name}' 不是可調(diào)用方法")
    else:
        raise AttributeError(f"對象沒有 '{method_name}' 方法")

# 安全調(diào)用示例
try:
    result = safe_dynamic_call(calc, "add", 10, 5)
    print(f"安全調(diào)用結(jié)果: {result}")
except AttributeError as e:
    print(f"調(diào)用失敗: {e}")

import operator

class TextProcessor:
    def __init__(self, text):
        self.text = text
    
    def uppercase(self):
        return self.text.upper()
    
    def lowercase(self):
        return self.text.lower()
    
    def reverse(self):
        return self.text[::-1]

processor = TextProcessor("Hello World")

# 使用methodcaller創(chuàng)建調(diào)用器
upper_caller = operator.methodcaller('uppercase')
lower_caller = operator.methodcaller('lowercase')
reverse_caller = operator.methodcaller('reverse')

print(upper_caller(processor))   # 輸出: HELLO WORLD
print(lower_caller(processor))   # 輸出: hello world
print(reverse_caller(processor)) # 輸出: dlroW olleH

# 在排序中使用methodcaller
class Person:
    def __init__(self, name, age):
        self.name = name
        self.age = age
    
    def get_name(self):
        return self.name
    
    def get_age(self):
        return self.age

people = [
    Person("Alice", 25),
    Person("Bob", 30),
    Person("Charlie", 20)
]

# 按年齡排序
people.sort(key=operator.methodcaller('get_age'))
for person in people:
    print(f"{person.name}: {person.age}")

class AdvancedExample:
    def __init__(self, value):
        self.value = value
        self._private_data = "敏感信息"
    
    def public_method(self):
        return f"公共方法: {self.value}"
    
    def _protected_method(self):
        return "受保護(hù)的方法"
    
    def __private_method(self):
        return "私有方法"
    
    def dynamic_processor(self, data):
        return f"處理數(shù)據(jù): {data}"

# 創(chuàng)建實(shí)例
obj = AdvancedExample(42)

# 使用dir()查看所有屬性和方法
print("所有屬性和方法:")
for attr in dir(obj):
    if not attr.startswith('__'):  # 過濾掉內(nèi)置特殊方法
        print(f" - {attr}")

# 檢查屬性存在性
print(f"是否有public_method: {hasattr(obj, 'public_method')}")
print(f"是否有不存在的方法: {hasattr(obj, 'nonexistent')}")

# 動態(tài)設(shè)置屬性
setattr(obj, 'new_attribute', '動態(tài)添加的屬性')
print(f"新屬性: {getattr(obj, 'new_attribute')}")

# 動態(tài)調(diào)用
if hasattr(obj, 'dynamic_processor'):
    method = getattr(obj, 'dynamic_processor')
    result = method("測試數(shù)據(jù)")
    print(result)

class SafeDynamicDispatcher:
    """安全的方法調(diào)用分發(fā)器"""
    
    def __init__(self, obj, allowed_methods=None):
        self.obj = obj
        # 設(shè)置允許調(diào)用的方法白名單
        self.allowed_methods = allowed_methods or []
        
        # 自動添加所有公共方法到白名單
        if not self.allowed_methods:
            for attr_name in dir(self.obj):
                attr = getattr(self.obj, attr_name)
                if (callable(attr) and 
                    not attr_name.startswith('_') and 
                    not attr_name.startswith('_')):
                    self.allowed_methods.append(attr_name)
    
    def call_method(self, method_name, *args, **kwargs):
        """安全地調(diào)用方法"""
        # 方法名驗(yàn)證
        if not isinstance(method_name, str):
            raise TypeError("方法名必須是字符串")
        
        # 白名單檢查
        if method_name not in self.allowed_methods:
            raise PermissionError(f"方法 '{method_name}' 不在允許的調(diào)用列表中")
        
        # 存在性檢查
        if not hasattr(self.obj, method_name):
            raise AttributeError(f"對象沒有 '{method_name}' 方法")
        
        method = getattr(self.obj, method_name)
        
        # 可調(diào)用性檢查
        if not callable(method):
            raise TypeError(f"'{method_name}' 不是可調(diào)用方法")
        
        # 執(zhí)行調(diào)用
        try:
            return method(*args, **kwargs)
        except Exception as e:
            raise RuntimeError(f"方法調(diào)用失敗: {e}") from e

# 使用安全分發(fā)器
calculator = Calculator()
dispatcher = SafeDynamicDispatcher(calculator)

try:
    result = dispatcher.call_method('add', 10, 5)
    print(f"安全調(diào)用結(jié)果: {result}")
    
    # 嘗試調(diào)用不存在的方法
    dispatcher.call_method('dangerous_method')
except (PermissionError, AttributeError, RuntimeError) as e:
    print(f"調(diào)用失敗: {e}")

class CommandProcessor:
    """基于動態(tài)方法調(diào)用的命令處理器"""
    
    def __init__(self, target_object):
        self.target = target_object
        self.history = []  # 命令歷史記錄
        self.undo_stack = []  # 撤銷棧
    
    def execute_command(self, command_name, *args, **kwargs):
        """執(zhí)行命令"""
        if not hasattr(self.target, command_name):
            raise ValueError(f"未知命令: {command_name}")
        
        method = getattr(self.target, command_name)
        
        try:
            # 執(zhí)行命令
            result = method(*args, **kwargs)
            # 記錄命令歷史
            self.history.append({
                'command': command_name,
                'args': args,
                'kwargs': kwargs,
                'result': result
            })
            return result
        except Exception as e:
            print(f"命令執(zhí)行失敗: {e}")
            raise
    
    def get_command_history(self):
        """獲取命令歷史"""
        return self.history.copy()
    
    def replay_commands(self, history_records):
        """重放命令序列"""
        results = []
        for record in history_records:
            result = self.execute_command(
                record['command'], 
                *record['args'], 
                **record['kwargs']
            )
            results.append(result)
        return results

# 應(yīng)用示例
class DocumentEditor:
    def __init__(self):
        self.content = ""
        self.clipboard = ""
    
    def insert_text(self, text, position=None):
        """插入文本"""
        if position is None:
            self.content += text
        else:
            self.content = self.content[:position] + text + self.content[position:]
        return self.content
    
    def delete_text(self, start, end):
        """刪除文本"""
        deleted = self.content[start:end]
        self.content = self.content[:start] + self.content[end:]
        return deleted
    
    def copy_text(self, start, end):
        """復(fù)制文本"""
        self.clipboard = self.content[start:end]
        return self.clipboard
    
    def paste_text(self, position=None):
        """粘貼文本"""
        if not self.clipboard:
            return self.content
        return self.insert_text(self.clipboard, position)

# 使用命令處理器
editor = DocumentEditor()
processor = CommandProcessor(editor)

# 執(zhí)行一系列編輯命令
processor.execute_command('insert_text', 'Hello, World!')
processor.execute_command('copy_text', 7, 12)  # 復(fù)制"World"
processor.execute_command('insert_text', ' Python', 12)

print(f"最終內(nèi)容: {editor.content}")
print(f"命令歷史: {[cmd['command'] for cmd in processor.get_command_history()]}")

import importlib
import inspect
from typing import Dict, List, Any

class PluginManager:
    """基于動態(tài)方法調(diào)用的插件管理器"""
    
    def __init__(self):
        self.plugins = {}
        self.hooks = {}
    
    def load_plugin(self, plugin_module_name, plugin_class_name):
        """動態(tài)加載插件"""
        try:
            # 動態(tài)導(dǎo)入模塊
            module = importlib.import_module(plugin_module_name)
            # 獲取插件類
            plugin_class = getattr(module, plugin_class_name)
            # 創(chuàng)建插件實(shí)例
            plugin_instance = plugin_class()
            
            # 注冊插件
            self.plugins[plugin_class_name] = plugin_instance
            
            # 自動注冊插件方法作為鉤子
            self._register_plugin_hooks(plugin_instance, plugin_class_name)
            
            print(f"插件加載成功: {plugin_class_name}")
            return True
            
        except (ImportError, AttributeError) as e:
            print(f"插件加載失敗: {e}")
            return False
    
    def _register_plugin_hooks(self, plugin_instance, plugin_name):
        """注冊插件鉤子方法"""
        for method_name in dir(plugin_instance):
            method = getattr(plugin_instance, method_name)
            
            # 檢查是否是鉤子方法（以hook_開頭）
            if (callable(method) and 
                method_name.startswith('hook_') and 
                not method_name.startswith('__')):
                
                hook_name = method_name[5:]  # 去掉'hook_'前綴
                
                if hook_name not in self.hooks:
                    self.hooks[hook_name] = []
                
                self.hooks[hook_name].append({
                    'plugin': plugin_name,
                    'method': method
                })
                
                print(f"注冊鉤子: {hook_name} -> {plugin_name}.{method_name}")
    
    def call_hook(self, hook_name, *args, **kwargs):
        """調(diào)用鉤子方法"""
        if hook_name not in self.hooks:
            print(f"警告: 未找到鉤子 '{hook_name}'")
            return []
        
        results = []
        for hook in self.hooks[hook_name]:
            try:
                result = hook['method'](*args, **kwargs)
                results.append({
                    'plugin': hook['plugin'],
                    'result': result
                })
            except Exception as e:
                print(f"鉤子執(zhí)行失敗 {hook['plugin']}.{hook_name}: {e}")
        
        return results

# 插件基類
class BasePlugin:
    """插件基類"""
    
    def __init__(self):
        self.name = self.__class__.__name__
    
    def plugin_init(self):
        """插件初始化方法"""
        pass
    
    def plugin_cleanup(self):
        """插件清理方法"""
        pass

# 示例插件
class TextFilterPlugin(BasePlugin):
    """文本過濾插件"""
    
    def hook_text_process(self, text):
        """文本處理鉤子"""
        # 簡單的文本過濾邏輯
        filtered_text = text.replace('不良詞匯', '***')
        return filtered_text
    
    def hook_text_analyze(self, text):
        """文本分析鉤子"""
        word_count = len(text.split())
        return {'word_count': word_count}

class FormatPlugin(BasePlugin):
    """格式處理插件"""
    
    def hook_text_format(self, text):
        """文本格式化鉤子"""
        return text.upper()

# 使用插件系統(tǒng)
def main():
    manager = PluginManager()
    
    # 加載插件
    manager.load_plugin(__name__, 'TextFilterPlugin')  # 加載當(dāng)前模塊的插件
    manager.load_plugin(__name__, 'FormatPlugin')
    
    # 處理文本
    sample_text = "這是一個(gè)包含不良詞匯的測試文本"
    
    # 調(diào)用文本處理鉤子
    processed_results = manager.call_hook('text_process', sample_text)
    for result in processed_results:
        print(f"{result['plugin']} 處理結(jié)果: {result['result']}")
    
    # 調(diào)用文本分析鉤子
    analysis_results = manager.call_hook('text_analyze', sample_text)
    for result in analysis_results:
        print(f"{result['plugin']} 分析結(jié)果: {result['result']}")

if __name__ == "__main__":
    main()

import time
from functools import lru_cache

class OptimizedDynamicCaller:
    """性能優(yōu)化的動態(tài)調(diào)用器"""
    
    def __init__(self, obj):
        self.obj = obj
        self._method_cache = {}
        self._attribute_cache = {}
    
    @lru_cache(maxsize=128)
    def get_cached_method(self, method_name):
        """緩存方法查找結(jié)果"""
        if hasattr(self.obj, method_name):
            method = getattr(self.obj, method_name)
            if callable(method):
                return method
        return None
    
    def call_method(self, method_name, *args, **kwargs):
        """優(yōu)化的方法調(diào)用"""
        # 首先嘗試緩存
        method = self.get_cached_method(method_name)
        
        if method is None:
            # 緩存未命中，重新查找
            if not hasattr(self.obj, method_name):
                raise AttributeError(f"方法不存在: {method_name}")
            
            method = getattr(self.obj, method_name)
            if not callable(method):
                raise TypeError(f"不是可調(diào)用方法: {method_name}")
        
        return method(*args, **kwargs)

class PerformanceCriticalClass:
    def method1(self, x):
        return x * 2
    
    def method2(self, x, y):
        return x + y

# 性能對比測試
obj = PerformanceCriticalClass()
optimized_caller = OptimizedDynamicCaller(obj)

# 測試標(biāo)準(zhǔn)getattr性能
start_time = time.time()
for i in range(10000):
    getattr(obj, 'method1')(i)
standard_time = time.time() - start_time

# 測試優(yōu)化調(diào)用器性能
start_time = time.time()
for i in range(10000):
    optimized_caller.call_method('method1', i)
optimized_time = time.time() - start_time

print(f"標(biāo)準(zhǔn)getattr耗時(shí): {standard_time:.4f}秒")
print(f"優(yōu)化調(diào)用器耗時(shí): {optimized_time:.4f}秒")
print(f"性能提升: {standard_time/optimized_time:.2f}倍")

class DynamicFactory:
    """基于動態(tài)調(diào)用的對象工廠"""
    
    def __init__(self):
        self.creators = {}
    
    def register_creator(self, product_type, creator_method):
        """注冊創(chuàng)建器"""
        self.creators[product_type] = creator_method
    
    def create_product(self, product_type, *args, **kwargs):
        """動態(tài)創(chuàng)建產(chǎn)品"""
        if product_type not in self.creators:
            raise ValueError(f"未知產(chǎn)品類型: {product_type}")
        
        creator = self.creators[product_type]
        return creator(*args, **kwargs)
    
    def create_from_config(self, config):
        """根據(jù)配置動態(tài)創(chuàng)建對象"""
        product_type = config.get('type')
        init_args = config.get('args', [])
        init_kwargs = config.get('kwargs', {})
        
        return self.create_product(product_type, *init_args, **init_kwargs)

# 使用示例
class ProductA:
    def __init__(self, name):
        self.name = name
    
    def operate(self):
        return f"ProductA操作: {self.name}"

class ProductB:
    def __init__(self, value, multiplier=1):
        self.value = value * multiplier
    
    def operate(self):
        return f"ProductB操作: {self.value}"

# 配置驅(qū)動創(chuàng)建
factory = DynamicFactory()
factory.register_creator('product_a', ProductA)
factory.register_creator('product_b', ProductB)

# 根據(jù)配置動態(tài)創(chuàng)建
configs = [
    {'type': 'product_a', 'args': ['測試產(chǎn)品']},
    {'type': 'product_b', 'kwargs': {'value': 10, 'multiplier': 2}}
]

products = []
for config in configs:
    product = factory.create_from_config(config)
    products.append(product)

for product in products:
    print(product.operate())

class WebRouter:
    """簡單的Web路由系統(tǒng)"""
    
    def __init__(self):
        self.routes = {}
        self.error_handlers = {}
    
    def route(self, path, methods=None):
        """路由裝飾器"""
        methods = methods or ['GET']
        
        def decorator(handler):
            if path not in self.routes:
                self.routes[path] = {}
            for method in methods:
                self.routes[path][method.upper()] = handler
            return handler
        return decorator
    
    def error_handler(self, status_code):
        """錯誤處理裝飾器"""
        def decorator(handler):
            self.error_handlers[status_code] = handler
            return handler
        return decorator
    
    def handle_request(self, path, method='GET', **params):
        """處理請求"""
        method = method.upper()
        
        # 查找路由處理器
        if path in self.routes and method in self.routes[path]:
            handler = self.routes[path][method]
            try:
                return handler(**params)
            except Exception as e:
                # 處理內(nèi)部錯誤
                if 500 in self.error_handlers:
                    return self.error_handlers[500](e)
                raise
        else:
            # 處理404錯誤
            if 404 in self.error_handlers:
                return self.error_handlers[404](path)
            return f"404 Not Found: {path}"

# 創(chuàng)建路由實(shí)例
router = WebRouter()

@router.route('/')
def index_handler():
    return "歡迎訪問首頁"

@router.route('/user', methods=['GET', 'POST'])
def user_handler():
    return "用戶頁面"

@router.route('/user/<user_id>')
def user_detail_handler(user_id):
    return f"用戶詳情: {user_id}"

@router.error_handler(404)
def not_found_handler(path):
    return f"頁面未找到: {path}"

@router.error_handler(500)
def internal_error_handler(error):
    return f"服務(wù)器內(nèi)部錯誤: {error}"

# 模擬請求處理
print(router.handle_request('/'))
print(router.handle_request('/user', 'POST'))
print(router.handle_request('/user/123'))
print(router.handle_request('/unknown'))

class DataPipeline:
    """可配置的數(shù)據(jù)處理管道"""
    
    def __init__(self):
        self.processors = []
        self.context = {}
    
    def add_processor(self, processor_name, *args, **kwargs):
        """添加處理器"""
        self.processors.append({
            'name': processor_name,
            'args': args,
            'kwargs': kwargs
        })
        return self  # 支持鏈?zhǔn)秸{(diào)用
    
    def set_context(self, key, value):
        """設(shè)置處理上下文"""
        self.context[key] = value
        return self
    
    def execute(self, data):
        """執(zhí)行數(shù)據(jù)處理管道"""
        current_data = data
        
        for processor_config in self.processors:
            processor_name = processor_config['name']
            
            # 動態(tài)獲取處理器方法
            if not hasattr(self, processor_name):
                raise AttributeError(f"未知處理器: {processor_name}")
            
            processor_method = getattr(self, processor_name)
            
            # 準(zhǔn)備參數(shù)
            args = processor_config['args']
            kwargs = processor_config['kwargs']
            
            # 執(zhí)行處理
            try:
                current_data = processor_method(current_data, *args, **kwargs)
                self.context['last_result'] = current_data
            except Exception as e:
                print(f"處理器執(zhí)行失敗 {processor_name}: {e}")
                raise
        
        return current_data
    
    # 定義各種處理器方法
    def filter_empty(self, data):
        """過濾空值"""
        if isinstance(data, list):
            return [item for item in data if item not in [None, '', []]]
        return data
    
    def transform_uppercase(self, data):
        """轉(zhuǎn)換為大寫"""
        if isinstance(data, str):
            return data.upper()
        elif isinstance(data, list):
            return [item.upper() if isinstance(item, str) else item for item in data]
        return data
    
    def calculate_statistics(self, data):
        """計(jì)算統(tǒng)計(jì)信息"""
        if isinstance(data, list):
            stats = {
                'count': len(data),
                'sum': sum(data) if all(isinstance(x, (int, float)) for x in data) else None,
                'average': sum(data)/len(data) if all(isinstance(x, (int, float)) for x in data) else None
            }
            return stats
        return data

# 使用數(shù)據(jù)處理管道
pipeline = DataPipeline()

# 構(gòu)建處理流程
result = (pipeline
    .add_processor('filter_empty')
    .add_processor('transform_uppercase')
    .set_context('source', 'dynamic_pipeline')
    .execute(['hello', '', 'world', None, 'python']))

print(f"處理結(jié)果: {result}")

# 數(shù)值數(shù)據(jù)處理
numbers = [1, 2, 3, 4, 5]
stats = (DataPipeline()
    .add_processor('calculate_statistics')
    .execute(numbers))

print(f"統(tǒng)計(jì)結(jié)果: {stats}")