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

# -*- coding: utf-8 -*-
"""
教务同步：学生异动视图只读 API（Python 3.7+）

配置：默认读取与本文件同目录下的 config.ini（INI 格式），路径可用环境变量 JW_CX_CONFIG 覆盖。
校验：请求头 X-Jw-Cx-Key 须与配置文件 [api] api_key 一致。
"""
import hmac
import os
import sys
from datetime import date, datetime
from decimal import Decimal

import configparser

import pymysql
from flask import Flask, jsonify, request
from pymysql.cursors import DictCursor

# 与调用方约定的 Header 名（勿随意改动，已写入 README）
HEADER_API_KEY = "X-Jw-Cx-Key"

VIEW_NAME = "vw_jw_student_change_export"

# 视图列名（与 vw_jw_student_change_export 一致；筛选近 N 天时使用）
COL_SUBMIT_DATE = "提交时间"

_BASE_DIR = os.path.dirname(os.path.abspath(__file__))
_DEFAULT_CONFIG_PATH = os.path.join(_BASE_DIR, "config.ini")

app = Flask(__name__)

_config_parser = None


def _config_path():
    return os.environ.get("JW_CX_CONFIG", _DEFAULT_CONFIG_PATH)


def _load_config():
    """加载并缓存 INI 配置；缺失文件时给出明确提示。"""
    global _config_parser
    if _config_parser is not None:
        return _config_parser

    path = _config_path()
    if not os.path.isfile(path):
        sys.stderr.write(
            "错误：未找到配置文件 %s\n"
            "请复制 config.ini.example 为 config.ini 并填写数据库等信息，"
            "或设置环境变量 JW_CX_CONFIG 指向您的配置文件路径。\n" % (path,)
        )
        sys.exit(1)

    parser = configparser.ConfigParser()
    # Python 3.2+：read 支持 encoding
    read_ok = parser.read(path, encoding="utf-8")
    if not read_ok:
        sys.stderr.write("错误：无法读取配置文件 %s\n" % (path,))
        sys.exit(1)

    _config_parser = parser
    return _config_parser


def _get(section, option, fallback=""):
    parser = _load_config()
    if parser.has_option(section, option):
        return parser.get(section, option).strip()
    return fallback


def _get_int(section, option, fallback):
    raw = _get(section, option, str(fallback))
    try:
        return int(raw)
    except ValueError:
        return int(fallback)


def configured_api_key():
    return _get("api", "api_key", "JWCX001")


def db_config():
    return {
        "host": _get("database", "host", "127.0.0.1"),
        "port": _get_int("database", "port", 3306),
        "user": _get("database", "user", "jw_cx"),
        "password": _get("database", "password", ""),
        "database": _get("database", "database", "hzdyjs_xg"),
    }


def server_bind():
    return {
        "host": _get("api", "bind_host", "0.0.0.0"),
        "port": _get_int("api", "bind_port", 5050),
    }


def _api_defaults_from_config():
    """从 [api] 读取 page_size、recent_days 默认值（INI 见 config*.ini）。"""
    page_size = _get_int("api", "page_size", 500)
    recent_days = _get_int("api", "recent_days", 30)
    if page_size < 1:
        page_size = 1
    if page_size > 5000:
        page_size = 5000
    if recent_days < 1:
        recent_days = 1
    if recent_days > 3660:
        recent_days = 3660
    return page_size, recent_days


def _clamp_int(value, lo, hi):
    v = int(value)
    if v < lo:
        return lo
    if v > hi:
        return hi
    return v


