content-find-agent-new/scripts/smoke_douyin_blogger.py

"""Manual live smoke for the Crawapi douyin blogger contract (V2-M5D).

Hits POST <CONTENTFIND_API_CRAWAPI_BASE_URL><CONTENTFIND_DOUYIN_BLOGGER_PATH>
with the fixed three-field payload (account_id / sort_type / cursor) and prints
a redacted summary only — never the raw response, never any credential.

Not part of default pytest. Run manually:

    uv run python scripts/smoke_douyin_blogger.py --author-id '<sec_uid>'
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

import httpx

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

from content_agent.integrations.douyin import _env, _load_env_file  # noqa: E402


def main() -> int:
    args = _parse_args()
    env = _load_env_file(args.env_file)
    base_url = _env("CONTENTFIND_API_CRAWAPI_BASE_URL", env, required=True)
    blogger_path = _env("CONTENTFIND_DOUYIN_BLOGGER_PATH", env, required=True)
    sort_type = args.sort_type or _env(
        "CONTENTFIND_DOUYIN_ACCOUNT_WORKS_DEFAULT_SORT_TYPE", env, default="最新"
    )

    url = base_url.rstrip("/") + "/" + blogger_path.lstrip("/")
    payload = {"account_id": args.author_id, "sort_type": sort_type, "cursor": args.cursor}
    response = httpx.post(
        url, json=payload, headers={"Content-Type": "application/json"}, timeout=60.0
    )
    try:
        data = response.json()
    except ValueError:
        data = {}
    if not isinstance(data, dict):
        data = {}
    data_block = data.get("data") if isinstance(data.get("data"), dict) else {}
    items = data_block.get("data") if isinstance(data_block.get("data"), list) else []

    summary = {
        "endpoint": "/" + blogger_path.lstrip("/"),
        "http_status": response.status_code,
        "business_code": data.get("code"),
        "result_count": len(items),
        "has_more": bool(data_block.get("has_more", False)),
        "next_cursor_present": bool(data_block.get("next_cursor")),
    }
    print(json.dumps(summary, ensure_ascii=False, indent=2))
    ok = (
        summary["http_status"] == 200
        and summary["business_code"] in (0, "0")
        and summary["result_count"] > 0
    )
    return 0 if ok else 1


def _parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--author-id", required=True, help="author platform_author_id (sec_uid)")
    parser.add_argument("--sort-type", default=None, help="override sort_type (default from env)")
    parser.add_argument("--cursor", default="", help="page cursor, empty for first page")
    parser.add_argument("--env-file", default=str(ROOT / ".env"))
    return parser.parse_args()


if __name__ == "__main__":
    sys.exit(main())