def process(data, options):
    ...

def process(data: list[dict], options: ProcessOptions) -> Result:
    ...

name: str = "咖飞"
age: int = 3
prices: list[float] = [3.14, 2.71]
data: dict[str, int] = {"a": 1, "b": 2}

def add(a: int, b: int) -> int:
    return a + b

def greet(name: str) -> None:
    print(f"hi {name}")

int          # 整数
float        # 浮点
str          # 字符串
bool         # 布尔
bytes        # 字节
None         # 空值
list[X]      # 元素为 X 的列表
tuple[X, Y]  # 定长元组
tuple[X, ...]  # 可变长同类型元组
dict[K, V]   # 字典
set[X]       # 集合
frozenset[X]

def find_user(id: int) -> Optional[User]:
    """找不到时返回 None。"""

# Python 3.10+ 更简洁：
def find_user(id: int) -> User | None:
    ...

def parse(x: Union[str, int]) -> str:
    return str(x)

# Python 3.10+
def parse(x: str | int) -> str:
    ...

from typing import Any

def call_anything(obj: Any) -> Any:
    return obj.something()   # 类型检查器不管

from typing import Callable

def apply(func: Callable[[int, int], int], a: int, b: int) -> int:
    return func(a, b)

# 参数任意
def apply_any(func: Callable[..., int]) -> int:
    return func()

from typing import Literal

def align(text: str, side: Literal["left", "right", "center"]) -> str:
    ...

align("hi", "left")     # ✅
align("hi", "top")      # ⚠️ 类型检查报错

from typing import TypedDict

class UserDict(TypedDict):
    name: str
    age: int
    email: str


def process(u: UserDict) -> None:
    print(u["name"], u["age"])


process({"name": "咖飞", "age": 3, "email": "x@y.com"})    # ✅
process({"name": "咖飞", "age": 3})                       # ⚠️ 缺 email

from typing import Protocol

class SupportsClose(Protocol):
    def close(self) -> None: ...


def cleanup(x: SupportsClose) -> None:
    x.close()

from typing import TypeVar

T = TypeVar("T")

def first(items: list[T]) -> T:
    return items[0]


x: int = first([1, 2, 3])         # 类型检查器知道返回 int
s: str = first(["a", "b", "c"])   # 知道返回 str

from typing import TypeVar

Number = TypeVar("Number", int, float)

def double(x: Number) -> Number:
    return x * 2

double(3)       # int
double(3.14)    # float
double("a")     # ⚠️ 报错

def first[T](items: list[T]) -> T:
    return items[0]

from typing import Generic, TypeVar

T = TypeVar("T")

class Stack(Generic[T]):
    def __init__(self) -> None:
        self._items: list[T] = []

    def push(self, item: T) -> None:
        self._items.append(item)

    def pop(self) -> T:
        return self._items.pop()


s: Stack[int] = Stack()
s.push(1)
s.push("hi")     # ⚠️ 报错

class Stack[T]:
    ...

from dataclasses import dataclass

@dataclass
class User:
    name: str
    age: int
    email: str = ""


u = User("咖飞", 3)
print(u.name)         # IDE 补全

from typing import Callable, TypeVar
from functools import wraps

F = TypeVar("F", bound=Callable)

def log(func: F) -> F:
    @wraps(func)
    def wrapper(*args, **kwargs):
        print(f"call {func.__name__}")
        return func(*args, **kwargs)
    return wrapper       # 类型检查器认为返回类型和 func 一样

from typing import ParamSpec, TypeVar, Callable

P = ParamSpec("P")
R = TypeVar("R")

def log(func: Callable[P, R]) -> Callable[P, R]:
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        return func(*args, **kwargs)
    return wrapper

from typing import TypeAlias, NewType

# 别名
Vector: TypeAlias = list[float]

# NewType 是"逻辑上不同"的类型
UserId = NewType("UserId", int)

def get_user(uid: UserId) -> None: ...

get_user(123)                # ⚠️ 是 int，不是 UserId
get_user(UserId(123))        # ✅

type Vector = list[float]

from typing import Self

class Builder:
    def add(self, x: int) -> Self:
        self.items.append(x)
        return self


b = Builder().add(1).add(2)   # 链式，类型检查器知道每步返回 Builder

from typing import get_type_hints

