Douyin Downloader 完整技术笔记

{
  "aweme_id": "7687202250602976234",
  "desc": "一曲佳人伴孤灯,愿听完的人万事渐顺 …",
  "author": { "nickname": "应瑾笛箫", "unique_id": "dabuliufenqi" },
  "statistics": { "digg_count": 105080 },
  "video": {
    "duration": 28908,                 // ms
    "width": 1920, "height": 1080,
    "play_addr": {
      "uri": "v0300fg10000dan6pbfog65rus09c6g0",   // ← CDN 视频 ID(核心)
      "url_list": ["https://aweme.snssdk.com/aweme/v1/playwm/?line=0&logo_name=aweme_diversion_search&ratio=720p&video_id=v0300..."]
    },
    "bit_rate": [ /* 网页 detail API 才有的多码率列表 */ ]
  },
  "images": [ /* 图集帖才有 */ ]
}

import json, re
from urllib.parse import unquote
import httpx

def register_ttwid(proxy: str | None = None) -> str | None:
    body = {
        "region": "cn", "aid": 1768, "needFid": False,
        "service": "www.douyin.com",
        "migrate_info": {"ticket": "", "source": "node"},
        "cbUrlProtocol": "https", "union": True,
    }
    kwargs = {}
    if proxy:
        kwargs["proxies"] = {"http://": proxy, "https://": proxy}  # httpx<0.27; >=0.27 用 proxy=
    r = httpx.post(
        "https://ttwid.bytedance.com/ttwid/union/register/",
        json=body,
        headers={"User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 13_2_3 like Mac OS X) "
                               "AppleWebKit/605.1.15 (KHTML, like Gecko) Version/13.0.3 "
                               "Mobile/15E148 Safari/604.1"},
        timeout=30, trust_env=False, **kwargs,
    )
    m = re.search(r"ttwid=([^;]+)", r.headers.get("set-cookie", ""))
    return unquote(m.group(1)) if m else None

ttwid = register_ttwid()
print(ttwid)   # 1|IfLTQStOIWS7ulSy_qkyJ2...|1792473323|09dd11e70d57...

# 用法:请求分享页时带上
html = httpx.get(
    f"https://www.iesdouyin.com/share/video/7687202250602976234/",
    headers={"Cookie": f"ttwid={ttwid}",
             "User-Agent": "Mozilla/5.0 (iPhone; ... )"},
    trust_env=False,
).text
# 此时 html 里的 window._ROUTER_DATA 包含完整 videoInfoRes

curl -s -D - -o /dev/null   -X POST "https://ttwid.bytedance.com/ttwid/union/register/"   -H 'Content-Type: application/json'   -d '{"region":"cn","aid":1768,"needFid":false,"service":"www.douyin.com",
       "migrate_info":{"ticket":"","source":"node"},"cbUrlProtocol":"https","union":true}'   | grep -i '^set-cookie'

fastapi>=0.110
uvicorn[standard]>=0.29
httpx>=0.27
playwright>=1.44

"""Runtime configuration, all overridable via environment variables.

Important (xray / proxy routing):
  Douyin hosts are domestic Chinese sites.  If the server runs inside
  mainland China, leave DOUYIN_PROXY empty so traffic goes DIRECT (this is
  what gives the best anti-bot results).  Only set DOUYIN_PROXY when the
  server is outside China (point it at a mainland exit node, e.g. your
  xray vless setup) or when your xray works in TUN mode and you want to
  force a specific route.
"""
import os


def _env(name, default):
    v = os.environ.get(name)
    if v is None or v.strip() == "":
        return default
    return v.strip()


def _env_bool(name, default):
    v = os.environ.get(name)
    if v is None or v.strip() == "":
        return default
    return v.strip().lower() in ("1", "true", "yes", "on")


class Config(object):
    # Outbound proxy used for ALL douyin-related traffic (share page,
    # metadata APIs and video CDN).  Empty string = direct connection.
    # Examples:  http://127.0.0.1:10809   socks5://127.0.0.1:1080
    DOUYIN_PROXY = _env("DOUYIN_PROXY", "")

    # Enable the headless-browser fallback layer.  Disable if playwright /
    # chromium is not installed on the server.
    BROWSER_ENABLED = _env_bool("BROWSER_ENABLED", True)

    # How long (seconds) the browser layer waits for page data.
    BROWSER_TIMEOUT = int(_env("BROWSER_TIMEOUT", "45"))

    # Explicit chromium/chrome binary for the browser layer.  Empty = use
    # playwright's own chromium (auto-falls back to a compatible system
    # browser when playwright's build cannot start, e.g. 64KB-page ARM
    # kernels with old playwright builds).
    CHROMIUM_PATH = _env("CHROMIUM_PATH", "")

    # Directory for downloaded files.
    TMP_DIR = _env("TMP_DIR", "/tmp/douyin-downloader")

    # How long downloaded files are kept before cleanup (hours).
    FILE_TTL_HOURS = float(_env("FILE_TTL_HOURS", "4"))

    # Default user agents.
    UA_MOBILE = _env(
        "UA_MOBILE",
        "Mozilla/5.0 (iPhone; CPU iPhone OS 13_2_3 like Mac OS X) "
        "AppleWebKit/605.1.15 (KHTML, like Gecko) Version/13.0.3 "
        "Mobile/15E148 Safari/604.1",
    )
    UA_DESKTOP = _env(
        "UA_DESKTOP",
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
        "(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36",
    )

    # Service
    HOST = _env("HOST", "0.0.0.0")
    PORT = int(_env("PORT", "8931"))

    # API access key. When non-empty, every /api/* endpoint requires the
    # client to present this key via the "X-Api-Key" header OR the "?key="
    # query param. /health stays open. Empty = no auth (local testing).
    API_KEY = _env("API_KEY", "")

    # Serve the web UI at "/". Set 0 for an API-only deployment.
    ENABLE_UI = _env_bool("ENABLE_UI", True)

    # Timeouts (seconds)
    HTTP_TIMEOUT = float(_env("HTTP_TIMEOUT", "30"))
    DOWNLOAD_TIMEOUT = float(_env("DOWNLOAD_TIMEOUT", "300"))


