import dataclasses, inspect
print(inspect.getsource(dataclasses.field)[:120])   # реальная реализация field()

from collections import Counter, defaultdict, deque, ChainMap

lines = ["GET /api 200", "GET /api 500", "POST /pay 200", "GET /api 500"]

codes = Counter(line.rsplit(maxsplit=1)[-1] for line in lines)   # частоты за один проход
print(codes.most_common(2), codes["404"])   # [('200', 2), ('500', 2)] 0 — без KeyError

by_method = defaultdict(list)                # группировка без проверок на существование
for method, path, _ in (l.split() for l in lines):
    by_method[method].append(path)
print(dict(by_method))   # {'GET': ['/api', '/api', '/api'], 'POST': ['/pay']}

window = deque(maxlen=3)                     # скользящее окно: старое вытесняется само
for x in range(6):
    window.append(x)
print(list(window))                          # [3, 4, 5]

cfg = ChainMap({"timeout": 5}, {"timeout": 30, "retries": 3})   # слои конфигурации
print(cfg["timeout"], cfg["retries"])        # 5 3 — поиск слева направо, без копирования

import heapq
from bisect import bisect_right

# Топ-K из потока: O(n log k) времени и O(k) памяти вместо O(n log n) и O(n).
# Не материализует поток — работает и на файле в 50 ГБ.
print(heapq.nlargest(2, [("a", 3), ("b", 9), ("c", 1), ("d", 7)], key=lambda r: r[1]))
# [('b', 9), ('d', 7)]

# Приоритетная очередь: (приоритет, счётчик, объект) — счётчик снимает сравнение
# несравнимых объектов при равных приоритетах.
pq: list[tuple[int, int, str]] = []
for i, (prio, task) in enumerate([(2, "email"), (1, "pay"), (2, "log")]):
    heapq.heappush(pq, (prio, i, task))               # O(log n)
print([heapq.heappop(pq)[2] for _ in range(3)])       # ['pay', 'email', 'log']

# bisect: попадание в интервал за O(log n) без единого if
print("FDCBA"[bisect_right([60, 70, 80, 90], 85)])    # 'B'

from itertools import batched, chain, islice, pairwise, accumulate, groupby

for chunk in batched(range(1000), 100):     # 3.12+: нарезка на пачки для bulk-insert
    ...                                     # chunk — кортеж, промежуточных списков нет

print([b - a for a, b in pairwise([10, 12, 11, 15])])   # [2, -1, 4] — дельты соседей
print(list(accumulate([100, -30, 50])))                 # [100, 70, 120] — нарастающий итог
print(list(islice(chain([1, 2], [3, 4, 5]), 3)))        # [1, 2, 3] — срез итератора

rows = [("db", 12), ("db", 3), ("api", 7), ("db", 1)]
for key, grp in groupby(rows, key=lambda r: r[0]):
    print(key, [v for _, v in grp])
# db [12, 3]
# api [7]
# db [1]     <- третья группа "db": вход не был отсортирован

import functools

@functools.cache                       # 3.9+: то же, что lru_cache(maxsize=None)
def fib(n: int) -> int:
    return n if n < 2 else fib(n - 1) + fib(n - 2)

print(fib(80), fib.cache_info())       # 23416728348467685 CacheInfo(hits=78, misses=81, ...)

@functools.singledispatch              # диспетчеризация по типу первого аргумента
def render(value) -> str:
    return str(value)

@render.register
def _(value: list) -> str:             # тип берётся из аннотации
    return ", ".join(render(v) for v in value)

print(render([1, "два", 3.0]))         # '1, два, 3.0'

from pathlib import Path

cfg = Path("/srv/app") / "conf" / "app.toml"     # оператор / вместо os.path.join
print(cfg.suffix, cfg.stem, cfg.with_suffix(".bak"))   # .toml app /srv/app/conf/app.bak
text = cfg.read_text(encoding="utf-8")           # чтение целиком без конструкции with

for dirpath, dirnames, filenames in Path("/srv/app").walk():   # Path.walk — 3.12+
    dirnames[:] = [d for d in dirnames if d != ".git"]         # обрезаем обход на месте
    for name in filenames:
        ...

from pathlib import Path
import csv, os, tempfile