def f(a: int, b: str) -> bool:
    return True

print(get_type_hints(f))     # {'a': <class 'int'>, 'b': <class 'str'>, 'return': <class 'bool'>}

# utils.py
"""一个类型注解齐全的工具库。"""

from __future__ import annotations

from collections.abc import Iterable, Callable
from typing import TypeVar, Protocol
from dataclasses import dataclass
from datetime import datetime


T = TypeVar("T")
K = TypeVar("K")


# ---------- 数据类 ----------

@dataclass
class Result[T]:                # Python 3.12+ 泛型语法
    ok: bool
    value: T | None = None
    error: str | None = None


# ---------- 协议 ----------

class Comparable(Protocol):
    def __lt__(self, other: object) -> bool: ...


# ---------- 泛型工具 ----------

def first(items: Iterable[T], default: T | None = None) -> T | None:
    for x in items:
        return x
    return default


def group_by[K, T](items: Iterable[T], key: Callable[[T], K]) -> dict[K, list[T]]:
    result: dict[K, list[T]] = {}
    for x in items:
        result.setdefault(key(x), []).append(x)
    return result


def find_max[T: Comparable](items: Iterable[T]) -> T | None:
    it = iter(items)
    try:
        best = next(it)
    except StopIteration:
        return None
    for x in it:
        if best < x:
            best = x
    return best


# ---------- 使用 ----------

if __name__ == "__main__":
    nums: list[int] = [3, 1, 4, 1, 5, 9, 2]
    print(find_max(nums))          # 9
    print(first([], default=-1))    # -1
    print(group_by(["ap", "bc", "ad", "bb"], key=lambda s: s[0]))
    # {'a': ['ap', 'ad'], 'b': ['bc', 'bb']}

import asyncio

async def hello():
    print("开始")
    await asyncio.sleep(1)      # 让出 1 秒
    print("结束")

asyncio.run(hello())

import asyncio
import time


async def task(name, delay):
    print(f"{name} 开始")
    await asyncio.sleep(delay)
    print(f"{name} 结束")
    return name


async def main():
    start = time.perf_counter()
    results = await asyncio.gather(
        task("A", 2),
        task("B", 1),
        task("C", 3),
    )
    print(f"耗时 {time.perf_counter() - start:.2f}s，结果 {results}")


asyncio.run(main())

async def main():
    task1 = asyncio.create_task(task("A", 2))     # 立即调度
    task2 = asyncio.create_task(task("B", 1))
    result1 = await task1
    result2 = await task2

async def bad():
    time.sleep(1)         # ❌ 整个事件循环卡 1 秒

async def good():
    await asyncio.sleep(1)   # ✅ 让出控制

async def main():
    time.sleep(1)           # ❌ 阻塞，不 await
    await time.sleep(1)     # ❌ time.sleep 返回 None，不能 await
    await asyncio.sleep(1)  # ✅

# 顺序 requests
import requests
for url in urls:
    r = requests.get(url)
# 100 秒

# 异步 httpx
import httpx
import asyncio

async def fetch(client, url):
    r = await client.get(url)
    return r.status_code

async def main():
    async with httpx.AsyncClient() as client:
        results = await asyncio.gather(*[fetch(client, u) for u in urls])
    return results

asyncio.run(main())
# 1-2 秒

try:
    await asyncio.wait_for(some_coro(), timeout=5)
except asyncio.TimeoutError:
    print("超时")


# Python 3.11+ 更优雅
async with asyncio.timeout(5):
    await some_coro()

async def bad():
    raise ValueError("boom")

async def main():
    try:
        await bad()
    except ValueError as e:
        print(f"抓到：{e}")

try:
    await asyncio.gather(bad(), good())
except ValueError:
    ...

# 不想让一个失败拖累全部：
results = await asyncio.gather(bad(), good(), return_exceptions=True)
# results 里失败的项是异常对象

async def main():
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(task("A", 2))
        t2 = tg.create_task(task("B", 1))
    # 出块时会等所有 task 完成
    # 一个抛异常会取消其他，最后一起 ExceptionGroup

sem = asyncio.Semaphore(20)      # 最多 20 个并发

async def fetch(url):
    async with sem:
        # 同时最多 20 个进这里
        return await client.get(url)


