直播事件监控
通过 WebSocket 实时获取任意 TikTok LIVE 主播的弹幕、礼物、点赞、关注 + 观众数。
Node.jsPythonHTML / JSLLM 提示词
// live-events.mjs
// Run: node live-events.mjs (random live creator)
// node live-events.mjs aljazeeraenglish (specific channel)
// Requires Node 22+ (native WebSocket). Zero npm install.
const KEY = 'your_api_key'; // <- swap for your free key at https://tik.tools
const USER = (process.argv[2] || '').replace(/^@/, '');
const API = 'https://api.tik.tools', WS = 'wss://api.tik.tools';
let user = USER;
if (!user) {
const r = await fetch(API + '/api/live/top-channels').then(r => r.json());
user = r.channels?.[0]?.uniqueId;
if (!user) { console.log('No live creators right now.'); process.exit(0); }
console.log('Random live pick:', user);
}
const j = await fetch(API + '/authentication/jwt?apiKey=' + KEY, {
method: 'POST', headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ allowed_creators: [user], expire_after: 600, max_websockets: 1 }),
}).then(r => r.json());
const token = j.data?.token;
if (!token) { console.log('Auth failed:', j); process.exit(1); }
console.log('Connecting to @' + user);
const ws = new WebSocket(`${WS}?uniqueId=${user}&jwtKey=${encodeURIComponent(token)}`);
ws.onmessage = ev => {
const e = JSON.parse(ev.data);
if (e.event === 'roomInfo') return;
const d = e.data || {};
console.log(`[${e.event}]`, d.user?.nickname || d.user_unique_id || '', d.comment || d.giftName || '');
};
ws.onclose = ev => { console.log('Closed:', ev.reason || ev.code); };# live_events.py
# Run: pip install websockets
# python live_events.py (random live creator)
# python live_events.py aljazeeraenglish (specific channel)
import asyncio, json, sys, urllib.request, websockets
KEY = 'your_api_key' # <- swap for your free key at https://tik.tools
USER = (sys.argv[1] if len(sys.argv) > 1 else '').lstrip('@')
API = 'https://api.tik.tools'; WS = 'wss://api.tik.tools'
def call(url, body=None):
req = urllib.request.Request(url, method='POST' if body else 'GET',
headers={'Content-Type': 'application/json'},
data=json.dumps(body).encode() if body else None)
return json.loads(urllib.request.urlopen(req).read())
async def main():
user = USER
if not user:
r = call(API + '/api/live/top-channels')
user = (r.get('channels') or [{}])[0].get('uniqueId', '')
if not user: print('No live creators right now.'); return
print('Random live pick:', user)
j = call(API + '/authentication/jwt?apiKey=' + KEY,
{'allowed_creators': [user], 'expire_after': 600, 'max_websockets': 1})
token = j.get('data', {}).get('token')
if not token: print('Auth failed:', j); return
print('Connecting to @' + user)
url = f"{WS}?uniqueId={user}&jwtKey={token}"
async with websockets.connect(url) as ws:
async for raw in ws:
e = json.loads(raw)
if e.get('event') == 'roomInfo': continue
d = e.get('data', {})
nm = (d.get('user') or {}).get('nickname') or d.get('user_unique_id') or ''
text = d.get('comment') or (f"sent {d['giftName']} x{d.get('repeatCount', 1)}" if d.get('giftName') else '')
print(f"[{e.get('event')}] {nm} {text}")
asyncio.run(main())<!DOCTYPE html>
<!-- Save as live-events.html and double-click. No build step. Swap KEY for your own key. -->
<meta charset="utf-8">
<title>Tik.Tools live events</title>
<style>
body{font:14px ui-monospace,Menlo,monospace;background:#0a0a0c;color:#e4e4e7;padding:24px;max-width:760px;margin:0 auto}
h1{font-size:18px;margin:0 0 8px}
.bar{display:flex;gap:8px;margin:8px 0 14px}
input,button{font:inherit;background:#14141a;color:#fff;border:1px solid #333;border-radius:6px;padding:6px 10px}
button{background:#2a2a2a;border:none;cursor:pointer;font-weight:700}
.row{padding:6px 10px;border-bottom:1px solid #222;display:flex;gap:10px;align-items:baseline}
.type{font-size:11px;text-transform:uppercase;color:#d1d3d4;width:60px;flex-shrink:0}
.user{color:#fff;font-weight:600}
.msg{color:#9aa0a6;flex:1;overflow:hidden;text-overflow:ellipsis}
</style>
<h1>Live events monitor</h1>
<div class="bar">
<input id="u" placeholder="username (empty = random live)" />
<button id="go">Connect</button>
</div>
<div id="state">Enter a username or click Connect for a random live creator.</div>
<div id="feed"></div>
<script>
const API = 'https://api.tik.tools', WS = 'wss://api.tik.tools';
const KEY = 'your_api_key'; // <- swap for your free key at https://tik.tools
let ws = null;
async function connect() {
if (ws) try { ws.close(); } catch {}
feed.innerHTML = '';
let user = u.value.trim().replace(/^@/, '');
if (!user) {
state.textContent = 'Picking a random live creator...';
try {
const r = await fetch(API + '/api/live/top-channels').then(r => r.json());
user = (r.channels && r.channels[0] && r.channels[0].uniqueId) || '';
} catch {}
if (!user) { state.textContent = 'No live creators right now, try again in a minute.'; return; }
}
state.textContent = 'Connecting to @' + user + '...';
const j = await fetch(API + '/authentication/jwt?apiKey=' + KEY, {
method: 'POST', headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ allowed_creators: [user], expire_after: 600, max_websockets: 1 }),
}).then(r => r.json());
const token = j.data && j.data.token;
if (!token) { state.textContent = 'Auth failed - get a free key at https://tik.tools'; return; }
ws = new WebSocket(WS + '?uniqueId=' + user + '&jwtKey=' + encodeURIComponent(token));
ws.onopen = () => state.textContent = '@' + user + ' - LIVE';
ws.onmessage = ev => {
const e = JSON.parse(ev.data);
if (e.event === 'roomInfo' || !e.event) return;
const d = e.data || {}, nm = (d.user && d.user.nickname) || d.user_unique_id || 'someone';
const text = d.comment || (d.giftName ? 'sent ' + d.giftName + ' x' + (d.repeatCount || 1) : '');
feed.insertAdjacentHTML('afterbegin',
'<div class="row"><span class="type">' + e.event + '</span><span class="user">' + nm + '</span><span class="msg">' + text + '</span></div>');
while (feed.children.length > 50) feed.removeChild(feed.lastChild);
};
ws.onclose = ev => { state.textContent = 'Closed: ' + (ev.reason || ev.code); };
}
go.onclick = connect;
// Auto-connect on load so the embedded preview starts streaming immediately.
connect();
</script>Write me a minimal one-file program that connects to a TikTok LIVE stream using the Tik.Tools API + WebSocket gateway and prints every chat / gift / like / member event to the console. The program must:
1. Accept an optional CLI argument: the TikTok username to watch. If empty, hit GET https://api.tik.tools/api/live/top-channels (no auth, public, returns {channels:[{uniqueId,...}]}) and use channels[0].uniqueId.
2. Mint a short-lived JWT: POST https://api.tik.tools/authentication/jwt?apiKey=your_api_key with body {"allowed_creators":["<user>"],"expire_after":600,"max_websockets":1}. Response: {data:{token:"..."}}.
3. Open a WebSocket: wss://api.tik.tools?uniqueId=<user>&jwtKey=<token>
4. Each WS message is JSON shaped {event, data:{user:{nickname,uniqueId}, comment, giftName, repeatCount, viewerCount}}.
Constraints:
- Zero npm install / pip install except 'websockets' (Python only).
- One file. No build step. Stupid simple to copy/paste and run.
- The API key 'your_api_key' is a shared public demo key with a small daily cap shared by every user, so it is often exhausted. Put the key in one clearly marked constant so I can swap in my own: a free Sandbox account at https://tik.tools gives a personal key for a 7-day evaluation (no credit card), and paid plans continue from there with higher limits.
Pick the language I asked for and give me ONE working file. Comment the swap-the-key line + the username arg.live-events-monitor.html
<!DOCTYPE html>
<!-- Save as live-events.html and double-click. No build step. Swap KEY for your own key. -->
<meta charset="utf-8">
<title>Tik.Tools live events</title>
<style>
body{font:14px ui-monospace,Menlo,monospace;background:#0a0a0c;color:#e4e4e7;padding:24px;max-width:760px;margin:0 auto}
h1{font-size:18px;margin:0 0 8px}
.bar{display:flex;gap:8px;margin:8px 0 14px}
input,button{font:inherit;background:#14141a;color:#fff;border:1px solid #333;border-radius:6px;padding:6px 10px}
button{background: ...