元类与抽象基类

# Python 3.10+ - 聚焦 type 字符串
class Person:
    """一个简单的类"""
    species = "Homo sapiens"

    def __init__(self, name: str):
        self.name = name

    def greet(self) -> str:
        return f"Hello, I'm {self.name}"

# 类是对象，因此有类型
print(type(Person))       # <class 'type'>
print(type(int))          # <class 'type'>
print(type(str))          # <class 'type'>
print(type(type))         # <class 'type'>  —— type 自己的类型也是 type

# 类可以赋值给变量、作为参数传递
MyClass = Person
alice = MyClass("Alice")
print(alice.greet())      # Hello, I'm Alice

# 用法一：查看对象类型（最常见）
print(type(42))           # <class 'int'>
print(type("hello"))      # <class 'str'>

# 用法二：动态创建类（元类用法）
# type(name, bases, namespace) -> 返回一个新类

Dog = type('Dog',                                    # 类名
           (object,),                                # 基类元组
           {'species': 'Canis',                      # 类属性和方法
            '__init__': lambda self, name: setattr(self, 'name', name),
            'bark': lambda self: f"{self.name} says woof!"})

buddy = Dog("Buddy")
print(type(buddy))         # <class 'Dog'>
print(buddy.bark())        # Buddy says woof!
print(isinstance(buddy, Dog))  # True

# Python 3.10+
class InspectMeta(type):
    """打印类创建过程中每一步的元类"""

    def __new__(mcs, name: str, bases: tuple, namespace: dict):
        print(f"[__new__]  创建类 '{name}'")
        print(f"          基类: {[b.__name__ for b in bases]}")
        print(f"          属性: {[k for k in namespace if not k.startswith('__')]}")
        cls = super().__new__(mcs, name, bases, namespace)
        return cls

    def __init__(cls, name: str, bases: tuple, namespace: dict):
        print(f"[__init__] 初始化类 '{name}'")
        super().__init__(name, bases, namespace)

    def __call__(cls, *args, **kwargs):
        print(f"[__call__] 调用 {cls.__name__}({args}) 创建实例")
        instance = super().__call__(*args, **kwargs)
        return instance


class Animal(metaclass=InspectMeta):
    """使用自定义元类观察创建过程"""
    species = "Animalia"

    def __init__(self, name: str):
        self.name = name

    def speak(self) -> str:
        return f"{self.name} makes a sound"


# 输出（class 语句被执行时）：
# [__new__]  创建类 'Animal'
#           基类: ['object']
#           属性: ['species', '__init__', 'speak']
# [__init__] 初始化类 'Animal'

print("--- 分割线 ---")

dog = Animal("Buddy")
# 输出：
# [__call__] 调用 Animal(('Buddy'), {}) 创建实例

print(type(Animal))  # <class '__main__.InspectMeta'>

# Python 3.10+
class MetaA(type):
    pass

class MetaB(MetaA):
    pass

class MetaC(type):
    pass

class BaseA(metaclass=MetaB):
    pass

class BaseB(metaclass=MetaA):
    pass

class Child1(BaseA):
    pass
# Child1 的元类是 MetaB（继承自 BaseA）

class Child2(BaseA, BaseB):
    pass
# Child2 的元类是 MetaB（MetaB 是 MetaA 的子类，最具体）

# class Child3(BaseA, metaclass=MetaC):
#     pass
# TypeError: metaclass conflict
# MetaC 不是 MetaB 的子类，无法兼容

# Python 3.10+ —— 三种方式的效果对比

# 方式一：类装饰器
def add_created_at(cls):
    """添加创建时间戳"""
    import time
    original_init = cls.__init__

    def new_init(self, *args, **kwargs):
        original_init(self, *args, **kwargs)
        self._created_at = time.time()

    cls.__init__ = new_init
    return cls

@add_created_at
class DecoratedClass:
    def __init__(self, name):
        self.name = name

# 方式二：__init_subclass__
class SubclassHook:
    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        cls._created_at = "子类创建时记录"

class HookedClass(SubclassHook):
    pass

