pip install smpplib

import smpplib.gsm
import smpplib.client
import smpplib.consts

# SMSC 連線資訊（需向電信商申請）
client = smpplib.client.Client('smpp.example.com', 2775)

# 綁定為發送者
client.connect()
client.bind_transmitter(system_id='user123', password='pass456')

# 發送簡訊
parts, encoding_flag, msg_type_flag = smpplib.gsm.make_parts('Hello World!')

for part in parts:
    pdu = client.send_message(
        source_addr_ton=smpplib.consts.SMPP_TON_INTL,  # 國際號碼
        source_addr='1234',                             # 發送者號碼
        dest_addr_ton=smpplib.consts.SMPP_TON_INTL,
        destination_addr='886912345678',                # 接收者號碼（台灣）
        short_message=part,
        data_coding=encoding_flag,
        esm_class=msg_type_flag,
        registered_delivery=True,  # 要求送達報告
    )

    print(f"Message ID: {pdu.message_id}")

# 解除綁定
client.unbind()
client.disconnect()

import smpplib.client
import smpplib.consts

def message_received_handler(pdu):
    """處理接收到的簡訊或送達報告"""

    if pdu.command == smpplib.consts.SMPP_CMD_DELIVER_SM:
        # 檢查是否為送達報告
        if pdu.receipted_message_id:
            print(f"送達報告：Message ID {pdu.receipted_message_id}")
            print(f"狀態：{pdu.message_state}")
        else:
            # 普通簡訊
            print(f"收到簡訊：{pdu.short_message.decode()}")
            print(f"發送者：{pdu.source_addr}")

client = smpplib.client.Client('smpp.example.com', 2775)

# 設定回調函數
client.set_message_received_handler(message_received_handler)

# 綁定為接收者
client.connect()
client.bind_receiver(system_id='user123', password='pass456')

# 監聽訊息
client.listen()

import smpplib.client
import smpplib.consts
import threading

class SMSGateway:
    def __init__(self, host, port, system_id, password):
        self.client = smpplib.client.Client(host, port)
        self.system_id = system_id
        self.password = password

    def connect(self):
        """連線並綁定為收發者"""
        self.client.connect()
        self.client.bind_transceiver(
            system_id=self.system_id,
            password=self.password
        )
        print("已連線到 SMSC")

    def send_sms(self, to, message):
        """發送簡訊"""
        parts, encoding, msg_type = smpplib.gsm.make_parts(message)

        for part in parts:
            pdu = self.client.send_message(
                source_addr_ton=smpplib.consts.SMPP_TON_INTL,
                source_addr='1234',
                dest_addr_ton=smpplib.consts.SMPP_TON_INTL,
                destination_addr=to,
                short_message=part,
                data_coding=encoding,
                esm_class=msg_type,
                registered_delivery=True
            )

            print(f"簡訊已發送：{pdu.message_id}")

    def on_receive(self, pdu):
        """接收簡訊回調"""
        if pdu.receipted_message_id:
            # 送達報告
            print(f"✅ 送達確認：{pdu.receipted_message_id}")
        else:
            # 接收到的簡訊
            sender = pdu.source_addr.decode()
            message = pdu.short_message.decode()
            print(f"📩 收到簡訊來自 {sender}: {message}")

            # 自動回覆
            self.send_sms(sender, f"已收到您的訊息：{message}")

    def start_listening(self):
        """開始監聽"""
        self.client.set_message_received_handler(self.on_receive)

        # 在背景執行緒中監聽
        listen_thread = threading.Thread(target=self.client.listen)
        listen_thread.daemon = True
        listen_thread.start()

    def disconnect(self):
        """斷開連線"""
        self.client.unbind()
        self.client.disconnect()

# 使用
gateway = SMSGateway('smpp.example.com', 2775, 'user123', 'pass456')
gateway.connect()
gateway.start_listening()

# 發送簡訊
gateway.send_sms('886912345678', 'Hello from SMPP!')

# 保持運行
import time
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    gateway.disconnect()

import random
import string

class SMS2FA:
    def __init__(self, sms_gateway):
        self.gateway = sms_gateway
        self.codes = {}  # 儲存驗證碼（實務應用應使用 Redis）

    def generate_code(self, length=6):
        """產生隨機驗證碼"""
        return ''.join(random.choices(string.digits, k=length))

    def send_verification_code(self, phone_number):
        """發送驗證碼"""
        code = self.generate_code()

        # 儲存驗證碼（5 分鐘有效）
        self.codes[phone_number] = {
            'code': code,
            'expires': time.time() + 300  # 5 分鐘
        }

        # 發送簡訊
        message = f"您的驗證碼是：{code}，5分鐘內有效。"
        self.gateway.send_sms(phone_number, message)

        print(f"驗證碼已發送給 {phone_number}")

    def verify_code(self, phone_number, code):
        """驗證驗證碼"""
        if phone_number not in self.codes:
            return False

        stored = self.codes[phone_number]

        # 檢查是否過期
        if time.time() > stored['expires']:
            del self.codes[phone_number]
            return False

        # 驗證碼是否正確
        if stored['code'] == code:
            del self.codes[phone_number]  # 使用後刪除
            return True

        return False

