04-行为型模式¶

from abc import ABC, abstractmethod
from typing import List

# ====== 策略接口 ======
class PricingStrategy(ABC):
    @abstractmethod
    def calculate(self, price: float) -> float:
        pass

# ====== 具体策略 ======
class NormalPricing(PricingStrategy):
    def calculate(self, price):
        return price

class VIPPricing(PricingStrategy):
    def calculate(self, price):
        return price * 0.8  # 8折

class PromotionPricing(PricingStrategy):
    def __init__(self, discount_rate):
        self.discount_rate = discount_rate

    def calculate(self, price):
        return price * self.discount_rate

class SecondHalfPrice(PricingStrategy):
    """第二件半价"""
    def calculate(self, price):
        return price * 0.75  # 两件均价

# ====== 上下文 ======
class ShoppingCart:
    def __init__(self, strategy: PricingStrategy = None):
        self._strategy = strategy or NormalPricing()
        self._items: List[tuple] = []

    def set_strategy(self, strategy: PricingStrategy):
        """运行时切换策略"""
        self._strategy = strategy

    def add_item(self, name, price):
        self._items.append((name, price))

    def total(self):
        return sum(self._strategy.calculate(p) for _, p in self._items)

# ====== 使用 ======
cart = ShoppingCart()
cart.add_item("Book", 100)
cart.add_item("Pen", 50)

print(f"Normal: ¥{cart.total()}")        # 150
cart.set_strategy(VIPPricing())
print(f"VIP: ¥{cart.total()}")            # 120
cart.set_strategy(PromotionPricing(0.6))
print(f"Promotion: ¥{cart.total()}")      # 90

from typing import Callable

def normal(price): return price
def vip(price): return price * 0.8
def promotion(price, rate=0.6): return price * rate

class ShoppingCartFP:
    def __init__(self, strategy: Callable = normal):
        self.strategy = strategy
        self.items = []

    def add(self, name, price):
        self.items.append((name, price))

    def total(self):
        return sum(self.strategy(p) for _, p in self.items)

cart = ShoppingCartFP(strategy=vip)
cart.add("Book", 100)
print(cart.total())  # 80.0

from abc import ABC, abstractmethod
from typing import Dict, List, Callable

# ====== 方式一：经典OOP实现 ======
class EventManager:
    """事件管理器（发布-订阅）"""
    def __init__(self):
        self._listeners: Dict[str, List[Callable]] = {}

    def subscribe(self, event_type: str, listener: Callable):
        if event_type not in self._listeners:
            self._listeners[event_type] = []
        self._listeners[event_type].append(listener)

    def unsubscribe(self, event_type: str, listener: Callable):
        if event_type in self._listeners:
            self._listeners[event_type].remove(listener)

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

# ====== 具体发布者 ======
class UserService:
    def __init__(self):
        self.events = EventManager()

    def register(self, username, email):
        user = {"username": username, "email": email}
        print(f"User registered: {username}")
        # 注册成功后发布事件
        self.events.notify("user_registered", user)

    def login(self, username):
        print(f"User logged in: {username}")
        self.events.notify("user_logged_in", {"username": username})

# ====== 具体订阅者 ======
class EmailService:
    def send_welcome(self, user):
        print(f"  📧 Sending welcome email to {user['email']}")

class AnalyticsService:
    def track_registration(self, user):
        print(f"  📊 Tracking registration: {user['username']}")

class NotificationService:
    def on_login(self, data):
        print(f"  🔔 Login notification: {data['username']}")

# ====== 组装 ======
user_service = UserService()

email = EmailService()
analytics = AnalyticsService()
notification = NotificationService()

# 订阅事件
user_service.events.subscribe("user_registered", email.send_welcome)
user_service.events.subscribe("user_registered", analytics.track_registration)
user_service.events.subscribe("user_logged_in", notification.on_login)

# 触发事件
user_service.register("alice", "alice@example.com")
# User registered: alice
#   📧 Sending welcome email to alice@example.com
#   📊 Tracking registration: alice

user_service.login("alice")
# User logged in: alice
#   🔔 Login notification: alice

# 使用 blinker 库实现信号（类似Django的signals）
from collections import defaultdict  # defaultdict带默认值的字典，避免KeyError

class Signal:
    """简化版信号实现"""
    def __init__(self, name=""):
        self.name = name
        self._receivers = []

    def connect(self, func):
        """装饰器方式注册"""
        self._receivers.append(func)
        return func

    def send(self, sender=None, **kwargs):  # *args接收任意位置参数；**kwargs接收任意关键字参数
        results = []
        for receiver in self._receivers:
            results.append(receiver(sender, **kwargs))
        return results

