from datetime import date, time, datetime, timedelta

d = date(2025, 9, 25)
t = time(14, 30, 0)
dt = datetime(2025, 9, 25, 14, 30, 0)
delta = timedelta(days=7, hours=3)

from datetime import date, datetime

date.today()               # 2025-09-25
datetime.now()             # 2025-09-25 14:30:15.123456
datetime.utcnow()          # UTC 时间（不带时区，不推荐用）
datetime.now(tz=...)       # 带时区的（推荐）

# datetime → 时间戳
datetime.now().timestamp()        # 1758787815.123456   秒

# 时间戳 → datetime
datetime.fromtimestamp(1758787815)
datetime.fromtimestamp(1758787815, tz=timezone.utc)

dt = datetime.now()

# 格式化
dt.strftime("%Y-%m-%d %H:%M:%S")
# '2025-09-25 14:30:15'

# 解析
datetime.strptime("2025-09-25 14:30", "%Y-%m-%d %H:%M")

# 输出
dt.isoformat()              # '2025-09-25T14:30:15.123456'

# 解析
datetime.fromisoformat("2025-09-25T14:30:15")

# date 也支持
date.fromisoformat("2025-09-25")

from datetime import timedelta

delta = timedelta(days=7, hours=3, minutes=30)
print(delta)                # 7 days, 3:30:00
print(delta.total_seconds())  # 613800.0

# 加减
now = datetime.now()
tomorrow = now + timedelta(days=1)
last_week = now - timedelta(weeks=1)

# 两个 datetime 相减 → timedelta
diff = datetime.now() - datetime(2025, 1, 1)
print(diff.days)

from dateutil.relativedelta import relativedelta
next_month = datetime.now() + relativedelta(months=1)
next_year = datetime.now() + relativedelta(years=1)

from datetime import datetime, timezone
from zoneinfo import ZoneInfo

# 建立带时区的时间
utc_now = datetime.now(tz=timezone.utc)
beijing = datetime.now(tz=ZoneInfo("Asia/Shanghai"))
tokyo = datetime.now(tz=ZoneInfo("Asia/Tokyo"))

print(utc_now.isoformat())        # 2025-09-25T06:30:00+00:00
print(beijing.isoformat())         # 2025-09-25T14:30:00+08:00

# 时区之间转换
utc_now.astimezone(ZoneInfo("America/New_York"))

naive = datetime.now()
aware = datetime.now(tz=timezone.utc)

naive < aware       # ❌ TypeError: 不能比较

import time

time.time()             # 当前 UNIX 时间戳（秒，float）
time.time_ns()          # 纳秒

time.sleep(2)           # 阻塞 2 秒
time.monotonic()        # 单调时钟，测耗时用
time.perf_counter()     # 高精度性能计时器（推荐用来测代码耗时）
time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())

import time

start = time.perf_counter()
do_stuff()
elapsed = time.perf_counter() - start
print(f"用了 {elapsed:.4f}s")

# 时间戳 → naive datetime（本地时区）
datetime.fromtimestamp(1758787815)

# 时间戳 → aware datetime（UTC）
datetime.fromtimestamp(1758787815, tz=timezone.utc)

# datetime → 时间戳
dt.timestamp()   # naive 会被当作本地时间；aware 会用其时区

# 加一天：安全
d + timedelta(days=1)

# 加一月：不安全！月长度不定
# 用 relativedelta
from dateutil.relativedelta import relativedelta
d + relativedelta(months=1)

# 计算月末
from calendar import monthrange
year, month = 2025, 9
_, last_day = monthrange(year, month)   # (0, 30)
month_end = date(year, month, last_day)

# 今天开始/结束
today_start = datetime.combine(date.today(), time.min)   # 00:00:00
today_end   = datetime.combine(date.today(), time.max)   # 23:59:59.999999

# 本周一
today = date.today()
monday = today - timedelta(days=today.weekday())   # 0=周一, 6=周日

# 本月 1 号
first_of_month = today.replace(day=1)

# 两个日期之间的天数
(date(2025, 12, 31) - date.today()).days

# 遍历日期区间
def date_range(start, end):
    d = start
    while d < end:
        yield d
        d += timedelta(days=1)

