aiosmsactivate.client

  1import asyncio
  2import json
  3import logging
  4import re
  5import time
  6from typing import Literal
  7
  8import aiohttp
  9
 10from .utils import is_json
 11from .exceptions import SmsActivateException
 12from .models import ActivationData, Number, SetActivationStatusResponse, Sms
 13from .types import SetActivationStatus, ActivationStatus
 14
 15__all__ = [
 16    "SmsActivate",
 17]
 18
 19
 20allowed_domains = [
 21    'https://api.sms-activate.ae/stubs/handler_api.php',
 22    'https://api.sms-activate.ru/stubs/handler_api.php',
 23    'https://api.sms-activate.io/stubs/handler_api.php',
 24    'https://api.sms-activate.page/stubs/handler_api.php',
 25]
 26
 27class SmsActivate:
 28    """
 29    RU  
 30    Спасибо за использование моей библиотеки, вы можете принять участие в развитии библиотеки  
 31      
 32    ВАЖНО
 33    библиотека полностью поддерживает все методы с оффициальной документации
 34    https://sms-activate.page/api2 на момент 08.07.2025  
 35      
 36    на git: https://github.com/AioSmsProviders/aiosmsactivate
 37    Так же можете писать в чат https://t.me/+5YQ8k6H02bkxZmRi
 38    или обратиться к главному разработчику с идеями, предложениями и багами: https://t.me/lolkof  
 39    
 40    EN  
 41    Thank you for using my library, you can participate in the development of the library.  
 42      
 43    important
 44    The library fully supports all methods from the official documentation
 45    https://sms-activate.page/api2 as of 07/08/2025  
 46      
 47    on git: https://github.com/AioSmsProviders/aiosmsactivate
 48    You can also write to the chat https://t.me/+5YQ8k6H02bkxZmRi
 49    or contact the main developer with ideas, suggestions, and bugs: https://t.me/lolkof
 50    
 51    SIMPLE USAGE
 52    ```python
 53    from aiosmsactivate import SmsActivate
 54    from aiosmsactivate.types import SetActivationStatus
 55
 56    import asyncio
 57
 58
 59    sa = SmsActivate('token')
 60
 61    async def main():
 62        balance = await sa.get_balance()
 63        print(balance) # 6.25
 64        
 65        number = await sa.purchase('ya')
 66        number.activation_id # 3807035855
 67        number.phone_number # '79238944456'
 68        number.operator # 'mtt'
 69        print(number)
 70        # activation_id=3807035855 phone_number='79238944456' activation_cost=0.2 
 71        # country_code='0' can_get_another_sms=True activation_time='2025-07-08 10:49:27' 
 72        # operator='mtt' 
 73        
 74        code = await number.wait_sms_code(timeout=300)
 75        print(code) # 1234
 76        
 77        status = await number.get_activation_status()
 78        
 79        await number.set_activation_status(SetActivationStatus.CANCEL) # Отменить номер || Cancel number
 80        await number.set_activation_status(8) # Отменить номер || Cancel number
 81        
 82    asyncio.run(main())
 83    ```
 84    """
 85
 86    def __init__(self, api_key: str, base_url: str | list = allowed_domains):
 87        """
 88        RU  
 89        api_key передавать api ключ, получить можно вот тут: https://sms-activate.page/profile
 90        В base_url можно указать список адресов, модуль будет проходиться по всем, пока не найдёт рабочий
 91        а можно указать один или вообще не указывать, если не указать будет браться из allowed_domains  
 92        
 93        EN  
 94        api_key to transfer the api key, you can get it here: https://sms-activate.page/profile
 95        You can specify a list of addresses in base_url, and the module will go through all of them until it finds a working one.
 96        or you can specify one or not at all, if not specified, it will be taken from allowed_domains.
 97        """
 98        self._api_key = api_key
 99        if isinstance(base_url, str):