# 使用
twofa = SMS2FA(gateway)

# 發送驗證碼
twofa.send_verification_code('886912345678')

# 驗證
is_valid = twofa.verify_code('886912345678', '123456')
if is_valid:
    print("驗證成功")
else:
    print("驗證失敗")

import time
from concurrent.futures import ThreadPoolExecutor

class BulkSMSService:
    def __init__(self, gateway):
        self.gateway = gateway

    def send_bulk(self, recipients, message, delay=1):
        """批次發送簡訊"""
        total = len(recipients)
        success = 0
        failed = 0

        for i, recipient in enumerate(recipients):
            try:
                # 發送簡訊
                self.gateway.send_sms(recipient['phone'], message)

                success += 1
                print(f"[{i+1}/{total}] 已發送給 {recipient['name']}")

                # 避免發送太快被限制
                time.sleep(delay)

            except Exception as e:
                failed += 1
                print(f"[{i+1}/{total}] 發送失敗：{recipient['name']} - {e}")

        print(f"\n統計：成功 {success}，失敗 {failed}")

    def send_personalized_bulk(self, recipients, template):
        """發送個人化簡訊"""
        for recipient in recipients:
            # 替換變數
            message = template.format(**recipient)
            self.gateway.send_sms(recipient['phone'], message)
            time.sleep(1)

# 使用
bulk_service = BulkSMSService(gateway)

# 收件人列表
recipients = [
    {'name': 'Alice', 'phone': '886912345678'},
    {'name': 'Bob', 'phone': '886987654321'},
    # ...
]

# 發送相同訊息
bulk_service.send_bulk(recipients, '新產品上市！限時優惠中！')

# 發送個人化訊息
recipients_with_data = [
    {'name': 'Alice', 'phone': '886912345678', 'points': 1000},
    {'name': 'Bob', 'phone': '886987654321', 'points': 500},
]

template = "親愛的 {name}，您目前有 {points} 點數可使用！"
bulk_service.send_personalized_bulk(recipients_with_data, template)

class TransactionNotifier:
    def __init__(self, gateway):
        self.gateway = gateway

    def notify_payment(self, phone, amount, merchant):
        """付款通知"""
        message = f"""
您已完成付款
金額：NT$ {amount:,}
商家：{merchant}
時間：{datetime.now().strftime('%Y-%m-%d %H:%M')}
        """.strip()

        self.gateway.send_sms(phone, message)

    def notify_withdrawal(self, phone, amount, atm_location):
        """提款通知"""
        message = f"""
提款交易成功
金額：NT$ {amount:,}
地點：{atm_location}
時間：{datetime.now().strftime('%Y-%m-%d %H:%M')}
        """.strip()

        self.gateway.send_sms(phone, message)

    def notify_suspicious_activity(self, phone, activity):
        """可疑活動警告"""
        message = f"""
【安全警示】
偵測到可疑活動：{activity}
如非本人操作，請立即聯絡客服。
        """.strip()

        self.gateway.send_sms(phone, message)

# 使用
notifier = TransactionNotifier(gateway)

# 付款通知
notifier.notify_payment('886912345678', 1500, '7-11 中正店')

# 提款通知
notifier.notify_withdrawal('886912345678', 5000, '台北火車站 ATM')

# 可疑活動警告
notifier.notify_suspicious_activity('886912345678', '境外登入')

# 建立長連線
client.connect()
client.bind_transmitter()

# 發送（即時）
client.send_message()

# 保持連線（心跳）
client.enquire_link()

優點：
✅ 即時（毫秒級）
✅ 高吞吐量（每秒數百則）
✅ 雙向通訊（收發簡訊）
✅ 送達報告即時

缺點：
❌ 複雜（需維護長連線）
❌ 成本高（需專線費用）
❌ 需電信商合作

# 每次發送都是新請求
import requests

response = requests.post('https://sms-api.com/send', data={
    'api_key': 'xxx',
    'to': '886912345678',
    'message': 'Hello'
})

優點：
✅ 簡單（HTTP 請求）
✅ 成本低（按量計費）
✅ 容易整合

缺點：
❌ 較慢（HTTP 開銷）
❌ 吞吐量低
❌ 送達報告需 Webhook

# TON (Type of Number)
smpplib.consts.SMPP_TON_INTL      # 國際號碼（+886）
smpplib.consts.SMPP_TON_NATIONAL  # 國內號碼（0912）

# NPI (Numbering Plan Indicator)
smpplib.consts.SMPP_NPI_ISDN      # 標準電話號碼

# 錯誤設定會導致發送失敗

import smpplib.gsm

# 自動處理編碼
parts, encoding, msg_type = smpplib.gsm.make_parts('中文測試')

# GSM 7-bit：160 字元
# UCS-2（中文）：70 字元