# 方式三：元类
class TimestampMeta(type):
    def __new__(mcs, name, bases, namespace):
        import time
        namespace['_created_at'] = time.time()
        return super().__new__(mcs, name, bases, namespace)

class MetaClass(metaclass=TimestampMeta):
    pass

# 关键区别：
# - 装饰器：在类创建之后修改，看不到"原始的"类
# - __init_subclass__：只能在子类创建时触发，基类定义时不会触发
# - 元类：在类创建的最底层介入，可以修改 namespace、改变类名、替换基类

# Python 3.10+
class PluginRegistryMeta(type):
    """元类：自动将所有子类注册到一个字典中"""
    registry: dict[str, type] = {}

    def __new__(mcs, name: str, bases: tuple, namespace: dict):
        cls = super().__new__(mcs, name, bases, namespace)
        # 跳过基类，不注册 Plugin 本身
        if name != "Plugin":
            mcs.registry[name] = cls
        return cls


class Plugin(metaclass=PluginRegistryMeta):
    """插件基类"""

    def process(self, data: str) -> str:
        raise NotImplementedError


class JSONPlugin(Plugin):
    def process(self, data: str) -> str:
        import json
        return json.dumps({"data": data})


class UpperPlugin(Plugin):
    def process(self, data: str) -> str:
        return data.upper()


class ReversePlugin(Plugin):
    def process(self, data: str) -> str:
        return data[::-1]


# 所有插件自动注册
print(PluginRegistryMeta.registry)
# {'JSONPlugin': <class '__main__.JSONPlugin'>,
#  'UpperPlugin': <class '__main__.UpperPlugin'>,
#  'ReversePlugin': <class '__main__.ReversePlugin'>}

# 使用注册的插件
for name, plugin_cls in PluginRegistryMeta.registry.items():
    plugin = plugin_cls()
    print(f"{name}: {plugin.process('Hello')}")

# JSONPlugin: {"data": "Hello"}
# UpperPlugin: HELLO
# ReversePlugin: olleH

# Python 3.10+
class SingletonMeta(type):
    """元类实现单例模式"""
    _instances: dict[type, object] = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super().__call__(*args, **kwargs)
        return cls._instances[cls]


class DatabaseConnection(metaclass=SingletonMeta):
    def __init__(self, host: str = "localhost"):
        self.host = host
        print(f"初始化连接 → {self.host}")

    def query(self, sql: str) -> str:
        return f"[{self.host}] 执行: {sql}"


# 多次"创建"，始终返回同一个实例
conn1 = DatabaseConnection("db1.example.com")
# 初始化连接 → db1.example.com
conn2 = DatabaseConnection("db2.example.com")  # host 参数被忽略
# 没有打印！因为 __init__ 没有被再次调用

print(conn1 is conn2)     # True
print(conn2.host)         # db1.example.com（注意不是 db2）

# Python 3.10+
import inspect


class InterfaceMeta(type):
    """元类：确保子类实现了指定接口的所有方法"""

    def __new__(mcs, name: str, bases: tuple, namespace: dict):
        cls = super().__new__(mcs, name, bases, namespace)
        # 只校验直接继承自 Service 的子类
        if name != "Service":
            required = getattr(mcs, '_required_methods', set())
            missing = required - set(namespace.keys())
            if missing:
                raise TypeError(
                    f"类 '{name}' 未实现以下方法: {missing}"
                )
        return cls


class Service(metaclass=InterfaceMeta):
    _required_methods = {"start", "stop", "status"}

    def start(self) -> None:
        raise NotImplementedError

    def stop(self) -> None:
        raise NotImplementedError

    def status(self) -> str:
        raise NotImplementedError


class WebService(Service):        # OK: 实现了所有方法
    def start(self) -> None:
        print("Web 服务启动")

    def stop(self) -> None:
        print("Web 服务停止")

    def status(self) -> str:
        return "running"


# class BadService(Service):      # TypeError!
#     def start(self) -> None:
#         pass
#     # 缺少 stop 和 status