CFG = Config()

"""Core (pure-HTTP) layer of the douyin downloader.

Verified flow (2026-10)
-----------------------
1. A valid ``ttwid`` cookie (registered via ttwid.bytedance.com) makes the
   share page server-side-render the full video metadata inside
   ``window._ROUTER_DATA`` (``videoInfoRes.item_list``) — from mainland
   China *and* overseas IPs.
2. ``video.play_addr.uri`` is the CDN video id.  The watermark-free video
   is served by ``https://aweme.snssdk.com/aweme/v1/play/?video_id={uri}``
   (the watermarked twin is ``playwm``).  ``ratio=1080p`` requests the
   highest quality the source has (CDN returns 404 if the source has no
   1080p encode — fall back to the listed ratio / default).
3. The play endpoint 302-redirects to the real CDN (douyinvod.com);
   httpx follows it automatically.

If the SSR data is missing (rare: anti-bot change, unusual video state),
the browser fallback layer (dy/browser.py) is used.
"""
import json
import os
import re
import time
import uuid
import zipfile
from typing import List, Optional, Tuple

import httpx

from .config import CFG


class DouyinError(Exception):
    """User-facing error with a readable message."""


# ---------------------------------------------------------------------------
# 1. Input handling
# ---------------------------------------------------------------------------

_URL_RE = re.compile(
    r"https?://(?:[a-z0-9-]+\.)?"
    r"(?:douyin\.com|iesdouyin.com|amemv\.com|snssdk\.com)/[^\s\"'<>)\]]+",
    re.I,
)
_AWEME_ID_RE = re.compile(r"\b(\d{17,22})\b")
_HTML_LINK_RE = re.compile(r"<a\s+href=\"([^\"]+)\"", re.I)


def extract_target(text: str) -> str:
    """Return a share URL to work with, from raw pasted share text."""
    if not text:
        raise DouyinError("empty input")
    m = _URL_RE.search(text)
    url = m.group(0).rstrip(".,;!?，。；！？") if m else None
    # desktop page URLs (www.douyin.com/video/<id>) are acrawler-walled;
    # the aweme_id works identically on the mobile share page, so convert.
    if url:
        m2 = re.match(
            r"https?://(?:www\.)?douyin\.com/(?:video|note)/(\d{17,22})(?:$|[?/])",
            url, re.I,
        )
        if m2:
            return "https://www.iesdouyin.com/share/video/%s/" % m2.group(1)
    if url:
        return url
    m = re.search(r"/(?:share/)?(?:video|note)/(\d{17,22})", text)
    if m:
        return "https://www.iesdouyin.com/share/video/%s/" % m.group(1)
    m = _AWEME_ID_RE.search(text)
    if m:
        return "https://www.iesdouyin.com/share/video/%s/" % m.group(1)
    raise DouyinError(
        "could not find a douyin link in the input "
        "(expected a v.douyin.com / douyin.com URL or a numeric video id)"
    )


def _is_final_share_url(url: str) -> bool:
    return bool(re.search(r"(?:share/video|/video|/note)/\d{17,22}", url))


# ---------------------------------------------------------------------------
# httpx plumbing
# ---------------------------------------------------------------------------

# httpx renamed its client kwarg: <0.27 uses proxies=, >=0.27 uses proxy=
_PROXY_KWARG = (
    "proxies"
    if tuple(int(x) for x in httpx.__version__.split(".")[:2]) < (0, 27)
    else "proxy"
)


def make_client(proxy: Optional[str] = None, follow: bool = True,
                cookie: Optional[str] = None) -> httpx.Client:
    kwargs = {}
    if proxy:
        if _PROXY_KWARG == "proxies":  # httpx < 0.27: needs a URL map
            kwargs["proxies"] = {"http://": proxy, "https://": proxy}
        else:
            kwargs["proxy"] = proxy
    headers = {
        "User-Agent": CFG.UA_MOBILE,
        "Accept-Language": "zh-CN,zh;q=0.9",
    }
    if cookie:
        headers["Cookie"] = cookie
    return httpx.Client(
        follow_redirects=follow,
        timeout=httpx.Timeout(CFG.HTTP_TIMEOUT, connect=15.0),
        headers=headers,
        # Never pick up ambient HTTP(S)_PROXY env vars: routing for this
        # service is explicit (DOUYIN_PROXY), which matters when the box
        # also runs xray for other traffic.
        trust_env=False,
        **kwargs,
    )