def send_with_retry(client, phone, message, max_retries=3):
    for attempt in range(max_retries):
        try:
            pdu = client.send_message(
                source_addr='1234',
                destination_addr=phone,
                short_message=message
            )
            print(f"發送成功：{pdu.message_id}")
            return True

        except Exception as e:
            print(f"嘗試 {attempt + 1} 失敗：{e}")
            time.sleep(5)  # 等待 5 秒後重試

    print("發送失敗")
    return False

def on_delivery_report(pdu):
    message_id = pdu.receipted_message_id
    state = pdu.message_state

    if state == smpplib.consts.SMPP_MSGSTATE_DELIVERED:
        print(f"✅ {message_id} 已送達")
    elif state == smpplib.consts.SMPP_MSGSTATE_UNDELIVERABLE:
        print(f"❌ {message_id} 無法送達")
        # 記錄失敗，稍後重試

client.set_message_received_handler(on_delivery_report)

import re

def validate_phone(phone):
    """驗證台灣手機號碼"""
    # 09XXXXXXXX 或 886 9XXXXXXXX
    pattern = r'^(09\d{8}|8869\d{8})$'

    if re.match(pattern, phone):
        # 統一格式為國際號碼
        if phone.startswith('09'):
            phone = '886' + phone[1:]
        return phone
    else:
        raise ValueError(f"無效號碼：{phone}")

# 使用
try:
    validated = validate_phone('0912345678')
    send_sms(validated, 'Hello')
except ValueError as e:
    print(e)

import smpplib.gsm

def calculate_sms_parts(message):
    """計算簡訊則數"""
    parts, encoding, msg_type = smpplib.gsm.make_parts(message)

    num_parts = len(parts)

    if encoding == smpplib.consts.SMPP_ENCODING_DEFAULT:
        # GSM 7-bit
        max_length = 160 if num_parts == 1 else 153
        encoding_name = 'GSM 7-bit'
    else:
        # UCS-2（中文）
        max_length = 70 if num_parts == 1 else 67
        encoding_name = 'UCS-2'

    return {
        'parts': num_parts,
        'encoding': encoding_name,
        'max_length': max_length,
        'total_chars': len(message)
    }

# 測試
messages = [
    "Hello World",                    # 11 字元 → 1 則
    "This is a very long message " * 10,  # 290 字元 → 2 則
    "中文測試",                       # 4 字元 → 1 則
    "這是一則很長的中文簡訊" * 10,    # 110 字元 → 2 則
]

for msg in messages:
    result = calculate_sms_parts(msg)
    print(f"訊息：{msg[:50]}...")
    print(f"則數：{result['parts']} 則")
    print(f"編碼：{result['encoding']}")
    print(f"字數：{result['total_chars']}/{result['max_length']}")
    print('-' * 50)

def calculate_cost(message, price_per_sms=1.5):
    """計算簡訊費用（台灣約 NT$1.5/則）"""
    result = calculate_sms_parts(message)
    total_cost = result['parts'] * price_per_sms

    return {
        'parts': result['parts'],
        'price_per_sms': price_per_sms,
        'total_cost': total_cost
    }

# 批次計算
recipients = 1000
message = "親愛的客戶，您的訂單已出貨！"

cost = calculate_cost(message)
total = cost['total_cost'] * recipients

print(f"單則簡訊：{cost['parts']} 則 × NT${cost['price_per_sms']} = NT${cost['total_cost']}")
print(f"發送 {recipients} 人：NT${total}")

def subscribe_sms(user):
    """使用者訂閱簡訊"""
    # 發送確認簡訊
    send_sms(user.phone, """
歡迎訂閱！
您將收到我們的最新優惠資訊。
如需取消，請回覆 STOP
    """)

    # 記錄同意
    user.sms_opt_in = True
    user.sms_opt_in_date = datetime.now()
    user.save()

def on_receive_sms(pdu):
    """處理接收到的簡訊"""
    sender = pdu.source_addr.decode()
    message = pdu.short_message.decode().upper()

    if message in ['STOP', 'UNSUBSCRIBE', '取消']:
        # 取消訂閱
        user = User.objects.get(phone=sender)
        user.sms_opt_in = False
        user.save()

        # 發送確認
        send_sms(sender, "您已成功取消訂閱。")

# ❌ 不好的做法
message = "!!!免費贈送!!! 立即點擊 http://bit.ly/xxx"

# ✅ 好的做法
message = """
親愛的會員，
您有一張 100 元折價券即將到期。
有效期限：2025/01/31
詳情：https://example.com/coupon/abc123
"""

from datetime import datetime, timedelta

def can_send_sms(user):
    """檢查是否可以發送（避免騷擾）"""
    # 最多每天 3 則
    today_count = SMSLog.objects.filter(
        phone=user.phone,
        sent_at__gte=datetime.now().date()
    ).count()

    if today_count >= 3:
        return False

    # 最少間隔 1 小時
    last_sms = SMSLog.objects.filter(
        phone=user.phone
    ).order_by('-sent_at').first()

    if last_sms and (datetime.now() - last_sms.sent_at) < timedelta(hours=1):
        return False

    return True