# Python 3.10+
class ModelMeta(type):
    """简化版 ORM 元类：自动收集字段定义"""

    def __new__(mcs, name: str, bases: tuple, namespace: dict):
        # 从类命名空间中提取字段定义
        fields = {}
        for key, value in namespace.items():
            if isinstance(value, Field):
                fields[key] = value
                value.name = key  # 绑定字段名

        cls = super().__new__(mcs, name, bases, namespace)
        cls._fields = fields  # 将字段信息附加到类上
        return cls


class Field:
    """数据库字段描述符"""
    def __init__(self, field_type: type = str, required: bool = False):
        self.field_type = field_type
        self.required = required
        self.name: str = ""


class Model(metaclass=ModelMeta):
    """ORM 基类"""

    def __init__(self, **kwargs):
        for name, field in self._fields.items():
            value = kwargs.get(name)
            if value is None and field.required:
                raise ValueError(f"字段 '{name}' 是必填的")
            setattr(self, name, value)

    def __repr__(self):
        field_values = ", ".join(
            f"{name}={getattr(self, name)!r}"
            for name in self._fields
        )
        return f"{self.__class__.__name__}({field_values})"


class User(Model):
    name = Field(str, required=True)
    age = Field(int)
    email = Field(str)


# 查看自动收集的字段
print(User._fields)
# {'name': <__main__.Field object at ...>,
#  'age': <__main__.Field object at ...>,
#  'email': <__main__.Field object at ...>}

alice = User(name="Alice", age=30, email="alice@example.com")
print(alice)  # User(name='Alice', age=30, email='alice@example.com')

# Python 3.10+
from abc import ABC, abstractmethod


class DatabaseClient(ABC):
    """数据库客户端抽象基类"""

    @abstractmethod
    def connect(self, connection_string: str) -> bool:
        """建立连接，返回 True 表示成功"""
        ...

    @abstractmethod
    def execute(self, query: str) -> list[dict]:
        """执行查询，返回结果列表"""
        ...

    @abstractmethod
    def close(self) -> None:
        """关闭连接"""
        ...

    # 非抽象方法：提供默认实现，子类可覆盖
    def execute_one(self, query: str) -> dict | None:
        """执行查询，只返回第一条结果"""
        results = self.execute(query)
        return results[0] if results else None


class MySQLClient(DatabaseClient):
    def connect(self, connection_string: str) -> bool:
        print(f"连接 MySQL: {connection_string}")
        return True

    def execute(self, query: str) -> list[dict]:
        return [{"id": 1, "name": "test"}]

    def close(self) -> None:
        print("关闭 MySQL 连接")


# 可以实例化
client = MySQLClient()

# 不能实例化抽象类
# client = DatabaseClient()
# TypeError: Can't instantiate abstract class DatabaseClient
# with abstract methods close, connect, execute

# 查看 ABC 内部机制
from abc import ABC, abstractmethod, ABCMeta


class Shape(ABC):
    @abstractmethod
    def area(self) -> float:
        ...

print(Shape.__abstractmethods__)  # frozenset({'area'})

class Circle(Shape):
    def area(self) -> float:
        return 3.14159 * 5 ** 2

print(Circle.__abstractmethods__)  # frozenset()

# 手动查看元类
print(type(Shape))    # <class 'abc.ABCMeta'>
print(type(Circle))   # <class 'abc.ABCMeta'>

# Python 3.10+
from abc import ABC, abstractmethod


class Validator(ABC):
    """数据校验器抽象基类"""

    @abstractmethod
    def validate(self, value) -> bool:
        """校验值是否合法"""
        ...

    # 抽象属性（读）
    @property
    @abstractmethod
    def error_message(self) -> str:
        """校验失败时的错误消息"""
        ...

    # 抽象类方法（Python 3.2+）
    @classmethod
    @abstractmethod
    def from_config(cls, config: dict) -> "Validator":
        """从配置字典创建校验器"""
        ...


class EmailValidator(Validator):
    def validate(self, value: str) -> bool:
        return "@" in value and "." in value

    @property
    def error_message(self) -> str:
        return "请输入有效的邮箱地址"

    @classmethod
    def from_config(cls, config: dict) -> "Validator":
        return cls()