# ---------------------------------------------------------------------------
# 2. ttwid
# ---------------------------------------------------------------------------

_TTWID_URL = "https://ttwid.bytedance.com/ttwid/union/register/"
_TTWID_TTL = 30 * 24 * 3600  # registered tokens live ~1 year; refresh at 30d


def _ttwid_file() -> str:
    d = CFG.TMP_DIR
    os.makedirs(d, exist_ok=True)
    return os.path.join(d, ".ttwid")


def register_ttwid(proxy: Optional[str] = None) -> Optional[str]:
    """Register a fresh ttwid via the bytedance union register endpoint."""
    body = {
        "region": "cn",
        "aid": 1768,
        "needFid": False,
        "service": "www.douyin.com",
        "migrate_info": {"ticket": "", "source": "node"},
        "cbUrlProtocol": "https",
        "union": True,
    }
    with make_client(proxy) as client:
        r = client.post(
            _TTWID_URL,
            json=body,
            headers={"User-Agent": CFG.UA_MOBILE},
        )
    if r.status_code != 200:
        return None
    m = re.search(r"ttwid=([^;]+)", r.headers.get("set-cookie", ""))
    if not m:
        return None
    from urllib.parse import unquote
    return unquote(m.group(1))


def get_ttwid(proxy: Optional[str] = None) -> Optional[str]:
    """Return a valid ttwid (file-cached), registering a new one if needed."""
    try:
        if os.path.isfile(_ttwid_file()):
            with open(_ttwid_file()) as f:
                raw = f.read().strip()
            if "|" in raw:
                return raw
    except OSError:
        pass
    token = register_ttwid(proxy)
    if token:
        try:
            with open(_ttwid_file(), "w") as f:
                f.write(token)
        except OSError:
            pass
    return token


# ---------------------------------------------------------------------------
# 3. Short link resolution + share page fetch
# ---------------------------------------------------------------------------

def resolve_share_url(url: str) -> Tuple[str, Optional[str]]:
    """Follow redirects (including <a href> interstitials); return
    (final_share_url, aweme_id)."""
    with make_client(CFG.DOUYIN_PROXY) as client:
        try:
            r = client.get(url)
        except httpx.HTTPError as e:
            raise DouyinError("cannot reach douyin short link: %s" % e)
        if r.status_code >= 400:
            raise DouyinError("short link returned HTTP %s" % r.status_code)
        final = str(r.url)
        # some edge responses answer with an HTML <a href> interstitial
        if not _is_final_share_url(final) and r.text:
            m = _HTML_LINK_RE.search(r.text)
            if m and m.group(1).startswith("http"):
                final = m.group(1).replace("&amp;", "&")
        m = re.search(r"(?:video|note)/(\d{17,22})", final) or re.search(
            r"[?&]modal_id=(\d{17,22})", final
        )
        return final, (m.group(1) if m else None)


def fetch_share_page(url: str, ttwid: Optional[str] = None) -> Tuple[
        str, Optional[str], Optional[dict]]:
    """GET the share page; return (html, aweme_id, item_or_None)."""
    if not _is_final_share_url(url):
        url, _ = resolve_share_url(url)
    cookie = ("ttwid=%s" % ttwid) if ttwid else None
    with make_client(CFG.DOUYIN_PROXY, cookie=cookie) as client:
        try:
            r = client.get(url)
        except httpx.HTTPError as e:
            raise DouyinError("cannot fetch share page: %s" % e)
        if r.status_code >= 400:
            raise DouyinError("share page returned HTTP %s" % r.status_code)
    html = r.text
    m = re.search(r"(?:video|note)/(\d{17,22})", url)
    aweme_id = m.group(1) if m else None
    return html, aweme_id, parse_router_data(html)


def parse_router_data(html: str) -> Optional[dict]:
    """Extract videoInfoRes.item_list[0] from page HTML (may be None)."""
    m = re.search(r"window\._ROUTER_DATA\s*=\s*(\{.*?\})\s*</script>", html, re.S)
    if not m:
        return None
    try:
        data = json.loads(m.group(1))
    except ValueError:
        return None
    loader = data.get("loaderData") or {}
    for page in loader.values():
        if not isinstance(page, dict):
            continue
        res = page.get("videoInfoRes")
        if not isinstance(res, dict):
            continue
        items = res.get("item_list") or []
        if items and isinstance(items[0], dict):
            return items[0]
        return {"_empty": True}
    return None


def fetch_item_http(share_url: str) -> Optional[dict]:
    """Return the video item dict (HTTP + ttwid flow), or None when the
    page had no data.  Raises DouyinError for a clearly dead/private video."""
    ttwid = get_ttwid(CFG.DOUYIN_PROXY)
    item = None
    for attempt in range(2):
        _html, _id, item = fetch_share_page(share_url, ttwid)
        if item is not None:
            break
        # no data: rotate the ttwid once and retry
        ttwid = register_ttwid(CFG.DOUYIN_PROXY)
        time.sleep(1)
    if item is None:
        return None
    if item.get("_empty"):
        raise DouyinError(
            "video %s is unavailable (deleted, private or region-restricted)"
            % _id
        )
    return item