def _resolve_query_paging():
    """
    合并 URL 参数与配置文件：
    - page_size：Query `page_size`，未传则用 [api] page_size
    - recent_days：Query `recent_days` 或 `days`（别名），未传则用 [api] recent_days
    范围：page_size 1～5000，recent_days 1～3660
    """
    cfg_ps, cfg_rd = _api_defaults_from_config()

    raw_ps = request.args.get("page_size")
    if raw_ps is not None and str(raw_ps).strip() != "":
        try:
            page_size = _clamp_int(raw_ps, 1, 5000)
        except (ValueError, TypeError):
            return None, None, "page_size 必须为整数"
    else:
        page_size = cfg_ps

    raw_rd = request.args.get("recent_days")
    if raw_rd is None or str(raw_rd).strip() == "":
        raw_rd = request.args.get("days")
    if raw_rd is not None and str(raw_rd).strip() != "":
        try:
            recent_days = _clamp_int(raw_rd, 1, 3660)
        except (ValueError, TypeError):
            return None, None, "recent_days / days 必须为整数"
    else:
        recent_days = cfg_rd

    return page_size, recent_days, None


def _check_api_key():
    sent = request.headers.get(HEADER_API_KEY, "")
    expected = configured_api_key()
    if not expected:
        return False
    try:
        return hmac.compare_digest(sent.encode("utf-8"), expected.encode("utf-8"))
    except Exception:
        return False


def _serialize_value(value):
    if value is None:
        return None
    if isinstance(value, datetime):
        return value.strftime("%Y-%m-%d %H:%M:%S")
    if isinstance(value, date):
        return value.strftime("%Y-%m-%d")
    if isinstance(value, Decimal):
        return str(value)
    return value


def _row_to_dict(row):
    return {k: _serialize_value(v) for k, v in row.items()}


def _db_connect():
    cfg = db_config()
    return pymysql.connect(
        host=cfg["host"],
        port=cfg["port"],
        user=cfg["user"],
        password=cfg["password"],
        database=cfg["database"],
        charset="utf8mb4",
        cursorclass=DictCursor,
        autocommit=True,
    )


@app.route("/health", methods=["GET"])
def health():
    """存活探测，不做密钥校验，便于负载均衡 / 运维。"""
    return jsonify({"status": "ok"})


@app.route("/api/v1/student-changes", methods=["GET"])
def list_student_changes():
    if not _check_api_key():
        return jsonify({"code": 401, "msg": "无效的密钥或未提供请求头 " + HEADER_API_KEY}), 401

    try:
        page = int(request.args.get("page", "1"))
    except ValueError:
        return jsonify({"code": 400, "msg": "page 必须为整数"}), 400

    if page < 1:
        page = 1

    page_size, recent_days, param_err = _resolve_query_paging()
    if param_err:
        return jsonify({"code": 400, "msg": param_err}), 400

    offset = (page - 1) * page_size

    # 近 recent_days 天：按视图中的「提交时间」筛选（登记/提交日期）
    where_recent = (
        "WHERE `{col}` >= DATE_SUB(CURDATE(), INTERVAL {days} DAY)".format(
            col=COL_SUBMIT_DATE, days=recent_days
        )
    )
    sql_count = "SELECT COUNT(*) AS cnt FROM `{view}` {where_clause}".format(
        view=VIEW_NAME, where_clause=where_recent
    )
    sql_list = (
        "SELECT * FROM `{view}` {where_clause} "
        "ORDER BY `来源记录id` ASC LIMIT %s OFFSET %s"
    ).format(view=VIEW_NAME, where_clause=where_recent)

    conn = None
    try:
        conn = _db_connect()
        with conn.cursor() as cur:
            cur.execute(sql_count)
            total = int(cur.fetchone()["cnt"])
            cur.execute(sql_list, (page_size, offset))
            rows = cur.fetchall()
    except pymysql.Error as e:
        return jsonify({"code": 500, "msg": "数据库查询失败", "detail": str(e)}), 500
    finally:
        if conn is not None:
            conn.close()

    data = [_row_to_dict(r) for r in rows]
    return jsonify(
        {
            "code": 200,
            "msg": "success",
            "data": data,
            "page": page,
            "page_size": page_size,
            "total": total,
            "filter_recent_days": recent_days,
            "filter_by_column": COL_SUBMIT_DATE,
        }
    )


if __name__ == "__main__":
    _load_config()
    bind = server_bind()
    app.run(host=bind["host"], port=bind["port"], threaded=True)