# 定义信号
user_registered = Signal("user_registered")
order_created = Signal("order_created")

# 注册处理器
@user_registered.connect
def handle_new_user(sender, **kwargs):
    print(f"New user: {kwargs.get('username')}")

@user_registered.connect
def send_email(sender, **kwargs):
    print(f"Sending email to: {kwargs.get('email')}")

# 触发
user_registered.send(None, username="bob", email="bob@test.com")

from abc import ABC, abstractmethod
from typing import List

# ====== 命令接口 ======
class Command(ABC):
    @abstractmethod
    def execute(self): pass

    @abstractmethod
    def undo(self): pass

# ====== 接收者 ======
class TextEditor:
    def __init__(self):
        self.content = ""

    def insert(self, text, position=-1):
        if position == -1:
            self.content += text
        else:
            self.content = self.content[:position] + text + self.content[position:]

    def delete(self, start, length):
        deleted = self.content[start:start + length]
        self.content = self.content[:start] + self.content[start + length:]
        return deleted

    def __str__(self):
        return f"'{self.content}'"

# ====== 具体命令 ======
class InsertCommand(Command):
    def __init__(self, editor: TextEditor, text: str, position: int = -1):
        self.editor = editor
        self.text = text
        self.position = position

    def execute(self):
        self.editor.insert(self.text, self.position)

    def undo(self):
        if self.position == -1:
            pos = len(self.editor.content) - len(self.text)
        else:
            pos = self.position
        self.editor.delete(pos, len(self.text))

class DeleteCommand(Command):
    def __init__(self, editor: TextEditor, start: int, length: int):
        self.editor = editor
        self.start = start
        self.length = length
        self._deleted = ""

    def execute(self):
        self._deleted = self.editor.delete(self.start, self.length)

    def undo(self):
        self.editor.insert(self._deleted, self.start)

# ====== 调用者（支持撤销/重做） ======
class CommandManager:
    def __init__(self):
        self._history: List[Command] = []
        self._redo_stack: List[Command] = []

    def execute(self, command: Command):
        command.execute()
        self._history.append(command)
        self._redo_stack.clear()

    def undo(self):
        if not self._history:
            print("Nothing to undo!")
            return
        command = self._history.pop()
        command.undo()
        self._redo_stack.append(command)

    def redo(self):
        if not self._redo_stack:
            print("Nothing to redo!")
            return
        command = self._redo_stack.pop()
        command.execute()
        self._history.append(command)

# ====== 使用 ======
editor = TextEditor()
manager = CommandManager()

manager.execute(InsertCommand(editor, "Hello"))
print(editor)  # 'Hello'

manager.execute(InsertCommand(editor, " World"))
print(editor)  # 'Hello World'

manager.undo()
print(editor)  # 'Hello'

manager.redo()
print(editor)  # 'Hello World'

manager.execute(DeleteCommand(editor, 5, 6))
print(editor)  # 'Hello'

from abc import ABC, abstractmethod

class DataProcessor(ABC):
    """模板方法定义处理流程骨架"""

    def process(self, source):
        """模板方法 — 定义算法骨架，不可覆盖"""
        data = self.read_data(source)
        parsed = self.parse_data(data)
        if self.should_validate():  # 钩子方法
            self.validate(parsed)
        result = self.transform(parsed)
        self.save(result)
        return result

    @abstractmethod
    def read_data(self, source):
        """抽象方法 — 子类必须实现"""
        pass

    @abstractmethod
    def parse_data(self, raw_data):
        pass

    def should_validate(self):
        """钩子方法 — 子类可选择覆盖"""
        return True

    def validate(self, data):
        print(f"Validating {len(data)} records")

    @abstractmethod
    def transform(self, data):
        pass

    def save(self, result):
        print(f"Saved {len(result)} records")

class CSVProcessor(DataProcessor):
    def read_data(self, source):
        print(f"Reading CSV: {source}")
        return "name,age\nAlice,25\nBob,30"

    def parse_data(self, raw_data):
        lines = raw_data.strip().split("\n")
        headers = lines[0].split(",")
        return [dict(zip(headers, line.split(","))) for line in lines[1:]]  # zip并行遍历多个可迭代对象

    def transform(self, data):
        return [{**d, "age": int(d["age"])} for d in data]