await asyncio.gather(*[fetch(u) for u in urls])

lock = asyncio.Lock()
event = asyncio.Event()
queue = asyncio.Queue()

async def worker(q):
    while True:
        item = await q.get()
        if item is None:
            break
        process(item)
        q.task_done()

# 生产者-消费者
async def main():
    q = asyncio.Queue()
    workers = [asyncio.create_task(worker(q)) for _ in range(5)]
    for item in items:
        await q.put(item)
    for _ in workers:
        await q.put(None)
    await asyncio.gather(*workers)

# ✅ 一次性入口
asyncio.run(main())

# ❌ 在已有 event loop 里再 asyncio.run 会报错

loop = asyncio.get_running_loop()
result = await loop.run_in_executor(None, blocking_func, arg1, arg2)

# 或者 Python 3.9+：
result = await asyncio.to_thread(blocking_func, arg1, arg2)

# crawler.py
"""异步并发爬取多个 URL，展示 asyncio 的力量。"""

import asyncio
import httpx
import time


async def fetch(client, url, sem):
    async with sem:
        try:
            r = await client.get(url, timeout=10)
            return url, r.status_code, len(r.text)
        except Exception as e:
            return url, None, str(e)


async def main(urls, concurrency=20):
    sem = asyncio.Semaphore(concurrency)
    async with httpx.AsyncClient() as client:
        tasks = [fetch(client, u, sem) for u in urls]
        return await asyncio.gather(*tasks)


if __name__ == "__main__":
    urls = [f"https://httpbin.org/status/{code}" for code in range(200, 300)]
    start = time.perf_counter()
    results = asyncio.run(main(urls))
    print(f"共 {len(results)} 个，耗时 {time.perf_counter() - start:.2f}s")
    ok = sum(1 for u, s, _ in results if s == 200)
    print(f"200 状态：{ok}")

from multiprocessing import Process

def worker(name):
    print(f"{name} in pid {os.getpid()}")

if __name__ == "__main__":
    p = Process(target=worker, args=("t1",))
    p.start()
    p.join()

from concurrent.futures import ProcessPoolExecutor
import math

def is_prime(n):
    if n < 2:
        return False
    for i in range(2, int(math.sqrt(n)) + 1):
        if n % i == 0:
            return False
    return True


if __name__ == "__main__":
    numbers = range(10**6, 10**6 + 100)
    with ProcessPoolExecutor(max_workers=4) as pool:
        results = list(pool.map(is_prime, numbers))
    print(sum(results))

from multiprocessing import Pool

with Pool(4) as p:
    result = p.map(is_prime, numbers)

# ❌ Pool 里报错
def outer():
    def inner(x):
        return x * 2
    with Pool(4) as p:
        p.map(inner, range(10))

# ✅ 顶层函数
def double(x):
    return x * 2

def outer():
    with Pool(4) as p:
        p.map(double, range(10))

from multiprocessing import Process, Queue

def worker(q):
    q.put(42)

if __name__ == "__main__":
    q = Queue()
    p = Process(target=worker, args=(q,))
    p.start()
    p.join()
    print(q.get())     # 42

from multiprocessing import Pipe

parent_conn, child_conn = Pipe()

from multiprocessing import Manager

with Manager() as m:
    shared_dict = m.dict()
    shared_list = m.list()
    # 多个进程可以像本地一样操作它们

from multiprocessing.shared_memory import SharedMemory

shm = SharedMemory(create=True, size=1024)
shm.buf[0] = 42       # 直接操作字节

import multiprocessing as mp
mp.set_start_method("spawn")     # 显式设置

import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

def heavy(n):
    return sum(i * i for i in range(n))


nums = [10**7] * 8

# 顺序
start = time.perf_counter()
list(map(heavy, nums))
print(f"顺序：{time.perf_counter() - start:.2f}s")

# 多线程（GIL 阻塞，基本没提升）
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=8) as p:
    list(p.map(heavy, nums))
print(f"多线程：{time.perf_counter() - start:.2f}s")

# 多进程（4 核 CPU 上大概快 3-4 倍）
if __name__ == "__main__":
    start = time.perf_counter()
    with ProcessPoolExecutor(max_workers=8) as p:
        list(p.map(heavy, nums))
    print(f"多进程：{time.perf_counter() - start:.2f}s")

