03-结构型模式¶

from abc import ABC, abstractmethod

# ====== 目标接口 ======
class PaymentGateway(ABC):
    @abstractmethod
    def pay(self, amount: float) -> bool:
        pass

# ====== 已有的第三方SDK（接口不同） ======
class AlipaySDK:
    def create_payment(self, total_fee, subject=""):
        print(f"Alipay: ¥{total_fee}")
        return {"status": "success"}

class WeChatPaySDK:
    def unified_order(self, total_amount_in_cents):
        print(f"WeChatPay: ¥{total_amount_in_cents / 100}")
        return {"return_code": "SUCCESS"}

class PayPalSDK:
    def make_payment(self, usd_amount):
        print(f"PayPal: ${usd_amount}")
        return True

# ====== 适配器 ======
class AlipayAdapter(PaymentGateway):
    def __init__(self, sdk: AlipaySDK):
        self._sdk = sdk

    def pay(self, amount):
        result = self._sdk.create_payment(total_fee=amount)
        return result.get("status") == "success"

class WeChatPayAdapter(PaymentGateway):
    def __init__(self, sdk: WeChatPaySDK):
        self._sdk = sdk

    def pay(self, amount):
        result = self._sdk.unified_order(total_amount_in_cents=int(amount * 100))
        return result.get("return_code") == "SUCCESS"

class PayPalAdapter(PaymentGateway):
    def __init__(self, sdk: PayPalSDK):
        self._sdk = sdk

    def pay(self, amount):
        return self._sdk.make_payment(usd_amount=amount)

# ====== 使用 ======
def checkout(gateway: PaymentGateway, amount: float):
    """客户端代码只依赖统一接口"""
    if gateway.pay(amount):
        print("Payment successful!")
    else:
        print("Payment failed!")

checkout(AlipayAdapter(AlipaySDK()), 99.9)
checkout(WeChatPayAdapter(WeChatPaySDK()), 99.9)

from abc import ABC, abstractmethod

# ====== 实现维度：渲染方式 ======
class Renderer(ABC):
    @abstractmethod
    def render_circle(self, x, y, radius): pass

    @abstractmethod
    def render_rectangle(self, x, y, w, h): pass

class VectorRenderer(Renderer):
    def render_circle(self, x, y, radius):
        print(f"Drawing circle at ({x},{y}) r={radius} as vector")

    def render_rectangle(self, x, y, w, h):
        print(f"Drawing rect at ({x},{y}) {w}x{h} as vector")

class RasterRenderer(Renderer):
    def render_circle(self, x, y, radius):
        print(f"Drawing circle at ({x},{y}) r={radius} as pixels")

    def render_rectangle(self, x, y, w, h):
        print(f"Drawing rect at ({x},{y}) {w}x{h} as pixels")

# ====== 抽象维度：图形类型 ======
class Shape(ABC):
    def __init__(self, renderer: Renderer):
        self.renderer = renderer  # 桥接：持有渲染器引用

    @abstractmethod
    def draw(self): pass

class Circle(Shape):
    def __init__(self, renderer, x, y, radius):
        super().__init__(renderer)
        self.x, self.y, self.radius = x, y, radius

    def draw(self):
        self.renderer.render_circle(self.x, self.y, self.radius)

class Rectangle(Shape):
    def __init__(self, renderer, x, y, w, h):
        super().__init__(renderer)
        self.x, self.y, self.w, self.h = x, y, w, h

    def draw(self):
        self.renderer.render_rectangle(self.x, self.y, self.w, self.h)

# 两个维度可以自由组合
circle_vector = Circle(VectorRenderer(), 10, 10, 5)
circle_raster = Circle(RasterRenderer(), 10, 10, 5)
circle_vector.draw()
circle_raster.draw()

from abc import ABC, abstractmethod
from typing import List

class FileSystemItem(ABC):
    def __init__(self, name):
        self.name = name

    @abstractmethod
    def get_size(self) -> int:
        pass

    @abstractmethod
    def display(self, indent=0):
        pass

class File(FileSystemItem):
    """叶子节点"""
    def __init__(self, name, size):
        super().__init__(name)
        self.size = size

    def get_size(self):
        return self.size

    def display(self, indent=0):
        print("  " * indent + f"📄 {self.name} ({self.size}KB)")