# 使用
v = EmailValidator()
print(v.validate("test@example.com"))  # True
print(v.validate("invalid"))           # False
print(v.error_message)                 # 请输入有效的邮箱地址

# Python 3.10+
from collections.abc import (
    Iterable, Iterator, Generator,
    Sequence, MutableSequence,
    Mapping, MutableMapping,
    Set, MutableSet,
    Callable, Container, Sized,
    Hashable, Collection,
)

# 类型检查示例
def print_iterable_info(obj):
    match obj:
        case _ if isinstance(obj, Sequence):
            print(f"序列: 长度 {len(obj)}, 第三个元素 {obj[2]}")
        case _ if isinstance(obj, Mapping):
            print(f"映射: 键 {list(obj.keys())[:3]}")
        case _ if isinstance(obj, Set):
            print(f"集合: 大小 {len(obj)}")
        case _ if isinstance(obj, Iterable):
            print(f"可迭代对象: 类型 {type(obj).__name__}")
        case _:
            print(f"不可迭代: {type(obj).__name__}")


print_iterable_info([1, 2, 3, 4, 5])       # 序列: 长度 5, 第三个元素 3
print_iterable_info({"a": 1, "b": 2})      # 映射: 键 ['a', 'b']
print_iterable_info({1, 2, 3})             # 集合: 大小 3
print_iterable_info(range(10))             # 序列: 长度 10, 第三个元素 2
print_iterable_info(42)                    # 不可迭代: int

# Python 3.10+
from collections.abc import MutableMapping


class LRUCache(MutableMapping):
    """实现 MutableMapping 接口，自动获得 .get(), .keys(), .items() 等方法"""
    def __init__(self, maxsize: int = 128):
        self.maxsize = maxsize
        self._data: dict = {}

    def __getitem__(self, key):
        value = self._data[key]
        # 移到末尾（最近使用）
        del self._data[key]
        self._data[key] = value
        return value

    def __setitem__(self, key, value):
        if key in self._data:
            del self._data[key]
        elif len(self._data) >= self.maxsize:
            # 删除最久未使用的项（第一个）
            oldest = next(iter(self._data))
            del self._data[oldest]
        self._data[key] = value

    def __delitem__(self, key):
        del self._data[key]

    def __iter__(self):
        return iter(self._data)

    def __len__(self):
        return len(self._data)

    def __repr__(self):
        return f"LRUCache({dict(self._data)})"


cache = LRUCache(maxsize=3)
cache["a"] = 1
cache["b"] = 2
cache["c"] = 3
print(cache)              # LRUCache({'a': 1, 'b': 2, 'c': 3})

cache["d"] = 4            # 触发淘汰，移除 "a"
print(cache)              # LRUCache({'b': 2, 'c': 3, 'd': 4})

# 自动获得的方法（来自 MutableMapping 的 Mixin）
print(cache.get("b"))     # 2
print(cache.get("x", 0))  # 0
print(list(cache.keys())) # ['b', 'c', 'd']

# Python 3.10+
from abc import ABC, abstractmethod
from collections.abc import Sequence


# 一个没有继承 Sequence 的自定义序列类
class MyRange:
    """自定义范围类，实现了 Sequence 协议，但没有继承 Sequence"""
    def __init__(self, start: int, end: int):
        self.start = start
        self.end = end

    def __getitem__(self, index: int) -> int:
        if index >= len(self):
            raise IndexError
        return self.start + index

    def __len__(self) -> int:
        return max(0, self.end - self.start)


r = MyRange(1, 10)

# 鸭子类型：行为上它就是一个序列
print(r[3])               # 4
print(len(r))             # 9
print(list(r))            # [1, 2, 3, 4, 5, 6, 7, 8, 9]

# 但 isinstance 检测不到
print(isinstance(r, Sequence))  # False —— 因为没有继承 Sequence

# 注册为虚拟子类
Sequence.register(MyRange)