for d in date_range(date(2025, 9, 1), date(2025, 9, 5)):
    print(d)

import locale
locale.setlocale(locale.LC_TIME, "zh_CN")
datetime.now().strftime("%A")   # '星期四'

locale.setlocale(locale.LC_TIME, "en_US")
datetime.now().strftime("%A")   # 'Thursday'

f"{datetime.now():%Y-%m-%d %H:%M:%S}"

# meeting.py
"""把一个会议时间显示给多个时区的参与者。"""

from datetime import datetime
from zoneinfo import ZoneInfo


def format_for(dt_utc: datetime, tz_name: str) -> str:
    tz = ZoneInfo(tz_name)
    local = dt_utc.astimezone(tz)
    return local.strftime(f"%Y-%m-%d %H:%M ({tz_name})")


def schedule_meeting():
    # 组织者在北京，定 2025-09-30 14:00 北京时间
    beijing_time = datetime(2025, 9, 30, 14, 0, tzinfo=ZoneInfo("Asia/Shanghai"))

    # 转 UTC 存储
    utc_time = beijing_time.astimezone(ZoneInfo("UTC"))
    print("存储用：", utc_time.isoformat())

    # 展示给不同参与者
    zones = [
        "Asia/Shanghai",
        "Asia/Tokyo",
        "Europe/London",
        "America/New_York",
        "America/Los_Angeles",
    ]
    for z in zones:
        print(format_for(utc_time, z))


if __name__ == "__main__":
    schedule_meeting()

import re

# 1. 匹配：在字符串开头找
re.match(r"\d+", "123abc")     # <Match object>
re.match(r"\d+", "abc123")     # None，不是开头

# 2. 搜索：在字符串任意位置找
re.search(r"\d+", "abc123def")  # <Match object>

# 3. 找全部
re.findall(r"\d+", "a1b22c333")  # ['1', '22', '333']

# 4. 替换
re.sub(r"\d+", "N", "a1b22c333")  # 'aNbNcN'

# 拆分
re.split(r"\s+", "hello   world  python")  # ['hello', 'world', 'python']

# 迭代匹配
for m in re.finditer(r"\d+", "a1b22c333"):
    print(m.start(), m.group())

re.findall(r"<.+>", "<a><b>")     # ['<a><b>']   贪婪
re.findall(r"<.+?>", "<a><b>")    # ['<a>', '<b>'] 非贪婪

m = re.match(r"(\d{4})-(\d{2})-(\d{2})", "2025-09-25")
m.group()       # '2025-09-25'（整体）
m.group(1)      # '2025'
m.group(2)      # '09'
m.groups()      # ('2025', '09', '25')

# 命名组
m = re.match(r"(?P<year>\d{4})-(?P<mon>\d{2})-(?P<day>\d{2})", "2025-09-25")
m.group("year")    # '2025'
m.groupdict()      # {'year': '2025', 'mon': '09', 'day': '25'}

re.findall("\\d+", s)     # 能用，但难读
re.findall(r"\d+", s)     # ✅ 推荐

pattern = re.compile(r"\d+")
pattern.findall("a1b22")
pattern.sub("N", "a1b22")

# 邮箱（简化版）
re.fullmatch(r"[\w.+-]+@[\w-]+\.[\w.-]+", email)

# 手机号（中国大陆 11 位）
re.fullmatch(r"1[3-9]\d{9}", phone)

# 身份证（18 位）
re.fullmatch(r"\d{17}[\dX]", id_card)

# IPv4
re.fullmatch(r"(\d{1,3}\.){3}\d{1,3}", ip)

# 从日志里抽 IP + 状态码 + 路径
log = '192.168.1.1 - - [2025-09-25:10:11:12] "GET /login HTTP/1.1" 200'
m = re.search(r'(\S+) .+? "(\S+) (\S+) .+?" (\d+)', log)
print(m.groups())
# ('192.168.1.1', 'GET', '/login', '200')

# 拆分句子（多种标点）
re.split(r"[,.!?、；。！？]\s*", "hi, 咖飞. how? 学 Python！")
# ['hi', '咖飞', 'how', '学 Python', '']

# 简单替换
re.sub(r"\d+", "N", "abc 123 def 456")     # 'abc N def N'

