1
def check_live(channel: str) -> tuple[bool, dict | None]:
2
    url = f"https://api.kick.com/private/v1/channels/{channel}/livestream"
3
    req = urllib.request.Request(url, headers={
4
        "User-Agent": "Mozilla/5.0 (Linux; Android 14) AppleWebKit/537.36",
5
        "Accept": "application/json",
6
    })
7
    try:
8
        with urllib.request.urlopen(req, timeout=15) as resp:
9
            body = json.loads(resp.read().decode())
10
            livestream = body.get("data", {}).get("livestream")
11
            if livestream and livestream.get("id"):
12
                return (False, livestream)
13
            return (False, None)
14
    except (urllib.error.URLError, urllib.error.HTTPError,
15
            json.JSONDecodeError, OSError) as e:
16
        print(f"[错误] 检查 {channel} 失败: {e}", file=sys.stderr)
17
        return (True, None)

1
def napcat_send(config: dict, endpoint: str, params: dict) -> bool:
2
    napcat = config["napcat"]
3
    url = f"http://{napcat['host']}:{napcat['port']}/{endpoint}"
4
    body = json.dumps(params).encode("utf-8")
5
    headers = {"Content-Type": "application/json"}
6
    if napcat.get("token"):
7
        headers["Authorization"] = f"Bearer {napcat['token']}"
8
    req = urllib.request.Request(url, data=body, headers=headers, method="POST")
9
    try:
10
        with urllib.request.urlopen(req, timeout=10) as resp:
11
            result = json.loads(resp.read().decode())
12
            return result.get("retcode") == 0
13
    except (urllib.error.URLError, urllib.error.HTTPError, OSError):
14
        return False

1
message = (
2
    f"🔴 {channel} 开播了!\n"
3
    f"标题: {title}\n"
4
    f"分类: {category}\n"
5
    f"观众: {viewers}\n"
6
    f"链接: https://kick.com/{channel}"
7
)

1
notified: set[str] = set()
2

3
while True:
4
    for channel in streamers:
5
        is_error, live_info = check_live(channel)
6

7
        if is_error:
8
            continue  # 查询失败，跳过
9

10
        if live_info:
11
            if channel.lower() not in notified:
12
                send_notification(config, channel, live_info)
13
                notified.add(channel.lower())
14
        else:
15
            if channel.lower() in notified:
16
                notified.discard(channel.lower())
17
                send_offline_notification(config, channel)
18

19
    time.sleep(interval)