tessl/pypi-qiniu
Comprehensive Python SDK for Qiniu Cloud Storage services enabling file upload, download, CDN management, SMS, and real-time communication features
—
Pending
communication-services.mddocs/
Communication Services
SMS messaging and real-time communication services for application integration. The communication platform provides comprehensive messaging capabilities and real-time communication features.
Capabilities
SMS Services
SMS message service operations including signature and template management through the Sms class.
class Sms:
def __init__(self, auth: Auth):
"""
Initialize SMS service client.
Args:
auth: Auth instance for authentication
"""
# Signature Management
def createSignature(self, signature: str, source: str, pics: list = None) -> tuple:
"""
Create SMS signature.
Args:
signature: Signature text content
source: Source type or identifier
pics: List of signature image URLs (optional)
Returns:
(dict, ResponseInfo): Creation result and response info
"""
def querySignature(self, audit_status: str = None, page: int = 1, page_size: int = 20) -> tuple:
"""
Query SMS signatures.
Args:
audit_status: Audit status filter ('pending', 'approved', 'rejected')
page: Page number (default: 1)
page_size: Number of items per page (default: 20)
Returns:
(dict, ResponseInfo): Signature list and response info
"""
def updateSignature(self, id: str, signature: str) -> tuple:
"""
Update SMS signature.
Args:
id: Signature ID
signature: Updated signature text
Returns:
(dict, ResponseInfo): Update result and response info
"""
def deleteSignature(self, id: str) -> tuple:
"""
Delete SMS signature.
Args:
id: Signature ID
Returns:
(dict, ResponseInfo): Deletion result and response info
"""
# Template Management
def createTemplate(self, name: str, template: str, type: str, description: str, signature_id: str) -> tuple:
"""
Create SMS template.
Args:
name: Template name
template: Template content with variable placeholders
type: Template type ('notification', 'verification', 'marketing')
description: Template description
signature_id: Associated signature ID
Returns:
(dict, ResponseInfo): Creation result and response info
"""
def queryTemplate(self, audit_status: str, page: int = 1, page_size: int = 20) -> tuple:
"""
Query SMS templates.
Args:
audit_status: Audit status filter ('pending', 'approved', 'rejected')
page: Page number (default: 1)
page_size: Number of items per page (default: 20)
Returns:
(dict, ResponseInfo): Template list and response info
"""
def updateTemplate(self, id: str, name: str, template: str, description: str, signature_id: str) -> tuple:
"""
Update SMS template.
Args:
id: Template ID
name: Updated template name
template: Updated template content
description: Updated description
signature_id: Updated signature ID
Returns:
(dict, ResponseInfo): Update result and response info
"""
def deleteTemplate(self, id: str) -> tuple:
"""
Delete SMS template.
Args:
id: Template ID
Returns:
(dict, ResponseInfo): Deletion result and response info
"""
# Message Operations
def sendMessage(self, template_id: str, mobiles: list, parameters: dict) -> tuple:
"""
Send SMS message.
Args:
template_id: Template ID to use
mobiles: List of mobile phone numbers
parameters: Template parameter values
Returns:
(dict, ResponseInfo): Send result and response info
"""
def get_messages_info(self) -> tuple:
"""
Query SMS send records.
Returns:
(dict, ResponseInfo): Message records and response info
"""Real-Time Communication
Real-time communication server management and room operations through the RtcServer class.
class RtcServer:
def __init__(self, auth: Auth):
"""
Initialize RTC server client.
Args:
auth: Auth instance for authentication
"""
def create_app(self, data: dict) -> tuple:
"""
Create RTC application.
Args:
data: Application configuration dictionary
Returns:
(dict, ResponseInfo): Creation result and response info
"""
def get_app(self, app_id: str = None) -> tuple:
"""
Get RTC application info or list all applications.
Args:
app_id: Application ID (optional, if None returns all apps)
Returns:
(dict, ResponseInfo): Application info and response info
"""
def delete_app(self, app_id: str) -> tuple:
"""
Delete RTC application.
Args:
app_id: Application ID
Returns:
(dict, ResponseInfo): Deletion result and response info
"""
def update_app(self, app_id: str, data: dict) -> tuple:
"""
Update RTC application configuration.
Args:
app_id: Application ID
data: Updated configuration dictionary
Returns:
(dict, ResponseInfo): Update result and response info
"""
def list_user(self, app_id: str, room_name: str) -> tuple:
"""
List users in RTC room.
Args:
app_id: Application ID
room_name: Room name
Returns:
(dict, ResponseInfo): User list and response info
"""
def kick_user(self, app_id: str, room_name: str, user_id: str) -> tuple:
"""
Kick user from RTC room.
Args:
app_id: Application ID
room_name: Room name
user_id: User ID to kick
Returns:
(dict, ResponseInfo): Operation result and response info
"""
def list_active_rooms(self, app_id: str, room_name_prefix: str = None) -> tuple:
"""
List active RTC rooms.
Args:
app_id: Application ID
room_name_prefix: Room name prefix filter (optional)
Returns:
(dict, ResponseInfo): Active room list and response info
"""RTC Token Generation
Utility function for generating room access tokens.
def get_room_token(access_key: str, secret_key: str, room_access: dict) -> str:
"""
Generate room access token for RTC client.
Args:
access_key: Qiniu access key
secret_key: Qiniu secret key
room_access: Room access configuration dictionary
Returns:
Room access token string
"""Usage Examples
SMS Signature Management
from qiniu import Auth, Sms
auth = Auth(access_key, secret_key)
sms = Sms(auth)
# Create SMS signature
ret, info = sms.createSignature(
signature='YourCompany',
source='website',
pics=['https://example.com/company-logo.png']
)
if info.ok():
signature_id = ret['id']
print(f"Signature created: {signature_id}")
else:
print(f"Signature creation failed: {info.error}")
# Query signatures
ret, info = sms.querySignature(audit_status='approved', page=1, page_size=10)
if info.ok():
for signature in ret['items']:
print(f"Signature: {signature['signature']} - Status: {signature['audit_status']}")
# Update signature
ret, info = sms.updateSignature(signature_id, 'UpdatedCompanyName')
if info.ok():
print("Signature updated successfully")SMS Template Management
from qiniu import Auth, Sms
auth = Auth(access_key, secret_key)
sms = Sms(auth)
# Create verification code template
template_content = "Your verification code is ${code}, valid for ${minutes} minutes."
ret, info = sms.createTemplate(
name='Verification Code',
template=template_content,
type='verification',
description='User verification code template',
signature_id='signature-id-123'
)
if info.ok():
template_id = ret['id']
print(f"Template created: {template_id}")
# Create notification template
notification_template = "Hello ${name}, your order ${order_id} has been ${status}."
ret, info = sms.createTemplate(
name='Order Notification',
template=notification_template,
type='notification',
description='Order status notification template',
signature_id='signature-id-123'
)
# Query templates
ret, info = sms.queryTemplate(audit_status='approved')
if info.ok():
for template in ret['items']:
print(f"Template: {template['name']} - {template['template']}")
print(f"Status: {template['audit_status']}")Sending SMS Messages
from qiniu import Auth, Sms
auth = Auth(access_key, secret_key)
sms = Sms(auth)
# Send verification code
mobile_numbers = ['+86-13812345678', '+86-13987654321']
parameters = {
'code': '123456',
'minutes': '5'
}
ret, info = sms.sendMessage(
template_id='template-verification-123',
mobiles=mobile_numbers,
parameters=parameters
)
if info.ok():
print(f"Messages sent successfully: {ret}")
for result in ret['items']:
mobile = result['mobile']
status = result['status']
print(f"Mobile: {mobile} - Status: {status}")
else:
print(f"Failed to send messages: {info.error}")
# Send order notification
notification_params = {
'name': 'John Doe',
'order_id': 'ORD-2024-001',
'status': 'shipped'
}
ret, info = sms.sendMessage(
template_id='template-notification-456',
mobiles=['+86-13812345678'],
parameters=notification_params
)
# Query message records
ret, info = sms.get_messages_info()
if info.ok():
for record in ret['items']:
print(f"Message ID: {record['id']}")
print(f"Mobile: {record['mobile']}")
print(f"Status: {record['status']}")
print(f"Send time: {record['created_at']}")RTC Application Management
from qiniu import Auth, RtcServer
auth = Auth(access_key, secret_key)
rtc_server = RtcServer(auth)
# Create RTC application
app_config = {
'name': 'video-chat-app',
'description': 'Video chat application',
'callback_url': 'https://api.example.com/rtc-callback'
}
ret, info = rtc_server.create_app(app_config)
if info.ok():
app_id = ret['appId']
print(f"RTC app created: {app_id}")
# Get application info
ret, info = rtc_server.get_app(app_id)
if info.ok():
app_info = ret
print(f"App name: {app_info['name']}")
print(f"Status: {app_info['status']}")
print(f"Created: {app_info['created_at']}")
# List all applications
ret, info = rtc_server.get_app() # No app_id returns all apps
if info.ok():
for app in ret['items']:
print(f"App: {app['name']} ({app['appId']})")
# Update application
update_config = {
'description': 'Updated video chat application',
'max_users_per_room': 50
}
ret, info = rtc_server.update_app(app_id, update_config)
if info.ok():
print("App updated successfully")Room and User Management
from qiniu import Auth, RtcServer, get_room_token
auth = Auth(access_key, secret_key)
rtc_server = RtcServer(auth)
app_id = 'your-app-id'
room_name = 'meeting-room-001'
# Generate room access token for client
room_access = {
'appId': app_id,
'roomName': room_name,
'userId': 'user-123',
'expireAt': 1640995200, # Unix timestamp
'permission': 'admin' # 'admin' or 'user'
}
room_token = get_room_token(access_key, secret_key, room_access)
print(f"Room token: {room_token}")
# List users in room
ret, info = rtc_server.list_user(app_id, room_name)
if info.ok():
users = ret['users']
print(f"Users in room '{room_name}': {len(users)}")
for user in users:
print(f" User ID: {user['userId']}")
print(f" Join time: {user['joinTime']}")
print(f" Status: {user['status']}")
# Kick a user from room
user_to_kick = 'disruptive-user-456'
ret, info = rtc_server.kick_user(app_id, room_name, user_to_kick)
if info.ok():
print(f"User {user_to_kick} kicked from room")
else:
print(f"Failed to kick user: {info.error}")
# List active rooms
ret, info = rtc_server.list_active_rooms(app_id, room_name_prefix='meeting-')
if info.ok():
active_rooms = ret['rooms']
print(f"Active meeting rooms: {len(active_rooms)}")
for room in active_rooms:
print(f" Room: {room['roomName']}")
print(f" Users: {room['userCount']}")
print(f" Created: {room['createdAt']}")Bulk SMS Operations
from qiniu import Auth, Sms
import csv
import time
auth = Auth(access_key, secret_key)
sms = Sms(auth)
def send_bulk_notifications(csv_file, template_id):
"""Send bulk SMS notifications from CSV file"""
with open(csv_file, 'r') as file:
reader = csv.DictReader(file)
batch_size = 100 # Send in batches
batch = []
for row in reader:
mobile = row['mobile']
name = row['name']
# Add to current batch
batch.append({
'mobile': mobile,
'parameters': {
'name': name,
'message': row.get('message', 'General notification')
}
})
# Send batch when full
if len(batch) >= batch_size:
send_batch(batch, template_id)
batch = []
time.sleep(1) # Rate limiting
# Send remaining items
if batch:
send_batch(batch, template_id)
def send_batch(batch, template_id):
"""Send SMS batch"""
mobiles = [item['mobile'] for item in batch]
# For simplicity, using first item's parameters
# In practice, you might need template per message
parameters = batch[0]['parameters']
ret, info = sms.sendMessage(template_id, mobiles, parameters)
if info.ok():
success_count = sum(1 for item in ret['items'] if item['status'] == 'success')
print(f"Batch sent: {success_count}/{len(batch)} successful")
else:
print(f"Batch failed: {info.error}")
# Usage
send_bulk_notifications('customers.csv', 'template-promotion-789')RTC Room Monitoring
from qiniu import Auth, RtcServer
import time
import threading
auth = Auth(access_key, secret_key)
rtc_server = RtcServer(auth)
def monitor_rtc_rooms(app_id):
"""Monitor RTC room activity"""
while True:
try:
# Get active rooms
ret, info = rtc_server.list_active_rooms(app_id)
if info.ok():
rooms = ret['rooms']
print(f"\n=== RTC Room Status ===")
print(f"Active rooms: {len(rooms)}")
total_users = 0
for room in rooms:
room_name = room['roomName']
user_count = room['userCount']
total_users += user_count
print(f"Room: {room_name} - Users: {user_count}")
# Get detailed user list for rooms with many users
if user_count > 10:
ret, info = rtc_server.list_user(app_id, room_name)
if info.ok():
users = ret['users']
print(f" High activity room - {len(users)} users:")
for user in users[:5]: # Show first 5 users
print(f" {user['userId']} (joined: {user['joinTime']})")
print(f"Total active users: {total_users}")
except Exception as e:
print(f"Monitoring error: {e}")
time.sleep(30) # Check every 30 seconds
# Start monitoring in background
app_id = 'your-rtc-app-id'
monitor_thread = threading.Thread(target=monitor_rtc_rooms, args=(app_id,))
monitor_thread.daemon = True
monitor_thread.start()
print(f"Started monitoring RTC app: {app_id}")SMS Campaign Management
from qiniu import Auth, Sms
import datetime
auth = Auth(access_key, secret_key)
sms = Sms(auth)
class SMSCampaign:
def __init__(self, sms_client):
self.sms = sms_client
self.campaign_stats = {}
def create_campaign_template(self, campaign_name, signature_id):
"""Create template for campaign"""
template_content = "Hi ${name}! ${campaign_message} Reply STOP to opt out."
ret, info = self.sms.createTemplate(
name=f'Campaign-{campaign_name}',
template=template_content,
type='marketing',
description=f'Marketing campaign: {campaign_name}',
signature_id=signature_id
)
if info.ok():
return ret['id']
else:
raise Exception(f"Template creation failed: {info.error}")
def send_campaign(self, template_id, recipients, campaign_message):
"""Send campaign to recipients"""
campaign_id = f"campaign-{datetime.datetime.now().strftime('%Y%m%d-%H%M%S')}"
self.campaign_stats[campaign_id] = {
'total': len(recipients),
'sent': 0,
'failed': 0,
'start_time': datetime.datetime.now()
}
batch_size = 50
for i in range(0, len(recipients), batch_size):
batch = recipients[i:i + batch_size]
mobiles = [r['mobile'] for r in batch]
# Send to batch
for recipient in batch:
parameters = {
'name': recipient['name'],
'campaign_message': campaign_message
}
ret, info = self.sms.sendMessage(template_id, [recipient['mobile']], parameters)
if info.ok():
self.campaign_stats[campaign_id]['sent'] += 1
else:
self.campaign_stats[campaign_id]['failed'] += 1
print(f"Failed to send to {recipient['mobile']}: {info.error}")
time.sleep(2) # Rate limiting between batches
return campaign_id
def get_campaign_stats(self, campaign_id):
"""Get campaign statistics"""
return self.campaign_stats.get(campaign_id, {})
# Usage
campaign_manager = SMSCampaign(sms)
# Create campaign template
template_id = campaign_manager.create_campaign_template('Spring Sale', 'signature-id-123')
# Define recipients
recipients = [
{'name': 'Alice', 'mobile': '+86-13812345678'},
{'name': 'Bob', 'mobile': '+86-13987654321'},
{'name': 'Carol', 'mobile': '+86-13611111111'}
]
# Send campaign
campaign_id = campaign_manager.send_campaign(
template_id=template_id,
recipients=recipients,
campaign_message='Save 30% on all items this week! Use code SPRING30'
)
# Check stats
stats = campaign_manager.get_campaign_stats(campaign_id)
print(f"Campaign {campaign_id} stats:")
print(f"Total: {stats['total']}")
print(f"Sent: {stats['sent']}")
print(f"Failed: {stats['failed']}")
print(f"Success rate: {stats['sent']/stats['total']*100:.1f}%")Install with Tessl CLI
npx tessl i tessl/pypi-qiniu