class JSONProcessor(DataProcessor):
    def read_data(self, source):
        print(f"Reading JSON: {source}")
        return '[{"name":"Alice","age":25}]'

    def parse_data(self, raw_data):
        import json
        return json.loads(raw_data)  # json.loads将JSON字符串转为Python对象

    def should_validate(self):
        return False  # JSON自带格式校验，跳过

    def transform(self, data):
        return data

# 使用
csv_processor = CSVProcessor()
csv_processor.process("data.csv")

json_processor = JSONProcessor()
json_processor.process("data.json")

class FibonacciIterator:
    """斐波那契数列迭代器"""
    def __init__(self, max_count):
        self.max_count = max_count
        self.count = 0
        self.a, self.b = 0, 1

    def __iter__(self):
        return self

    def __next__(self):
        if self.count >= self.max_count:
            raise StopIteration
        self.count += 1
        result = self.a
        self.a, self.b = self.b, self.a + self.b
        return result

# 使用
for num in FibonacciIterator(10):
    print(num, end=" ")
# 0 1 1 2 3 5 8 13 21 34

# ====== 自定义可迭代集合 ======
class TreeNode:
    def __init__(self, value, children=None):
        self.value = value
        self.children = children or []

class TreeIterator:
    """树的深度优先迭代器"""
    def __init__(self, root: TreeNode):
        self._stack = [root]

    def __iter__(self):
        return self

    def __next__(self):
        if not self._stack:
            raise StopIteration
        node = self._stack.pop()
        # 逆序压栈，保证左子节点先出
        self._stack.extend(reversed(node.children))
        return node.value

class Tree:
    def __init__(self, root: TreeNode):
        self.root = root

    def __iter__(self):
        return TreeIterator(self.root)

# 构建树并遍历
tree = Tree(TreeNode("root", [
    TreeNode("A", [TreeNode("A1"), TreeNode("A2")]),
    TreeNode("B", [TreeNode("B1")]),
]))

print(list(tree))  # ['root', 'A', 'A1', 'A2', 'B', 'B1']

def fibonacci(max_count):
    """生成器函数 — 等价于FibonacciIterator但更简洁"""
    a, b = 0, 1
    for _ in range(max_count):
        yield a  # yield生成器：惰性产出值，节省内存
        a, b = b, a + b

# 树的深度优先遍历（生成器版）
def dfs(node):
    yield node.value
    for child in node.children:
        yield from dfs(child)  # yield from 递归委托

print(list(fibonacci(8)))  # [0, 1, 1, 2, 3, 5, 8, 13]

from abc import ABC, abstractmethod

# ====== 状态接口 ======
class OrderState(ABC):
    @abstractmethod
    def pay(self, order): pass

    @abstractmethod
    def ship(self, order): pass

    @abstractmethod
    def deliver(self, order): pass

    @abstractmethod
    def cancel(self, order): pass

# ====== 具体状态 ======
class PendingState(OrderState):
    def pay(self, order):
        print("✅ Payment received!")
        order.state = PaidState()

    def ship(self, order):
        print("❌ Cannot ship: not paid yet")

    def deliver(self, order):
        print("❌ Cannot deliver: not paid yet")

    def cancel(self, order):
        print("✅ Order cancelled")
        order.state = CancelledState()

class PaidState(OrderState):
    def pay(self, order):
        print("❌ Already paid")

    def ship(self, order):
        print("✅ Order shipped!")
        order.state = ShippedState()

    def deliver(self, order):
        print("❌ Cannot deliver: not shipped yet")

    def cancel(self, order):
        print("✅ Refunding and cancelling")
        order.state = CancelledState()

class ShippedState(OrderState):
    def pay(self, order): print("❌ Already paid")
    def ship(self, order): print("❌ Already shipped")

    def deliver(self, order):
        print("✅ Order delivered!")
        order.state = DeliveredState()

    def cancel(self, order):
        print("❌ Cannot cancel: already shipped")

class DeliveredState(OrderState):
    def pay(self, order): print("❌ Already completed")
    def ship(self, order): print("❌ Already completed")
    def deliver(self, order): print("❌ Already delivered")
    def cancel(self, order): print("❌ Cannot cancel: already delivered")

class CancelledState(OrderState):
    def pay(self, order): print("❌ Order cancelled")
    def ship(self, order): print("❌ Order cancelled")
    def deliver(self, order): print("❌ Order cancelled")
    def cancel(self, order): print("❌ Already cancelled")