# 引用捕获组：\1 或 \g<name>
re.sub(r"(\w+)@\w+", r"\1@匿名", "user@example.com")
# 'user@匿名'

# 用函数做动态替换
def to_upper(m):
    return m.group().upper()

re.sub(r"[a-z]+", to_upper, "hi 咖飞 abc")     # 'HI 咖飞 ABC'

# 只替换前 N 次
re.sub(r"\d", "N", "1234", count=2)     # 'NN34'

# 交换年月
re.sub(r"(?P<y>\d{4})-(?P<m>\d{2})", r"\g<m>/\g<y>", "2025-09")
# '09/2025'

re.IGNORECASE / re.I     # 忽略大小写
re.MULTILINE / re.M      # ^ 和 $ 匹配每行的开头结尾
re.DOTALL / re.S         # . 匹配换行
re.VERBOSE / re.X        # 允许写多行 + 注释

# 多个 flag 用 | 组合
re.findall(r"hello", "Hello", re.I)   # ['Hello']

# 写多行正则
pattern = re.compile(r"""
    (\d{4})           # 年
    -
    (\d{2})           # 月
    -
    (\d{2})           # 日
""", re.VERBOSE)

re.findall(r"<.+>", "<h1>hello</h1>")   # ['<h1>hello</h1>']  整个吃掉
re.findall(r"<.+?>", "<h1>hello</h1>")  # ['<h1>', '</h1>']   非贪婪

re.findall(r"\d", "1 二 3")            # ['1', '3']
re.findall(r"\d", "1 二 3", re.ASCII)  # ['1', '3']
re.findall(r"[0-9]", "1 二 3")         # ['1', '3']   更明确

re.findall(r"[一-龥]+", "hi 咖飞 abc 学 Python")
# ['咖飞', '学']

if m := re.match(r"(\d+)-(\d+)", s):
    print(m.group(1), m.group(2))
else:
    print("不匹配")

# 去除多余空白
re.sub(r"\s+", " ", text).strip()

# 只保留字母数字
re.sub(r"[^a-zA-Z0-9]", "", text)

# 只保留中文
re.sub(r"[^一-龥]", "", text)

# 驼峰转下划线
re.sub(r"([A-Z])", r"_\1", "UserProfileID").lower().lstrip("_")
# 'user_profile_i_d'   最后一个 D 也被分开，业务上一般够用

# 更精细的驼峰
re.sub(r"(?<!^)(?=[A-Z])", "_", "UserProfileID").lower()
# 'user_profile_i_d'

# masker.py
"""对日志行做敏感信息脱敏。"""

import re

PATTERNS = [
    # 手机号
    (re.compile(r"1[3-9]\d{9}"),
     lambda m: m.group()[:3] + "****" + m.group()[-4:]),

    # 邮箱：xxx@yyy → x***@yyy
    (re.compile(r"([\w.+-])[\w.+-]*(@[\w.-]+)"),
     r"\1***\2"),

    # 身份证 18 位
    (re.compile(r"\d{6}(\d{8})\d{3}[\dX]"),
     lambda m: m.group()[:6] + "********" + m.group()[-4:]),

    # 银行卡（简化：连续 16-19 位数字）
    (re.compile(r"\b\d{16,19}\b"),
     lambda m: m.group()[:4] + " **** **** " + m.group()[-4:]),
]


def mask(line: str) -> str:
    for pattern, repl in PATTERNS:
        line = pattern.sub(repl, line)
    return line


if __name__ == "__main__":
    tests = [
        "user phone 13812345678 login",
        "email kafei.gk@example.com",
        "id: 410102199001011234",
        "card 6222 0212 3456 7890",
    ]
    for t in tests:
        print(mask(t))

import functools

def retry(times=3, delay=1):
    """外层：接收装饰器参数，返回真正的装饰器。"""
    def decorator(func):
        """真正的装饰器：接收函数，返回包装。"""
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            """包装：真正执行的地方。"""
            for i in range(times):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if i == times - 1:
                        raise
                    print(f"[{i+1}] 失败：{e}")
                    import time; time.sleep(delay)
        return wrapper
    return decorator