# ---------------------------------------------------------------------------
# 4. Video metadata -> candidate URLs
# ---------------------------------------------------------------------------

def _set_ratio(url: str, ratio: str) -> str:
    if "ratio=" in url:
        return re.sub(r"ratio=[^&]*", "ratio=%s" % ratio, url)
    return url + "&ratio=%s" % ratio


def _clean(url: str) -> str:
    # watermark-free flavour of the CDN endpoint
    return url.replace("playwm", "play")


def _strip_logo(url: str) -> str:
    return re.sub(r"[&?]logo_name=[^&]*", "", url)


def _listed_ratio(item: dict) -> Optional[str]:
    play = (item.get("video") or {}).get("play_addr") or {}
    for u in play.get("url_list") or []:
        m = re.search(r"ratio=(\w+p)", u)
        if m:
            return m.group(1)
    return None


def candidate_urls(item: dict) -> List[Tuple[str, str, int]]:
    """Return [(url, label, priority)] for direct watermark-free download.

    Higher priority is tried first; each candidate is verified (HTTP 200 +
    MP4 header) and the biggest valid file wins, so ordering mainly saves
    time on the common happy path.
    """
    out: List[Tuple[str, str, int]] = []
    vid = item.get("video") or {}
    play = vid.get("play_addr") or {}
    uri = play.get("uri") or play.get("url_key") or None
    listed = _listed_ratio(item)

    # a) multi-bitrate list (web detail API provides this)
    for br in vid.get("bit_rate") or []:
        bp = (br or {}).get("play_addr") or {}
        for u in bp.get("url_list") or []:
            out.append((_clean(_strip_logo(u)),
                        "bitrate %s" % br.get("gear_name", "?"), 40))

    # b) play_addr.url_list entries (share-page data)
    for u in play.get("url_list") or []:
        u = _clean(_strip_logo(u))
        out.append((_set_ratio(u, "1080p"), "play 1080p", 25))
        out.append((u, "play orig", 15))

    # c) constructed CDN play endpoints from the raw video uri
    if uri:
        base = "https://aweme.snssdk.com/aweme/v1/play/?video_id=%s&line=0" % uri
        out.append((base + "&ratio=1080p", "api 1080p", 30))
        if listed and listed != "1080p":
            out.append((base + "&ratio=%s" % listed, "api %s" % listed, 28))
        out.append((base, "api default", 10))
        base2 = "https://www.douyin.com/aweme/v1/play/?video_id=%s&line=0" % uri
        out.append((base2 + "&ratio=1080p", "web 1080p", 12))
        if listed and listed != "1080p":
            out.append((base2 + "&ratio=%s" % listed, "web %s" % listed, 11))

    seen = set()
    deduped = []
    for u, label, pr in out:
        if u and u not in seen:
            seen.add(u)
            deduped.append((u, label, pr))
    deduped.sort(key=lambda t: -t[2])
    return deduped


def image_urls(item: dict) -> List[List[str]]:
    """For image (photo post) items: per-image list of URL candidates."""
    result = []
    for img in item.get("images") or []:
        urls = []
        for key in ("url_list", "web_url_list"):
            u = img.get(key)
            if isinstance(u, list):
                urls.extend(u)
            elif isinstance(u, str):
                urls.append(u)
        di = img.get("display_image")
        if isinstance(di, dict):
            urls.extend(di.get("url_list") or [])
        urls = [x for x in urls if x]
        if urls:
            result.append(urls)
    return result


# ---------------------------------------------------------------------------
# 5. Download
# ---------------------------------------------------------------------------

def _looks_like_mp4(head: bytes) -> bool:
    return len(head) >= 12 and head[4:8] == b"ftyp"


def probe_url(client: httpx.Client, url: str) -> Optional[Tuple[int, str]]:
    """Range-probe a CDN URL (fetches only the first 64 KB).

    Returns (total_size, content_type) when the URL serves an MP4, else None.
    """
    try:
        with client.stream(
            "GET",
            url,
            headers={"Range": "bytes=0-65535", "User-Agent": CFG.UA_MOBILE,
                     "Referer": "https://www.iesdouyin.com/"},
            timeout=httpx.Timeout(CFG.DOWNLOAD_TIMEOUT, connect=20.0),
        ) as r:
            if r.status_code not in (200, 206):
                return None
            head = next(r.iter_bytes(16), b"")[:16]
            if not _looks_like_mp4(head):
                return None
            m = re.search(r"/(\d+)\s*$", r.headers.get("content-range") or "")
            total = int(m.group(1)) if m else int(
                r.headers.get("content-length") or 0
            )
            return (total, r.headers.get("content-type") or "video/mp4")
    except (httpx.HTTPError, StopIteration):
        return None


