test_ifind_http.py

import json
import sys
import urllib.error
import urllib.request


def post_json(url, headers, payload=None, timeout=30):
    data = None if payload is None else json.dumps(payload).encode("utf-8")
    req = urllib.request.Request(url, data=data, method="POST")
    for key, value in headers.items():
        req.add_header(key, value)
    with urllib.request.urlopen(req, timeout=timeout) as response:
        body = response.read()
        return response.status, body


def load_json(body):
    text = body.decode("utf-8", "replace")
    try:
        return json.loads(text), text
    except json.JSONDecodeError:
        return None, text


def main():
    refresh_token = sys.stdin.read().strip()
    if not refresh_token:
        print("missing refresh token")
        return 2

    try:
        status, body = post_json(
            "https://quantapi.51ifind.com/api/v1/get_access_token",
            {"Content-Type": "application/json", "refresh_token": refresh_token},
        )
    except urllib.error.HTTPError as exc:
        error_body = exc.read().decode("utf-8", "replace")
        print(f"access_token_http_error={exc.code}")
        print(error_body[:1000])
        return 1
    except Exception as exc:
        print(f"access_token_exception={type(exc).__name__}: {exc}")
        return 1

    obj, text = load_json(body)
    print(f"access_token_status={status}")
    if obj is None:
        print(text[:1000])
        return 1

    access_token = (obj.get("data") or {}).get("access_token")
    safe_obj = json.loads(json.dumps(obj, ensure_ascii=False))
    if isinstance(safe_obj.get("data"), dict) and "access_token" in safe_obj["data"]:
        safe_obj["data"]["access_token"] = "<redacted>"
    print(json.dumps(safe_obj, ensure_ascii=False, indent=2)[:2000])

    if not access_token:
        return 1

    try:
        status, body = post_json(
            "https://quantapi.51ifind.com/api/v1/real_time_quotation",
            {"Content-Type": "application/json", "access_token": access_token},
            {"codes": "300033.SZ", "indicators": "latest"},
        )
    except urllib.error.HTTPError as exc:
        error_body = exc.read().decode("utf-8", "replace")
        print(f"quotation_http_error={exc.code}")
        print(error_body[:1000])
        return 1
    except Exception as exc:
        print(f"quotation_exception={type(exc).__name__}: {exc}")
        return 1

    obj, text = load_json(body)
    print(f"quotation_status={status}")
    if obj is None:
        print(text[:2000])
        return 1
    print(json.dumps(obj, ensure_ascii=False, indent=2)[:4000])
    return 0


if __name__ == "__main__":
    raise SystemExit(main())