def init_worker():
    global model
    model = load_expensive_model()

def worker(x):
    return model.predict(x)

with ProcessPoolExecutor(max_workers=4, initializer=init_worker) as pool:
    ...

from joblib import Parallel, delayed

results = Parallel(n_jobs=-1)(delayed(heavy)(n) for n in nums)

# imgproc.py
"""对一个目录里的所有 JPG 做缩略图，多进程加速。"""

from PIL import Image
from pathlib import Path
from concurrent.futures import ProcessPoolExecutor, as_completed
import os


def make_thumb(src: Path, dst_dir: Path, size=(300, 300)) -> Path:
    dst = dst_dir / src.name
    with Image.open(src) as img:
        img.thumbnail(size)
        img.save(dst)
    print(f"[pid={os.getpid()}] {src.name} → {dst}")
    return dst


def batch(src_dir: Path, dst_dir: Path, workers: int = None):
    dst_dir.mkdir(parents=True, exist_ok=True)
    imgs = list(src_dir.glob("*.jpg"))

    with ProcessPoolExecutor(max_workers=workers) as pool:
        futures = {pool.submit(make_thumb, p, dst_dir): p for p in imgs}
        for f in as_completed(futures):
            try:
                f.result()
            except Exception as e:
                print(f"失败：{futures[f]} - {e}")


if __name__ == "__main__":
    batch(Path("photos"), Path("thumbs"))

import threading

def worker(name):
    print(f"{name} 开始")
    ...
    print(f"{name} 结束")


t = threading.Thread(target=worker, args=("t1",))
t.start()          # 启动
t.join()           # 等待结束

threads = [threading.Thread(target=worker, args=(f"t{i}",)) for i in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

from concurrent.futures import ThreadPoolExecutor, as_completed
import requests


def fetch(url):
    r = requests.get(url, timeout=5)
    return url, r.status_code


urls = ["https://httpbin.org/get"] * 20

with ThreadPoolExecutor(max_workers=10) as pool:
    # map：按顺序返回
    for url, status in pool.map(fetch, urls):
        print(url, status)

    # 或者用 submit + as_completed（完成一个处理一个）
    futures = [pool.submit(fetch, url) for url in urls]
    for future in as_completed(futures):
        url, status = future.result()
        print(url, status)

future = pool.submit(fetch, url)

future.done()             # 是否完成
future.cancel()           # 尝试取消
future.result(timeout=5)  # 阻塞取结果，抛异常也会传上来
future.exception()        # 取异常（如果有）
future.add_done_callback(lambda f: print("完成"))

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(100000):
        with lock:                   # 加锁块
            counter += 1

threads = [threading.Thread(target=increment) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()

print(counter)     # 应该是 400000

import queue
import threading
import time


def producer(q):
    for i in range(10):
        print(f"生产 {i}")
        q.put(i)
        time.sleep(0.1)
    q.put(None)      # 结束信号


def consumer(q):
    while True:
        item = q.get()
        if item is None:
            break
        print(f"消费 {item}")


q = queue.Queue()
t1 = threading.Thread(target=producer, args=(q,))
t2 = threading.Thread(target=consumer, args=(q,))
t1.start(); t2.start()
t1.join(); t2.join()

t = threading.Thread(target=worker, daemon=True)
t.start()
# 主线程退出时，t 也跟着挂（不管是否完成）

sem = threading.Semaphore(5)

def limited_task():
    with sem:
        # 同一时刻最多 5 个在这里
        do_stuff()

local = threading.local()

def worker():
    local.value = threading.current_thread().name    # 每个线程独立
    print(local.value)

def bad():
    raise RuntimeError("boom")

t = threading.Thread(target=bad)
t.start()
t.join()          # 什么都不发生，异常"消失"了

import time
import requests
from concurrent.futures import ThreadPoolExecutor

urls = ["https://httpbin.org/delay/1"] * 20

# 顺序：约 22 秒
start = time.perf_counter()
for u in urls:
    requests.get(u)
print(f"顺序：{time.perf_counter() - start:.1f}s")

# 多线程：约 1-2 秒
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=20) as pool:
    list(pool.map(requests.get, urls))
print(f"多线程：{time.perf_counter() - start:.1f}s")

# downloader.py
"""并发下载多个 URL 到本地文件。"""

import requests
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed
import logging

logger = logging.getLogger(__name__)


def download(url: str, out_dir: Path, timeout: float = 30) -> Path | None:
    """下载单个 URL 到 out_dir，返回本地路径。"""
    try:
        with requests.get(url, timeout=timeout, stream=True) as r:
            r.raise_for_status()
            fname = url.split("/")[-1] or "index.html"
            out = out_dir / fname
            with out.open("wb") as f:
                for chunk in r.iter_content(chunk_size=8192):
                    f.write(chunk)
            logger.info("下载 %s → %s", url, out)
            return out
    except Exception as e:
        logger.error("下载 %s 失败：%s", url, e)
        return None


def batch_download(urls: list[str], out_dir: Path, workers: int = 8) -> list[Path]:
    out_dir.mkdir(parents=True, exist_ok=True)
    results = []

    with ThreadPoolExecutor(max_workers=workers) as pool:
        futures = {pool.submit(download, u, out_dir): u for u in urls}
        for future in as_completed(futures):
            path = future.result()
            if path:
                results.append(path)

    return results


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO,
                        format="%(asctime)s %(threadName)s %(message)s")

    urls = [
        "https://httpbin.org/uuid",
        "https://httpbin.org/ip",
        "https://httpbin.org/user-agent",
    ] * 5

    downloaded = batch_download(urls, Path("downloads"), workers=5)
    print(f"共下载 {len(downloaded)} 个文件")

