"""Отмена в TaskGroup: что происходит с соседями и что видно в ExceptionGroup.

Показывает запуском:
  * одна упала — остальные отменены, их finally выполнился, тело async with
    тоже прервано;
  * две упали — обе приехали в одном ExceptionGroup;
  * CancelledError от ребёнка группу НЕ роняет (в отличие от любого Exception);
  * внешняя отмена родителя выходит как CancelledError, а не ExceptionGroup;
  * ошибка в теле async with попадает в ту же группу;
  * create_task после аварии — RuntimeError;
  * то же на gather для контраста: соседей никто не отменяет.

    python3.13 bench/cancellation/05_taskgroup_cancellation.py
"""

import asyncio
import sys


def row(name, value):
    print(f"  {name:<46} {value}")


def head(title):
    print()
    print(f"== {title}")


async def spin(n=3):
    for _ in range(n):
        await asyncio.sleep(0)


def describe(exc, indent="    "):
    """Дерево ExceptionGroup в виде строк."""
    lines = []

    def walk(e, depth):
        pad = indent + "  " * depth
        if isinstance(e, BaseExceptionGroup):
            lines.append(f"{pad}{type(e).__name__}: {e.message!r} "
                         f"({len(e.exceptions)} шт.)")
            for sub in e.exceptions:
                walk(sub, depth + 1)
        else:
            lines.append(f"{pad}{type(e).__name__}: {e}")

    walk(exc, 0)
    return lines


def states(tasks):
    out = []
    for t in tasks:
        if not t.done():
            state = "ещё работает"
        elif t.cancelled():
            state = "cancelled"
        elif t.exception() is not None:
            state = f"exception {type(t.exception()).__name__}"
        else:
            state = f"result {t.result()!r}"
        out.append(f"{t.get_name():<12} {state}")
    return out


# ---------------------------------------------------------------------------

async def good(log, name, seconds):
    try:
        await asyncio.sleep(seconds)
    except asyncio.CancelledError:
        log.append(f"{name}: получил отмену")
        raise
    finally:
        log.append(f"{name}: finally")
    log.append(f"{name}: доработал")
    return name


async def bad(log, name, seconds, exc):
    await asyncio.sleep(seconds)
    log.append(f"{name}: бросает {type(exc).__name__}")
    raise exc


async def self_cancelling(log, name, seconds):
    await asyncio.sleep(seconds)
    log.append(f"{name}: сам бросает CancelledError")
    raise asyncio.CancelledError()


# ---------------------------------------------------------------------------

