Merge pull request #472 from tutorcruncher/message_bird_pricing_through_webhooks

Author: tomhamiltonstubber

src/schemas/webhooks.py

@@ -98,13 +98,19 @@ class MessageBirdWebHook(BaseWebhook):
     status: MessageBirdMessageStatus
     message_id: IDStr
     error_code: str = None
+    price_amount: float
 
     def extra_json(self, sort_keys=False):
         return json.dumps({'error_code': self.error_code} if self.error_code else {}, sort_keys=sort_keys)
 
     class Config:
         ignore_extra = True
-        fields = {'message_id': 'id', 'ts': 'statusDatetime', 'error_code': 'statusErrorCode'}
+        fields = {
+            'message_id': 'id',
+            'ts': 'statusDatetime',
+            'error_code': 'statusErrorCode',
+            'price_amount': 'price[amount]',
+        }
 
 
 ID_REGEX = re.compile(r'[/<>= ]')

src/views/messages.py

@@ -97,7 +97,7 @@ async def messages_list(
         data['previous'] = f"{this_url}?offset={max(offset - LIST_PAGE_SIZE, 0)}"
     if 'sms' in method:
         start, end = month_interval()
-        data['spend'] = await get_sms_spend(conn, company_id=company_id, start=start, end=end, method=method)
+        data['spend'] = await get_sms_spend(conn, company_id=company_id, start=start, end=end, method=method) or 0
     return data
 
 

src/views/webhooks.py

@@ -56,5 +56,8 @@ async def messagebird_webhook_view(request: Request):
         event = MessageBirdWebHook(**request.query_params)
     except ValidationError as e:
         raise HttpUnprocessableEntity(e.args[0])
-    await glove.redis.enqueue_job('update_message_status', SendMethod.sms_messagebird, event)
+    method = SendMethod.sms_messagebird
+    if (test := request.query_params.get('test')) and test.lower() == 'true':
+        method = SendMethod.sms_test
+    await glove.redis.enqueue_job('update_message_status', method, event)
     return 'message status updated\n'

src/worker/sms.py

@@ -1,6 +1,5 @@
 from dataclasses import asdict, dataclass
 
-import asyncio
 import chevron
 import json
 import logging
@@ -20,7 +19,7 @@
 from phonenumbers.geocoder import country_name_for_number, description_for_number
 from typing import Optional
 
-from src.ext import ApiError, MessageBird
+from src.ext import MessageBird
 from src.render.main import MessageTooLong, SmsLength, apply_short_links, sms_length
 from src.schemas.messages import MessageStatus, SmsRecipientModel, SmsSendMethod, SmsSendModel
 from src.settings import Settings
@@ -127,7 +126,6 @@ async def _test_send_sms(self, sms_data: SmsData):
         # remove the + from the beginning of the number
         msg_id = f'{self.m.uid}-{sms_data.number.number[1:]}'
         send_ts = utcnow()
-        cost = 0.012 * sms_data.length.parts
         output = (
             f'to: {sms_data.number}\n'
             f'msg id: {msg_id}\n'
@@ -136,7 +134,6 @@ async def _test_send_sms(self, sms_data: SmsData):
             f'tags: {self.tags}\n'
             f'company_code: {self.m.company_code}\n'
             f'from_name: {self.from_name}\n'
-            f'cost: {cost}\n'
             f'length: {sms_data.length}\n'
             f'message:\n'
             f'{sms_data.message}\n'
@@ -146,85 +143,12 @@ async def _test_send_sms(self, sms_data: SmsData):
             save_path = self.settings.test_output / f'{msg_id}.txt'
             test_logger.info('sending message: %s (saved to %s)', output, save_path)
             save_path.write_text(output)
-        await self._store_sms(msg_id, send_ts, sms_data, cost)
-
-    async def _messagebird_get_mcc_cost(self, redis, mcc):
-        rates_key = 'messagebird-rates'
-        if not await redis.exists(rates_key):
-            # get fresh data on rates by mcc
-            main_logger.info('getting fresh pricing data from messagebird...')
-            r = await self.messagebird.get('pricing/sms/outbound')
-            if r.status_code != 200:
-                response = r.text
-                main_logger.error(
-                    'error getting messagebird api', extra={'status': r.status_code, 'response': response}
-                )
-                raise MessageBirdExternalError((r.status_code, response))
-            data = r.json()
-            prices = data['prices']
-            if not next((1 for g in prices if g['mcc'] == '0'), None):
-                main_logger.error('no default messagebird pricing with mcc "0"', extra={'prices': prices})
-            prices = {g['mcc']: f'{float(g["price"]):0.5f}' for g in prices}
-            await asyncio.gather(redis.hmset_dict(rates_key, prices), redis.expire(rates_key, ONE_DAY))
-        rate = await redis.hget(rates_key, mcc, encoding='utf8')
-        if not rate:
-            main_logger.warning('no rate found for mcc: "%s", using default', mcc)
-            rate = await redis.hget(rates_key, '0', encoding='utf8')
-        assert rate, f'no rate found for mcc: {mcc}'
-        return float(rate)
-
-    async def _messagebird_get_number_cost(self, number: Number):
-        cc_mcc_key = f'messagebird-cc:{number.country_code}'
-        with await self.ctx['redis'] as redis:
-            mcc = await redis.get(cc_mcc_key)
-            if mcc is None:
-                main_logger.info('no mcc for %s, doing HLR lookup...', number.number)
-                api_number = number.number.replace('+', '')
-
-                data = {'msisdn': api_number}
-                response = await self.messagebird.post('hlr', **data)
-                hlr_data = response.json()
-                hlr_id = hlr_data.get('id')
-
-                network, hlr = None, None
-                for i in range(60):
-                    try:
-                        r = await self.messagebird.get(f'hlr/{hlr_id}')
-                    except ApiError:
-                        main_logger.info('Error retrieving HLR data for %s, attempt %d', number.number, i)
-                        continue
-                    hlr = r.json()
-                    if hlr.get('errors'):
-                        continue
-                    network = hlr.get('network')
-                    if not network:
-                        continue
-                    elif hlr['status'] == 'active':
-                        main_logger.info(
-                            'found result for %s after %d attempts %s', number.number, i, json.dumps(hlr, indent=2)
-                        )
-                        break
-                    await asyncio.sleep(1)
-                if not hlr or not network:
-                    main_logger.warning('No HLR result found for %s after 30 attempts', number.number, extra=hlr)
-                    return
-                mcc = str(network)[:3]
-                await redis.setex(cc_mcc_key, ONE_YEAR, mcc)
-            return await self._messagebird_get_mcc_cost(redis, mcc)
+        await self._store_sms(msg_id, send_ts, sms_data)
 
     async def _messagebird_send_sms(self, sms_data: SmsData):
-        try:
-            msg_cost = await self._messagebird_get_number_cost(sms_data.number)
-        except MessageBirdExternalError:
-            msg_cost = 0  # Set to SMS cost to 0 until cost API is working/changed
-        if msg_cost is None:
-            return
-
-        cost = sms_data.length.parts * msg_cost
         send_ts = utcnow()
-        main_logger.info(
-            'sending SMS to %s, parts: %d, cost: %0.2fp', sms_data.number.number, sms_data.length.parts, cost * 100
-        )
+        main_logger.info('sending SMS to %s, parts: %d', sms_data.number.number, sms_data.length.parts)
+
         r = await self.messagebird.post(
             'messages',
             originator=self.from_name,
@@ -237,9 +161,9 @@ async def _messagebird_send_sms(self, sms_data: SmsData):
         data = r.json()
         if data['recipients']['totalCount'] != 1:
             main_logger.error('not one recipients in send response', extra={'data': data})
-        await self._store_sms(data['id'], send_ts, sms_data, cost)
+        await self._store_sms(data['id'], send_ts, sms_data)
 
-    async def _store_sms(self, external_id, send_ts, sms_data: SmsData, cost: float):
+    async def _store_sms(self, external_id, send_ts, sms_data: SmsData):
         message_id = await glove.pg.fetchval_b(
             'insert into messages (:values__names) values :values returning id',
             values=Values(
@@ -255,7 +179,6 @@ async def _store_sms(self, external_id, send_ts, sms_data: SmsData, cost: float)
                 to_address=sms_data.number.number_formatted,
                 tags=self.tags,
                 body=sms_data.message,
-                cost=cost,
                 extra=json.dumps(asdict(sms_data.length)),
             ),
         )

src/worker/webhooks.py

@@ -1,3 +1,4 @@
+import asyncio
 import hashlib
 import json
 import logging
@@ -10,7 +11,7 @@
 from ua_parser.user_agent_parser import Parse as ParseUserAgent
 
 from src.schemas.messages import SendMethod
-from src.schemas.webhooks import BaseWebhook, MandrillWebhook
+from src.schemas.webhooks import BaseWebhook, MandrillWebhook, MessageBirdWebHook
 
 main_logger = logging.getLogger('worker.webhooks')
 
@@ -81,6 +82,7 @@ async def update_message_status(ctx, send_method: SendMethod, m: BaseWebhook, lo
     message_id = await glove.pg.fetchval_b(
         'select id from messages where :where', where=(V('external_id') == m.message_id) & (V('method') == send_method)
     )
+
     if not message_id:
         return UpdateStatus.missing
 
@@ -90,8 +92,17 @@ async def update_message_status(ctx, send_method: SendMethod, m: BaseWebhook, lo
     if log_each:
         main_logger.info('adding event %s, ts: %s, status: %s', m.message_id, m.ts, m.status)
 
-    await glove.pg.execute_b(
-        'insert into events (:values__names) values :values',
-        values=Values(message_id=message_id, status=m.status, ts=m.ts, extra=m.extra_json()),
-    )
+    qs = [
+        glove.pg.execute_b(
+            'insert into events (:values__names) values :values',
+            values=Values(message_id=message_id, status=m.status, ts=m.ts, extra=m.extra_json()),
+        ),
+    ]
+    if isinstance(m, MessageBirdWebHook):
+        cost = m.price_amount
+        qs.append(
+            glove.pg.execute_b('update messages set cost=:cost where id=:message_id', cost=cost, message_id=message_id)
+        )
+    await asyncio.gather(*qs)
+
     return UpdateStatus.added

tests/conftest.py

@@ -14,6 +14,7 @@
 from pathlib import Path
 from starlette.testclient import TestClient
 from typing import Any, Callable
+from urllib.parse import urlencode
 
 from src.schemas.messages import EmailSendModel, SendMethod
 from src.settings import Settings
@@ -231,6 +232,27 @@ def _send_message(**extra):
     return _send_message
 
 
+@pytest.fixture
+def send_webhook(cli, worker, loop):
+    def _send_webhook(ext_id, price, **extra):
+        url_args = {
+            'id': ext_id,
+            'reference': 'morpheus',
+            'recipient': '447896541236',
+            'status': 'delivered',
+            'statusDatetime': '2032-06-06T12:00:00',
+            'price[amount]': price,
+            'test': True,
+        }
+
+        url_args.update(**extra)
+        r = cli.get(f'/webhook/messagebird/?{urlencode(url_args)}')
+        assert r.status_code == 200
+        worker.test_run()
+
+    return _send_webhook
+
+
 @pytest.fixture(name='call_send_emails')
 def _fix_call_send_emails(glove, sync_db):
     def run(**kwargs):