import requests

r = requests.get("https://httpbin.org/get")

print(r.status_code)     # 200
print(r.text)            # 响应内容（字符串）
print(r.json())          # 直接解析成 dict

r = requests.get(url)

r.status_code       # 200
r.reason            # 'OK'
r.headers           # 响应头字典
r.text              # 文本内容（str）
r.content           # 二进制内容（bytes）
r.json()            # JSON 解析
r.url               # 最终 URL（重定向后）
r.encoding          # 猜测的编码
r.cookies           # 服务器 set 的 cookies
r.elapsed           # 耗时（timedelta）
r.ok                # True 如果 status_code < 400
r.raise_for_status()   # 4xx/5xx 抛异常

# ❌ 手动拼
requests.get(f"https://api.com/search?q={q}&page=1")

# ✅ 用 params
r = requests.get("https://api.com/search", params={
    "q": "python",
    "page": 1,
    "tags": ["web", "async"],   # 会展开成多个 tags=web&tags=async
})
print(r.url)

# 表单
r = requests.post("https://httpbin.org/post", data={"name": "咖飞", "age": 3})

# JSON body（更常见）
r = requests.post("https://httpbin.org/post", json={"name": "咖飞", "age": 3})

# 上传文件
with open("photo.jpg", "rb") as f:
    r = requests.post("https://httpbin.org/post", files={"photo": f})

# 组合
r = requests.post(url,
    files={"photo": open("photo.jpg", "rb")},
    data={"caption": "咖飞的照片"},
)

r = requests.get(url, headers={
    "User-Agent": "MyApp/1.0",
    "Accept": "application/json",
    "Authorization": "Bearer xxx",
})

r = requests.get(url, timeout=5)         # 总超时 5 秒
r = requests.get(url, timeout=(3, 10))   # (连接超时, 读取超时)

try:
    r = requests.get(url, timeout=5)
    r.raise_for_status()             # 4xx / 5xx 抛 HTTPError
except requests.Timeout:
    print("超时")
except requests.ConnectionError:
    print("连接失败")
except requests.HTTPError as e:
    print(f"HTTP {e.response.status_code}")
except requests.RequestException as e:
    print(f"其他 requests 异常：{e}")

# Basic Auth
r = requests.get(url, auth=("user", "password"))

# Bearer token
r = requests.get(url, headers={"Authorization": "Bearer xxx"})

# 复杂 auth 用 HTTPDigestAuth / HTTPProxyAuth 等

with requests.Session() as s:
    # 登录：cookie 自动保存
    s.post("https://example.com/login", data={"user": "x", "pass": "y"})

    # 后续请求自动带 cookie
    r = s.get("https://example.com/profile")

    # 全局默认 headers
    s.headers.update({"User-Agent": "MyBot/1.0"})