async def main():
    print(f"Python {sys.version.split()[0]}")

    head("1. Одна задача упала — соседей отменяют")
    log = []
    handles = []
    try:
        async with asyncio.TaskGroup() as tg:
            handles.append(tg.create_task(good(log, "мирная-1", 5.0),
                                          name="мирная-1"))
            handles.append(tg.create_task(good(log, "мирная-2", 5.0),
                                          name="мирная-2"))
            handles.append(tg.create_task(bad(log, "падучая", 0.05,
                                              ValueError("бух")),
                                          name="падучая"))
            log.append("тело async with: жду 5 с")
            await asyncio.sleep(5.0)
            log.append("тело async with: доспало (не должно)")
    except* ValueError as eg:
        log.append(f"except* ValueError: {len(eg.exceptions)} шт.")
        caught = eg
    for line in log:
        print(f"    | {line}")
    for line in states(handles):
        print(f"    | {line}")
    print("    ExceptionGroup:")
    for line in describe(caught):
        print(line)

    head("2. Две задачи упали — обе в одном ExceptionGroup")
    log = []
    handles = []
    caught = None
    try:
        async with asyncio.TaskGroup() as tg:
            handles.append(tg.create_task(bad(log, "первая", 0.05,
                                              ValueError("V")), name="первая"))
            handles.append(tg.create_task(bad(log, "вторая", 0.05,
                                              KeyError("K")), name="вторая"))
            handles.append(tg.create_task(good(log, "мирная", 5.0),
                                          name="мирная"))
            await asyncio.sleep(5.0)
    except BaseExceptionGroup as eg:
        caught = eg
    for line in log:
        print(f"    | {line}")
    for line in states(handles):
        print(f"    | {line}")
    print("    ExceptionGroup:")
    for line in describe(caught):
        print(line)
    row("тип", type(caught).__name__)
    row("isinstance(eg, Exception)", isinstance(caught, Exception))
    print("    разбор через except*:")
    try:
        raise caught
    except* ValueError as sub:
        print(f"      except* ValueError -> {type(sub).__name__} "
              f"{[type(e).__name__ for e in sub.exceptions]}")
    except* KeyError as sub:
        print(f"      except* KeyError   -> {type(sub).__name__} "
              f"{[type(e).__name__ for e in sub.exceptions]}")

    head("3. CancelledError от ребёнка группу НЕ роняет")
    log = []
    handles = []
    result = "группа вышла штатно"
    try:
        async with asyncio.TaskGroup() as tg:
            handles.append(tg.create_task(self_cancelling(log, "самоотменяющаяся",
                                                          0.05),
                                          name="самоотмена"))
            handles.append(tg.create_task(good(log, "мирная", 0.10),
                                          name="мирная"))
    except BaseExceptionGroup as eg:
        result = f"ExceptionGroup {[type(e).__name__ for e in eg.exceptions]}"
    except asyncio.CancelledError:
        result = "CancelledError"
    for line in log:
        print(f"    | {line}")
    for line in states(handles):
        print(f"    | {line}")
    row("итог", result)
    print("  taskgroups.py:235 — `if task.cancelled(): return`, отменённый")
    print("  ребёнок в _errors не попадает и соседей не роняет.")

    head("4. Внешняя отмена родителя: наружу CancelledError, не ExceptionGroup")
    log = []
    handles = []

    async def parent():
        try:
            async with asyncio.TaskGroup() as tg:
                handles.append(tg.create_task(good(log, "дитя-1", 5.0),
                                              name="дитя-1"))
                handles.append(tg.create_task(good(log, "дитя-2", 5.0),
                                              name="дитя-2"))
                await asyncio.sleep(5.0)
        except BaseExceptionGroup as eg:
            log.append(f"родитель: ExceptionGroup {len(eg.exceptions)}")
            raise
        except asyncio.CancelledError:
            log.append("родитель: CancelledError")
            raise

    ptask = asyncio.ensure_future(parent())
    await spin(4)
    ptask.cancel()
    try:
        await ptask
        verdict = "вернулся нормально"
    except BaseExceptionGroup as eg:
        verdict = f"ExceptionGroup {[type(e).__name__ for e in eg.exceptions]}"
    except asyncio.CancelledError:
        verdict = "CancelledError"
    for line in log:
        print(f"    | {line}")
    for line in states(handles):
        print(f"    | {line}")
    row("await parent ->", verdict)
    row("parent.cancelled()", ptask.cancelled())

    head("5. Ошибка в теле async with едет в ту же группу")
    log = []
    handles = []
    caught = None
    try:
        async with asyncio.TaskGroup() as tg:
            handles.append(tg.create_task(bad(log, "задача", 0.05,
                                              ValueError("из задачи")),
                                          name="задача"))
            await asyncio.sleep(5.0)
    except BaseExceptionGroup as eg:
        caught = eg
    print("    ExceptionGroup:")
    for line in describe(caught):
        print(line)
    log = []
    handles = []
    caught = None
    try:
        async with asyncio.TaskGroup() as tg:
            handles.append(tg.create_task(good(log, "мирная", 5.0),
                                          name="мирная"))
            await asyncio.sleep(0.05)
            raise RuntimeError("из тела async with")
    except BaseExceptionGroup as eg:
        caught = eg
    print("    ExceptionGroup:")
    for line in describe(caught):
        print(line)
    for line in states(handles):
        print(f"    | {line}")

    head("6. create_task у аварийной/законченной группы")
    async def noop():
        return None
    outcome = []
    try:
        async with asyncio.TaskGroup() as tg:
            saved = tg
            tg.create_task(bad([], "падучая", 0.05, ValueError("бух")))
            await asyncio.sleep(5.0)
    except* ValueError:
        pass
    import gc
    import warnings
    with warnings.catch_warnings(record=True) as caught:
        warnings.simplefilter("always")
        coro = noop()
        try:
            saved.create_task(coro)
            outcome.append("RuntimeError НЕ бросили")
        except RuntimeError as exc:
            outcome.append(f"RuntimeError: {exc}")
        del coro
        gc.collect()
    warned = [str(w.message) for w in caught if "never awaited" in str(w.message)]
    outcome.append(f"RuntimeWarning 'never awaited': {warned or 'нет'}")
    for line in outcome:
        print(f"    | {line}")
    print("  taskgroups.py:186-194 — три отдельные проверки, и каждая перед")
    print("  RuntimeError делает coro.close(), чтобы не осталось «never awaited».")
    print("  coro.close() тут появился в 3.13 — на 3.11/3.12 предупреждение есть.")

    head("7. Для контраста: gather соседей не отменяет")
    log = []
    tasks = [
        asyncio.create_task(good(log, "мирная-1", 0.20), name="мирная-1"),
        asyncio.create_task(good(log, "мирная-2", 0.20), name="мирная-2"),
        asyncio.create_task(bad(log, "падучая", 0.05, ValueError("бух")),
                            name="падучая"),
    ]
    try:
        await asyncio.gather(*tasks)
        verdict = "gather вернулся"
    except ValueError as exc:
        verdict = f"gather бросил {type(exc).__name__}: {exc}"
    row("сразу после gather", verdict)
    for line in states(tasks):
        print(f"    | {line}")
    await asyncio.sleep(0.25)
    print("    ...через 0.25 с:")
    for line in states(tasks):
        print(f"    | {line}")
    for line in log:
        print(f"    | {line}")


if __name__ == "__main__":
    asyncio.run(main())