271 lines
7.6 KiB
Python
271 lines
7.6 KiB
Python
# -*- 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)
|