def download_to_file(url: str, path: str, proxy: Optional[str] = None) -> int:
    """Stream ``url`` to ``path`` (following CDN redirects); returns bytes."""
    with make_client(proxy, follow=True) as client:
        client.headers["Referer"] = "https://www.iesdouyin.com/"
        with client.stream(
            "GET",
            url,
            headers={"User-Agent": CFG.UA_MOBILE,
                     "Referer": "https://www.iesdouyin.com/"},
            timeout=httpx.Timeout(CFG.DOWNLOAD_TIMEOUT, connect=20.0),
        ) as r:
            if r.status_code != 200:
                raise DouyinError("download failed: HTTP %s" % r.status_code)
            n = 0
            with open(path, "wb") as f:
                for chunk in r.iter_bytes(1 << 16):
                    f.write(chunk)
                    n += len(chunk)
    return n


def download_bytes(url: str, proxy: Optional[str] = None) -> bytes:
    """Small-file download (images etc.)."""
    with make_client(proxy, follow=True) as client:
        r = client.get(
            url,
            headers={"User-Agent": CFG.UA_MOBILE,
                     "Referer": "https://www.iesdouyin.com/"},
            timeout=httpx.Timeout(CFG.DOWNLOAD_TIMEOUT, connect=20.0),
        )
        if r.status_code != 200:
            raise DouyinError("download failed: HTTP %s" % r.status_code)
        return r.content


def save_video(item: dict, workdir: str) -> dict:
    """Download the best available video for ``item`` (biggest valid MP4)."""
    aweme_id = str(item.get("aweme_id") or "video")
    os.makedirs(workdir, exist_ok=True)
    desc = (item.get("desc") or "").strip().replace("\n", " ")
    title = re.sub(r"[\\/:*?\"<>|\r\n\t#]", "", desc)[:50].strip() or aweme_id
    filename = "%s_%s.mp4" % (aweme_id, title)

    cands = candidate_urls(item)
    if not cands:
        raise DouyinError("no downloadable video URL found in metadata")

    best = None  # (size, url, label, tmp_path)
    with make_client(CFG.DOUYIN_PROXY) as client:
        for url, label, _pr in cands:
            probe = probe_url(client, url)
            if probe is None:
                continue
            tmp = os.path.join(
                workdir, "dl_%s_%s.mp4" % (aweme_id, uuid.uuid4().hex[:8])
            )
            try:
                written = download_to_file(url, tmp, CFG.DOUYIN_PROXY)
            except (DouyinError, OSError):
                continue
            with open(tmp, "rb") as f:
                head = f.read(16)
            if not _looks_like_mp4(head):
                try:
                    os.remove(tmp)
                except OSError:
                    pass
                continue
            if best is None or written > best[0]:
                if best and os.path.exists(best[3]):
                    try:
                        os.remove(best[3])
                    except OSError:
                        pass
                best = (written, url, label, tmp)
            else:
                try:
                    os.remove(tmp)
                except OSError:
                    pass
            # stop after the first 1080p-tier success to save bandwidth
            if "1080p" in label and best and best[0] > 0:
                break

    if best is None:
        raise DouyinError(
            "all video URLs failed (CDN may be geo-restricted; "
            "try running on a mainland-China IP or set DOUYIN_PROXY)"
        )

    final_path = os.path.join(workdir, filename)
    os.replace(best[3], final_path)
    return {
        "aweme_id": aweme_id,
        "filename": filename,
        "path": final_path,
        "size": best[0],
        "url_used": best[1],
        "variant": best[2],
        "kind": "video",
        "meta": video_meta(item),
    }


def save_images(item: dict, workdir: str) -> dict:
    aweme_id = str(item.get("aweme_id") or "photos")
    os.makedirs(workdir, exist_ok=True)
    per_image = image_urls(item)
    if not per_image:
        raise DouyinError("no images found in metadata")
    files = []
    for i, urls in enumerate(per_image, 1):
        data = None
        for u in urls:
            try:
                data = download_bytes(u, CFG.DOUYIN_PROXY)
                if data[:3] == b"\xff\xd8\xff" or data[:4] == b"\x89PNG":
                    break
                data = None
            except DouyinError:
                continue
        if data is None:
            raise DouyinError(
                "failed to download image %d/%d" % (i, len(per_image))
            )
        ext = "jpg" if data[:3] == b"\xff\xd8\xff" else "png"
        p = os.path.join(workdir, "%s_%02d.%s" % (aweme_id, i, ext))
        with open(p, "wb") as f:
            f.write(data)
        files.append(p)
    zip_path = os.path.join(workdir, "%s_photos.zip" % aweme_id)
    with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as z:
        for p in files:
            z.write(p, arcname=os.path.basename(p))
    return {
        "aweme_id": aweme_id,
        "filename": os.path.basename(zip_path),
        "path": zip_path,
        "size": os.path.getsize(zip_path),
        "url_used": "",
        "variant": "%d photos" % len(files),
        "kind": "images",
        "meta": video_meta(item),
    }


def video_meta(item: dict) -> dict:
    vid = item.get("video") or {}
    author = item.get("author") or {}
    music = item.get("music") or {}
    stats = item.get("statistics") or {}
    return {
        "title": (item.get("desc") or "").strip(),
        "author": author.get("nickname") or "",
        "author_id": author.get("unique_id") or "",
        "duration_ms": vid.get("duration"),
        "width": vid.get("width"),
        "height": vid.get("height"),
        "music": music.get("title") or music.get("author") or "",
        "likes": stats.get("digg_count"),
        "play_addr_uri": (vid.get("play_addr") or {}).get("uri"),
    }


