webapp/app/services/sms_service.py

import json
import time
import requests
from typing import List
from app.schemas.sms import ShortMessageDetail


def save_short_message(redis_client, phone: str, message: str, received_at: str, max_ttl: int) -> ShortMessageDetail:
    """
    将短信保存到 Redis。
    键格式: sms:{phone}
    值为 JSON 数组（保存最近多条短信）
    """
    key = f"sms:{phone}"

    # 取出已有数据
    existing_data = redis_client.get(key)
    if existing_data:
        messages = json.loads(existing_data)
    else:
        messages = []

    # 添加新短信
    new_msg = ShortMessageDetail(phone=phone, message=message, received_at=received_at)
    messages.append(new_msg.dict())

    # 最多保留最近 20 条（可调整）
    messages = messages[-20:]

    # 保存回 Redis
    redis_client.setex(key, max_ttl, json.dumps(messages))

    return new_msg


def query_short_message(redis_client, phone: str, keyword: str, sent_at: str) -> List[ShortMessageDetail]:
    """
    从 Redis 查询短信。
    支持按关键字和时间过滤。
    """
    key = f"sms:{phone}"
    existing_data = redis_client.get(key)
    if not existing_data:
        return []

    messages = [ShortMessageDetail(**m) for m in json.loads(existing_data)]

    # 关键字过滤
    if keyword:
        messages = [m for m in messages if keyword in m.message]

    # 时间过滤（如果需要更复杂可转为时间戳比较）
    if sent_at:
        messages = [m for m in messages if m.received_at >= sent_at]

    return messages