# ====== 上下文 ======
class Order:
    def __init__(self, order_id):
        self.order_id = order_id
        self.state: OrderState = PendingState()

    def pay(self): self.state.pay(self)
    def ship(self): self.state.ship(self)
    def deliver(self): self.state.deliver(self)
    def cancel(self): self.state.cancel(self)

    @property
    def status(self):
        return self.state.__class__.__name__.replace("State", "")

# 使用
order = Order("ORD-001")
print(f"Status: {order.status}")  # Pending
order.ship()     # ❌ Cannot ship: not paid yet
order.pay()      # ✅ Payment received!
print(f"Status: {order.status}")  # Paid
order.ship()     # ✅ Order shipped!
order.deliver()  # ✅ Order delivered!
order.cancel()   # ❌ Cannot cancel: already delivered

from abc import ABC, abstractmethod
from typing import Optional

# ====== 中间件/处理器基类 ======
class Middleware(ABC):
    def __init__(self):
        self._next: Optional[Middleware] = None

    def set_next(self, handler: 'Middleware') -> 'Middleware':
        self._next = handler
        return handler  # 支持链式设置

    def handle(self, request: dict) -> dict:
        if self._next:
            return self._next.handle(request)
        return request

# ====== 具体中间件 ======
class AuthMiddleware(Middleware):
    def handle(self, request):
        token = request.get("token")
        if not token:
            return {"error": "401 Unauthorized"}
        if token != "valid_token":
            return {"error": "403 Forbidden"}
        print("  ✅ Auth passed")
        request["user"] = "alice"
        return super().handle(request)  # super()调用父类方法

class RateLimitMiddleware(Middleware):
    def __init__(self, max_requests=100):
        super().__init__()
        self.max_requests = max_requests
        self._count = 0

    def handle(self, request):
        self._count += 1
        if self._count > self.max_requests:
            return {"error": "429 Too Many Requests"}
        print(f"  ✅ Rate limit OK ({self._count}/{self.max_requests})")
        return super().handle(request)

class LoggingMiddleware(Middleware):
    def handle(self, request):
        print(f"  📝 Request: {request.get('path', '/')}")
        response = super().handle(request)
        print(f"  📝 Response: {response}")
        return response

class ValidationMiddleware(Middleware):
    def handle(self, request):
        if not request.get("path"):
            return {"error": "400 Bad Request: missing path"}
        print("  ✅ Validation passed")
        return super().handle(request)

# ====== 构建中间件链 ======
logging = LoggingMiddleware()
rate_limit = RateLimitMiddleware(max_requests=10)
auth = AuthMiddleware()
validation = ValidationMiddleware()

# 设置链: Logging → RateLimit → Auth → Validation
logging.set_next(rate_limit).set_next(auth).set_next(validation)

# 处理请求
print("--- Request 1 (valid) ---")
result = logging.handle({"path": "/api/data", "token": "valid_token"})

print("\n--- Request 2 (no token) ---")
result = logging.handle({"path": "/api/data"})

from abc import ABC, abstractmethod

class ChatRoom:
    """中介者：聊天室"""
    def __init__(self, name):
        self.name = name
        self._users = {}

    def join(self, user):
        self._users[user.name] = user
        user.room = self
        self._broadcast(f"[{user.name} joined the room]", sender=None)

    def send(self, message, sender, target=None):
        if target:
            # 私聊
            if target in self._users:
                self._users[target].receive(message, sender.name, private=True)
        else:
            # 群发
            self._broadcast(message, sender)

    def _broadcast(self, message, sender):
        for name, user in self._users.items():
            if sender is None or name != sender.name:
                user.receive(message, sender.name if sender else "System")

class User:
    def __init__(self, name):
        self.name = name
        self.room: ChatRoom = None

    def send(self, message, target=None):
        if self.room:
            self.room.send(message, self, target)

    def receive(self, message, sender, private=False):
        tag = "[DM]" if private else ""
        print(f"  {self.name} received {tag} from {sender}: {message}")

# 使用
room = ChatRoom("dev-chat")
alice = User("Alice")
bob = User("Bob")
charlie = User("Charlie")

room.join(alice)
room.join(bob)
room.join(charlie)

alice.send("Hello everyone!")   # Bob和Charlie收到
bob.send("Hi Alice!", "Alice")  # 仅Alice收到（私聊）

import copy
from datetime import datetime

class EditorMemento:
    """备忘录：保存编辑器状态快照"""
    def __init__(self, content, cursor_pos, selection):
        self._content = content
        self._cursor_pos = cursor_pos
        self._selection = selection
        self._timestamp = datetime.now()

    @property  # @property将方法变为属性访问
    def timestamp(self):
        return self._timestamp

    @property
    def description(self):
        preview = self._content[:30] + "..." if len(self._content) > 30 else self._content  # 切片操作：[start:end:step]提取子序列
        return f"[{self._timestamp.strftime('%H:%M:%S')}] '{preview}'"