# 不跟随重定向
r = requests.get(url, allow_redirects=False)

# 代理
r = requests.get(url, proxies={
    "http": "http://127.0.0.1:7890",
    "https": "http://127.0.0.1:7890",
})

# SSL 校验（默认开，某些自签证书场景可关，但要谨慎）
r = requests.get(url, verify=False)

with requests.get(url, stream=True) as r:
    r.raise_for_status()
    with open("bigfile.zip", "wb") as f:
        for chunk in r.iter_content(chunk_size=8192):
            f.write(chunk)

# 内存不够时不能 `open(path).read()`
with open("bigfile.zip", "rb") as f:
    r = requests.post(url, data=f)   # 直接传文件对象

from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


def make_session():
    s = requests.Session()
    retry = Retry(
        total=3,
        backoff_factor=1,         # 1s, 2s, 4s
        status_forcelist=[500, 502, 503, 504],
        allowed_methods=["GET", "POST"],
    )
    adapter = HTTPAdapter(max_retries=retry)
    s.mount("http://", adapter)
    s.mount("https://", adapter)
    return s

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
def fetch(url):
    r = requests.get(url, timeout=5)
    r.raise_for_status()
    return r

# httpx 同步
import httpx
r = httpx.get(url)

# httpx 异步
async with httpx.AsyncClient() as client:
    r = await client.get(url)

# github.py
"""用 requests 从 GitHub API 获取用户仓库信息。"""

import requests
from pathlib import Path
import json
import time


class GitHubClient:
    BASE = "https://api.github.com"

    def __init__(self, token: str | None = None):
        self.s = requests.Session()
        self.s.headers.update({
            "Accept": "application/vnd.github+json",
            "User-Agent": "MyPythonScript/1.0",
        })
        if token:
            self.s.headers["Authorization"] = f"Bearer {token}"

    def get(self, path: str, params: dict = None) -> dict | list:
        r = self.s.get(f"{self.BASE}{path}", params=params, timeout=10)
        # 处理限流
        if r.status_code == 403 and "rate limit" in r.text.lower():
            reset = int(r.headers.get("X-RateLimit-Reset", 0))
            wait = max(0, reset - int(time.time()))
            print(f"限流，等待 {wait} 秒")
            time.sleep(wait + 1)
            return self.get(path, params)
        r.raise_for_status()
        return r.json()

    def user_repos(self, username: str) -> list:
        return self.get(f"/users/{username}/repos", params={"per_page": 100})

    def repo_readme(self, owner: str, repo: str) -> str:
        r = self.get(f"/repos/{owner}/{repo}/readme")
        # readme 内容是 base64
        import base64
        return base64.b64decode(r["content"]).decode("utf-8")


if __name__ == "__main__":
    import os
    client = GitHubClient(token=os.getenv("GITHUB_TOKEN"))
    repos = client.user_repos("python")
    for r in repos[:5]:
        print(f"{r['name']:30} ⭐ {r['stargazers_count']:>5} - {r.get('description', '')}")

def do_task(user_id):
    print(f"开始处理用户 {user_id}")
    ...
    print(f"处理完成")

import logging
logger = logging.getLogger(__name__)

def do_task(user_id):
    logger.info(f"开始处理用户 {user_id}")
    ...
    logger.info("处理完成")

logger.debug("详细的调试信息")
logger.info("正常的运行信息")
logger.warning("警告，程序还能跑")
logger.error("错误，某个功能挂了")
logger.critical("严重错误，程序快挂了")

import logging

logging.basicConfig(level=logging.INFO)
logging.info("hello")
# INFO:root:hello

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)

# mymodule.py
import logging
logger = logging.getLogger(__name__)     # __name__ = 'mymodule'

def foo():
    logger.info("foo called")

format = "%(asctime)s %(levelname)-8s [%(name)s] %(message)s"

import logging

logger = logging.getLogger("myapp")
logger.setLevel(logging.DEBUG)

# 终端：INFO 及以上
console = logging.StreamHandler()
console.setLevel(logging.INFO)
console.setFormatter(logging.Formatter("%(levelname)s: %(message)s"))
logger.addHandler(console)