# ---------------------------------------------------------------------------
# 6. Top-level pipeline
# ---------------------------------------------------------------------------

def handle(text: str, workdir: str, browser_fetch=None) -> dict:
    """Full pipeline used by the API layer.

    ``browser_fetch(url)`` is an optional callable (the playwright layer)
    used as fallback when the HTTP layer finds no data.
    """
    url = extract_target(text)
    if not _is_final_share_url(url):
        url, aweme_id = resolve_share_url(url)
        if aweme_id and not url.startswith("https://www.iesdouyin.com/share/"):
            # desktop redirect target – convert to the mobile share page
            url = "https://www.iesdouyin.com/share/video/%s/" % aweme_id

    item = None
    try:
        item = fetch_item_http(url)
    except DouyinError as e:
        if "unavailable" not in str(e):
            raise
        item = None  # give the browser layer a chance

    source = "http"
    if item is None and browser_fetch is not None:
        source = "browser"
        item = browser_fetch(url)
    if item is None:
        raise DouyinError(
            "could not read video data from douyin "
            "(deleted / private / region-restricted, or anti-bot change)"
        )
    if item.get("_empty"):
        raise DouyinError(
            "video is unavailable (deleted, private or region-restricted)"
        )

    if item.get("images"):
        result = save_images(item, workdir)
    else:
        result = save_video(item, workdir)
    result["source"] = source
    result["share_url"] = url
    return result


def cleanup_old_files(workdir: str, ttl_hours: float) -> None:
    """Remove downloaded files older than ttl_hours (background janitor)."""
    if not os.path.isdir(workdir):
        return
    cutoff = time.time() - ttl_hours * 3600
    for name in os.listdir(workdir):
        if name.startswith("."):
            continue
        p = os.path.join(workdir, name)
        try:
            if os.path.isfile(p) and os.path.getmtime(p) < cutoff:
                os.remove(p)
        except OSError:
            pass

"""Headless-browser fallback layer (playwright + chromium).

Used when the pure-HTTP layer (ttwid + server-side-rendered share page)
cannot read the video metadata.  A real browser runs douyin's own
JavaScript (secsdk/bdms signing, msToken, reflow params), which makes the
client-side API calls succeed in cases the HTTP layer cannot cover.

Architecture: all playwright calls run inside ONE dedicated worker thread
(playwright's sync API is not thread-safe).  API handlers submit jobs to
its queue and block on a per-job event.

Strategy (in order):
  1. load the share page with a valid ttwid cookie pre-set — the page may
     then server-render the data straight into ``window._ROUTER_DATA``;
  2. capture the page's own iteminfo API response from network traffic;
  3. inject an iteminfo fetch in the page context as a last resort.
"""
import asyncio
import glob
import json
import os
import queue
import re
import sys
import threading
import time
import uuid
from typing import Optional

from .config import CFG

from . import core as _core

try:
    from playwright.sync_api import sync_playwright
except Exception:  # pragma: no cover  (missing / broken playwright)
    sync_playwright = None


