下载地址:https://www.pan38.com/dow/share.php?code=JCnzE 提取密码:1133
该工具包含三个核心模块:短信发送核心功能、配置管理系统和命令行界面。使用时需先配置API密钥和短信模板,支持自定义号码前缀生成和详细的发送日志记录。实际部署时请确保遵守相关法律法规,该代码示例仅用于技术学习目的。
import random
import time
import csv
import requests
from datetime import datetime
class PhoneGenerator:
def init(self, prefix_list=['138','139','150']):
self.prefixes = prefix_list
def generate(self, count=100):
phones = []
for _ in range(count):
prefix = random.choice(self.prefixes)
suffix = ''.join([str(random.randint(0,9)) for _ in range(8)])
phones.append(f"{prefix}{suffix}")
return phones
class SMSSender:
API_URL = "https://api.sms-service.com/v2/send"
def __init__(self, api_key):
self.api_key = api_key
self.session = requests.Session()
def send_single(self, phone, content, retry=3):
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"to": phone,
"text": content,
"sender": "NOTICE"
}
for attempt in range(retry):
try:
response = self.session.post(
self.API_URL,
json=payload,
headers=headers,
timeout=10
)
if response.status_code == 200:
return True, response.json()
else:
time.sleep(2**attempt)
except Exception as e:
print(f"Attempt {attempt+1} failed: {str(e)}")
return False, None
class BatchProcessor:
def init(self, sender):
self.sender = sender
self.logger = SMSLogger()
def process_batch(self, phone_list, content_template, params_list=None):
results = []
for i, phone in enumerate(phone_list):
if params_list:
content = content_template.format(**params_list[i])
else:
content = content_template
success, response = self.sender.send_single(phone, content)
log_data = {
"phone": phone,
"content": content,
"timestamp": datetime.now().isoformat(),
"status": "SUCCESS" if success else "FAILED",
"response": str(response)
}
self.logger.write(log_data)
results.append(log_data)
return results
class SMSLogger:
def init(self, filename="sms_log.csv"):
self.filename = filename
self._init_file()
def _init_file(self):
with open(self.filename, 'a+', newline='') as f:
writer = csv.DictWriter(f, fieldnames=[
"timestamp", "phone", "content",
"status", "response"
])
if f.tell() == 0:
writer.writeheader()
def write(self, record):
with open(self.filename, 'a', newline='') as f:
writer = csv.DictWriter(f, fieldnames=[
"timestamp", "phone", "content",
"status", "response"
])
writer.writerow(record)
if name == "main":
# 示例用法
gen = PhoneGenerator()
phones = gen.generate(10)
sender = SMSSender(api_key="your_api_key_here")
processor = BatchProcessor(sender)
template = "尊敬的{}用户,您的验证码是{}"
params = [
{"name": "客户A", "code": random.randint(1000,9999)},
# 其他参数...
]
processor.process_batch(phones[:5], template, params)
processor.process_batch(phones[5:], "系统通知:您的订单已发货")
configparser
import json
from pathlib import Path
class ConfigManager:
def init(self, config_path="config.ini"):
self.config_path = Path(config_path)
self._ensure_config_exists()
def _ensure_config_exists(self):
if not self.config_path.exists():
default_config = {
"API": {
"key": "your_api_key",
"endpoint": "https://api.sms-service.com/v2/send"
},
"PHONE": {
"prefixes": "138,139,150,151,152"
}
}
self._save_config(default_config)
def _save_config(self, config_dict):
config = configparser.ConfigParser()
for section, options in config_dict.items():
config[section] = options
with open(self.config_path, 'w') as f:
config.write(f)
def load_config(self):
config = configparser.ConfigParser()
config.read(self.config_path)
return {
"api_key": config["API"]["key"],
"api_endpoint": config["API"]["endpoint"],
"phone_prefixes": [
p.strip() for p in config["PHONE"]["prefixes"].split(",")
]
}
class TemplateManager:
@staticmethod
def load_templates(file_path="templates.json"):
try:
with open(file_path, 'r', encoding='utf-8') as f:
return json.load(f)
except FileNotFoundError:
default_templates = {
"verification": "验证码:{code},有效期5分钟",
"notification": "亲爱的{name},您预约的{service}将于{time}开始"
}
with open(file_path, 'w', encoding='utf-8') as f:
json.dump(default_templates, f, ensure_ascii=False, indent=2)
return default_templates
argparse
from sms_toolkit import PhoneGenerator, SMSSender, BatchProcessor
from config_loader import ConfigManager, TemplateManager
def main():
parser = argparse.ArgumentParser(
description="批量短信发送工具 v1.0",
formatter_class=argparse.ArgumentDefaultsHelpFormatter
)
subparsers = parser.add_subparsers(dest='command', required=True)
# 生成号码命令
gen_parser = subparsers.add_parser('generate', help='生成手机号码')
gen_parser.add_argument(
'-n', '--number',
type=int,
default=100,
help='要生成的号码数量'
)
gen_parser.add_argument(
'-o', '--output',
default='phones.csv',
help='输出文件路径'
)
# 发送短信命令
send_parser = subparsers.add_parser('send', help='发送批量短信')
send_parser.add_argument(
'-f', '--file',
required=True,
help='包含号码列表的CSV文件'
)
send_parser.add_argument(
'-t', '--template',
required=True,
help='短信模板名称或内容'
)
send_parser.add_argument(
'-p', '--params',
help='模板参数JSON文件'
)
args = parser.parse_args()
config = ConfigManager().load_config()
if args.command == 'generate':
generator = PhoneGenerator(config['phone_prefixes'])
phones = generator.generate(args.number)
with open(args.output, 'w') as f:
f.write("\n".join(phones))
print(f"已生成 {args.number} 个号码到 {args.output}")
elif args.command == 'send':
sender = SMSSender(config['api_key'])
processor = BatchProcessor(sender)
with open(args.file, 'r') as f:
phones = [line.strip() for line in f if line.strip()]
templates = TemplateManager.load_templates()
template = templates.get(args.template, args.template)
params = None
if args.params:
with open(args.params, 'r') as f:
params = json.load(f)
print(f"开始向 {len(phones)} 个号码发送短信...")
results = processor.process_batch(phones, template, params)
success = sum(1 for r in results if r['status'] == 'SUCCESS')
print(f"发送完成,成功 {success} 条,失败 {len(phones)-success} 条")
if name == "main":
main()