245 lines
7.8 KiB
Python
245 lines
7.8 KiB
Python
"""Push-уведомления Seaca через Firebase Cloud Messaging (HTTP v1)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.config import get_settings
|
|
from app.models import Event, Problem
|
|
from app.models.mobile_device import MobileDevice
|
|
from app.services.notification_severity import severity_meets_minimum
|
|
from app.services.notification_policy import get_effective_notification_policy
|
|
from app.services.telegram_templates import format_event_mobile_push
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
FCM_SCOPE = "https://www.googleapis.com/auth/firebase.messaging"
|
|
|
|
|
|
class MobileNotConfiguredError(Exception):
|
|
"""FCM отключён или не задан service account."""
|
|
|
|
|
|
class MobileSendError(Exception):
|
|
def __init__(self, message: str, *, status_code: int | None = None) -> None:
|
|
super().__init__(message)
|
|
self.status_code = status_code
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FcmConfig:
|
|
enabled: bool
|
|
project_id: str
|
|
service_account_path: str
|
|
|
|
@property
|
|
def configured(self) -> bool:
|
|
return bool(self.project_id.strip() and self.service_account_path.strip())
|
|
|
|
@property
|
|
def active(self) -> bool:
|
|
return self.enabled and self.configured
|
|
|
|
|
|
def get_effective_fcm_config() -> FcmConfig:
|
|
settings = get_settings()
|
|
return FcmConfig(
|
|
enabled=bool(settings.sac_fcm_enabled),
|
|
project_id=(settings.sac_fcm_project_id or "").strip(),
|
|
service_account_path=(settings.sac_fcm_service_account_json or "").strip(),
|
|
)
|
|
|
|
|
|
def _load_service_account(path: str) -> dict:
|
|
file_path = Path(path)
|
|
if not file_path.is_file():
|
|
raise MobileNotConfiguredError(f"FCM service account file not found: {path}")
|
|
with file_path.open(encoding="utf-8") as handle:
|
|
return json.load(handle)
|
|
|
|
|
|
def _fcm_access_token(service_account_path: str) -> str:
|
|
try:
|
|
from google.oauth2 import service_account
|
|
from google.auth.transport.requests import Request
|
|
except ImportError as exc:
|
|
msg = str(exc).lower()
|
|
if "requests" in msg:
|
|
raise MobileNotConfiguredError(
|
|
"FCM: установите пакет requests (pip install requests)"
|
|
) from exc
|
|
raise MobileNotConfiguredError("FCM: установите google-auth") from exc
|
|
|
|
creds = service_account.Credentials.from_service_account_file(
|
|
service_account_path,
|
|
scopes=[FCM_SCOPE],
|
|
)
|
|
creds.refresh(Request())
|
|
if not creds.token:
|
|
raise MobileSendError("Failed to obtain FCM access token")
|
|
return creds.token
|
|
|
|
|
|
def _active_device_tokens(db: Session | None) -> list[str]:
|
|
if db is None:
|
|
return []
|
|
rows = db.scalars(
|
|
select(MobileDevice).where(
|
|
MobileDevice.revoked_at.is_(None),
|
|
MobileDevice.fcm_token.is_not(None),
|
|
)
|
|
).all()
|
|
tokens: list[str] = []
|
|
for row in rows:
|
|
token = (row.fcm_token or "").strip()
|
|
if token:
|
|
tokens.append(token)
|
|
return tokens
|
|
|
|
|
|
def _send_fcm_data(
|
|
tokens: list[str],
|
|
data: dict[str, str],
|
|
*,
|
|
title: str,
|
|
body: str,
|
|
require_success: bool = False,
|
|
) -> None:
|
|
if not tokens:
|
|
if require_success:
|
|
raise MobileSendError("Нет FCM-токенов для отправки")
|
|
return
|
|
cfg = get_effective_fcm_config()
|
|
if not cfg.active:
|
|
if require_success:
|
|
raise MobileNotConfiguredError("FCM не настроен на сервере")
|
|
return
|
|
|
|
try:
|
|
access_token = _fcm_access_token(cfg.service_account_path)
|
|
except (MobileNotConfiguredError, MobileSendError) as exc:
|
|
logger.warning("FCM skipped: %s", exc)
|
|
if require_success:
|
|
raise
|
|
return
|
|
except Exception:
|
|
logger.exception("FCM token acquisition failed")
|
|
if require_success:
|
|
raise MobileSendError("Failed to obtain FCM access token", status_code=502)
|
|
return
|
|
|
|
url = f"https://fcm.googleapis.com/v1/projects/{cfg.project_id}/messages:send"
|
|
headers = {
|
|
"Authorization": f"Bearer {access_token}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
errors: list[str] = []
|
|
|
|
# Data-only: Android always calls SeacaMessagingService.onMessageReceived so the app
|
|
# can honour per-device notification mode (sound / silent / off).
|
|
message_data = {**data, "title": title, "body": body}
|
|
|
|
with httpx.Client(timeout=12.0) as client:
|
|
for token in tokens:
|
|
payload = {
|
|
"message": {
|
|
"token": token,
|
|
"data": message_data,
|
|
"android": {"priority": "high"},
|
|
}
|
|
}
|
|
try:
|
|
response = client.post(url, headers=headers, json=payload)
|
|
response.raise_for_status()
|
|
except httpx.HTTPStatusError as exc:
|
|
detail = exc.response.text[:200] if exc.response is not None else str(exc)
|
|
logger.warning("FCM send failed for token prefix %s: %s", token[:8], detail)
|
|
errors.append(detail or f"HTTP {exc.response.status_code if exc.response else '?'}")
|
|
except Exception as exc:
|
|
logger.exception("FCM send failed")
|
|
errors.append(str(exc))
|
|
|
|
if require_success and errors:
|
|
raise MobileSendError(errors[0], status_code=502)
|
|
|
|
|
|
def _event_payload(event: Event) -> tuple[str, str, dict[str, str]]:
|
|
title, body = format_event_mobile_push(event)
|
|
data = {
|
|
"kind": "event",
|
|
"id": str(event.id),
|
|
"severity": event.severity,
|
|
"type": event.type,
|
|
}
|
|
return title, body, data
|
|
|
|
|
|
def _problem_payload(problem: Problem) -> tuple[str, str, dict[str, str]]:
|
|
title = f"[{problem.severity}] {problem.title}"
|
|
body = (problem.summary or "")[:200]
|
|
data = {
|
|
"kind": "problem",
|
|
"id": str(problem.id),
|
|
"severity": problem.severity,
|
|
"status": problem.status,
|
|
}
|
|
return title, body, data
|
|
|
|
|
|
def notify_event(event: Event, *, db: Session | None = None, apply_policy_gate: bool = True) -> None:
|
|
policy = get_effective_notification_policy(db)
|
|
if apply_policy_gate and not severity_meets_minimum(event.severity, policy.min_severity):
|
|
return
|
|
tokens = _active_device_tokens(db)
|
|
if not tokens:
|
|
return
|
|
title, body, data = _event_payload(event)
|
|
_send_fcm_data(tokens, data, title=title, body=body)
|
|
|
|
|
|
def notify_problem(
|
|
problem: Problem,
|
|
event: Event | None = None,
|
|
*,
|
|
db: Session | None = None,
|
|
apply_policy_gate: bool = True,
|
|
) -> None:
|
|
policy = get_effective_notification_policy(db)
|
|
if apply_policy_gate and not severity_meets_minimum(problem.severity, policy.min_severity):
|
|
return
|
|
tokens = _active_device_tokens(db)
|
|
if not tokens:
|
|
return
|
|
title, body, data = _problem_payload(problem)
|
|
_send_fcm_data(tokens, data, title=title, body=body)
|
|
|
|
|
|
def send_test_push(*, db: Session, device_id: int) -> None:
|
|
cfg = get_effective_fcm_config()
|
|
if not cfg.enabled:
|
|
raise MobileNotConfiguredError("FCM отключён (SAC_FCM_ENABLED=false)")
|
|
if not cfg.configured:
|
|
raise MobileNotConfiguredError("Не задан SAC_FCM_PROJECT_ID или SAC_FCM_SERVICE_ACCOUNT_JSON")
|
|
|
|
device = db.get(MobileDevice, device_id)
|
|
if device is None or device.revoked_at is not None:
|
|
raise ValueError("device not found or revoked")
|
|
token = (device.fcm_token or "").strip()
|
|
if not token:
|
|
raise ValueError("device has no FCM token")
|
|
|
|
_send_fcm_data(
|
|
[token],
|
|
{"kind": "test", "id": "0", "severity": "info"},
|
|
title="SAC: тестовое push",
|
|
body="Канал Seaca (FCM) работает.",
|
|
require_success=True,
|
|
)
|