yingshiyun/probe.py

#!/usr/bin/env python3
"""
第一步：探测萤石API，确认录像列表和回放流接口的实际返回格式
"""

import json, os, requests
from datetime import datetime, timedelta
from config import *

BASE = "https://open.ys7.com/api/lapp"


def get_token():
    """获取AccessToken，缓存到本地（有效期7天）"""
    if os.path.exists(TOKEN_CACHE):
        cache = json.load(open(TOKEN_CACHE))
        if datetime.now().timestamp() < cache["expire_ts"] - 3600:
            print(f"使用缓存Token，有效期至: {cache['expire_str']}")
            return cache["token"]

    r = requests.post(f"{BASE}/token/get",
                      data={"appKey": APP_KEY, "appSecret": APP_SECRET})
    d = r.json()
    if d["code"] != "200":
        raise Exception(f"Token获取失败: {d}")

    token     = d["data"]["accessToken"]
    expire_ts = d["data"]["expireTime"] / 1000
    expire_str = datetime.fromtimestamp(expire_ts).strftime("%Y-%m-%d %H:%M:%S")
    json.dump({"token": token, "expire_ts": expire_ts, "expire_str": expire_str},
              open(TOKEN_CACHE, "w"))
    print(f"Token已刷新，有效期至: {expire_str}")
    return token


def probe_record_search(token, date_str):
    """探测：查询SD卡录像列表（试几个端点）"""
    params = {
        "accessToken": token,
        "deviceSerial": DEVICE_SERIAL,
        "channelNo":    CHANNEL_NO,
        "startTime":    f"{date_str} 00:00:00",
        "endTime":      f"{date_str} 23:59:59",
        "recType":      0,
        "pageStart":    0,
        "pageSize":     10
    }
    endpoints = [
        "/record/search",
        "/storage/recording/day/times",
        "/device/record/list",
        "/video/rec/list",
    ]
    for ep in endpoints:
        print(f"\n{'='*50}")
        print(f"【接口1变体】{ep}  日期={date_str}")
        r = requests.post(f"{BASE}{ep}", data=params)
        print(f"HTTP {r.status_code}")
        print(r.text[:500])
    return None


def probe_video_by_time(token, date_str):
    """探测：尝试多种时间格式"""
    from datetime import timezone
    dt_start = datetime.strptime(f"{date_str} 00:00:00", "%Y-%m-%d %H:%M:%S")
    dt_end   = datetime.strptime(f"{date_str} 01:00:00", "%Y-%m-%d %H:%M:%S")
    ts_start = int(dt_start.timestamp() * 1000)  # 毫秒时间戳
    ts_end   = int(dt_end.timestamp() * 1000)

    formats = [
        ("yyyy-MM-dd HH:mm:ss", f"{date_str} 00:00:00", f"{date_str} 01:00:00"),
        ("yyyyMMddHHmmss",      date_str.replace("-","")+"000000", date_str.replace("-","")+"010000"),
        ("毫秒时间戳",           str(ts_start), str(ts_end)),
        ("秒时间戳",             str(ts_start//1000), str(ts_end//1000)),
    ]
    for fmt_name, s, e in formats:
        print(f"\n{'='*50}")
        print(f"【接口2】video/by/time 时间格式={fmt_name}  start={s}")
        r = requests.post(f"{BASE}/video/by/time", data={
            "accessToken":  token,
            "deviceSerial": DEVICE_SERIAL,
            "channelNo":    CHANNEL_NO,
            "startTime":    s,
            "endTime":      e
        })
        print(f"HTTP {r.status_code}  {r.text[:200]}")
    return None


def probe_live_address(token):
    """探测：获取直播流地址（顺带看看协议格式）"""
    print(f"\n{'='*50}")
    print("【接口3】直播流地址（探测协议格式）")
    r = requests.post(f"{BASE}/v2/live/address/get", data={
        "accessToken":  token,
        "deviceSerial": DEVICE_SERIAL,
        "channelNo":    CHANNEL_NO,
        "protocol":     2,   # 1=ezopen 2=HLS 3=RTMP 4=FLV
        "quality":      1
    })
    print(f"HTTP {r.status_code}")
    print(r.text[:500])


def probe_playback_address(token):
    """探测回放流地址（已知录像时间段的毫秒时间戳）"""
    rec_start = 1781179564000
    rec_end   = 1781211539000
    endpoints = [
        ("/v2/live/address/get",     {"startTime": rec_start, "endTime": rec_end, "protocol": 2, "quality": 1}),
        ("/record/playback/address", {"startTime": rec_start, "endTime": rec_end}),
        ("/v3/live/address/get",     {"startTime": rec_start, "endTime": rec_end, "protocol": 2}),
        ("/video/playback",          {"startTime": rec_start, "endTime": rec_end}),
    ]
    for ep, extra in endpoints:
        params = {"accessToken": token, "deviceSerial": DEVICE_SERIAL, "channelNo": CHANNEL_NO}
        params.update(extra)
        print(f"\n{'='*50}")
        print(f"【回放】{ep}")
        r = requests.post(f"{BASE}{ep}", data=params)
        print(f"HTTP {r.status_code}  {r.text[:300]}")


if __name__ == "__main__":
    token = get_token()

    yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
    y_start_ms = int(datetime.strptime(f"{yesterday} 00:00:00", "%Y-%m-%d %H:%M:%S").timestamp() * 1000)
    y_end_ms   = int(datetime.strptime(f"{yesterday} 23:59:59", "%Y-%m-%d %H:%M:%S").timestamp() * 1000)

    print("\n=== 1. 查录像列表 ===")
    r = requests.post(f"{BASE}/video/by/time", data={
        "accessToken": token, "deviceSerial": DEVICE_SERIAL,
        "channelNo": CHANNEL_NO,
        "startTime": y_start_ms, "endTime": y_end_ms
    })
    print(r.text[:800])

    print("\n=== 2. 探测回放流地址 ===")
    probe_playback_address(token)