class Directory(FileSystemItem):
    """组合节点"""
    def __init__(self, name):
        super().__init__(name)
        self.children: List[FileSystemItem] = []

    def add(self, item: FileSystemItem):
        self.children.append(item)
        return self

    def remove(self, item: FileSystemItem):
        self.children.remove(item)

    def get_size(self):
        return sum(child.get_size() for child in self.children)

    def display(self, indent=0):
        print("  " * indent + f"📁 {self.name}/ ({self.get_size()}KB)")
        for child in self.children:
            child.display(indent + 1)

# 构建树形结构
root = Directory("project")
src = Directory("src")
src.add(File("main.py", 15))
src.add(File("utils.py", 8))
root.add(src)
root.add(File("README.md", 3))
root.add(File("requirements.txt", 1))

root.display()
# 📁 project/ (27KB)
#   📁 src/ (23KB)
#     📄 main.py (15KB)
#     📄 utils.py (8KB)
#   📄 README.md (3KB)
#   📄 requirements.txt (1KB)

from abc import ABC, abstractmethod

# ====== 组件接口 ======
class DataSource(ABC):
    @abstractmethod
    def write(self, data: str): pass

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

# ====== 具体组件 ======
class FileDataSource(DataSource):
    def __init__(self, filename):
        self.filename = filename
        self._data = ""

    def write(self, data):
        self._data = data
        print(f"Writing to {self.filename}: {data[:50]}...")  # 切片操作：[start:end:step]提取子序列

    def read(self):
        return self._data

# ====== 装饰器基类 ======
class DataSourceDecorator(DataSource):
    def __init__(self, source: DataSource):
        self._source = source

    def write(self, data):
        self._source.write(data)

    def read(self):
        return self._source.read()

# ====== 具体装饰器 ======
class EncryptionDecorator(DataSourceDecorator):
    """加密装饰器"""
    def write(self, data):
        encrypted = self._encrypt(data)
        super().write(encrypted)  # super()调用父类方法

    def read(self):
        data = super().read()
        return self._decrypt(data)

    def _encrypt(self, data):
        return ''.join(chr(ord(c) + 1) for c in data)

    def _decrypt(self, data):
        return ''.join(chr(ord(c) - 1) for c in data)

class CompressionDecorator(DataSourceDecorator):
    """压缩装饰器"""
    def write(self, data):
        compressed = self._compress(data)
        super().write(compressed)

    def read(self):
        data = super().read()
        return self._decompress(data)

    def _compress(self, data):
        return f"[compressed:{len(data)}]{data[:10]}..."

    def _decompress(self, data):
        return data  # 简化实现

class LoggingDecorator(DataSourceDecorator):
    """日志装饰器"""
    def write(self, data):
        print(f"[LOG] Writing {len(data)} bytes")
        super().write(data)

    def read(self):
        print(f"[LOG] Reading data")
        return super().read()

# ====== 灵活组合 ======
# 基础写入
source = FileDataSource("data.txt")

# 加密 + 压缩 + 日志（像套娃一样层层包装）
source = LoggingDecorator(
    CompressionDecorator(
        EncryptionDecorator(source)
    )
)

source.write("Hello, Design Patterns!")
data = source.read()

import functools
import time

# Python函数装饰器 — 装饰器模式的函数式实现
def timer(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):  # *args接收任意位置参数；**kwargs接收任意关键字参数
        start = time.time()
        result = func(*args, **kwargs)
        elapsed = time.time() - start
        print(f"{func.__name__} took {elapsed:.4f}s")
        return result
    return wrapper

def retry(max_attempts=3):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:  # try/except捕获异常
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise
                    print(f"Retry {attempt + 1}/{max_attempts}: {e}")
        return wrapper
    return decorator

@timer
@retry(max_attempts=3)
def fetch_data(url):
    # 模拟网络请求
    return f"Data from {url}"

class VideoDecoder:
    def decode(self, filename):
        print(f"Decoding video: {filename}")
        return "video_data"

class AudioDecoder:
    def decode(self, filename):
        print(f"Decoding audio: {filename}")
        return "audio_data"

class SubtitleParser:
    def parse(self, filename):
        print(f"Parsing subtitles: {filename}")
        return "subtitle_data"

