@thepappo said in Нормальное завершение потока, из которого запускается асинхронная функция:
В скрипте нужно отменять активации номеров. Если отменить номер незадолго после взятия, то сервис СМС вернет ошибку.
Я написал отдельный скрипт, который по API получает номера, которые вот-вот истекут и отменяет их. Скрипт запускается в цикле каждую минуту.
Например, номер дается на 20 минут, поток в скрипте у меня не может взять номер на это время, а может, к примеру на 5 минут максимум, значит я могу смело отменять все номера, которые были взяты более 15 минут назад.
5 минут взяты для примера. Я знаю, что я беру номер из сервиса, что-то там делаю, жду смс-ку и все это занимает у меня максимум эти самые 5 минут.
Сам скрипт ни о каких потоках, скриптах и так далее понятия не имеет и работает автономно.
import logging
import os
import datetime
import time
from dotenv import load_dotenv
from smsactivate.api import SMSActivateAPI
# Configure logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("[sms_activate]")
# Load environment variables from .env file
ABS_PATH = os.path.dirname(os.path.abspath(__file__))
load_dotenv()
def main(api_key: str, threshold_seconds: int = 15 * 60):
"""
Main function to check and process active activations.
Args:
api_key (str): API key for SMSActivate service.
threshold_seconds (int): Threshold in seconds to consider activation time.
Returns:
None
"""
sa = SMSActivateAPI(api_key)
sa.debug_mode = True
# Get active activations
activations = sa.getActiveActivations()
if activations.get("error") == 'NO_ACTIVATIONS':
logger.info("No pending activations found.")
return False
# Check if activations retrieval was successful
if activations.get("status") != 'success':
raise Exception("Activation status is not success")
# Loop through active activations
for one in activations["activeActivations"]:
if one["smsCode"]:
# Skip if activation is already used
continue
# Parse activation time
activation_time_str = one["activationTime"]
activation_time = datetime.datetime.strptime(activation_time_str, '%Y-%m-%d %H:%M:%S')
current_time = datetime.datetime.now()
# Calculate time difference in seconds
time_difference = current_time - activation_time
seconds_difference = time_difference.total_seconds()
# Log the time difference
logger.info("Seconds difference from activation time to now: %s" % seconds_difference)
# Check if activation time exceeds threshold
if seconds_difference > threshold_seconds:
logger.info("Activation time exceeds threshold for: %s" % one)
# Mark activation as used
sa.setStatus(id=one["activationId"], status=8)
if __name__ == '__main__':
# Get SMS Activate API key from environment variables
sms_activate_api_key = os.getenv("SMS_ACTIVATE_API_KEY")
if not sms_activate_api_key:
raise Exception('SMS_ACTIVATE_API_KEY environment variable not set')
sms_activate_api_key = sms_activate_api_key.strip()
time_to_sleep = 60
while True:
try:
main(api_key=sms_activate_api_key)
except Exception as e:
# Log any exceptions
logger.error(e)
# Wait for 60 seconds before checking again
logger.info(f"Sleeping for {time_to_sleep} seconds")
time.sleep(time_to_sleep)
