82 lines
2.7 KiB
Python
Executable File
82 lines
2.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import os
|
|
from collections.abc import Iterable
|
|
|
|
import httpx
|
|
from sqlalchemy import select
|
|
|
|
from app.core.config import settings
|
|
from app.db.session import async_session_factory
|
|
from app.models import car, expense, gamification, push # noqa: F401
|
|
from app.models.user import User
|
|
|
|
REPORT_ROLES = {"admin", "super_admin", "moderator", "support"}
|
|
|
|
|
|
def env_recipients() -> list[str]:
|
|
recipients: list[str] = []
|
|
if settings.admin_notification_chat_id:
|
|
recipients.append(settings.admin_notification_chat_id)
|
|
recipients.extend(str(item) for item in settings.admin_telegram_id_list)
|
|
return recipients
|
|
|
|
|
|
async def db_recipients() -> list[str]:
|
|
async with async_session_factory() as session:
|
|
result = await session.execute(
|
|
select(User.telegram_id).where(User.platform_role.in_(REPORT_ROLES))
|
|
)
|
|
return [str(row[0]) for row in result.all() if row[0]]
|
|
|
|
|
|
def unique(values: Iterable[str]) -> list[str]:
|
|
return list(dict.fromkeys(item.strip() for item in values if item and item.strip()))
|
|
|
|
|
|
async def send_report(text: str, *, dry_run: bool = False) -> int:
|
|
recipients = unique([*env_recipients(), *(await db_recipients())])
|
|
if dry_run:
|
|
print(f"telegram_report_dry_run recipients={len(recipients)}")
|
|
return len(recipients)
|
|
if not settings.bot_token or not recipients:
|
|
print("telegram_report_skipped")
|
|
return 0
|
|
|
|
sent = 0
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
for chat_id in recipients:
|
|
try:
|
|
response = await client.post(
|
|
f"https://api.telegram.org/bot{settings.bot_token}/sendMessage",
|
|
json={"chat_id": chat_id, "text": text, "disable_web_page_preview": True},
|
|
)
|
|
response.raise_for_status()
|
|
sent += 1
|
|
except Exception as exc: # noqa: BLE001 - deploy report must never fail deploy
|
|
print(f"telegram_report_failed chat_id={chat_id} error={type(exc).__name__}")
|
|
print(f"telegram_report_sent_count {sent}")
|
|
return sent
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Send a CarPass operational Telegram report.")
|
|
parser.add_argument("--text", help="Report text. Defaults to CARPASS_REPORT_TEXT.")
|
|
parser.add_argument("--dry-run", action="store_true")
|
|
return parser.parse_args()
|
|
|
|
|
|
async def main() -> None:
|
|
args = parse_args()
|
|
text = args.text or os.getenv("CARPASS_REPORT_TEXT") or ""
|
|
if not text.strip():
|
|
raise SystemExit("Report text is required")
|
|
await send_report(text, dry_run=args.dry_run)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|