coordinate/test/test_publish_slot.py

import json
import redis

# Redis 连接
r = redis.Redis(
    host="text.skin",
    port=6379,
    db=0,
    password="STEs2x6ML0U1HlpE9SojM6YU7QPhqzY8",
    decode_responses=True
)

# 使用的键名（原 Channel 名称）
key = "vs:signal:slot.lon.fr.tourist"

# 消息体
message = {
    "group_id": "tls.gb.fr",
    "apt_type": {
        "weight": 10,
        "routing_key": "slot.lon.fr.tourist",
        "city": "London",
        "visa_type": "Tourist",
        "country": "France"
    },
    "query_result": {
        "routing_key": "slot.lon.fr.tourist",
        "country": "France",
        "city": "London",
        "visa_type": "Tourist",
        "availability_status": "Available",
        "earliest_date": "2026-06-05",
        "availability": [
            {
                "date": "2026-06-05",
                "times": [
                    {
                        "time": "14:00",
                        "label": ""
                    }
                ]
            }
        ],
        "snapshot_source": "worker",
        "snapshot_at": "2026-05-02T14:23:45.123456+00:00"
    },
    "timestamp": 1777745853
}

# 写入 Redis 键，有效期 30 秒
r.setex(key, 30, json.dumps(message, ensure_ascii=False))

print(f"写入成功 -> 键: {key}")
print(f"有效期: 30 秒")