100            base_url = [base_url]
101        self._base_urls = base_url
102        self._accept_url = None
103
104    async def __send_request(self, action: str, **kwargs):
105        last_exception = None
106
107        for url in self._base_urls:
108            try:
109                url = self._accept_url if self._accept_url else url
110                params = kwargs.get('params')
111                if params:
112                    kwargs.pop('params')
113                async with aiohttp.ClientSession() as session:
114                    async with session.request(
115                        'POST',
116                        url,
117                        **kwargs,
118                        params={
119                            'api_key': self._api_key,
120                            'action': action,
121                            **(params if params else {})
122                        }
123                    ) as response:
124                        response.raise_for_status()
125                        logging.debug(response.real_url)
126                        return await response.text()
127            except Exception as e:
128                last_exception = e
129                continue
130            self._accept_url = url
131            break
132
133        raise last_exception
134
135    async def get_balance(self, cashback: bool = False) -> float:
136        pattern = re.compile(r'ACCESS_BALANCE:(\d+\.\d{2})')
137        response = await self.__send_request('getBalance' if not cashback else 'getBalanceAndCashBack')
138        match = pattern.match(response)
139        if not match:
140            raise SmsActivateException('Invalid response sequence')
141
142        return float(match.group(1))
143    async def get_balance_and_cashback(self):
144        return await self.get_balance(cashback=True)
145
146    async def get_available_countries(self, service: str, freePrice: bool | str) -> dict[str, ...]:
147        response = await self.__send_request('getTopCountriesByService', params={
148            'service': service,
149            'freePrice': str(freePrice).lower()
150        })
151        
152        if not is_json(response):
153            return response
154        
155        return json.loads(response)
156    
157    async def get_count_numbers(self, country: str, operator: str) -> dict[str, ...]:
158        response = await self.__send_request('getNumbersStatus', params={
159            'country': country,
160            'operator': operator
161        })
162        
163        if not is_json(response):
164            return response
165        
166        return json.loads(response)
167    
168    async def get_operators(self, country: str = None) -> dict[str, ...]:
169        params = {}
170        if country is not None:
171            params["country"] = country
172        response = await self.__send_request('getOperators', params=params)
173        
174        if not is_json(response):
175            return response
176        
177        return json.loads(response)
178    
179    async def get_active_activations(self) -> dict[str, ...]:
180        response = await self.__send_request('getActiveActivations')
181        
182        if not is_json(response):
183            return response
184        
185        return json.loads(response)
186
187    async def get_activation_status_v1(self, id: str) -> tuple[ActivationStatus, str | None]:
188        response = await self.__send_request('getStatus', params={
189            'id': id
190        })
191
192        data = response.split(':')
193
194        match data[0]:
195            case 'STATUS_WAIT_CODE':
196                return ActivationStatus.WAIT, None
197            case 'STATUS_WAIT_RETRY':
198                return ActivationStatus.RETRY, data[1]
199            case 'STATUS_WAIT_RESEND':
200                return ActivationStatus.RESEND, None
201            case 'STATUS_CANCEL':
202                return ActivationStatus.CANCEL, None
203            case 'STATUS_OK':
204                return ActivationStatus.OK, data[1]
205            case _:
206                raise SmsActivateException('Invalid response sequence')
207    
208    async def get_activation_status(self, activation_id: str | int | Number) -> ActivationData | str:
209        if isinstance(activation_id, Number):
210            activation_id = activation_id.activation_id
211        response = await self.__send_request('getStatusV2', params={
212            'id': activation_id
213        })
214
215        if not is_json(response):
216            return response
217        
218        return ActivationData(**json.loads(response))
219    
220    async def wait_sms_code(self, activation_id: str | int | Number, timeout: int = 60*5, per_attempt: int = 5) -> Sms | str | int | None:
221        """
222        Ожидание смс кода
223        Wait sms code
224
225        Аргументы:
226            activation_id: activation_id номера или целый объект номера
227            timeout: максимальное время ожидание смс в секундах, по умолчанию 5 минут 
228            per_attempt: время между попыткой получить смс, по умолчанию 5 секунд
229            
230        Args:
231            activation_id: activation_id of number or Number object
232            timeout: maximum time to wait sms code 
233            per_attempt: time per attempt
234            
235        Returns: Sms
236        """
237        activation_id = activation_id.activation_id if isinstance(activation_id, Number) else activation_id
238        if not self._api_key:
239            raise ValueError('API key is required for this method')
240
241        try:
242            await self.set_activation_status(activation_id=activation_id, status=SetActivationStatus.READY)
243        except:
244            pass
245
246        start_time = time.time()
247        
248        while time.time() - start_time < timeout:
249            await asyncio.sleep(per_attempt)
250            status, code = await self.get_activation_status_v1(activation_id)
251            if status == ActivationStatus.OK:
252                try:
253                    await self.set_activation_status(activation_id, SetActivationStatus.AGAIN)
254                except:
255                    pass
256                return code
257        
258        return None
259
260    async def purchase(self, service: str, forward: bool | None = None, maxPrice: float | None = None,
261                       phoneException: str | None = None, operator: str | None = None,
262                       activationType: int | str | None = None, language: str | None = None,
263                       userId: str | int | None = None,
264                       ref: str | None = None, country: str | None = None,
265                       useCashBack: bool | None = None,
266                       orderId: str | int | None = None,
267                       _is_v2: bool = True
268                       ) -> Number | str:
269        response = await self.__send_request('getNumber' if not _is_v2 else 'getNumberV2', params={
270            'service': service,
271            **({'forward': 1 if forward else 0} if forward is not None else {}),
272            **({'maxPrice': str(maxPrice)} if maxPrice is not None else {}),
273            **({'phoneException': phoneException} if phoneException is not None else {}),
274            **({'operator': operator} if operator is not None else {}),
275            **({'activationType': str(activationType)} if activationType is not None else {}),
276            **({'language': str(language)} if language is not None else {}),
277            **({'userId': str(userId)} if userId is not None else {}),
278            **({'orderId': str(orderId)} if orderId is not None and _is_v2 else {}),
279            **({'ref': ref} if ref is not None else {}),
280            **({'country ': country} if country is not None else {}),
281            **({'useCashBack': str(useCashBack).lower()} if useCashBack is not None else {}),
282        })
283
284        if not is_json(response):
285            return response
286        
287        return Number.from_response(self, json.loads(response))
288    
289    async def get_number(self, *args, **kwargs):
290        kwargs["_is_v2"] = False
291        return await self.purchase(*args, **kwargs)
292    
293    async def get_multi_service_number(self, 
294                        multiService: str, multiForward: str | None = None,
295                        operator: str | None = None,
296                        ref: str | None = None, country: str | None = None,
297                       ) -> dict:
298        """
299        Get multiservice number.
300
301        :param multiService: service1,service2,service3 (Services separated by commas)
302        :param multiForward: 1,0,1 (forwards separated by commas, forwards count equal services count)
303        :return: dict object of response
304        """
305        response = await self.__send_request('getMultiServiceNumber', params={
306            'multiService': multiService,
307            **({'multiForward': multiForward} if multiForward is not None else {}),
308            **({'operator': operator} if operator is not None else {}),
309            **({'ref': ref} if ref is not None else {}),
310            **({'country ': country} if country is not None else {}),
311        })
312
313        if not is_json(response):
314            return response
315        
316        return json.loads(response)
317    
318
319    async def set_activation_status(self, activation_id: str | int, status: SetActivationStatus | int,
320                                    forward: str | None = None) -> SetActivationStatusResponse:
321        members = {member.value: member for member in SetActivationStatusResponse}
322        response = await self.__send_request('setStatus', params={
323            'id': activation_id,
324            'status': status.value if isinstance(status, SetActivationStatus) else status,
325            **({'forward': forward} if forward is not None else {})
326        })
327
328        return members[response]
329
330    async def get_history(self, 
331                          start: str | int = None,
332                          end: str | int = None,
333                          offset: str | int = None,
334                          limit: str | int = None,
335                       ) -> dict | list:
336        response = await self.__send_request('getHistory', params={
337            **({'start': str(start)} if start is not None else {}),
338            **({'end': str(end)} if end is not None else {}),
339            **({'offset': str(offset)} if offset is not None else {}),
340            **({'limit': str(limit)} if limit is not None else {}),
341        })
342
343        if not is_json(response):
344            return response
345        
346        return json.loads(response)
347    
348    async def get_list_top_countries(self, 
349                          service: str,
350                       ) -> dict | list:
351        response = await self.__send_request('getListOfTopCountriesByService', params={
352            'service': service
353        })
354
355        if not is_json(response):
356            return response
357        
358        return json.loads(response)
359    
360    async def get_incoming_call_status(self, 
361                          id: str | int = None,
362                       ) -> dict | list:
363        response = await self.__send_request('getIncomingCallStatus', params={
364            'activationId': id
365        })
366
367        if not is_json(response):
368            return response
369        
370        return json.loads(response)
371    
372    async def get_prices(self, 
373                          service: str = None,
374                          country: str = None,
375                       ) -> dict | list:
376        response = await self.__send_request('getPrices', params={
377            **({'service': str(service)} if service is not None else {}),
378            **({'country': str(country)} if country is not None else {}),
379        })
380
381        if not is_json(response):
382            return response
383        
384        return json.loads(response)
385    
386    async def get_prices_verification(self, 
387                          service: str = None,
388                       ) -> dict | list:
389        response = await self.__send_request('getPricesVerification', params={
390            **({'service': str(service)} if service is not None else {}),
391        })
392
393        if not is_json(response):
394            return response
395        
396        return json.loads(response)
397    
398    async def get_countries(self,
399                       ) -> dict | list:
400        response = await self.__send_request('getCountries', params={
401        })
402
403        if not is_json(response):
404            return response
405        
406        return json.loads(response)
407    
408    async def get_service_list(self, 
409                          country: str = None,
410                          lang: Literal['ru', 'en', 'es', 'cn'] = None,
411                       ) -> dict | list:
412        response = await self.__send_request('getServicesList', params={
413            **({'country': str(country)} if country is not None else {}),
414            **({'lang': str(lang)} if lang is not None else {}),
415        })
416
417        if not is_json(response):
418            return response
419        
420        return json.loads(response)
421    
422    async def get_additional_service(self, 
423                          service: str = None,
424                          id: str = None,
425                       ):
426        """
427        Get additional service to activation its cost 5rub
428        return 2 values: addition activation id and phone number
429        
430        use like this: 
431        activation_id, phone_number = await getAdditionalService(service, activation id)
432        """
433        response = await self.__send_request('getAdditionalService', params={
434            'service': service,
435            'id':id
436        })
437
438        data = response.split(':')
439        if len(data) > 2:
440            return data[1], data[2]
441        
442        return data
443    
444    async def get_extra_activation(self, 
445                          id: str = None,
446                       ):
447        """
448        return 2 values: addition activation id and phone number
449        
450        use like this: 
451        activation_id, phone_number = await getExtraActivation(activation id)
452        """
453        response = await self.__send_request('getExtraActivation', params={
454            'id':id
455        })
456
457        data = response.split(':')
458        if len(data) > 2:
459            return data[1], data[2]
460        
461        return data
462    
463    async def check_extra_activation(self, 
464                          activationId: str | int
465                       ) -> dict | list:
466        response = await self.__send_request('checkExtraActivation', params={
467            'activationId': str(activationId)
468        })
469
470        if not is_json(response):
471            return response
472        
473        return json.loads(response)
474    
475    async def parse_call(self, 
476                          id: str | int,
477                          newLang: str,
478                       ) -> dict | list:
479        response = await self.__send_request('parseCall', params={
480            "id": id,
481            "newLang": newLang,
482        })
483
484        if not is_json(response):
485            return response
486        
487        return json.loads(response)
488    
489    # !!! BOTTOM IT IS RENT API
490    async def get_rent_services_and_countries(self,
491                       time: int | str | None = None,
492                       operator: str | None = None,
493                       country: str | None = None,
494                       currency: str | None = None,
495                       incomingCall: bool | None = None,
496                       ) -> dict | str:
497        response = await self.__send_request('getRentServicesAndCountries', params={
498            **({'time ': str(time )} if time is not None else {}),
499            **({'operator ': str(operator )} if operator is not None else {}),
500            **({'country ': str(country )} if country is not None else {}),
501            **({'currency ': str(currency )} if currency is not None else {}),
502            **({'incomingCall': str(incomingCall).lower()} if incomingCall is not None else {}),
503        })
504
505        if not is_json(response):
506            return response
507        
508        return json.loads(response)
509    
510    async def get_rent_number(self,
511                        service: str,
512                        time: int | str | None = None,
513                        operator: str | None = None,
514                        country: str | None = None,
515                        url: str | None = None,
516                        incomingCall: bool | None = None,
517                       ) -> dict | str:
518        response = await self.__send_request('getRentNumber', params={
519            'service': service,
520            **({'time ': str(time )} if time is not None else {}),
521            **({'operator ': str(operator )} if operator is not None else {}),
522            **({'country ': str(country )} if country is not None else {}),
523            **({'url ': str(url )} if url is not None else {}),
524            **({'incomingCall': str(incomingCall).lower()} if incomingCall is not None else {}),
525        })
526
527        if not is_json(response):
528            return response
529        
530        return json.loads(response)
531    
532    async def get_rent_status(self,
533                        id: str,
534                        page: int | str | None = None,
535                        size: int | str | None = None,
536                       ) -> dict | str:
537        response = await self.__send_request('getRentStatus', params={
538            'id': id,
539            **({'page ': str(page)} if page is not None else {}),
540            **({'size ': str(size )} if size is not None else {}),
541        })
542
543        if not is_json(response):
544            return response
545        
546        return json.loads(response)
547    
548    async def set_rent_status(self,
549                        id: str,
550                        status: Literal[1, 2, '1', '2'],
551                       ) -> dict | str:
552        response = await self.__send_request('getRentStatus', params={
553            'id': str(id),
554            'status': str(status),
555        })
556
557        if not is_json(response):
558            return response
559        
560        return json.loads(response)
561    
562    async def get_rent_list(self,
563                       ) -> dict | str:
564        response = await self.__send_request('getRentList', params={
565        })
566
567        if not is_json(response):
568            return response
569        
570        return json.loads(response)
571    
572    async def continue_rent_number(self,
573                        id: str,
574                        rent_time: int | str | None = 4,
575                       ) -> dict | str:
576        response = await self.__send_request('continueRentNumber', params={
577            'id': id,
578            'rent_time': str(rent_time)
579        })
580
581        if not is_json(response):
582            return response
583        
584        return json.loads(response)
585    
586    async def get_continue_rent_price_number(self,
587                        id: str,
588                        rent_time: int | str | None = 4,
589                        currency: str | None = None
590                       ) -> dict | str:
591        response = await self.__send_request('getContinueRentPriceNumber', params={
592            'id': id,
593            'rent_time': str(rent_time),
594            **({'currency ': str(currency )} if currency is not None else {}),
595        })
596
597        if not is_json(response):
598            return response
599        
600        return json.loads(response)
601    
602    # !!! BOTTOM IS IT PARTNER SOFT API
603    async def buy_partner_product(self,
604                        id: str,
605                       ) -> dict | str:
606        response = await self.__send_request('buyPartnerProduct', params={
607            'id': id,
608        })
609
610        if not is_json(response):
611            return response
612        
613        return json.loads(response)
614    
615
616# === Method Aliases (outside class for pdoc) ===
617SmsActivate.getBalance = SmsActivate.get_balance
618SmsActivate.getBalanceAndCashBack = SmsActivate.get_balance_and_cashback
619SmsActivate.getTopCountriesByService = SmsActivate.get_available_countries
620SmsActivate.getNumbersStatus = SmsActivate.get_count_numbers
621SmsActivate.getOperators = SmsActivate.get_operators
622SmsActivate.getActiveActivations = SmsActivate.get_active_activations
623SmsActivate.getStatus = SmsActivate.get_activation_status_v1
624SmsActivate.getStatusV2 = SmsActivate.get_activation_status
625SmsActivate.getNumberV2 = SmsActivate.purchase
626SmsActivate.purchase_v1 = SmsActivate.get_number
627SmsActivate.getNumber = SmsActivate.get_number
628SmsActivate.getMultiServiceNumber = SmsActivate.get_multi_service_number
629SmsActivate.setStatus = SmsActivate.set_activation_status
630SmsActivate.getHistory = SmsActivate.get_history
631SmsActivate.getListOfTopCountriesByService = SmsActivate.get_list_top_countries
632SmsActivate.getIncomingCallStatus = SmsActivate.get_incoming_call_status
633SmsActivate.getPrices = SmsActivate.get_prices
634SmsActivate.getPricesVerification = SmsActivate.get_prices_verification
635SmsActivate.getCountries = SmsActivate.get_countries
636SmsActivate.getServicesList = SmsActivate.get_service_list
637SmsActivate.getAdditionalService = SmsActivate.get_additional_service
638SmsActivate.getExtraActivation = SmsActivate.get_extra_activation
639SmsActivate.checkExtraActivation = SmsActivate.check_extra_activation
640SmsActivate.parseCall = SmsActivate.parse_call
641SmsActivate.getRentServicesAndCountries = SmsActivate.get_rent_services_and_countries
642SmsActivate.getRentNumber = SmsActivate.get_rent_number
643SmsActivate.getRentStatus = SmsActivate.get_rent_status
644SmsActivate.getRentStatus = SmsActivate.get_rent_status
645SmsActivate.getRentList = SmsActivate.get_rent_list
646SmsActivate.continueRentNumber = SmsActivate.continue_rent_number
647SmsActivate.getContinueRentPriceNumber = SmsActivate.get_continue_rent_price_number
648SmsActivate.buyPartnerProduct = SmsActivate.buy_partner_product