抓取 API 接入文档

"""MIYAIP crawler example. Python 3.8+, standard library only.

Set MIYA_CONSOLE_TOKEN, then run:
  python crawler_api.py INTERFACE_KEY params.json
Inspect available interfaces first with the cURL request in the public docs.
"""
import argparse
import json
import math
import os
import sys
import time
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode, urlsplit
from urllib.request import HTTPRedirectHandler, Request, build_opener


class NoRedirect(HTTPRedirectHandler):
    # Do not forward login credentials to a redirected host.
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


def field(value, name, default=None):
    return value.get(name, value.get(name[0].upper() + name[1:], default))


def request(base, token, path, data=None, timeout=30):
    req = Request(base + path, headers={"Authorization": "Bearer " + token})
    if data is not None:
        req.data = json.dumps(data).encode("utf-8")
        req.add_header("Content-Type", "application/json")
    try:
        with build_opener(NoRedirect()).open(req, timeout=timeout) as response:
            payload = json.load(response)
    except HTTPError as error:
        raise RuntimeError("HTTP %s; check authentication (401), permissions or service availability. Do not blindly resubmit." % error.code) from None
    except (URLError, TimeoutError, OSError):
        raise RuntimeError("Network error; check call logs before resubmitting an uncertain creation request.") from None
    except (ValueError, UnicodeError):
        raise RuntimeError("Invalid JSON response") from None
    if isinstance(payload, dict):
        if "code" in payload and (type(payload["code"]) is not int or payload["code"] not in (0, 200)):
            raise RuntimeError("Business error %s: %s" % (payload.get("code"), payload.get("message", "Request failed")))
        return payload.get("body", payload.get("Body", payload))
    return payload


def run(base, token, interface_key, params, wait_seconds=600, interval=5):
    parsed = urlsplit(base)
    if parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in ("", "/"):
        raise ValueError("Base URL must be an origin without credentials, path or query")
    if parsed.scheme != "https" and not (parsed.scheme == "http" and parsed.hostname in ("localhost", "127.0.0.1")):
        raise ValueError("HTTPS is required (HTTP loopback is allowed for local tests)")
    if not token.strip() or not interface_key.strip() or not isinstance(params, dict):
        raise ValueError("Token, interface key and a JSON parameter object are required")
    if not math.isfinite(wait_seconds) or not math.isfinite(interval) or wait_seconds <= 0 or interval <= 0:
        raise ValueError("Wait and interval must be finite and positive")
    base = base.rstrip("/")
    mapping = request(base, token, "/api/CrawlerInterface/Detail?" + urlencode({"interfaceKey": interface_key}))
    if not isinstance(mapping, dict) or field(mapping, "interfaceKey") != interface_key:
        raise RuntimeError("Interface definition missing or mismatched")
    if mapping.get("isEnabled") is False or mapping.get("IsEnabled") is False:
        raise RuntimeError("Interface is disabled")
    # Params must follow the returned ParamSchema. The server validates the contract.
    created = request(base, token, "/api/Crawler/Invoke", {"InterfaceKey": interface_key, "Params": params})
    task_id = field(created, "taskId") if isinstance(created, dict) else None
    if not isinstance(task_id, str) or not task_id.strip():
        raise RuntimeError("Missing taskId; inspect call logs before resubmitting")
    print("Task ID: " + task_id, file=sys.stderr)
    deadline = time.monotonic() + wait_seconds
    while time.monotonic() < deadline:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            break
        result = request(base, token, "/api/Crawler/TaskResult?" + urlencode({"taskId": task_id}), timeout=min(30, remaining))
        if not isinstance(result, dict) or not isinstance(field(result, "status"), str):
            raise RuntimeError("Invalid task response; query the existing task ID later")
        status = field(result, "status").strip().lower()
        if status in ("success", "succeeded", "completed", "complete"):
            return result
        if status in ("failed", "fail", "error", "timeout", "timedout", "time_out"):
            raise RuntimeError("Task %s: %s" % (status, field(result, "error", "No error detail")))
        if status != "running":
            raise RuntimeError("Unknown task status: " + status)
        time.sleep(min(interval, max(0, deadline - time.monotonic())))
    raise RuntimeError("Waiting timed out. Task %s may still be running; check TaskResult or call logs. No new task was submitted." % task_id)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("interface_key")
    parser.add_argument("params_file")
    parser.add_argument("--wait-seconds", type=float, default=600)
    args = parser.parse_args()
    try:
        with open(args.params_file, encoding="utf-8") as source:
            params = json.load(source)
        result = run("https://miyaip.com", os.environ.get("MIYA_CONSOLE_TOKEN", ""), args.interface_key, params, args.wait_seconds)
        print(json.dumps(result, ensure_ascii=False, indent=2))
    except (OSError, ValueError, RuntimeError) as error:
        print(str(error), file=sys.stderr)
        return 1
    return 0


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