tiktok-download

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):

1. 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.

2. 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.csv
     • tikhub_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:

1. First check whether this skill file already contains a usable API key value in the API_KEY default section.
2. If the key is missing, empty, placeholder-only, or otherwise unavailable, stop the current task immediately.
3. 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
4. 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.

#!/usr/bin/env python3
"""
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)

# one video; raw JSON -> ./tikhub_api_raw/raw_<aweme_id>_....json
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.

#!/usr/bin/env python3
"""
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)

1. pip install httpx
2. Copy Part A and Part B scripts next to each other (any folder; no dependency on this repo’s Python packages).
3. 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.
4. 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.
5. 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.

---