all-project/跟教务系统同步学生异动信息/api_service/test_jw_cx_api.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
教务同步只读 API 联调脚本（Python 3.7+，仅标准库）

调用两个业务接口并打印成功结果：
  - GET /api/v1/students          学生基本信息（第 1 页）
  - GET /api/v1/student-changes   学生异动（按年）

用法：
  python test_jw_cx_api.py --prod
  python test_jw_cx_api.py --base-url http://web.hzdyjsxy.cn:18081/jw-cx-api --year 2026

环境变量：JW_CX_BASE_URL、JW_CX_API_KEY
"""
from __future__ import print_function

import argparse
import json
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

HEADER_API_KEY = "X-Jw-Cx-Key"
DEFAULT_BASE_URL = "http://127.0.0.1:5050"
PROD_BASE_URL = "http://web.hzdyjsxy.cn:18081"
DEFAULT_API_KEY = "JWCX001"


def _get_json(url, api_key, timeout=120):
    req = urllib.request.Request(url, method="GET")
    req.add_header(HEADER_API_KEY, api_key)
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        body = resp.read().decode("utf-8")
        return resp.status, json.loads(body)


def _join_url(base, path):
    return base.rstrip("/") + "/" + path.lstrip("/")


def fetch_students(base_url, api_key, page=1, page_size=10):
    qs = urllib.parse.urlencode({"page": page, "page_size": page_size})
    url = _join_url(base_url, "/api/v1/students?" + qs)
    status, data = _get_json(url, api_key)
    if status != 200 or data.get("code") != 200:
        raise RuntimeError("学生基本信息接口失败: status=%s body=%s" % (status, data))
    return data


def fetch_student_changes(base_url, api_key, year):
    qs = urllib.parse.urlencode({"year": year})
    url = _join_url(base_url, "/api/v1/student-changes?" + qs)
    status, data = _get_json(url, api_key)
    if status != 200 or data.get("code") != 200:
        raise RuntimeError("学生异动接口失败: status=%s body=%s" % (status, data))
    return data


def main():
    parser = argparse.ArgumentParser(description="教务同步 API：获取学生信息与学生异动")
    parser.add_argument("--base-url", default=None, help="API 根地址")
    parser.add_argument("--prod", action="store_true", help="使用 %s" % PROD_BASE_URL)
    parser.add_argument(
        "--api-key",
        default=os.environ.get("JW_CX_API_KEY", DEFAULT_API_KEY),
        help="请求头 %s" % HEADER_API_KEY,
    )
    parser.add_argument(
        "--year",
        type=int,
        default=None,
        help="异动筛选年份，默认当前自然年",
    )
    parser.add_argument(
        "--page-size",
        type=int,
        default=10,
        help="学生基本信息每页条数，默认 10",
    )
    args = parser.parse_args()

    if args.base_url:
        base_url = args.base_url
    elif args.prod:
        base_url = PROD_BASE_URL
    else:
        base_url = os.environ.get("JW_CX_BASE_URL", DEFAULT_BASE_URL)

    year = args.year if args.year is not None else time.localtime().tm_year

    print("BASE_URL: %s" % base_url)
    print("")

    try:
        students = fetch_students(base_url, args.api_key, page=1, page_size=args.page_size)
        print("=== 学生基本信息 GET /api/v1/students ===")
        print(
            "total=%s page=%s/%s 本页=%s条"
            % (
                students.get("total"),
                students.get("page"),
                students.get("total_pages"),
                len(students.get("data") or []),
            )
        )
        if students.get("data"):
            print("首条:", json.dumps(students["data"][0], ensure_ascii=False, indent=2))
        print("")

        changes = fetch_student_changes(base_url, args.api_key, year)
        print("=== 学生异动 GET /api/v1/student-changes?year=%s ===" % year)
        print("total=%s filter_year=%s" % (changes.get("total"), changes.get("filter_year")))
        if changes.get("data"):
            print("首条:", json.dumps(changes["data"][0], ensure_ascii=False, indent=2))
        elif changes.get("total") == 0:
            print("(当年无异动记录)")

    except (urllib.error.URLError, urllib.error.HTTPError, RuntimeError, ValueError) as e:
        print("调用失败: %s" % e, file=sys.stderr)
        sys.exit(1)

    print("")
    print("调用成功")
    sys.exit(0)


if __name__ == "__main__":
    main()