"""SMS.ir provider implementation. API documentation: https://sms.ir/rest-api/ Supports Iranian numbers (+98) only. Credentials (stored in SmsProvider.credentials JSON): api_key : str — X-API-KEY header value line_number: str — sender line number (e.g. "30004505001175") Extra params (stored in SmsProvider.extra_params JSON): otp_template_id: int — SMS.ir template ID for OTP messages (e.g. 453080) otp_param_name : str — parameter name in the template (default: "Code") """ from __future__ import annotations import structlog from decimal import Decimal from typing import Any from urllib import request as urllib_request from urllib.error import URLError, HTTPError import json as json_lib from simorgh.apps.notifications.sms.base import SmsSendResult _log = structlog.get_logger("simorgh.notifications.sms.sms_ir") _BASE_URL = "https://api.sms.ir/v1" _TIMEOUT = 15 # seconds def _http_post(url: str, payload: dict, api_key: str) -> dict: """Make a JSON POST request using stdlib urllib (no extra dependencies).""" data = json_lib.dumps(payload).encode("utf-8") req = urllib_request.Request( url, data=data, method="POST", headers={ "Content-Type": "application/json", "Accept": "application/json", "X-API-KEY": api_key, }, ) try: with urllib_request.urlopen(req, timeout=_TIMEOUT) as resp: return json_lib.loads(resp.read().decode("utf-8")) except HTTPError as exc: body = exc.read().decode("utf-8", errors="replace") try: return json_lib.loads(body) except Exception: raise RuntimeError(f"SMS.ir HTTP {exc.code}: {body}") from exc except URLError as exc: raise RuntimeError(f"SMS.ir connection error: {exc.reason}") from exc class SmsIrProvider: """SMS.ir — Iranian SMS provider (خط خدماتی for OTP, خط اختصاصی for bulk).""" slug = "sms_ir" name = "SMS.ir" supported_country_codes: list[str] = ["+98"] priority: int = 100 # highest priority for Iranian numbers def __init__( self, api_key: str, line_number: str, otp_template_id: int, otp_param_name: str = "Code", ) -> None: self._api_key = api_key self._line_number = line_number self._otp_template_id = otp_template_id self._otp_param_name = otp_param_name # ------------------------------------------------------------------ # Public interface # ------------------------------------------------------------------ def send( self, *, mobile: str, text: str, tenant_id: int | None = None, ) -> SmsSendResult: """Send a plain-text SMS from the dedicated line (خط اختصاصی).""" url = f"{_BASE_URL}/send/bulk" payload: dict[str, Any] = { "lineNumber": int(self._line_number), "messageText": text, "mobiles": [mobile], } _log.info("sms_ir.send", mobile=mobile, tenant_id=tenant_id) try: resp = _http_post(url, payload, self._api_key) except Exception as exc: _log.error("sms_ir.send_error", error=str(exc), mobile=mobile) return SmsSendResult(success=False, error=str(exc), raw_response={}) return self._parse_bulk_response(resp) def send_otp( self, *, mobile: str, code: str, tenant_id: int | None = None, extra: dict | None = None, ) -> SmsSendResult: """Send an OTP code via SMS.ir Verify endpoint (خط خدماتی).""" url = f"{_BASE_URL}/send/verify" payload: dict[str, Any] = { "mobile": mobile, "templateId": self._otp_template_id, "parameters": [ {"name": self._otp_param_name, "value": code}, ], } # Allow callers to inject extra template parameters if extra: for name, value in extra.items(): payload["parameters"].append({"name": name, "value": str(value)}) _log.info("sms_ir.send_otp", mobile=mobile, tenant_id=tenant_id) try: resp = _http_post(url, payload, self._api_key) except Exception as exc: _log.error("sms_ir.send_otp_error", error=str(exc), mobile=mobile) return SmsSendResult(success=False, error=str(exc), raw_response={}) return self._parse_verify_response(resp) def send_template( self, *, mobile: str, template_id: str, params: dict, tenant_id: int | None = None, ) -> SmsSendResult: """Send a template-based SMS via sms.ir Verify endpoint. Uses the same ``POST /v1/send/verify`` API as OTP but accepts an arbitrary ``template_id`` (the numeric ID the admin sets in the sms.ir panel) and a ``params`` dict of named substitution values. Admin can register different templates for each notification kind via the ``SmsProviderTemplate`` model in Django admin. Args: mobile: Recipient E.164 number, e.g. ``+989123456789``. template_id: Numeric template ID from sms.ir panel (passed as str, will be cast to int for the API payload). params: Dict of parameter name → value pairs that the template expects, e.g. ``{"ticket_id": "abc", "subject": "..."}``. tenant_id: Optional tenant for rate-limit / logging context. """ url = f"{_BASE_URL}/send/verify" payload: dict[str, Any] = { "mobile": mobile, "templateId": int(template_id), "parameters": [ {"name": k, "value": str(v)} for k, v in params.items() ], } _log.info( "sms_ir.send_template", mobile=mobile, template_id=template_id, tenant_id=tenant_id, ) try: resp = _http_post(url, payload, self._api_key) except Exception as exc: _log.error( "sms_ir.send_template_error", error=str(exc), mobile=mobile, template_id=template_id, ) return SmsSendResult(success=False, error=str(exc), raw_response={}) return self._parse_verify_response(resp) def supports_country_code(self, dial_code: str) -> bool: return dial_code in self.supported_country_codes # ------------------------------------------------------------------ # Response parsers # ------------------------------------------------------------------ @staticmethod def _parse_verify_response(resp: dict) -> SmsSendResult: status = resp.get("status", 0) if status != 1: return SmsSendResult( success=False, error=resp.get("message", "Unknown error"), raw_response=resp, ) data = resp.get("data", {}) cost_raw = data.get("cost") return SmsSendResult( success=True, provider_message_id=str(data.get("messageId", "")), cost=Decimal(str(cost_raw)) if cost_raw is not None else None, raw_response=resp, ) @staticmethod def _parse_bulk_response(resp: dict) -> SmsSendResult: status = resp.get("status", 0) if status != 1: return SmsSendResult( success=False, error=resp.get("message", "Unknown error"), raw_response=resp, ) data = resp.get("data", {}) message_ids = data.get("messageIds", []) cost_raw = data.get("cost") return SmsSendResult( success=True, provider_message_id=str(message_ids[0]) if message_ids else "", cost=Decimal(str(cost_raw)) if cost_raw is not None else None, raw_response=resp, )