class BrowserLayer(object):
    def __init__(self):
        self._worker_lock = threading.Lock()
        self._worker = None
        self._queue = None
        self._events = {}
        self._results = {}
        self._pw = None
        self._browser = None
        self._worker_ready = threading.Event()
        self._worker_error = None
        self._closed = False

    # -- lifecycle ---------------------------------------------------------
    def _ensure_worker(self):
        """Start the dedicated worker thread (first request triggers it)."""
        if sync_playwright is None:
            return False
        with self._worker_lock:
            if self._worker is not None:
                if not self._worker_ready.wait(90):
                    return False
                return self._worker_error is None
            self._queue = queue.Queue()
            t = threading.Thread(target=self._worker_main, name="dy-browser",
                                 daemon=True)
            self._worker = t
            t.start()
            return self._worker_ready.wait(90) and self._worker_error is None

    def _worker_main(self):
        try:
            self._prepare_thread()
            self._pw = sync_playwright().start()
            self._browser = self._launch_browser()
            self._worker_ready.set()
        except Exception as e:  # startup failure — report to callers
            self._worker_error = e
            self._worker_ready.set()
            return
        while True:
            job = self._queue.get()
            if job is None:  # shutdown sentinel
                break
            job_id, fn_name, arg = job
            try:
                result = getattr(self, fn_name)(arg)
                err = None
            except Exception as e:
                result, err = None, e
            self._results[job_id] = (result, err)
            self._events[job_id].set()
        self.close_browser()

    def _submit(self, fn_name: str, arg) -> Optional[object]:
        if not CFG.BROWSER_ENABLED or sync_playwright is None:
            return None
        if not self._ensure_worker():
            return None
        job_id = uuid.uuid4().hex
        ev = threading.Event()
        self._events[job_id] = ev
        self._queue.put((job_id, fn_name, arg))
        # job budget: page timeout + generous slack
        if not ev.wait(timeout=CFG.BROWSER_TIMEOUT * 2 + 120):
            return None
        entry = self._results.pop(job_id, (None, None))
        self._events.pop(job_id, None)
        return entry[0]

    # -- worker-thread prep & browser launch ---------------------------------
    def _prepare_thread(self):
        """Make playwright's sync API usable from this non-main thread.

        On Python 3.7 playwright's start() installs a ThreadedChildWatcher by
        closing the thread's current SafeChildWatcher, whose close() calls
        signal.signal(SIGCHLD) — main-thread only.  Pre-setting a
        ThreadedChildWatcher as the current child watcher makes playwright's
        isinstance(SafeChildWatcher) check fail, so the whole block is skipped.
        Handles both the stock per-thread-local watcher and Kylin's
        policy-level ``_watcher`` attribute.
        """
        if sys.version_info[:2] != (3, 7) or sys.platform == "win32":
            return
        try:
            from playwright.sync_api._py37ThreadedChildWatcher import ThreadedChildWatcher
        except Exception:
            return
        watcher = ThreadedChildWatcher()
        try:
            policy = asyncio.get_event_loop_policy()
        except Exception:
            return
        try:
            policy._local.child_watcher = watcher  # stock CPython 3.7
        except Exception:
            pass
        try:
            policy._watcher = watcher  # Kylin V10 style (shared attribute)
        except Exception:
            pass

    def _chromium_candidates(self):
        cands = []
        if CFG.CHROMIUM_PATH:
            cands.append(CFG.CHROMIUM_PATH)
        base = os.path.expanduser("~/.cache/ms-playwright")
        dirs = sorted(glob.glob(os.path.join(base, "chromium-*")), reverse=True)
        for d in dirs:
            for sub in ("chrome-linux-arm64", "chrome-linux", "chrome-linux64"):
                p = os.path.join(d, sub, "chrome")
                if os.path.isfile(p):
                    cands.append(p)
        for p in (
            "/usr/bin/chromium",
            "/usr/bin/chromium-browser",
            "/usr/bin/google-chrome",
            "/usr/bin/google-chrome-stable",
        ):
            if os.path.isfile(p):
                cands.append(p)
        return cands

    def _launch_browser(self):
        """Launch the first chromium that actually starts.

        Playwright's pinned chromium can be too old for some kernels (e.g.
        64KB-page ARM), so fall back to a newer playwright/system build.
        """
        args = [
            "--no-sandbox",
            "--disable-dev-shm-usage",
            "--disable-blink-features=AutomationControlled",
        ]
        kwargs = {
            "headless": True,
            # must not be True off the main thread (signal registration)
            "handle_sigint": False,
            "handle_sigterm": False,
            "handle_sighup": False,
            "args": args,
        }
        cands = self._chromium_candidates()
        last_err = None
        for p in cands:
            try:
                return self._pw.chromium.launch(executable_path=p, **kwargs)
            except Exception as e:
                last_err = e
        if last_err is not None:
            raise last_err
        return self._pw.chromium.launch(**kwargs)  # let playwright pick

    def close_browser(self):
        if self._browser is not None:
            try:
                self._browser.close()
            except Exception:
                pass
            self._browser = None

    def close(self):
        if self._queue is not None:
            try:
                self._queue.put(None)
            except Exception:
                pass
        if self._pw is not None:
            try:
                self._pw.stop()
            except Exception:
                pass
            self._pw = None

    # -- main entry ---------------------------------------------------------
    def fetch_item(self, share_url: str) -> Optional[dict]:
        """Return the video item dict, or None when nothing could be read."""
        return self._submit("_job_fetch_item", share_url)

    def fetch_web_detail(self, aweme_id: str) -> Optional[dict]:
        return self._submit("_job_fetch_web_detail", aweme_id)

    # -- jobs (run in the worker thread) ------------------------------------
    def _new_context(self, user_agent: str, mobile: bool):
        kwargs = {
            "user_agent": user_agent,
            "locale": "zh-CN",
            "viewport": (
                {"width": 390, "height": 844}
                if mobile
                else {"width": 1366, "height": 900}
            ),
            "extra_http_headers": {"Accept-Language": "zh-CN,zh;q=0.9"},
        }
        if CFG.DOUYIN_PROXY:
            kwargs["proxy"] = {"server": CFG.DOUYIN_PROXY}
        ctx = self._browser.new_context(**kwargs)
        # A valid ttwid makes the page server-render the data into the DOM,
        # which is the single most reliable way to get it (works from any IP).
        try:
            ttwid = _core.get_ttwid(CFG.DOUYIN_PROXY)
            if ttwid:
                cookies = []
                for domain in (".iesdouyin.com", ".douyin.com",
                               ".bytedance.com", ".snssdk.com"):
                    cookies.append({
                        "name": "ttwid",
                        "value": ttwid,
                        "domain": domain,
                        "path": "/",
                    })
                ctx.add_cookies(cookies)
        except Exception:
            pass
        return ctx

    def _job_fetch_item(self, share_url: str) -> Optional[dict]:
        ctx = None
        try:
            ctx = self._new_context(CFG.UA_MOBILE, mobile=True)
            page = ctx.new_page()
            captured = {"iteminfo": None}

            def on_response(resp):
                try:
                    u = resp.url
                except Exception:
                    return
                if "iteminfo" in u:
                    try:
                        j = resp.json()
                    except Exception:
                        return
                    items = (j or {}).get("item_list") if isinstance(j, dict) else None
                    if items and items[0]:
                        captured["iteminfo"] = j

            page.on("response", on_response)
            try:
                page.goto(share_url, wait_until="domcontentloaded",
                          timeout=CFG.BROWSER_TIMEOUT * 1000)
            except Exception:
                pass  # navigation hiccups are fine; we only need the API call

            start = time.time()
            deadline = start + CFG.BROWSER_TIMEOUT
            tried_inject = False
            while time.time() < deadline:
                if captured["iteminfo"]:
                    return self._item_from_response(captured["iteminfo"])
                # data may have been server-rendered into the DOM (ttwid)
                item = self._parse_dom_router_data(page)
                if item:
                    return item
                if not tried_inject and time.time() > start + 12:
                    tried_inject = True
                    j = self._inject_iteminfo_fetch(page)
                    if j:
                        return self._item_from_response(j)
                time.sleep(1)
            return None
        except Exception:
            return None
        finally:
            if ctx is not None:
                try:
                    ctx.close()
                except Exception:
                    pass

    def _item_from_response(self, j: dict) -> Optional[dict]:
        items = j.get("item_list") or []
        if items and items[0]:
            return items[0]
        if j.get("filter_list") or not items:
            return {"_empty": True}
        return None

    def _inject_iteminfo_fetch(self, page) -> Optional[dict]:
        """Run the iteminfo request inside the page context (its cookies/SDK apply)."""
        js = r"""
        async () => {
          let webId = "";
          const m = document.documentElement.innerHTML.match(/"webId":"(\d+)"/);
          if (m) webId = m[1];
          const q = new URLSearchParams(location.search);
          const url = '/web/api/v2/aweme/iteminfo/?reflow_source=reflow_page'
            + '&web_id=' + webId + '&device_id=' + webId
            + '&from_did=' + (q.get('did') || '')
            + '&user_cip='
            + '&use_new_select_scope=0'
            + '&item_ids=' + ((location.pathname.match(/(\d{17,22})/) || [])[1] || '')
            + '&aweme_type=' + (q.get('aweme_type') || '')
            + '&reflow_id='
            + '&share_scene=' + (q.get('share_scene') || '')
            + '&share_token=' + (q.get('share_token') || '')
            + '&scene_from=';
          try {
            const r = await fetch(url, { credentials: 'include',
              headers: { 'Accept': 'application/json, text/plain, */*' } });
            const t = await r.text();
            return t ? JSON.parse(t) : null;
          } catch (e) { return null; }
        }
        """
        try:
            j = page.evaluate(js)
            return j if isinstance(j, dict) else None
        except Exception:
            return None

    def _parse_dom_router_data(self, page) -> Optional[dict]:
        try:
            html = page.content()
        except Exception:
            return None
        m = re.search(r"window\._ROUTER_DATA\s*=\s*(\{.*?\})\s*</script>", html, re.S)
        if not m:
            return None
        try:
            data = json.loads(m.group(1))
        except ValueError:
            return None
        loader = data.get("loaderData") or {}
        for page_data in loader.values():
            if not isinstance(page_data, dict):
                continue
            res = page_data.get("videoInfoRes")
            if not isinstance(res, dict):
                continue
            items = res.get("item_list") or []
            if items and items[0]:
                return items[0]
        return None

    def _job_fetch_web_detail(self, aweme_id: str) -> Optional[dict]:
        """Load www.douyin.com/video/<id> in a desktop context and capture
        the /aweme/v1/web/aweme/detail/ response.  The page's acrawler
        challenge runs for real, so the API call usually succeeds."""
        ctx = None
        try:
            ctx = self._new_context(CFG.UA_DESKTOP, mobile=False)
            page = ctx.new_page()
            captured = {}

            def on_response(resp):
                u = resp.url
                if "aweme/v1/web/aweme/detail/" in u and "aweme_id" in u:
                    try:
                        j = resp.json()
                    except Exception:
                        return
                    if isinstance(j, dict) and j.get("aweme_detail"):
                        captured["detail"] = j["aweme_detail"]

            page.on("response", on_response)
            url = "https://www.douyin.com/video/%s" % aweme_id
            try:
                page.goto(url, wait_until="domcontentloaded",
                          timeout=CFG.BROWSER_TIMEOUT * 1000)
            except Exception:
                pass
            deadline = time.time() + CFG.BROWSER_TIMEOUT
            while time.time() < deadline:
                if "detail" in captured:
                    return captured["detail"]
                time.sleep(1)
            return None
        except Exception:
            return None
        finally:
            if ctx is not None:
                try:
                    ctx.close()
                except Exception:
                    pass


_LAYER = BrowserLayer()


def get_browser_layer() -> BrowserLayer:
    return _LAYER


def browser_fetch_item(share_url: str) -> Optional[dict]:
    """Convenience: try the mobile share flow, then the web detail flow."""
    layer = get_browser_layer()
    item = layer.fetch_item(share_url)
    if item:
        return item
    m = re.search(r"(?:video|note)/(\d{17,22})", share_url)
    if m:
        detail = layer.fetch_web_detail(m.group(1))
        if detail:
            return detail
    return None