class VideoRenderer:
    def render(self, video, audio, subtitles):
        print(f"Rendering video with audio and subtitles")

# 外观类：简化复杂操作
class VideoPlayerFacade:
    def __init__(self):
        self._video_decoder = VideoDecoder()
        self._audio_decoder = AudioDecoder()
        self._subtitle_parser = SubtitleParser()
        self._renderer = VideoRenderer()

    def play(self, filename):
        """一个方法搞定复杂的播放流程"""
        video = self._video_decoder.decode(filename)
        audio = self._audio_decoder.decode(filename)
        subtitles = self._subtitle_parser.parse(filename)
        self._renderer.render(video, audio, subtitles)

# 使用：简单调用
player = VideoPlayerFacade()
player.play("movie.mp4")

class TreeType:
    """享元对象：共享的内部状态"""
    def __init__(self, name, color, texture):
        self.name = name
        self.color = color
        self.texture = texture

    def draw(self, x, y):
        print(f"Drawing {self.name} tree at ({x},{y})")

class TreeFactory:
    """享元工厂：管理共享对象"""
    _types = {}

    @classmethod  # @classmethod类方法，第一个参数为类本身
    def get_tree_type(cls, name, color, texture) -> TreeType:
        key = f"{name}_{color}_{texture}"
        if key not in cls._types:
            cls._types[key] = TreeType(name, color, texture)
            print(f"  [New TreeType created: {name}]")
        return cls._types[key]

class Tree:
    """包含外部状态（位置）的具体树"""
    def __init__(self, x, y, tree_type: TreeType):
        self.x = x
        self.y = y
        self.type = tree_type  # 引用共享的享元

    def draw(self):
        self.type.draw(self.x, self.y)

# 创建森林：100万棵树，但只有少数几种类型
import random

forest = []
for _ in range(10):
    tree_type = TreeFactory.get_tree_type(
        random.choice(["Oak", "Pine", "Birch"]),
        random.choice(["Green", "DarkGreen"]),
        "standard"
    )
    forest.append(Tree(random.randint(0, 1000), random.randint(0, 1000), tree_type))

print(f"\nTrees: {len(forest)}, Shared types: {len(TreeFactory._types)}")

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

# ====== 接口 ======
class DataService(ABC):
    @abstractmethod
    def get_data(self, key: str) -> str:
        pass

# ====== 真实服务 ======
class RemoteDataService(DataService):
    def get_data(self, key):
        print(f"  [Remote] Fetching data for '{key}'... (slow)")
        time.sleep(0.5)  # 模拟网络延迟
        return f"Data-{key}-{time.time():.0f}"

# ====== 缓存代理 ======
class CachingProxy(DataService):
    def __init__(self, service: DataService, ttl=60):
        self._service = service
        self._cache = {}
        self._ttl = ttl

    def get_data(self, key):
        now = time.time()
        if key in self._cache:
            data, timestamp = self._cache[key]
            if now - timestamp < self._ttl:
                print(f"  [Cache HIT] '{key}'")
                return data

        print(f"  [Cache MISS] '{key}'")
        data = self._service.get_data(key)
        self._cache[key] = (data, now)
        return data

# ====== 保护代理 ======
class AuthProxy(DataService):
    def __init__(self, service: DataService, allowed_roles: list):
        self._service = service
        self._allowed_roles = allowed_roles
        self._current_role = "guest"

    def login(self, role):
        self._current_role = role

    def get_data(self, key):
        if self._current_role not in self._allowed_roles:
            raise PermissionError(f"Role '{self._current_role}' not allowed!")
        return self._service.get_data(key)

# ====== 日志代理 ======
class LoggingProxy(DataService):
    def __init__(self, service: DataService):
        self._service = service

    def get_data(self, key):
        print(f"[LOG] get_data('{key}') called at {time.strftime('%H:%M:%S')}")
        start = time.time()
        result = self._service.get_data(key)
        elapsed = time.time() - start
        print(f"[LOG] get_data('{key}') returned in {elapsed:.3f}s")
        return result

# ====== 组合使用 ======
service = LoggingProxy(
    CachingProxy(
        RemoteDataService(),
        ttl=10
    )
)

print(service.get_data("user:1"))  # Cache MISS → 远程调用
print(service.get_data("user:1"))  # Cache HIT → 直接返回