微博数据怎么采集？按时间搜索帖子、获取详情与评论

python3 -m pip install requests
export JUSTONEAPI_TOKEN='your_token'
export WEIBO_START_DAY='2026-10-01'
export WEIBO_END_DAY='2026-10-07'

import json
import os
from datetime import datetime, timezone
from pathlib import Path

import requests

BASE_URL = "https://api.justoneapi.com"
TOKEN = os.environ.get("JUSTONEAPI_TOKEN", "")


def call_api(path, **params):
    try:
        response = requests.get(
            BASE_URL + path,
            params={"token": TOKEN, **params},
            timeout=120, allow_redirects=False,
        )
    except requests.RequestException:
        raise RuntimeError("API network error or timeout") from None
    if response.status_code != 200:
        raise RuntimeError(f"API HTTP {response.status_code}")
    try:
        result = response.json()
    except ValueError:
        raise RuntimeError("API response is not JSON") from None
    if not isinstance(result, dict) or result.get("code") != 0:
        code = result.get("code") if isinstance(result, dict) else None
        raise RuntimeError(f"API failed: code={code}; check the docs")
    data = result.get("data")
    if not isinstance(data, dict):
        raise RuntimeError("Unexpected data shape; inspect the docs")
    inner = data.get("base_resp")
    if isinstance(inner, dict) and inner.get("status_code") not in (0, "0"):
        raise RuntimeError("Platform response was unsuccessful")
    return result


def save_json(folder, name, value):
    (folder / name).write_text(
        json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8"
    )


def require_list(data, key):
    value = data.get(key)
    if not isinstance(value, list):
        raise RuntimeError(f"Expected a list at {key}; inspect saved JSON")
    return value


def output_folder(topic):
    if not TOKEN or TOKEN == "your_token":
        raise RuntimeError("Set JUSTONEAPI_TOKEN before running")
    run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
    folder = Path(topic + "-output") / run_id
    folder.mkdir(parents=True, exist_ok=False)
    return folder


KEYWORD = "露营"
START_DAY = os.environ.get("WEIBO_START_DAY", "")
END_DAY = os.environ.get("WEIBO_END_DAY", "")
MAX_SEARCH_PAGES = 2
MAX_POSTS = 3
MAX_COMMENT_PAGES = 2


def main():
    try:
        start = datetime.strptime(START_DAY, "%Y-%m-%d")
        end = datetime.strptime(END_DAY, "%Y-%m-%d")
    except ValueError:
        raise RuntimeError("Set WEIBO_START_DAY and WEIBO_END_DAY: YYYY-MM-DD") from None
    if start > end:
        raise RuntimeError("Start date must not be after end date")
    folder = output_folder("weibo")
    ids, seen = [], set()
    for page in range(1, MAX_SEARCH_PAGES + 1):
        result = call_api(
            "/api/weibo/search-all/v2", q=KEYWORD,
            startDay=START_DAY, startHour=0, endDay=END_DAY, endHour=23,
            hotSort="false", contains="ALL", page=page,
        )
        save_json(folder, f"search-{page}.json", result)
        data = result["data"]
        rows = require_list(data, "weibo_list")
        if not rows:
            break
        previous_count = len(ids)
        for row in rows:
            post_id = row.get("id") if isinstance(row, dict) else None
            if post_id and str(post_id) not in seen:
                seen.add(str(post_id))
                ids.append(str(post_id))
        if len(ids) == previous_count:
            break
        if data.get("has_next_page") not in (True, 1, "1"):
            break

    selected = ids[:MAX_POSTS]
    save_json(folder, "selected-posts.json", selected)
    for index, post_id in enumerate(selected, start=1):
        detail = call_api("/api/weibo/get-weibo-detail/v1", id=post_id)
        save_json(folder, f"post-{index}-detail.json", detail)
        status = detail["data"].get("status")
        mid = status.get("mid") if isinstance(status, dict) else None
        if not mid:
            raise RuntimeError("Missing post MID; inspect the saved detail")
        cursor, cursors, comments, comment_ids = "", set(), [], set()
        for page in range(1, MAX_COMMENT_PAGES + 1):
            params = {"mid": str(mid), "sort": "TIME"}
            if cursor:
                params["maxId"] = cursor
            result = call_api("/api/weibo/get-post-comments/v1", **params)
            save_json(folder, f"post-{index}-comments-{page}.json", result)
            data = result["data"]
            if data.get("ok") not in (1, "1"):
                raise RuntimeError("Comment response was unsuccessful")
            rows = require_list(data, "data")
            if not rows:
                break
            for row in rows:
                if not isinstance(row, dict):
                    raise RuntimeError("Unexpected comment item")
                cid = row.get("idstr") or row.get("id")
                if cid is None:
                    raise RuntimeError("Missing comment ID; inspect saved JSON")
                if str(cid) not in comment_ids:
                    comment_ids.add(str(cid))
                    comments.append(row)
            next_cursor = data.get("max_id")
            if next_cursor in (None, "", 0, "0"):
                break
            cursor = str(next_cursor)
            if cursor in cursors:
                break
            cursors.add(cursor)
        save_json(folder, f"post-{index}-comments.json", comments)
    print(f"Saved {len(selected)} post samples to {folder}")


if __name__ == "__main__":
    try:
        main()
    except RuntimeError as error:
        raise SystemExit(str(error)) from None