# 现在 isinstance 可以识别了
print(isinstance(r, Sequence))  # True
print(issubclass(MyRange, Sequence))  # True

# Python 3.10+ —— 自定义 ABC 的虚拟子类注册
from abc import ABC, abstractmethod


class JSONSerializable(ABC):
    """可 JSON 序列化的对象"""

    @abstractmethod
    def to_json(self) -> str:
        ...

    @classmethod
    def __subclasshook__(cls, subclass):
        """自定义 issubclass 检查逻辑"""
        # 只要类有 to_json 方法，就认为它是 JSONSerializable 的子类
        if cls is JSONSerializable:
            if hasattr(subclass, 'to_json') and callable(subclass.to_json):
                return True
        return NotImplemented


class User:
    """这个类没有继承 JSONSerializable，但有 to_json 方法"""
    def __init__(self, name: str, age: int):
        self.name = name
        self.age = age

    def to_json(self) -> str:
        import json
        return json.dumps({"name": self.name, "age": self.age})


# __subclasshook__ 让 isinstance 自动生效
print(isinstance(User("Alice", 30), JSONSerializable))  # True
print(issubclass(User, JSONSerializable))                # True


class PlainStr(str):
    """这个类没有 to_json 方法"""
    pass

print(isinstance(PlainStr("hello"), JSONSerializable))   # False

# Python 3.10+
class PluginBase:
    """插件基类 —— 使用 __init_subclass__ 替代元类"""
    registry: dict[str, type] = {}

    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        # 自动注册子类
        PluginBase.registry[cls.__name__] = cls
        print(f"注册插件: {cls.__name__}")

    def process(self, data: str) -> str:
        raise NotImplementedError


class JSONHandler(PluginBase):
    def process(self, data: str) -> str:
        import json
        return json.dumps({"data": data})


class XMLHandler(PluginBase):
    def process(self, data: str) -> str:
        return f"<data>{data}</data>"


class CSVHandler(PluginBase):
    def process(self, data: str) -> str:
        return f"value\n{data}"


# 输出：
# 注册插件: JSONHandler
# 注册插件: XMLHandler
# 注册插件: CSVHandler

print(PluginBase.registry)
# {'JSONHandler': ..., 'XMLHandler': ..., 'CSVHandler': ...}

# Python 3.10+
class ORMModel:
    """使用 __init_subclass__ 实现 ORM 模式"""
    _table_name: str = ""
    _fields: dict[str, str] = {}

    def __init_subclass__(cls, *,
                          table_name: str | None = None,
                          abstract: bool = False,
                          **kwargs):
        super().__init_subclass__(**kwargs)

        if abstract:
            return  # 抽象模型不处理

        # 自动设置表名
        cls._table_name = table_name or cls.__name__.lower() + "s"

        # 收集类型注解作为字段定义
        cls._fields = {}
        for base in reversed(cls.__mro__):
            if hasattr(base, '__annotations__'):
                cls._fields.update(base.__annotations__)

        print(f"[ORM] 模型 '{cls.__name__}' → 表 '{cls._table_name}'")
        print(f"       字段: {list(cls._fields.keys())}")


class BaseModel(ORMModel, abstract=True):
    """抽象基模型，不映射到数据库表"""
    id: int
    created_at: str


class User(BaseModel, table_name="users"):
    name: str
    email: str
    age: int


class Post(BaseModel):
    title: str
    content: str
    author_id: int


# 输出：
# [ORM] 模型 'User' → 表 'users'
#        字段: ['id', 'created_at', 'name', 'email', 'age']
# [ORM] 模型 'Post' → 表 'posts'
#        字段: ['id', 'created_at', 'title', 'content', 'author_id']

# Python 3.10+
class MetaA(type):
    pass

class MetaB(type):
    pass

class BaseA(metaclass=MetaA):
    pass

class BaseB(metaclass=MetaB):
    pass

# class C(BaseA, BaseB):  # TypeError!
#     pass
# TypeError: metaclass conflict:
# the metaclass of a derived class must be a (non-strict)
# subclass of the metaclasses of all its bases

