256 lines
6.9 KiB
Python
256 lines
6.9 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"
|
||
|
||
# 视图列名：API 按「异动生效时间」所在自然年筛选
|
||
COL_EFFECTIVE_TIME = "异动生效时间"
|
||
|
||
# 对外返回字段（与教务对接清单一致；不含视图内部列）
|
||
EXPORT_COLUMNS = (
|
||
"提交时间",
|
||
"异动生效学期",
|
||
"异动类型",
|
||
"申请人学号",
|
||
"姓名",
|
||
"异动前所在年级",
|
||
"异动前所在校区",
|
||
"异动前所在院系",
|
||
"异动前所在专业",
|
||
"异动前所在班级",
|
||
"异动前学制",
|
||
"异动后所在校区",
|
||
"异动后所在年级",
|
||
"异动后所在班级",
|
||
"异动后所在院系",
|
||
"异动后所在专业",
|
||
"异动后学制",
|
||
"异动后班主任工号",
|
||
"异动后班主任姓名",
|
||
"新名字",
|
||
"审核状态",
|
||
)
|
||
|
||
_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 _default_filter_year():
|
||
"""配置文件 default_year；未配置或无效时使用当前自然年。"""
|
||
raw = _get("api", "default_year", "")
|
||
if raw:
|
||
try:
|
||
year = int(raw)
|
||
if 2000 <= year <= 2100:
|
||
return year
|
||
except ValueError:
|
||
pass
|
||
return date.today().year
|
||
|
||
|
||
def _resolve_filter_year():
|
||
"""
|
||
合并 URL 参数与配置文件：
|
||
- year：Query `year`，未传则用 [api] default_year 或当前自然年
|
||
"""
|
||
default_year = _default_filter_year()
|
||
raw_year = request.args.get("year")
|
||
if raw_year is not None and str(raw_year).strip() != "":
|
||
try:
|
||
year = int(raw_year)
|
||
except (ValueError, TypeError):
|
||
return None, "year 必须为整数"
|
||
if year < 2000 or year > 2100:
|
||
return None, "year 有效范围为 2000～2100"
|
||
else:
|
||
year = default_year
|
||
|
||
return year, 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_export_dict(row):
|
||
return {col: _serialize_value(row.get(col)) for col in EXPORT_COLUMNS}
|
||
|
||
|
||
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
|
||
|
||
filter_year, param_err = _resolve_filter_year()
|
||
if param_err:
|
||
return jsonify({"code": 400, "msg": param_err}), 400
|
||
|
||
select_cols = ", ".join("`%s`" % col for col in EXPORT_COLUMNS)
|
||
where_year = (
|
||
"WHERE `{col}` IS NOT NULL AND YEAR(`{col}`) = %s".format(
|
||
col=COL_EFFECTIVE_TIME
|
||
)
|
||
)
|
||
sql_list = (
|
||
"SELECT {cols} FROM `{view}` {where_clause} "
|
||
"ORDER BY `来源记录id` ASC"
|
||
).format(cols=select_cols, view=VIEW_NAME, where_clause=where_year)
|
||
|
||
conn = None
|
||
try:
|
||
conn = _db_connect()
|
||
with conn.cursor() as cur:
|
||
cur.execute(sql_list, (filter_year,))
|
||
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_export_dict(r) for r in rows]
|
||
return jsonify(
|
||
{
|
||
"code": 200,
|
||
"msg": "success",
|
||
"data": data,
|
||
"total": len(data),
|
||
"filter_year": filter_year,
|
||
"filter_by_column": COL_EFFECTIVE_TIME,
|
||
}
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
_load_config()
|
||
bind = server_bind()
|
||
app.run(host=bind["host"], port=bind["port"], threaded=True)
|