| name | tiktok_download |
| description | Single-file TikTok/Douyin video download and traffic metrics via TikHub API using only httpx; optional persisted raw API JSON plus a stdlib post-processor emitting CSV and simplified JSON. Supports one URL or concurrent batch (max 10 workers). No dependency on any project codebase. |
TikHub Download Independent
What this skill gives you
Two small artifacts (all copy-pasteable from this file):
-
tikhub_independent.py (single file, httpx only)
- Single video or batch (parallel, max 10 workers): download MP4 + print metrics
Raw API JSON: by default every successful API response is written to disk as a full JSON file (same shape as e.g. raw_api_response.json in this repo — top-level code, request_id, params, data.aweme_detail, etc.). Use --no-save-raw to skip.
-
postprocess_tikhub_raw.py (stdlib only: json, csv, argparse, glob)
- Reads one raw file or a directory of raw JSON files
- Writes into the current working directory (or
--out-dir):
tikhub_videos_summary.csvtikhub_videos_summary.json (list of simplified records; one object per video)
Structured-data reference (field meanings, nesting): see project doc
TikHub_API_数据格式说明.md (same repo, path from project root). Example raw payload: raw_api_response.json.
Does not import TikHub-Multi-Functional-Downloader or any other project package.
Install
pip install httpx
Post-processor: no extra packages.
API (for reference)
- TikTok:
GET https://api.tikhub.io/api/v1/tiktok/app/v3/fetch_one_video_by_share_url?share_url=...
- Douyin:
GET https://api.tikhub.io/api/v1/douyin/app/v3/fetch_one_video_by_share_url?share_url=...
- Header:
Authorization: Bearer <API_KEY>
Mandatory prerequisite: valid TikHub API key
This skill may be used only when a valid TikHub API key is available.
Agent rule:
- First check whether this skill file already contains a usable API key value in the
API_KEY default section.
- If the key is missing, empty, placeholder-only, or otherwise unavailable, stop the current task immediately.
- The agent must then tell the human user:
- they need to go to https://tikhub.io/ and apply for their own paid personal API key
- then fill that key into this file:
/AWorld/aworld-skills/tikhub_download/SKILL.md
- Until the human has provided and filled in a valid key, do not continue with download, data fetching, retry, workaround, or fabricated-output behavior.
Concrete example:
- If this file no longer contains a real key in the relevant
API_KEY location, treat the key as unavailable and follow the stop-and-notify rule above.
Part A — tikhub_independent.py (download + optional raw JSON)
Save the following as tikhub_independent.py.
Behavior note: After each fetch_video_info call, if saving is enabled (default), the entire parsed JSON object is written with json.dump(..., indent=2, ensure_ascii=False) — this is the audit / replay artifact for downstream tooling, not the simplified extract.
"""
TikTok/Douyin: download MP4 + metrics via TikHub API. Optional: save full raw API JSON per request.
Requires: pip install httpx
Usage:
python tikhub_independent.py one "https://www.tiktok.com/@user/video/123"
python tikhub_independent.py one "URL" --no-save-raw
python tikhub_independent.py batch urls.txt
python tikhub_independent.py batch urls.txt --raw-dir my_raw_dir --max-workers 4
Raw JSON default directory (relative to current working directory): ./tikhub_api_raw
"""
from __future__ import annotations
import argparse
import concurrent.futures
import hashlib
import json
import os
import re
import sys
from typing import Any, Dict, List
from urllib.parse import urlparse
import httpx
API_KEY = os.getenv(
"TIKHUB_API_KEY",
"",
).strip()
MAX_WORKERS_CAP = 10
DEFAULT_OUT = os.path.expanduser("~/Downloads/tikhub_independent")
DEFAULT_RAW_DIR = "tikhub_api_raw"
def clean_name(name: str, max_len: int = 60) -> str:
name = re.sub(r'[\\/:*?"<>|]+', "_", (name or "").strip())
name = re.sub(r"\s+", " ", name).strip()
return (name[:max_len] or "video").strip(" ._")
def platform_from_url() -> :
host = (urlparse(url).netloc ).lower()
host:
() -> :
platform = platform_from_url(share_url)
endpoint =
headers = {: , : }
params = {: share_url}
httpx.Client(timeout=, follow_redirects=) client:
resp = client.get(endpoint, headers=headers, params=params)
resp.raise_for_status()
resp.json()
() -> :
data = raw.get() {}
detail = data.get()
detail data.get():
detail = (data.get() [])[]
aid = (detail {}).get()
rid = (raw.get() ).replace(, )
rid = rid[:] (rid) > rid
aid == :
h = hashlib.sha256(share_url.encode()).hexdigest()[:]
() -> :
os.makedirs(raw_dir, exist_ok=)
path = os.path.join(raw_dir, safe_raw_filename(raw, share_url))
(path, , encoding=) f:
json.dump(raw, f, ensure_ascii=, indent=)
path
() -> :
data = raw.get(, {})
detail = data.get()
detail data.get():
detail = data[][]
detail:
{}
video = detail.get(, {})
play = video.get(, {}) {}
url_list = play.get() []
video_url = url_list[] url_list
author = detail.get(, {}) {}
stats = detail.get(, {}) {}
{
: detail.get(, ),
: detail.get(, ),
: author.get(, ),
: detail.get(, ),
: video_url,
: stats.get(, ),
: stats.get(, ),
: stats.get(, ),
: stats.get(, ),
: video.get(, ),
: play.get(, ),
: play.get(, ),
}
() -> :
headers = {
: (
)
}
httpx.Client(timeout=httpx.Timeout(, read=), follow_redirects=) client:
client.stream(, url, headers=headers) r:
r.raise_for_status()
(output_path, ) f:
chunk r.iter_bytes(chunk_size=):
chunk:
f.write(chunk)
() -> [, ]:
{
: info[],
: info.get(, ),
: (info.get() ),
: (info.get() ),
: (info.get() ),
: (info.get() ),
: (info.get() ),
: ,
}
() -> :
API_KEY:
(
,
file=sys.stderr,
)
os.makedirs(out_dir, exist_ok=)
raw = fetch_video_info(API_KEY, share_url)
raw_path =
save_raw:
raw_path = save_raw_json(raw, share_url, raw_dir)
(, raw_path)
info = extract_clean_data(raw)
info info.get() info.get():
(, file=sys.stderr)
(, raw.get(), raw.get(), file=sys.stderr)
base =
output_path = os.path.join(out_dir, )
download_file(info[], output_path)
()
(, output_path)
()
m = metrics_dict(info)
k, v m.items():
()
() -> [, ]:
:
raw = fetch_video_info(API_KEY, share_url)
raw_path =
save_raw:
raw_path = save_raw_json(raw, share_url, raw_dir)
info = extract_clean_data(raw)
info info.get() info.get():
{
: ,
: share_url,
: raw_path,
: ,
}
base =
output_path = os.path.join(out_dir, )
download_file(info[], output_path)
{
: ,
: share_url,
: output_path,
: raw_path,
: metrics_dict(info),
}
Exception e:
{: , : share_url, : (e), : }
() -> []:
urls: [] = []
(path, , encoding=) f:
line f:
u = line.strip()
u u.startswith():
urls.append(u)
urls
() -> :
API_KEY:
(
,
file=sys.stderr,
)
urls = read_urls(urls_file)
urls:
(, file=sys.stderr)
os.makedirs(out_dir, exist_ok=)
workers = (, (MAX_WORKERS_CAP, max_workers, (urls)))
results: [[, ]] = []
concurrent.futures.ThreadPoolExecutor(max_workers=workers) ex:
futs = [
ex.submit(process_one_job, u, out_dir, raw_dir, save_raw) u urls
]
fut concurrent.futures.as_completed(futs):
results.append(fut.result())
ok = ( r results r.get())
fail = (results) - ok
()
()
save_raw:
()
r (results, key= x: x.get(, )):
r.get():
()
()
r.get():
()
()
:
()
r.get():
()
()
fail ==
() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(
description=
)
sub = p.add_subparsers(dest=, required=)
p_one = sub.add_parser(, =)
p_one.add_argument(, =)
p_one.add_argument(
,
,
default=DEFAULT_OUT,
=,
)
p_one.add_argument(
,
action=,
=,
)
p_one.add_argument(
,
default=DEFAULT_RAW_DIR,
=,
)
p_batch = sub.add_parser(, =)
p_batch.add_argument(, =)
p_batch.add_argument(
,
,
default=DEFAULT_OUT,
=,
)
p_batch.add_argument(
,
action=,
=,
)
p_batch.add_argument(
,
default=DEFAULT_RAW_DIR,
=,
)
p_batch.add_argument(
,
=,
default=,
=,
)
p
() -> :
parser = build_parser()
args = parser.parse_args()
save_raw = args.no_save_raw
raw_dir = os.path.abspath(os.path.join(os.getcwd(), args.raw_dir))
args.command == :
run_one(
args.url.strip(),
os.path.expanduser(args.output_dir),
raw_dir,
save_raw,
)
args.command == :
mw = (, (MAX_WORKERS_CAP, (args.max_workers)))
run_batch(
args.urls_file,
os.path.expanduser(args.output_dir),
mw,
raw_dir,
save_raw,
)
__name__ == :
SystemExit(main())
Commands (tikhub_independent.py)
python tikhub_independent.py one "https://www.tiktok.com/@mumumelon67/video/7484120063369415978"
python tikhub_independent.py one "URL" --no-save-raw
python tikhub_independent.py one "URL" --raw-dir ./my_raw_exports
python tikhub_independent.py batch urls.txt
python tikhub_independent.py batch urls.txt --max-workers 4 --raw-dir ./batch_raw
Concurrent batch: API call + optional raw write + download all run in parallel threads (cap 10). Raw files are written under --raw-dir (absolute path resolved from cwd).
Part B — postprocess_tikhub_raw.py (CSV + simplified JSON)
Save as postprocess_tikhub_raw.py (stdlib only).
Input: one raw file like raw_api_response.json, or a directory containing multiple raw_*.json / any *.json from this workflow.
Output in --out-dir (default: current working directory):
| File | Purpose |
|---|
tikhub_videos_summary.csv | Flat columns for spreadsheets |
tikhub_videos_summary.json | {"generated_from": ..., "videos": [ {...}, ... ]} simplified records |
Field selection follows TikHub_API_数据格式说明.md: ids, text, type, time, author strip, statistics, video basics, music/commerce/AIGC flags where present.
"""
Post-process TikHub raw API JSON files -> CSV + simplified JSON (stdlib only).
Usage:
python postprocess_tikhub_raw.py --input raw_api_response.json
python postprocess_tikhub_raw.py --input ./tikhub_api_raw --out-dir .
See TikHub_API_数据格式说明.md for full field documentation.
"""
from __future__ import annotations
import argparse
import csv
import json
import os
import sys
from glob import glob
from typing import Any, Dict, List, Optional
def get_aweme_detail(raw: dict) -> Optional[dict]:
data = raw.get("data") or {}
d = data.get("aweme_detail")
if d:
return d
ads = data.get("aweme_details")
if isinstance(ads, list) and ads:
return ads[0]
return None
def first_url(obj: Any) -> str:
if not obj or not isinstance(obj, dict):
return
lst = obj.get()
(lst, ) lst:
(lst[])
() -> [, ]:
params = raw.get() {}
share_url = params.get()
detail = get_aweme_detail(raw)
detail:
{
: source_file,
: raw.get(),
: raw.get(),
: raw.get(),
: share_url,
: ,
: ,
}
stats = detail.get() {}
author = detail.get() {}
video = detail.get() {}
play = video.get() {}
music = detail.get() {}
aigc = detail.get() {}
hashtag_names = []
x detail.get() []:
(x, ) x.get():
hashtag_names.append(x[])
commerce = detail.get()
commerce_min =
(commerce, ):
commerce_min = {
k: commerce.get(k)
k (, , )
k commerce
}
commerce_min:
commerce_min = {: }
row: [, ] = {
: os.path.basename(source_file),
: ,
: raw.get(),
: raw.get(),
: raw.get(),
: raw.get(),
: share_url,
: raw.get(),
: detail.get(),
: (detail.get() )[:],
: detail.get(),
: detail.get(),
: detail.get() ,
: author.get(),
: author.get(),
: author.get(),
: author.get(),
: author.get(),
: author.get(),
: author.get(),
: author.get(),
: author.get(),
: author.get(),
: stats.get(),
: stats.get(),
: stats.get(),
: stats.get(),
: stats.get(),
: stats.get(),
: stats.get(),
: stats.get(),
: video.get(),
: play.get(),
: play.get(),
: video.get(),
: video.get(),
: first_url(video.get()),
: first_url(play),
: music.get() music.get(),
: music.get(),
: music.get() music.get(),
: music.get(),
: music.get(),
: music.get(),
: music.get(),
: detail.get(),
: commerce_min,
: aigc.get(),
: aigc.get(),
: .join(hashtag_names) hashtag_names ,
}
row
() -> []:
os.path.isfile(path):
[path]
os.path.isdir(path):
files = (glob(os.path.join(path, )))
files
FileNotFoundError(path)
() -> [, ]:
out = {}
k, v row.items():
v :
out[k] =
(v, (, )):
out[k] = json.dumps(v, ensure_ascii=)
:
out[k] = v
out
() -> :
ap = argparse.ArgumentParser(description=)
ap.add_argument(, , required=, =)
ap.add_argument(
,
,
default=,
=,
)
args = ap.parse_args()
:
files = collect_inputs(args.)
FileNotFoundError e:
(, e, file=sys.stderr)
files:
(, file=sys.stderr)
out_dir = os.path.abspath(args.out_dir)
os.makedirs(out_dir, exist_ok=)
csv_path = os.path.join(out_dir, )
json_path = os.path.join(out_dir, )
videos: [[, ]] = []
fp files:
:
(fp, , encoding=) f:
raw = json.load(f)
Exception ex:
videos.append(
{
: os.path.basename(fp),
: ,
: ,
}
)
videos.append(simplify_raw(raw, fp))
payload = {
: os.path.abspath(args.),
: (videos),
: videos,
}
(json_path, , encoding=) f:
json.dump(payload, f, ensure_ascii=, indent=)
videos:
flat = [flatten_for_csv(v) v videos]
fieldnames: [] = ({k row flat k row.keys()})
(csv_path, , encoding=, newline=) f:
w = csv.DictWriter(f, fieldnames=fieldnames, extrasaction=)
w.writeheader()
row flat:
w.writerow({k: row.get(k, ) k fieldnames})
(, csv_path)
(, json_path)
__name__ == :
SystemExit(main())
Commands (postprocess_tikhub_raw.py)
From the directory where you want tikhub_videos_summary.* (e.g. project root or a report folder):
python postprocess_tikhub_raw.py --input ./tikhub_api_raw
python postprocess_tikhub_raw.py --input /path/to/raw_api_response.json --out-dir .
End-to-end workflow (for others)
pip install httpx- Copy Part A and Part B scripts next to each other (any folder; no dependency on this repo’s Python packages).
- Run
tikhub_independent.py (one or batch) so tikhub_api_raw/ (or --raw-dir) contains full API responses — same idea as raw_api_response.json.
- Run
postprocess_tikhub_raw.py --input <raw file or dir> --out-dir . → get tikhub_videos_summary.csv and tikhub_videos_summary.json in the chosen working directory.
- For field-level meaning of nested keys, open
TikHub_API_数据格式说明.md in this repository (project root).
Troubleshooting
401/403: invalid key or missing scopes.- No valid API key configured in this skill: stop immediately and tell the human user to apply for a paid personal API key at https://tikhub.io/, then fill it into
/AWorld/aworld-skills/tikhub_download/SKILL.md before retrying.
429: rate limit; in batch, reduce --max-workers or retry later.- No
video_url / parse fail: video private, removed, or bad URL; a raw file may still be written if the HTTP response was JSON but content incomplete — check api_code / parse_ok in post-process output.
- Post-process
parse_ok: false: file is not a TikHub fetch_one_video payload or damaged JSON.
- Mainland TikTok: may need proxy (not in these scripts).
What this skill does not cover
User-profile crawling, non-URL workflows, image-only carousels as first-class exports, and any endpoint other than fetch_one_video_by_share_url.