# 解决方案：创建一个合成元类
class CombinedMeta(MetaA, MetaB):
    pass

class C(BaseA, BaseB, metaclass=CombinedMeta):
    pass  # OK

# Python 3.10+
class LoggingMixin:
    def __init_subclass__(cls, **kwargs):
        print(f"[LoggingMixin] 子类创建: {cls.__name__}")
        # 忘记调用 super().__init_subclass__(**kwargs) ！！！

class ValidationMixin:
    def __init_subclass__(cls, **kwargs):
        print(f"[ValidationMixin] 子类创建: {cls.__name__}")
        super().__init_subclass__(**kwargs)  # 正确


class Base(LoggingMixin, ValidationMixin):
    pass

# 输出：
# [LoggingMixin] 子类创建: Base
# [ValidationMixin] 子类创建: Base

# 注意：如果 LoggingMixin 中调用了 super().__init_subclass__(**kwargs)，
# 那么 ValidationMixin 的钩子才会被调用。
# 但上例中 LoggingMixin 没调用 super()，所以 ValidationMixin 的钩子被跳过了！

# Python 3.10+
# ❌ 元类版本：过度设计
class ValidateMeta(type):
    def __new__(mcs, name, bases, namespace):
        cls = super().__new__(mcs, name, bases, namespace)
        # 检查某些条件
        if name != "Base":
            for key in ("process", "validate"):
                if key not in namespace:
                    raise TypeError(f"类必须定义 '{key}' 方法")
        return cls

class Base(metaclass=ValidateMeta):
    pass

# ✅ __init_subclass__ 版本：简洁清晰
class Base:
    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        for key in ("process", "validate"):
            if key not in cls.__dict__:
                raise TypeError(f"类 '{cls.__name__}' 必须定义 '{key}' 方法")

# Python 3.10+
import functools
from typing import TypeVar

T = TypeVar('T', bound=type)

# ✅ 类装饰器：对类做后处理
def add_repr(cls: T) -> T:
    """为类添加一个自动化 __repr__"""
    def __repr__(self):
        attrs = ", ".join(f"{k}={v!r}" for k, v in self.__dict__.items())
        return f"{cls.__name__}({attrs})"
    cls.__repr__ = __repr__
    return cls

@add_repr
class Point:
    def __init__(self, x: float, y: float):
        self.x = x
        self.y = y

p = Point(3, 4)
print(p)  # Point(x=3, y=4)

# Python 3.10+
from abc import ABCMeta, abstractmethod

# ✅ 自定义元类兼容 ABC
class ServiceMeta(ABCMeta):
    """服务元类：既支持 abstractmethod，又支持自动注册"""
    registry: dict[str, type] = {}

    def __new__(mcs, name, bases, namespace):
        cls = super().__new__(mcs, name, bases, namespace)
        if name != "BaseService":
            mcs.registry[name] = cls
        return cls

class BaseService(metaclass=ServiceMeta):
    @abstractmethod
    def start(self) -> None:
        ...

class UserService(BaseService):
    def start(self) -> None:
        print("用户服务启动")

# 既有自动注册
print(ServiceMeta.registry)  # {'UserService': ...}

# 又能用 abstractmethod 约束
# class BadService(BaseService):
#     pass
# TypeError: Can't instantiate abstract class BadService with abstract method start

# Python 3.10+
from abc import ABC, abstractmethod

class Worker(ABC):
    """工作器：抽象方法约束 + 子类创建时校验"""

    @abstractmethod
    def execute(self, task: dict) -> dict:
        ...

    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        # 额外的编译时检查
        execute = cls.__dict__.get('execute')
        if execute and hasattr(execute, '__annotations__'):
            print(f"✓ Worker 子类 '{cls.__name__}' 已定义 execute，"
                  f"返回类型: {execute.__annotations__.get('return', 'Any')}")

class ImageProcessor(Worker):
    def execute(self, task: dict) -> dict:
        return {"status": "done", "image": task.get("path")}

# 输出：
# ✓ Worker 子类 'ImageProcessor' 已定义 execute，返回类型: <class 'dict'>