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): 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.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: 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. #!/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 ( url: str ) -> str : host = (urlparse(url).netloc or "" ).lower() if "douyin.com" in host: return "douyin" return "tiktok" def fetch_video_info ( api_key: str , share_url: str ) -> dict : platform = platform_from_url(share_url) endpoint = f"https://api.tikhub.io/api/v1/ {platform} /app/v3/fetch_one_video_by_share_url" headers = { "Authorization" : f"Bearer {api_key} " , "Accept" : "*/*" } params = { "share_url" : share_url} with httpx.Client(timeout= 30.0 , follow_redirects= True ) as client: resp = client.get(endpoint, headers=headers, params=params) resp.raise_for_status() return resp.json() def safe_raw_filename ( raw: dict , share_url: str ) -> str : data = raw.get( "data" ) or {} detail = data.get( "aweme_detail" ) if not detail and data.get( "aweme_details" ): detail = (data.get( "aweme_details" ) or [ None ])[ 0 ] aid = (detail or {}).get( "aweme_id" ) or "unknown" rid = (raw.get( "request_id" ) or "noreq" ).replace( "-" , "" ) rid = rid[: 16 ] if len (rid) > 16 else rid if aid == "unknown" : h = hashlib.sha256(share_url.encode( "utf-8" )).hexdigest()[: 10 ] return f"raw_unknown_ {h} _ {rid} .json" return f"raw_ {aid} _ {rid} .json" def save_raw_json ( raw: dict , share_url: str , raw_dir: str ) -> str : os.makedirs(raw_dir, exist_ok= True ) path = os.path.join(raw_dir, safe_raw_filename(raw, share_url)) with open (path, "w" , encoding= "utf-8" ) as f: json.dump(raw, f, ensure_ascii= False , indent= 2 ) return path def extract_clean_data ( raw: dict ) -> dict : data = raw.get( "data" , {}) detail = data.get( "aweme_detail" ) if not detail and data.get( "aweme_details" ): detail = data[ "aweme_details" ][ 0 ] if not detail: return {} video = detail.get( "video" , {}) play = video.get( "play_addr" , {}) or {} url_list = play.get( "url_list" ) or [] video_url = url_list[ 0 ] if url_list else "" author = detail.get( "author" , {}) or {} stats = detail.get( "statistics" , {}) or {} return { "id" : detail.get( "aweme_id" , "" ), "desc" : detail.get( "desc" , "" ), "author_name" : author.get( "nickname" , "" ), "create_time" : detail.get( "create_time" , 0 ), "video_url" : video_url, "like_count" : stats.get( "digg_count" , 0 ), "comment_count" : stats.get( "comment_count" , 0 ), "share_count" : stats.get( "share_count" , 0 ), "play_count" : stats.get( "play_count" , 0 ), "duration" : video.get( "duration" , 0 ), "width" : play.get( "width" , 0 ), "height" : play.get( "height" , 0 ), } def download_file ( url: str , output_path: str ) -> None : headers = { "User-Agent" : ( "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36" ) } with httpx.Client(timeout=httpx.Timeout( 60.0 , read= 300.0 ), follow_redirects= True ) as client: with client.stream( "GET" , url, headers=headers) as r: r.raise_for_status() with open (output_path, "wb" ) as f: for chunk in r.iter_bytes(chunk_size= 8192 ): if chunk: f.write(chunk) def metrics_dict ( info: dict ) -> Dict [ str , Any ]: return { "video_id" : info[ "id" ], "author" : info.get( "author_name" , "" ), "likes" : int (info.get( "like_count" ) or 0 ), "comments" : int (info.get( "comment_count" ) or 0 ), "shares" : int (info.get( "share_count" ) or 0 ), "plays" : int (info.get( "play_count" ) or 0 ), "duration_ms" : int (info.get( "duration" ) or 0 ), "resolution" : f" { int (info.get( 'width' ) or 0 )} x { int (info.get( 'height' ) or 0 )} " , } def run_one ( share_url: str , out_dir: str , raw_dir: str , save_raw: bool , ) -> int : if not API_KEY: print ( "Error: no valid TikHub API key is available. Stop this task and ask the human user " "to apply for a paid personal API key at https://tikhub.io/ and fill it into " "/AWorld/aworld-skills/tikhub_download/SKILL.md." , file=sys.stderr, ) return 3 os.makedirs(out_dir, exist_ok= True ) raw = fetch_video_info(API_KEY, share_url) raw_path = None if save_raw: raw_path = save_raw_json(raw, share_url, raw_dir) print ( "Raw API JSON:" , raw_path) info = extract_clean_data(raw) if not info or not info.get( "id" ) or not info.get( "video_url" ): print ( "Error: parse failed or video_url missing." , file=sys.stderr) print ( "Raw code/message:" , raw.get( "code" ), raw.get( "message" ), file=sys.stderr) return 4 base = f" {platform_from_url(share_url)} _ {clean_name(info.get( 'author_name' ))} _ {info[ 'id' ]} " output_path = os.path.join(out_dir, f" {base} .mp4" ) download_file(info[ "video_url" ], output_path) print ( "Download OK" ) print ( "Saved:" , output_path) print ( "\n=== Metrics ===" ) m = metrics_dict(info) for k, v in m.items(): print ( f" {k} : {v} " ) return 0 def process_one_job ( share_url: str , out_dir: str , raw_dir: str , save_raw: bool , ) -> Dict [ str , Any ]: try : raw = fetch_video_info(API_KEY, share_url) raw_path = None if save_raw: raw_path = save_raw_json(raw, share_url, raw_dir) info = extract_clean_data(raw) if not info or not info.get( "id" ) or not info.get( "video_url" ): return { "ok" : False , "url" : share_url, "raw_json_path" : raw_path, "error" : f"parse failed code= {raw.get( 'code' )} msg= {raw.get( 'message' )} " , } base = f" {platform_from_url(share_url)} _ {clean_name(info.get( 'author_name' ))} _ {info[ 'id' ]} " output_path = os.path.join(out_dir, f" {base} .mp4" ) download_file(info[ "video_url" ], output_path) return { "ok" : True ,