05-设计模式实战¶

from abc import ABC, abstractmethod
from typing import Dict, List, Callable
from dataclasses import dataclass
from enum import Enum

# ================================
# 1. 策略模式：促销规则
# ================================
class PromotionStrategy(ABC):
    @abstractmethod
    def calculate(self, original_price: float) -> float:
        pass

    @abstractmethod
    def description(self) -> str:
        pass

class NoPromotion(PromotionStrategy):
    def calculate(self, price): return price
    def description(self): return "无优惠"

class PercentDiscount(PromotionStrategy):
    def __init__(self, percent):
        self.percent = percent

    def calculate(self, price):
        return price * (1 - self.percent / 100)

    def description(self):
        return f"{self.percent}%折扣"

class FullReduction(PromotionStrategy):
    """满减"""
    def __init__(self, threshold, reduction):
        self.threshold = threshold
        self.reduction = reduction

    def calculate(self, price):
        if price >= self.threshold:
            return price - self.reduction
        return price

    def description(self):
        return f"满{self.threshold}减{self.reduction}"

class StackedPromotion(PromotionStrategy):
    """组合优惠：可叠加多种策略"""
    def __init__(self, strategies: List[PromotionStrategy]):
        self.strategies = strategies

    def calculate(self, price):
        result = price
        for s in self.strategies:
            result = s.calculate(result)
        return result

    def description(self):
        return " + ".join(s.description() for s in self.strategies)

# ================================
# 2. 工厂模式：支付渠道
# ================================
class PaymentProcessor(ABC):
    @abstractmethod
    def process(self, amount: float, order_id: str) -> bool:
        pass

class AlipayProcessor(PaymentProcessor):
    def process(self, amount, order_id):
        print(f"  💳 Alipay: Processing ¥{amount:.2f} for {order_id}")
        return True

class WeChatPayProcessor(PaymentProcessor):
    def process(self, amount, order_id):
        print(f"  💳 WeChatPay: Processing ¥{amount:.2f} for {order_id}")
        return True

class CreditCardProcessor(PaymentProcessor):
    def process(self, amount, order_id):
        print(f"  💳 CreditCard: Processing ¥{amount:.2f} for {order_id}")
        return True

class PaymentFactory:
    """支付处理器工厂"""
    _processors = {
        "alipay": AlipayProcessor,
        "wechat": WeChatPayProcessor,
        "credit_card": CreditCardProcessor,
    }

    @classmethod
    def create(cls, method: str) -> PaymentProcessor:
        processor_cls = cls._processors.get(method)
        if not processor_cls:
            raise ValueError(f"Unsupported payment: {method}")
        return processor_cls()

    @classmethod
    def register(cls, name: str, processor_cls):
        """注册新支付方式，无需修改工厂代码"""
        cls._processors[name] = processor_cls

# ================================
# 3. 观察者模式：订单事件
# ================================
class EventBus:
    """全局事件总线"""
    _instance = None

    def __new__(cls):
        if not cls._instance:
            cls._instance = super().__new__(cls)  # super()调用父类方法
            cls._instance._listeners = {}
        return cls._instance

    def on(self, event: str, listener: Callable):
        self._listeners.setdefault(event, []).append(listener)

    def emit(self, event: str, data=None):
        for listener in self._listeners.get(event, []):
            listener(data)

# ================================
# 整合：OrderService
# ================================
@dataclass
class OrderItem:
    name: str
    price: float
    quantity: int

class OrderService:
    def __init__(self):
        self.event_bus = EventBus()

    def create_order(self, items: List[OrderItem], promotion: PromotionStrategy,
                     payment_method: str) -> dict:
        # 1. 计算价格
        subtotal = sum(item.price * item.quantity for item in items)
        total = promotion.calculate(subtotal)

        order = {
            "id": f"ORD-{id(items) % 10000:04d}",
            "items": items,
            "subtotal": subtotal,
            "promotion": promotion.description(),
            "total": total,
        }

        print(f"\n📦 Order {order['id']}:")
        print(f"   Subtotal: ¥{subtotal:.2f}")
        print(f"   Promotion: {promotion.description()}")
        print(f"   Total: ¥{total:.2f}")

        # 2. 支付（工厂模式创建处理器）
        processor = PaymentFactory.create(payment_method)
        if not processor.process(total, order["id"]):
            raise Exception("Payment failed")

        # 3. 发布事件（观察者模式解耦后续操作）
        self.event_bus.emit("order_created", order)
        self.event_bus.emit("payment_completed", {"order_id": order["id"], "amount": total})

        return order

