lib/notify.py - master - beyondviolet/edw_prefect

lib/notify.py
144 строки4 KB
get458
first_commit
04 авг 2026, 07:38
04 авг 2026, 07:38b3e3bb3
# import asyncio
# import json as _json
# import logging
# import os
# import subprocess
#
# bv_log = logging.getLogger("bviolet-notify")
#
# TELEGRAM_TOKEN = "1899043817:AAHWU04YoRjtKDkudR28C5qk0veTWxso1EQ"
# TELEGRAM_CHAT_ID = "-4702713956"
#
#
# def _tg_send_sync(text: str):
#     env = os.environ.copy()
#     env['HTTPS_PROXY'] = 'http://127.0.0.1:12334'
#     subprocess.run([
#         "curl", "-s", "-X", "POST",
#         f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage",
#         "-H", "Content-Type: application/json",
#         "-d", _json.dumps({"chat_id": TELEGRAM_CHAT_ID, "text": text}, ensure_ascii=False),
#     ], timeout=10, check=True, env=env)
#
#
# async def tg_notify(text: str):
#     try:
#         loop = asyncio.get_event_loop()
#         await asyncio.wait_for(loop.run_in_executor(None, _tg_send_sync, text), timeout=15)
#     except (Exception, asyncio.CancelledError):
#         bv_log.error("Failed to send Telegram notification")
#


import asyncio
import json as _json
import logging
import os
import subprocess
import traceback

bv_log = logging.getLogger("bviolet-notify")

TELEGRAM_TOKEN = os.getenv("TELEGRAM_TOKEN", "1899043817:AAHWU04YoRjtKDkudR28C5qk0veTWxso1EQ")
TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID", "-4702713956")
HTTPS_PROXY = os.getenv("HTTPS_PROXY") or os.getenv("TG_HTTPS_PROXY")


def _build_env() -> dict:
    env = os.environ.copy()

    if HTTPS_PROXY:
        env["HTTPS_PROXY"] = HTTPS_PROXY
        env["https_proxy"] = HTTPS_PROXY
    else:
        env.pop("HTTPS_PROXY", None)
        env.pop("https_proxy", None)

    return env


def _tg_send_sync(text: str):
    if not TELEGRAM_TOKEN:
        raise RuntimeError("Не задан TELEGRAM_TOKEN")

    if not TELEGRAM_CHAT_ID:
        raise RuntimeError("Не задан TELEGRAM_CHAT_ID")

    payload = {
        "chat_id": TELEGRAM_CHAT_ID,
        "text": text,
    }

    cmd = [
        "curl",
        "-sS",
        "-X",
        "POST",
        f"https://tg.claude-access.ru/bot{TELEGRAM_TOKEN}/sendMessage",
        "-H",
        "Content-Type: application/json",
        "-d",
        _json.dumps(payload, ensure_ascii=False),
    ]

    result = subprocess.run(
        cmd,
        timeout=10,
        check=False,
        env=_build_env(),
        capture_output=True,
        text=True,
    )

    if result.returncode != 0:
        raise RuntimeError(
            "Ошибка отправки в Telegram через curl.\n"
            f"returncode={result.returncode}\n"
            f"stdout={result.stdout.strip()}\n"
            f"stderr={result.stderr.strip()}\n"
            f"proxy={HTTPS_PROXY or 'не используется'}"
        )

    response_text = (result.stdout or "").strip()

    if not response_text:
        raise RuntimeError("Telegram API вернул пустой ответ")

    try:
        response_json = _json.loads(response_text)
    except Exception as exc:
        raise RuntimeError(
            "Не удалось распарсить ответ Telegram API как JSON.\n"
            f"raw_response={response_text}"
        ) from exc

    if not response_json.get("ok"):
        raise RuntimeError(
            "Telegram API вернул ошибку.\n"
            f"response={response_json}"
        )

    return response_json


async def tg_notify(text: str):
    try:
        loop = asyncio.get_running_loop()
        response = await asyncio.wait_for(
            loop.run_in_executor(None, _tg_send_sync, text),
            timeout=15,
        )
        bv_log.info(
            "Уведомление в Telegram отправлено успешно. message_id=%s",
            response.get("result", {}).get("message_id"),
        )
    except asyncio.TimeoutError:
        bv_log.error("Таймаут при отправке уведомления в Telegram")
        bv_log.error(traceback.format_exc())
    except asyncio.CancelledError:
        bv_log.error("Отправка уведомления в Telegram была отменена")
        bv_log.error(traceback.format_exc())
        raise
    except Exception as exc:
        bv_log.error("Не удалось отправить уведомление в Telegram: %s", exc)
        bv_log.error(traceback.format_exc())