class Editor:
    """发起者：文本编辑器"""
    def __init__(self):
        self.content = ""
        self.cursor_pos = 0
        self.selection = ""

    def type_text(self, text):
        self.content += text
        self.cursor_pos = len(self.content)

    def save(self) -> EditorMemento:
        return EditorMemento(
            copy.copy(self.content),
            self.cursor_pos,
            self.selection
        )

    def restore(self, memento: EditorMemento):
        self.content = memento._content
        self.cursor_pos = memento._cursor_pos
        self.selection = memento._selection

class History:
    """管理者：保存备忘录历史"""
    def __init__(self, editor: Editor):
        self._editor = editor
        self._snapshots = []

    def backup(self):
        self._snapshots.append(self._editor.save())

    def undo(self):
        if not self._snapshots:
            return
        memento = self._snapshots.pop()
        self._editor.restore(memento)

    def show_history(self):
        for i, s in enumerate(self._snapshots):  # enumerate同时获取索引和值
            print(f"  {i}: {s.description}")

# 使用
editor = Editor()
history = History(editor)

editor.type_text("Hello")
history.backup()

editor.type_text(" World")
history.backup()

editor.type_text("!!!")
print(f"Current: '{editor.content}'")  # Hello World!!!

history.undo()
print(f"After undo: '{editor.content}'")  # Hello World

history.undo()
print(f"After undo: '{editor.content}'")  # Hello

from abc import ABC, abstractmethod

# ====== 元素 ======
class Shape(ABC):
    @abstractmethod
    def accept(self, visitor): pass

class Circle(Shape):
    def __init__(self, radius):
        self.radius = radius

    def accept(self, visitor):
        return visitor.visit_circle(self)

class Rectangle(Shape):
    def __init__(self, width, height):
        self.width = width
        self.height = height

    def accept(self, visitor):
        return visitor.visit_rectangle(self)

# ====== 访问者 ======
class ShapeVisitor(ABC):
    @abstractmethod
    def visit_circle(self, circle): pass
    @abstractmethod
    def visit_rectangle(self, rect): pass

class AreaCalculator(ShapeVisitor):
    """计算面积的访问者"""
    def visit_circle(self, c):
        import math
        return math.pi * c.radius ** 2

    def visit_rectangle(self, r):
        return r.width * r.height

class DrawVisitor(ShapeVisitor):
    """绘制的访问者"""
    def visit_circle(self, c):
        return f"Drawing circle with r={c.radius}"

    def visit_rectangle(self, r):
        return f"Drawing rectangle {r.width}x{r.height}"

# 新增操作不需要修改Shape类
shapes = [Circle(5), Rectangle(3, 4), Circle(10)]

area_calc = AreaCalculator()
total = sum(s.accept(area_calc) for s in shapes)
print(f"Total area: {total:.2f}")

drawer = DrawVisitor()
for s in shapes:
    print(s.accept(drawer))

from abc import ABC, abstractmethod  # ABC抽象基类；abstractmethod强制子类实现

class Expression(ABC):
    @abstractmethod
    def interpret(self, context: dict) -> bool:
        pass

class VariableExpression(Expression):
    def __init__(self, name):
        self.name = name

    def interpret(self, context):
        return context.get(self.name, False)

class AndExpression(Expression):
    def __init__(self, left, right):
        self.left = left
        self.right = right

    def interpret(self, context):
        return self.left.interpret(context) and self.right.interpret(context)

class OrExpression(Expression):
    def __init__(self, left, right):
        self.left = left
        self.right = right

    def interpret(self, context):
        return self.left.interpret(context) or self.right.interpret(context)

class NotExpression(Expression):
    def __init__(self, expr):
        self.expr = expr

    def interpret(self, context):
        return not self.expr.interpret(context)

# 构建规则: (is_vip AND has_balance) OR is_admin
rule = OrExpression(
    AndExpression(VariableExpression("is_vip"), VariableExpression("has_balance")),
    VariableExpression("is_admin")
)

print(rule.interpret({"is_vip": True, "has_balance": True, "is_admin": False}))   # True
print(rule.interpret({"is_vip": True, "has_balance": False, "is_admin": False}))  # False
print(rule.interpret({"is_vip": False, "has_balance": False, "is_admin": True}))  # True