# ====== 注册事件处理器（各服务独立） ======
event_bus = EventBus()

# 库存服务
event_bus.on("order_created", lambda order:
    print(f"  📦 Inventory: Deducting stock for {order['id']}"))

# 通知服务
event_bus.on("order_created", lambda order:
    print(f"  📧 Notification: Sending confirmation for {order['id']}"))

# 积分服务
event_bus.on("payment_completed", lambda data:
    print(f"  ⭐ Points: Adding {int(data['amount'])} points"))

# 日志服务
event_bus.on("order_created", lambda order:
    print(f"  📝 Logger: Order {order['id']} logged"))

# ====== 使用 ======
service = OrderService()

items = [
    OrderItem("Python Book", 89.0, 1),
    OrderItem("Keyboard", 299.0, 1),
]

# 满300减50 + 会员9折
promo = StackedPromotion([
    FullReduction(300, 50),
    PercentDiscount(10),
])

order = service.create_order(items, promo, "alipay")

from flask import Flask, request, jsonify
from functools import wraps

app = Flask(__name__)

# ====== 装饰器模式：认证 ======
def require_auth(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        token = request.headers.get("Authorization")
        if not token or token != "Bearer valid_token":
            return jsonify({"error": "Unauthorized"}), 401
        return f(*args, **kwargs)
    return decorated

# ====== 策略模式：序列化 ======
class ResponseSerializer:
    """不同格式的响应序列化"""
    serializers = {
        "json": lambda data: jsonify(data),
        "xml": lambda data: f"<response>{data}</response>",
    }

    @classmethod
    def serialize(cls, data, fmt="json"):
        serializer = cls.serializers.get(fmt, cls.serializers["json"])
        return serializer(data)

# ====== 责任链模式：中间件 ======
@app.before_request
def log_request():
    print(f"[LOG] {request.method} {request.path}")

@app.before_request
def check_rate_limit():
    # 简化的限流检查
    pass

# ====== 使用 ======
@app.route("/api/users")
@require_auth
def get_users():
    users = [{"id": 1, "name": "Alice"}]
    fmt = request.args.get("format", "json")
    return ResponseSerializer.serialize(users, fmt)

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import List, Dict
from enum import Enum
import copy

# ================================
# 1. 状态模式：角色状态机
# ================================
class CharacterState(ABC):
    @abstractmethod
    def handle_input(self, character, action): pass

    @abstractmethod
    def update(self, character): pass

class IdleState(CharacterState):
    def handle_input(self, character, action):
        if action == "move":
            character.state = MovingState()
        elif action == "attack":
            character.state = AttackingState()

    def update(self, character):
        character.stamina = min(100, character.stamina + 1)  # 休息恢复

class MovingState(CharacterState):
    def handle_input(self, character, action):
        if action == "stop":
            character.state = IdleState()
        elif action == "attack":
            character.state = AttackingState()

    def update(self, character):
        character.stamina -= 0.5

class AttackingState(CharacterState):
    def __init__(self):
        self.frames = 0

    def handle_input(self, character, action):
        pass  # 攻击中不接受输入

    def update(self, character):
        self.frames += 1
        character.stamina -= 2
        if self.frames >= 10:  # 攻击动画结束
            character.state = IdleState()

# ================================
# 2. 策略模式：攻击方式
# ================================
class AttackStrategy(ABC):
    @abstractmethod
    def calculate_damage(self, attacker, target) -> int:
        pass

class MeleeAttack(AttackStrategy):
    def calculate_damage(self, attacker, target):
        damage = attacker.strength * 2 - target.defense
        return max(1, damage)

class MagicAttack(AttackStrategy):
    def calculate_damage(self, attacker, target):
        damage = attacker.intelligence * 3 - target.magic_defense
        return max(1, damage)

class RangedAttack(AttackStrategy):
    def calculate_damage(self, attacker, target):
        damage = attacker.dexterity * 2.5 - target.defense * 0.5
        return max(1, int(damage))

# ================================
# 3. 观察者模式：游戏事件
# ================================
class GameEventSystem:
    _handlers: Dict[str, List] = {}

    @classmethod  # @classmethod类方法，第一个参数为类本身
    def on(cls, event, handler):
        cls._handlers.setdefault(event, []).append(handler)

    @classmethod
    def emit(cls, event, **kwargs):  # *args接收任意位置参数；**kwargs接收任意关键字参数
        for handler in cls._handlers.get(event, []):
            handler(**kwargs)

# ================================
# 4. 命令模式：可撤销操作
# ================================
class GameCommand(ABC):
    @abstractmethod
    def execute(self): pass
    @abstractmethod
    def undo(self): pass

class MoveCommand(GameCommand):
    def __init__(self, character, dx, dy):
        self.character = character
        self.dx, self.dy = dx, dy

    def execute(self):
        self.character.x += self.dx
        self.character.y += self.dy

    def undo(self):
        self.character.x -= self.dx
        self.character.y -= self.dy

# ================================
# 5. 原型模式：怪物批量生成
# ================================
@dataclass  # 自动生成__init__等方法
class Monster:
    name: str
    hp: int
    attack: int
    defense: int
    loot: List[str] = field(default_factory=list)

    def clone(self):
        return copy.deepcopy(self)

class MonsterFactory:
    """原型工厂：预设怪物模板"""
    _prototypes: Dict[str, Monster] = {}

    @classmethod
    def register(cls, name, prototype: Monster):
        cls._prototypes[name] = prototype

    @classmethod
    def create(cls, name, **overrides) -> Monster:
        proto = cls._prototypes.get(name)
        if not proto:
            raise ValueError(f"Unknown monster: {name}")
        monster = proto.clone()
        for key, value in overrides.items():
            setattr(monster, key, value)  # hasattr/getattr/setattr动态操作对象属性
        return monster

# ====== 注册怪物模板 ======
MonsterFactory.register("slime", Monster("Slime", hp=30, attack=5, defense=2, loot=["jelly"]))
MonsterFactory.register("goblin", Monster("Goblin", hp=50, attack=12, defense=5, loot=["gold", "dagger"]))
MonsterFactory.register("dragon", Monster("Dragon", hp=500, attack=80, defense=40, loot=["dragon_scale"]))

# 批量生成：基于原型快速创建不同变种
slime1 = MonsterFactory.create("slime")
slime_boss = MonsterFactory.create("slime", name="King Slime", hp=200, attack=25)
goblin = MonsterFactory.create("goblin")

print(f"{slime1.name}: HP={slime1.hp}")          # Slime: HP=30
print(f"{slime_boss.name}: HP={slime_boss.hp}")   # King Slime: HP=200

# ====== 注册事件处理 ======
GameEventSystem.on("monster_killed", lambda **kw:  # lambda匿名函数：简洁的单行函数
    print(f"  🎉 {kw['monster']} defeated! Loot: {kw.get('loot', [])}"))

GameEventSystem.on("monster_killed", lambda **kw:
    print(f"  ⭐ +{kw.get('exp', 0)} EXP"))

# 触发事件
GameEventSystem.emit("monster_killed", monster="Goblin", loot=["gold"], exp=50)

# ❌ 反模式：一个类做所有事
class Application:
    def handle_request(self): ...
    def connect_database(self): ...
    def send_email(self): ...
    def render_template(self): ...
    def validate_input(self): ...
    def generate_report(self): ...
    def process_payment(self): ...
    # 几千行代码...

# ❌ 反模式：把所有共享状态都做成单例
class UserManager: ...      # 单例
class OrderManager: ...     # 单例
class ProductManager: ...   # 单例
class ConfigManager: ...    # 单例
# 本质上是全局变量，不利于测试和并发

# ✅ 应该用依赖注入
class OrderService:
    def __init__(self, user_repo, product_repo, config):
        self.user_repo = user_repo
        self.product_repo = product_repo
        self.config = config

# ❌ 过度设计：一个简单功能用了3个模式
class AbstractLoggerFactory(ABC): ...  # ABC抽象基类；abstractmethod强制子类实现
class ConsoleLoggerFactory(AbstractLoggerFactory): ...
class FileLoggerFactory(AbstractLoggerFactory): ...
class LoggerDecorator: ...
class TimestampDecorator(LoggerDecorator): ...
class LevelDecorator(LoggerDecorator): ...
# 实际只需要打印日志...

# ✅ 简单实现
import logging
logger = logging.getLogger(__name__)
logger.info("This is enough!")