@retry(times=3, delay=0.5)
def flaky():
    import random
    if random.random() < 0.7:
        raise RuntimeError("挂了")
    return "OK"

import functools

def log(_func=None, *, level="INFO"):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            print(f"[{level}] 调用 {func.__name__}")
            return func(*args, **kwargs)
        return wrapper

    if _func is None:
        # 用 @log(level=...) 形式：只传关键字参数，_func 是 None
        return decorator
    # 用 @log 形式：_func 是被装饰的函数
    return decorator(_func)


@log
def a():
    print("a")


@log(level="DEBUG")
def b():
    print("b")


a()      # [INFO] 调用 a  → a
b()      # [DEBUG] 调用 b  → b

class CountCalls:
    def __init__(self, func):
        functools.update_wrapper(self, func)   # 类似 wraps
        self.func = func
        self.count = 0

    def __call__(self, *args, **kwargs):
        self.count += 1
        return self.func(*args, **kwargs)


@CountCalls
def greet(name):
    print(f"你好，{name}")


greet("咖飞")
greet("小王")
print(greet.count)     # 2

class Retry:
    def __init__(self, times=3):
        self.times = times

    def __call__(self, func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for _ in range(self.times):
                try:
                    return func(*args, **kwargs)
                except Exception:
                    pass
            raise
        return wrapper


@Retry(times=5)
def foo():
    ...

def add_repr(cls):
    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, y):
        self.x, self.y = x, y


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

from functools import lru_cache, cache

@lru_cache(maxsize=128)
def fib(n):
    if n < 2:
        return n
    return fib(n-1) + fib(n-2)


# Python 3.9+ 无界版
@cache
def fib(n): ...


fib(50)         # 秒出，不用 lru_cache 要算好久
fib.cache_info()   # CacheInfo(hits=X, misses=Y, maxsize=128, currsize=Z)
fib.cache_clear()  # 清缓存