# 1. ВСЕГДА указывайте encoding: без него берётся кодировка локали —
#    UTF-8 на Linux и cp1251 на Windows с русской локалью.
Path("out.txt").write_text("привет", encoding="utf-8")

# 2. Атомарная запись конфига: временный файл рядом, fsync, os.replace.
def write_atomic(path: Path, data: str) -> None:
    fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=path.name, suffix=".tmp")
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            f.write(data)
            f.flush()
            os.fsync(f.fileno())     # без этого при потере питания файл будет пустым
        os.replace(tmp, path)        # атомарно в пределах ФС: читатель видит старое или новое
    except BaseException:
        os.unlink(tmp)
        raise

# 3. Для CSV обязателен newline="" — иначе на Windows появятся пустые строки.
with Path("report.csv").open("w", encoding="utf-8", newline="") as f:
    csv.writer(f).writerows([("id", "sum"), (1, "3.14")])

from datetime import datetime, timedelta, UTC
from zoneinfo import ZoneInfo
import time

# 1. Момент времени всегда aware. datetime.utcnow() устарел в 3.12 именно потому,
#    что возвращал naive-объект, притворявшийся UTC.
now = datetime.now(UTC)
msk = now.astimezone(ZoneInfo("Europe/Moscow"))     # база IANA, PEP 615

# 2. Арифметика с tzinfo идёт «по стенным часам», а не по абсолютному времени.
tz = ZoneInfo("Europe/Berlin")
start = datetime(2026, 3, 28, 23, 30, tzinfo=tz)
print(start + timedelta(hours=3))
# 2026-03-29 02:30:00+01:00 — такого локального времени в этот день не существует
print((start.astimezone(UTC) + timedelta(hours=3)).astimezone(tz))
# 2026-03-29 03:30:00+02:00 — правильно: считать в UTC, конвертировать в конце

# 3. Интервалы — только по монотонным часам: time.time() прыгает после синхронизации NTP.
t0 = time.perf_counter()
elapsed = time.perf_counter() - t0

import json
from decimal import Decimal

print(json.dumps({1: "a", "amount": 0.1 + 0.2, "nan": float("nan")}))
# {"1": "a", "amount": 0.30000000000000004, "nan": NaN}
#  ^ ключ стал строкой         ^ двоичный float       ^ невалидный JSON по RFC 8259

print(json.loads('{"sum": 10.10}', parse_float=Decimal)["sum"] * 3)   # 30.30 — точно

import sqlite3

conn = sqlite3.connect("cache.db", isolation_level=None)   # автокоммит, ручные BEGIN
conn.execute("PRAGMA journal_mode=WAL")        # параллельное чтение во время записи
conn.execute("CREATE TABLE IF NOT EXISTS kv(k TEXT PRIMARY KEY, v TEXT)")

with conn:                                     # это транзакция, а НЕ закрытие соединения!
    conn.executemany("INSERT OR REPLACE INTO kv VALUES (?, ?)", [("a", "1"), ("b", "2")])

print(conn.execute("SELECT v FROM kv WHERE k = ?", ("a",)).fetchone()[0])   # 1
conn.close()

import hashlib, hmac, secrets, uuid

token = secrets.token_urlsafe(32)     # random — это Mersenne Twister, он предсказуем
ok = hmac.compare_digest(token, "подобранное")    # сравнение за постоянное время
key = hashlib.scrypt(b"pass", salt=secrets.token_bytes(16), n=2**14, r=8, p=1)  # не sha256!
print(uuid.uuid4())                   # случайный, плохо ложится в B-tree индекс
print(uuid.uuid7())                   # 3.14+: сортируемый по времени

import subprocess

res = subprocess.run(                 # канонический вызов: список аргументов, без shell
    ["git", "rev-parse", "--short", "HEAD"],
    capture_output=True, text=True, timeout=10, check=True,
)
print(res.stdout.strip())             # e3f1a2c

import logging.config, contextvars

request_id = contextvars.ContextVar("request_id", default="-")


class RequestIdFilter(logging.Filter):
    """Фильтр как способ обогатить запись, а не отбросить её."""

    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id.get()
        return True