# 文件：DEBUG 及以上，滚动
from logging.handlers import RotatingFileHandler
file_handler = RotatingFileHandler("app.log", maxBytes=10*1024*1024, backupCount=5, encoding="utf-8")
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(logging.Formatter(
    "%(asctime)s %(levelname)s [%(name)s:%(lineno)d] %(message)s"
))
logger.addHandler(file_handler)


logger.info("hello")     # 终端 + 文件都有
logger.debug("debug")    # 只有文件

try:
    do_stuff()
except Exception:
    logger.exception("do_stuff 出错")

# ✅ 推荐
logger.info("user %s login from %s", user_id, ip)

# ❌ 不推荐（每次都格式化，即使日志级别没开）
logger.info(f"user {user_id} login from {ip}")

import logging.config

LOGGING = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "default": {
            "format": "%(asctime)s %(levelname)-8s [%(name)s] %(message)s",
        },
        "detailed": {
            "format": "%(asctime)s %(levelname)-8s [%(name)s:%(lineno)d] %(funcName)s(): %(message)s",
        },
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "level": "INFO",
            "formatter": "default",
        },
        "file": {
            "class": "logging.handlers.RotatingFileHandler",
            "filename": "app.log",
            "maxBytes": 10 * 1024 * 1024,
            "backupCount": 5,
            "encoding": "utf-8",
            "level": "DEBUG",
            "formatter": "detailed",
        },
    },
    "loggers": {
        "myapp": {
            "level": "DEBUG",
            "handlers": ["console", "file"],
            "propagate": False,
        },
        "myapp.db": {
            "level": "INFO",
        },
    },
    "root": {
        "level": "WARNING",
        "handlers": ["console"],
    },
}

logging.config.dictConfig(LOGGING)

from loguru import logger

logger.info("hello")
logger.debug("debug 信息")
logger.add("app.log", rotation="10 MB")   # 一行加文件 handler
logger.exception("出错了")

class RequestAdapter(logging.LoggerAdapter):
    def process(self, msg, kwargs):
        return f"[req={self.extra['request_id']}] {msg}", kwargs


logger = logging.getLogger("myapp")
req_logger = RequestAdapter(logger, {"request_id": "abc-123"})
req_logger.info("处理请求")
# [req=abc-123] 处理请求

# main.py
"""一个生产级的日志初始化脚本。"""

import logging
import logging.config
import sys
from pathlib import Path


def setup_logging(log_dir: Path, verbose: bool = False):
    log_dir.mkdir(parents=True, exist_ok=True)

    LOGGING = {
        "version": 1,
        "disable_existing_loggers": False,
        "formatters": {
            "simple": {
                "format": "%(asctime)s %(levelname)-8s %(name)s: %(message)s",
            },
            "detailed": {
                "format": "%(asctime)s %(levelname)-8s [%(name)s:%(lineno)d] "
                          "%(funcName)s(): %(message)s",
            },
        },
        "handlers": {
            "console": {
                "class": "logging.StreamHandler",
                "level": "DEBUG" if verbose else "INFO",
                "formatter": "simple",
                "stream": sys.stdout,
            },
            "app_file": {
                "class": "logging.handlers.TimedRotatingFileHandler",
                "filename": str(log_dir / "app.log"),
                "when": "midnight",
                "backupCount": 30,
                "encoding": "utf-8",
                "level": "DEBUG",
                "formatter": "detailed",
            },
            "error_file": {
                "class": "logging.handlers.RotatingFileHandler",
                "filename": str(log_dir / "error.log"),
                "maxBytes": 20 * 1024 * 1024,
                "backupCount": 10,
                "encoding": "utf-8",
                "level": "ERROR",
                "formatter": "detailed",
            },
        },
        "loggers": {
            "myapp": {
                "level": "DEBUG",
                "handlers": ["console", "app_file", "error_file"],
                "propagate": False,
            },
            "urllib3": {
                "level": "WARNING",     # 屏蔽 requests 的 DEBUG 噪音
            },
        },
        "root": {
            "level": "WARNING",
            "handlers": ["console"],
        },
    }

    logging.config.dictConfig(LOGGING)


if __name__ == "__main__":
    setup_logging(Path("logs"), verbose=True)

    logger = logging.getLogger("myapp")
    logger.debug("debug 消息")
    logger.info("服务启动")
    try:
        1 / 0
    except Exception:
        logger.exception("除零错误")