def logged(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        print(f"[LOG] {func.__name__} 被调")
        return func(*args, **kwargs)
    return wrapper


class Service:
    @logged
    def run(self):
        print("running")


Service().run()

import functools

def deco(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

import functools
import inspect

def keep_signature(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    # 手动设 signature
    wrapper.__signature__ = inspect.signature(func)
    return wrapper

def deco(func):
    def wrapper(x, y):
        return func(x + 1, y + 1)
    return wrapper


@deco
def add(x, y, z=10):
    return x + y + z


add(1, 2)          # 5
add(1, 2, z=100)   # ❌ wrapper 不接受 z

def deco(func):
    def wrapper(*args, **kwargs):
        # 处理 args/kwargs
        return func(*args, **kwargs)
    return wrapper

@a
@b
@c
def f(): ...

def log(func):
    @functools.wraps(func)
    def w(*args, **kw):
        print(f"log begin {func.__name__}")
        r = func(*args, **kw)
        print(f"log end {func.__name__}")
        return r
    return w

def timed(func):
    @functools.wraps(func)
    def w(*args, **kw):
        print(f"timer start")
        r = func(*args, **kw)
        print(f"timer end")
        return r
    return w


@log
@timed
def work():
    print("working")


work()
# log begin work
# timer start
# working
# timer end
# log end work

class register:
    """把方法注册到某个字典。"""
    _registry = {}

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

    def __call__(self, func):
        type(self)._registry[self.name] = func
        return func


@register("add")
def do_add(a, b):
    return a + b


@register("mul")
def do_mul(a, b):
    return a * b


print(register._registry["add"](3, 4))   # 7
print(register._registry["mul"](3, 4))   # 12

def require_role(role):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            if role not in current_user["roles"]:
                raise PermissionError()
            return func(*args, **kwargs)
        return wrapper
    return decorator

def validate_positive(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        for a in args:
            if isinstance(a, (int, float)) and a < 0:
                raise ValueError(f"参数不能为负：{a}")
        return func(*args, **kwargs)
    return wrapper

import time
from functools import wraps

def ttl_cache(ttl=60):
    def decorator(func):
        cache = {}
        @wraps(func)
        def wrapper(*args, **kwargs):
            key = (args, tuple(sorted(kwargs.items())))
            now = time.time()
            if key in cache:
                val, ts = cache[key]
                if now - ts < ttl:
                    return val
            val = func(*args, **kwargs)
            cache[key] = (val, now)
            return val
        return wrapper
    return decorator

def singleton(cls):
    instances = {}
    def get_instance(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]
    return get_instance


@singleton
class Config:
    pass


Config() is Config()    # True

# webapi.py
"""一个 mini 版的 Flask-like 路由装饰器。"""

import functools
from typing import Callable


class App:
    def __init__(self):
        self.routes: dict[tuple[str, str], Callable] = {}

    def route(self, path: str, method: str = "GET"):
        def decorator(func):
            @functools.wraps(func)
            def wrapper(*args, **kwargs):
                print(f"[{method}] {path}")
                return func(*args, **kwargs)
            self.routes[(method, path)] = wrapper
            return wrapper
        return decorator

    def dispatch(self, method: str, path: str):
        handler = self.routes.get((method, path))
        if handler is None:
            return 404, "Not Found"
        return 200, handler()


app = App()


@app.route("/")
def home():
    return "首页"


@app.route("/hello", method="POST")
def hello():
    return "hi 咖飞"


if __name__ == "__main__":
    print(app.dispatch("GET", "/"))
    print(app.dispatch("POST", "/hello"))
    print(app.dispatch("GET", "/miss"))

def count_down(n):
    while n > 0:
        yield n
        n -= 1

g = count_down(3)
print(g)                  # <generator object count_down at ...>

print(next(g))            # 3
print(next(g))            # 2
print(next(g))            # 1
print(next(g))            # StopIteration

for x in count_down(3):
    print(x)      # 3 2 1

def demo():
    print("A")
    yield 1
    print("B")
    yield 2
    print("C")


g = demo()
print(next(g))    # A → yield 1
print(next(g))    # B → yield 2
print(next(g))    # C → StopIteration

# 列表推导式：立即算出全部
squares_list = [x*x for x in range(10)]

# 生成表达式：惰性
squares_gen = (x*x for x in range(10))

print(next(squares_gen))    # 0
print(next(squares_gen))    # 1

sum(x*x for x in range(1_000_000))   # 内存 O(1)
max(u.age for u in users)

def read_large_file(path):
    with open(path, encoding="utf-8") as f:
        for line in f:
            yield line.rstrip()


# 处理一个 100GB 的日志文件
for line in read_large_file("huge.log"):
    if "ERROR" in line:
        print(line)
        # 想早退随时 break

def sub():
    yield 1
    yield 2

def main():
    yield 0
    yield from sub()      # 委托：把 sub 产出的一次性 yield 出去
    yield 3


list(main())     # [0, 1, 2, 3]

for v in x:
    yield v

def flatten(nested):
    for item in nested:
        if isinstance(item, list):
            yield from flatten(item)     # 递归 yield
        else:
            yield item


list(flatten([1, [2, [3, [4, 5]]], 6]))   # [1, 2, 3, 4, 5, 6]

def read_lines(path):
    with open(path, encoding="utf-8") as f:
        for line in f:
            yield line

def filter_errors(lines):
    for line in lines:
        if "ERROR" in line:
            yield line

def parse(lines):
    for line in lines:
        parts = line.split()
        yield {"time": parts[0], "msg": " ".join(parts[1:])}


# 链起来
pipeline = parse(filter_errors(read_lines("huge.log")))
for event in pipeline:
    print(event)

def echo():
    while True:
        received = yield
        print("收到：", received)


g = echo()
next(g)               # 启动：跑到第一个 yield 处停下
g.send("hi")          # 收到：hi
g.send("hello")       # 收到：hello

def gen():
    yield 1
    yield 2
    return "done"


g = gen()
next(g)          # 1
next(g)          # 2
try:
    next(g)      # StopIteration
except StopIteration as e:
    print(e.value)     # 'done'

def outer():
    result = yield from gen()
    print(f"内层返回：{result}")


list(outer())    # [1, 2] + 打印"内层返回：done"

g = count_down(10)
next(g)
g.close()        # 关闭生成器，触发 GeneratorExit

g.throw(ValueError, "错")     # 在生成器内部抛异常

g = (x*x for x in range(5))
list(g)     # [0, 1, 4, 9, 16]
list(g)     # []

g = (x for x in range(5))
g[0]     # ❌ TypeError

len((x for x in range(5)))    # ❌

def naturals():
    n = 1
    while True:
        yield n
        n += 1


from itertools import islice
list(islice(naturals(), 10))    # [1..10]

from itertools import islice

def chunks(seq, n):
    it = iter(seq)
    while chunk := list(islice(it, n)):
        yield chunk


list(chunks(range(10), 3))    # [[0,1,2], [3,4,5], [6,7,8], [9]]

from collections import deque

def window(seq, size):
    it = iter(seq)
    buf = deque(islice(it, size), maxlen=size)
    if len(buf) < size:
        return
    yield tuple(buf)
    for x in it:
        buf.append(x)
        yield tuple(buf)


list(window([1, 2, 3, 4, 5], 3))
# [(1, 2, 3), (2, 3, 4), (3, 4, 5)]

def dedupe(seq):
    seen = set()
    for x in seq:
        if x not in seen:
            seen.add(x)
            yield x


list(dedupe([1, 2, 2, 3, 1, 4]))    # [1, 2, 3, 4]

# events.py
"""从日志流里按'空行分隔'的规则切分事件。"""


def read_events(lines):
    """lines 是行迭代器；产出一批批事件（一批 = 空行前的所有行）。"""
    buffer = []
    for line in lines:
        line = line.rstrip()
        if not line:
            if buffer:
                yield buffer
                buffer = []
        else:
            buffer.append(line)
    if buffer:
        yield buffer


def enrich(events):
    """给每个事件加上一个 id。"""
    for i, ev in enumerate(events, 1):
        yield {"id": i, "lines": ev}


def keep_errors(events):
    """只保留含 ERROR 关键字的事件。"""
    for ev in events:
        if any("ERROR" in ln for ln in ev["lines"]):
            yield ev


if __name__ == "__main__":
    sample = [
        "INFO start",
        "INFO loading",
        "",
        "ERROR db failure",
        "INFO retry",
        "",
        "INFO shutdown",
    ]
    pipeline = keep_errors(enrich(read_events(iter(sample))))
    for e in pipeline:
        print(e)
    # {'id': 2, 'lines': ['ERROR db failure', 'INFO retry']}

lst = [1, 2, 3]
it = iter(lst)          # 拿到迭代器

print(next(it))         # 1
print(next(it))         # 2
print(next(it))         # 3
print(next(it))         # StopIteration

for x in [10, 20, 30]:
    print(x)

_iter = iter([10, 20, 30])     # 拿迭代器
while True:
    try:
        x = next(_iter)         # 取下一个
    except StopIteration:
        break
    print(x)

class Counter:
    def __init__(self, start, stop):
        self.i = start
        self.stop = stop

    def __iter__(self):
        return self             # 自己就是迭代器

    def __next__(self):
        if self.i >= self.stop:
            raise StopIteration
        v = self.i
        self.i += 1
        return v


for x in Counter(1, 5):
    print(x)
# 1 2 3 4

c = Counter(1, 5)
list(c)     # [1, 2, 3, 4]
list(c)     # []  已经走完了！

class Range:
    def __init__(self, start, stop):
        self.start, self.stop = start, stop

    def __iter__(self):
        return RangeIter(self.start, self.stop)   # 每次返回新迭代器


class RangeIter:
    def __init__(self, start, stop):
        self.i, self.stop = start, stop

    def __iter__(self):
        return self

    def __next__(self):
        if self.i >= self.stop:
            raise StopIteration
        v = self.i
        self.i += 1
        return v


r = Range(1, 5)
list(r)     # [1, 2, 3, 4]
list(r)     # [1, 2, 3, 4]  可以反复遍历

def counter(start, stop):
    while start < stop:
        yield start
        start += 1


for x in counter(1, 5):
    print(x)
# 1 2 3 4

# 读文件直到遇到空行
with open("data.txt") as f:
    for line in iter(f.readline, ""):
        print(line.rstrip())

# 内存占 8 * 10^9 = 8GB
sum([x*x for x in range(10**9)])       # ❌ OOM

# 内存 O(1)
sum(x*x for x in range(10**9))         # ✅ 生成表达式，慢但能跑完

it = iter([1, 2, 3])
list(it)      # [1, 2, 3]
list(it)      # []  空了

from itertools import count, cycle, chain, islice, repeat

# 无限数字
for i in islice(count(10), 5):
    print(i)
# 10 11 12 13 14

# 无限循环
for x in islice(cycle([1, 2, 3]), 8):
    print(x)
# 1 2 3 1 2 3 1 2

# 拼接多个序列
list(chain([1, 2], [3, 4], [5]))       # [1, 2, 3, 4, 5]

# 截取
list(islice(range(100), 10, 20))       # [10..19]

from collections.abc import Iterable, Iterator

isinstance([1, 2], Iterable)     # True
isinstance([1, 2], Iterator)     # False
isinstance(iter([1, 2]), Iterator)   # True

# enumerate 加索引
for i, x in enumerate(items, start=1):
    print(f"{i}. {x}")

# zip 并行迭代
for a, b in zip(names, ages):
    print(a, b)

# zip 长度不等：以最短为准，严格模式
for a, b in zip(names, ages, strict=True):   # 3.10+
    ...

# 找第一个满足条件的
def first(pred, iterable, default=None):
    for x in iterable:
        if pred(x):
            return x
    return default

# 或者用 next + 生成表达式
first_neg = next((x for x in nums if x < 0), None)

# chunker.py
"""把一个可迭代对象按 n 个一组切分。"""

from itertools import islice
from collections.abc import Iterator, Iterable


def chunked(iterable: Iterable, size: int) -> Iterator[list]:
    """惰性地把 iterable 按 size 分组。"""
    it = iter(iterable)
    while True:
        chunk = list(islice(it, size))
        if not chunk:
            break
        yield chunk


def paginate(items: Iterable, page_size: int, page: int) -> list:
    """获取第 page 页（从 1 开始）。"""
    it = iter(items)
    # 跳过前 (page-1) * page_size 个
    for _ in range((page - 1) * page_size):
        next(it, None)
    return list(islice(it, page_size))


class PeekableIterator:
    """支持 peek 的迭代器包装。"""

    _SENTINEL = object()

    def __init__(self, source):
        self._it = iter(source)
        self._cache = self._SENTINEL

    def __iter__(self):
        return self

    def peek(self, default=None):
        if self._cache is self._SENTINEL:
            try:
                self._cache = next(self._it)
            except StopIteration:
                return default
        return self._cache

    def __next__(self):
        if self._cache is not self._SENTINEL:
            v = self._cache
            self._cache = self._SENTINEL
            return v
        return next(self._it)


if __name__ == "__main__":
    # 分块
    for chunk in chunked(range(11), 3):
        print(chunk)
    # [0, 1, 2] [3, 4, 5] [6, 7, 8] [9, 10]

    # 分页
    print(paginate(range(20), page_size=5, page=3))  # [10, 11, 12, 13, 14]

    # peek
    p = PeekableIterator([1, 2, 3])
    print(p.peek())       # 1
    print(p.peek())       # 1（没消耗）
    print(next(p))        # 1
    print(next(p))        # 2

first, *rest = range(5)         # first=0, rest=[1, 2, 3, 4]
*init, last = "abcde"           # init=['a', 'b', 'c', 'd'], last='e'

with open("data.txt", encoding="utf-8") as f:
    data = f.read()
# 到这里 f 已经自动关闭了

f = open("data.txt", encoding="utf-8")
try:
    data = f.read()
finally:
    f.close()

class Timer:
    def __enter__(self):
        import time
        self.start = time.perf_counter()
        return self          # 会赋给 as 后的名字

    def __exit__(self, exc_type, exc_val, exc_tb):
        import time
        self.end = time.perf_counter()
        print(f"耗时 {self.end - self.start:.4f}s")
        # 返回 True 会吞掉异常，返回 False/None 让异常继续传播

with Timer():
    sum(i * i for i in range(1_000_000))
# 耗时 0.0234s

def __exit__(self, exc_type, exc_val, exc_tb):
    ...

class SafeAPI:
    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is ConnectionError:
            print(f"连接问题：{exc_val}，已重试")
            return True         # 吞掉
        return False            # 其他异常继续抛