logging.config.dictConfig({
    "version": 1,
    "disable_existing_loggers": False,      # иначе замолчат логгеры библиотек
    "filters": {"rid": {"()": RequestIdFilter}},
    "formatters": {"plain": {
        "format": "%(asctime)s %(levelname)-8s [%(request_id)s] %(name)s: %(message)s"}},
    "handlers": {"console": {"class": "logging.StreamHandler", "formatter": "plain",
                             "filters": ["rid"], "stream": "ext://sys.stderr"}},
    "root": {"level": "INFO", "handlers": ["console"]},
    "loggers": {"urllib3": {"level": "WARNING"}},   # чужой шум — отдельным уровнем
})

log = logging.getLogger(__name__)           # ВСЕГДА __name__, никогда не root напрямую
request_id.set("7f3c")
log.info("заказ обработан: %s за %.3f c", "A-17", 0.412)     # %s, а не f-строка!
# 2026-07-16 10:00:00,123 INFO     [7f3c] __main__: заказ обработан: A-17 за 0.412 c

try:
    1 / 0
except ZeroDivisionError:
    log.exception("не смог посчитать конверсию")   # ERROR + полный traceback

import timeit
setup = "from collections import deque; d = deque(range(10_000)); l = list(range(10_000))"
print(timeit.timeit("l.pop(0)", setup, number=10_000))     # ~0.006 c  — O(n) на вызов
print(timeit.timeit("d.popleft()", setup, number=10_000))  # ~0.0002 c — O(1), в 30 раз быстрее

#!/usr/bin/env python3
"""inv.py — инвентаризация каталога. Зависимостей: ноль."""

from __future__ import annotations

import argparse, csv, hashlib, logging, os, sqlite3, sys
from collections.abc import Iterator
from concurrent.futures import ProcessPoolExecutor
from pathlib import Path

log = logging.getLogger("inv")


def digest(path: Path) -> tuple[str, int, str]:
    """Потоковый sha256: файл на 10 ГБ не окажется в памяти целиком."""
    h = hashlib.sha256()
    with path.open("rb") as f:
        while chunk := f.read(1 << 20):          # читаем по мегабайту
            h.update(chunk)
    return str(path), path.stat().st_size, h.hexdigest()


def iter_files(root: Path, skip: set[str]) -> Iterator[Path]:
    for dirpath, dirnames, filenames in root.walk():          # Path.walk — 3.12+
        dirnames[:] = [d for d in dirnames if d not in skip]  # обрезаем обход на месте
        for name in filenames:
            yield dirpath / name


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(prog="inv", description="Инвентаризация каталога")
    p.add_argument("root", type=Path)
    p.add_argument("--db", type=Path, default=Path("inv.sqlite3"))
    p.add_argument("--csv", type=Path, help="выгрузить результат в CSV")
    p.add_argument("--skip", action="append", default=[".git", "__pycache__"])
    p.add_argument("-v", "--verbose", action="count", default=0)
    args = p.parse_args(argv)

    logging.basicConfig(
        level=logging.WARNING - 10 * min(args.verbose, 2),    # -v → INFO, -vv → DEBUG
        format="%(asctime)s %(levelname)-8s %(message)s", stream=sys.stderr,
    )

    conn = sqlite3.connect(args.db)
    conn.execute("CREATE TABLE IF NOT EXISTS files("
                 "path TEXT PRIMARY KEY, size INTEGER, sha256 TEXT)")
    files = list(iter_files(args.root, set(args.skip)))
    log.info("найдено файлов: %d", len(files))

    workers = min(os.process_cpu_count() or 1, 8)     # 3.13+; ранее os.cpu_count()
    with ProcessPoolExecutor(max_workers=workers) as pool, conn:
        # хеширование CPU-bound → процессы, а не потоки;
        # chunksize амортизирует накладные расходы на межпроцессное общение
        rows = pool.map(digest, files, chunksize=16)
        conn.executemany("INSERT OR REPLACE INTO files VALUES (?, ?, ?)", rows)

    if args.csv:
        with args.csv.open("w", encoding="utf-8", newline="") as f:
            w = csv.writer(f)
            w.writerow(["path", "size", "sha256"])
            w.writerows(conn.execute("SELECT * FROM files ORDER BY size DESC"))
    conn.close()
    return 0


if __name__ == "__main__":       # обязателен: пул импортирует модуль в дочерних процессах
    raise SystemExit(main())