# -*- coding: utf-8 -*- """ hr-roster 花名册字段体系 (F03) 后端服务 ======================================== - 端口 9280 (ocai 模式) - 数据访问: sqlor - 权限: rbac - 写操作审计: roster_op_log 落库 端点清单 (18 个): GET /api/roster/field-groups 字段分组列表 POST /api/roster/field-groups 新增字段分组 PUT /api/roster/field-groups/ 更新字段分组 DELETE /api/roster/field-groups/ 删除字段分组 GET /api/roster/field-defs 字段定义列表(支持分组筛选) POST /api/roster/field-defs 新增字段定义 PUT /api/roster/field-defs/ 更新字段定义 DELETE /api/roster/field-defs/ 删除字段定义 POST /api/roster/field-defs/sort 字段拖拽排序(批量更新sort_order) POST /api/roster/field-defs/toggle 字段启停 GET /api/roster/type-rules 员工类型规则列表 POST /api/roster/type-rules 新增员工类型规则 PUT /api/roster/type-rules/ 更新员工类型规则 DELETE /api/roster/type-rules/ 删除员工类型规则 GET /api/roster/empno-rules 工号规则列表 POST /api/roster/empno-rules 新增工号规则 PUT /api/roster/empno-rules/ 更新工号规则 DELETE /api/roster/empno-rules/ 删除工号规则 POST /api/roster/empno-rules/generate 生成工号(冲突校验) GET /api/roster/options/ 字段选项端点(下拉联动) """ from __future__ import annotations import json import re import time from datetime import datetime # 基础模块引用 (已存在, 直接使用) from appbase import get_config # 配置 from apppublic import logger, now # 工具 from ahserver import http_router, json_response, request # HTTP 框架 from sqlor import db # 数据访问 # --------------------------------------------------------------------------- # 常量 # --------------------------------------------------------------------------- APP_PORT = 9280 OP_TYPE_CREATE = "create" OP_TYPE_UPDATE = "update" OP_TYPE_DELETE = "delete" OP_TYPE_ENABLE = "enable" OP_TYPE_DISABLE = "disable" OP_TYPE_SORT = "sort" OP_TYPE_GEN_EMPNO = "gen_empno" TABLE_GROUP = "roster_field_group" TABLE_DEF = "roster_field_def" TABLE_TYPE_RULE = "roster_type_rule" TABLE_EMPNO_RULE = "roster_empno_rule" TABLE_OP_LOG = "roster_op_log" ALLOWED_FIELD_TYPES = { "text", "number", "date", "datetime", "select", "multi_select", "attachment", "bool", } # --------------------------------------------------------------------------- # 审计 # --------------------------------------------------------------------------- def audit(op_type: str, target_table: str, target_id, before=None, after=None, operator_id=None, operator_name=None, client_ip=None, remark=None): """写操作审计落库""" try: db.insert(TABLE_OP_LOG, { "op_type": op_type, "target_table": target_table, "target_id": target_id, "before_json": json.dumps(before, ensure_ascii=False) if before is not None else None, "after_json": json.dumps(after, ensure_ascii=False) if after is not None else None, "operator_id": operator_id, "operator_name": operator_name, "client_ip": client_ip, "remark": remark, "created_at": now(), }) except Exception as exc: # 审计失败不阻断主流程 logger.error("audit fail: %s", exc) def _operator(): """从请求上下文取操作人信息(由 rbac/ahserver 注入)""" ctx = getattr(request, "ctx", {}) or {} return { "operator_id": ctx.get("user_id"), "operator_name": ctx.get("user_name"), "client_ip": ctx.get("client_ip"), } # --------------------------------------------------------------------------- # 通用校验 # --------------------------------------------------------------------------- def _require(cond: bool, msg: str): if not cond: raise ValueError(msg) def _valid_field_code(code): return bool(re.match(r"^[a-zA-Z][a-zA-Z0-9_]{1,63}$", code or "")) def _parse_options(options): """选项 JSON 校验, 返回 list[dict]""" if options is None: return None if isinstance(options, str): options = json.loads(options) if not isinstance(options, list): raise ValueError("options 必须是数组") for item in options: if not isinstance(item, dict) or "label" not in item or "value" not in item: raise ValueError("options 每项必须含 label/value") return options # --------------------------------------------------------------------------- # 1. 字段分组 # --------------------------------------------------------------------------- @http_router.get("/api/roster/field-groups") def list_field_groups(): rows = db.select( "SELECT * FROM roster_field_group " "WHERE deleted_at IS NULL ORDER BY sort_order ASC, id ASC" ) return json_response({"code": 0, "data": rows}) @http_router.post("/api/roster/field-groups") def create_field_group(): body = request.json() _require(body.get("group_code"), "group_code 必填") _require(body.get("group_name"), "group_name 必填") _require(_valid_field_code(body["group_code"]), "group_code 非法") exists = db.select_one( "SELECT id FROM roster_field_group WHERE group_code=%s AND deleted_at IS NULL", (body["group_code"],)) _require(not exists, "分组编码已存在") data = { "group_code": body["group_code"], "group_name": body["group_name"], "group_desc": body.get("group_desc"), "is_builtin": int(bool(body.get("is_builtin", 0))), "sort_order": int(body.get("sort_order", 0)), "is_enabled": int(bool(body.get("is_enabled", 1))), "created_by": _operator()["operator_id"], "created_at": now(), } new_id = db.insert(TABLE_GROUP, data) audit(OP_TYPE_CREATE, TABLE_GROUP, new_id, None, data, **_operator()) return json_response({"code": 0, "data": {"id": new_id}}) @http_router.put("/api/roster/field-groups/") def update_field_group(gid): body = request.json() before = db.select_one( "SELECT * FROM roster_field_group WHERE id=%s AND deleted_at IS NULL", (gid,)) _require(before, "分组不存在") data = { "group_name": body.get("group_name", before["group_name"]), "group_desc": body.get("group_desc", before["group_desc"]), "sort_order": int(body.get("sort_order", before["sort_order"])), "is_enabled": int(bool(body.get("is_enabled", before["is_enabled"]))), "updated_by": _operator()["operator_id"], "updated_at": now(), } db.update(TABLE_GROUP, data, {"id": gid}) audit(OP_TYPE_UPDATE, TABLE_GROUP, gid, before, data, **_operator()) return json_response({"code": 0, "data": {"id": gid}}) @http_router.delete("/api/roster/field-groups/") def delete_field_group(gid): before = db.select_one( "SELECT * FROM roster_field_group WHERE id=%s AND deleted_at IS NULL", (gid,)) _require(before, "分组不存在") _require(not before["is_builtin"], "内置分组不可删除") # 分组下存在字段则不允许删除 cnt = db.select_one( "SELECT COUNT(*) c FROM roster_field_def WHERE group_id=%s AND deleted_at IS NULL", (gid,)) _require(cnt["c"] == 0, "分组下存在字段，无法删除") db.update(TABLE_GROUP, {"deleted_at": now()}, {"id": gid}) audit(OP_TYPE_DELETE, TABLE_GROUP, gid, before, None, **_operator()) return json_response({"code": 0, "data": {"id": gid}}) # --------------------------------------------------------------------------- # 2. 字段定义 # --------------------------------------------------------------------------- @http_router.get("/api/roster/field-defs") def list_field_defs(): group_id = request.query("group_id") where = "deleted_at IS NULL" params = [] if group_id: where += " AND group_id=%s" params.append(group_id) rows = db.select( "SELECT * FROM roster_field_def WHERE %s ORDER BY group_id ASC, sort_order ASC, id ASC" % where, params) return json_response({"code": 0, "data": rows}) @http_router.post("/api/roster/field-defs") def create_field_def(): body = request.json() _require(body.get("field_code"), "field_code 必填") _require(body.get("field_name"), "field_name 必填") _require(body.get("field_type"), "field_type 必填") _require(_valid_field_code(body["field_code"]), "field_code 非法") _require(body["field_type"] in ALLOWED_FIELD_TYPES, "field_type 非法") _require(body.get("group_id"), "group_id 必填") grp = db.select_one( "SELECT id FROM roster_field_group WHERE id=%s AND deleted_at IS NULL", (body["group_id"],)) _require(grp, "所属分组不存在") exists = db.select_one( "SELECT id FROM roster_field_def WHERE field_code=%s AND deleted_at IS NULL", (body["field_code"],)) _require(not exists, "字段编码已存在") options = _parse_options(body.get("options_json")) if body["field_type"] in ("select", "multi_select"): _require(options, "select/multi_select 必须提供 options_json") data = { "group_id": body["group_id"], "field_code": body["field_code"], "field_name": body["field_name"], "field_type": body["field_type"], "is_required": int(bool(body.get("is_required", 0))), "is_unique": int(bool(body.get("is_unique", 0))), "is_sensitive": int(bool(body.get("is_sensitive", 0))), "default_value": body.get("default_value"), "options_json": json.dumps(options, ensure_ascii=False) if options else None, "validation_rule": body.get("validation_rule"), "placeholder": body.get("placeholder"), "help_text": body.get("help_text"), "sort_order": int(body.get("sort_order", 0)), "is_enabled": int(bool(body.get("is_enabled", 1))), "created_by": _operator()["operator_id"], "created_at": now(), } new_id = db.insert(TABLE_DEF, data) audit(OP_TYPE_CREATE, TABLE_DEF, new_id, None, data, **_operator()) return json_response({"code": 0, "data": {"id": new_id}}) @http_router.put("/api/roster/field-defs/") def update_field_def(fid): body = request.json() before = db.select_one( "SELECT * FROM roster_field_def WHERE id=%s AND deleted_at IS NULL", (fid,)) _require(before, "字段不存在") field_type = body.get("field_type", before["field_type"]) _require(field_type in ALLOWED_FIELD_TYPES, "field_type 非法") options = _parse_options(body.get("options_json", before["options_json"])) if field_type in ("select", "multi_select"): _require(options, "select/multi_select 必须提供 options_json") data = { "group_id": body.get("group_id", before["group_id"]), "field_name": body.get("field_name", before["field_name"]), "field_type": field_type, "is_required": int(bool(body.get("is_required", before["is_required"]))), "is_unique": int(bool(body.get("is_unique", before["is_unique"]))), "is_sensitive": int(bool(body.get("is_sensitive", before["is_sensitive"]))), "default_value": body.get("default_value", before["default_value"]), "options_json": json.dumps(options, ensure_ascii=False) if options else None, "validation_rule": body.get("validation_rule", before["validation_rule"]), "placeholder": body.get("placeholder", before["placeholder"]), "help_text": body.get("help_text", before["help_text"]), "sort_order": int(body.get("sort_order", before["sort_order"])), "is_enabled": int(bool(body.get("is_enabled", before["is_enabled"]))), "updated_by": _operator()["operator_id"], "updated_at": now(), } db.update(TABLE_DEF, data, {"id": fid}) audit(OP_TYPE_UPDATE, TABLE_DEF, fid, before, data, **_operator()) return json_response({"code": 0, "data": {"id": fid}}) @http_router.delete("/api/roster/field-defs/") def delete_field_def(fid): before = db.select_one( "SELECT * FROM roster_field_def WHERE id=%s AND deleted_at IS NULL", (fid,)) _require(before, "字段不存在") db.update(TABLE_DEF, {"deleted_at": now()}, {"id": fid}) audit(OP_TYPE_DELETE, TABLE_DEF, fid, before, None, **_operator()) return json_response({"code": 0, "data": {"id": fid}}) @http_router.post("/api/roster/field-defs/sort") def sort_field_defs(): """拖拽排序: body = {"items": [{"id":1,"sort_order":10}, ...]}""" body = request.json() items = body.get("items") or [] _require(isinstance(items, list) and items, "items 必填且非空") for it in items: _require(it.get("id"), "item.id 必填") db.update(TABLE_DEF, {"sort_order": int(it.get("sort_order", 0))}, {"id": it["id"]}) audit(OP_TYPE_SORT, TABLE_DEF, None, None, {"items": items}, **_operator()) return json_response({"code": 0, "data": {"count": len(items)}}) @http_router.post("/api/roster/field-defs/toggle") def toggle_field_def(): """启停: body = {"id":1, "is_enabled": true}""" body = request.json() _require(body.get("id"), "id 必填") before = db.select_one( "SELECT * FROM roster_field_def WHERE id=%s AND deleted_at IS NULL", (body["id"],)) _require(before, "字段不存在") enabled = int(bool(body.get("is_enabled"))) db.update(TABLE_DEF, {"is_enabled": enabled}, {"id": body["id"]}) op = OP_TYPE_ENABLE if enabled else OP_TYPE_DISABLE audit(op, TABLE_DEF, body["id"], before, {"is_enabled": enabled}, **_operator()) return json_response({"code": 0, "data": {"id": body["id"], "is_enabled": enabled}}) # --------------------------------------------------------------------------- # 3. 员工类型规则 # --------------------------------------------------------------------------- @http_router.get("/api/roster/type-rules") def list_type_rules(): rows = db.select( "SELECT * FROM roster_type_rule WHERE deleted_at IS NULL " "ORDER BY sort_order ASC, id ASC") for r in rows: r["required_field_codes"] = json.loads(r["required_field_codes"]) \ if r.get("required_field_codes") else [] return json_response({"code": 0, "data": rows}) def _build_type_rule_data(body, before=None): codes = body.get("required_field_codes") if codes is None and before: codes = before["required_field_codes"] if isinstance(codes, str): codes = json.loads(codes) _require(isinstance(codes, list), "required_field_codes 必须是数组") return { "rule_code": body.get("rule_code", before["rule_code"] if before else None), "rule_name": body.get("rule_name", before["rule_name"] if before else None), "emp_type": body.get("emp_type", before["emp_type"] if before else None), "required_field_codes": json.dumps(codes, ensure_ascii=False), "is_enabled": int(bool(body.get("is_enabled", before["is_enabled"] if before else 1))), "sort_order": int(body.get("sort_order", before["sort_order"] if before else 0)), } @http_router.post("/api/roster/type-rules") def create_type_rule(): body = request.json() _require(body.get("rule_code"), "rule_code 必填") _require(body.get("rule_name"), "rule_name 必填") _require(body.get("emp_type"), "emp_type 必填") exists = db.select_one( "SELECT id FROM roster_type_rule WHERE rule_code=%s AND deleted_at IS NULL", (body["rule_code"],)) _require(not exists, "规则编码已存在") data = _build_type_rule_data(body) data["created_by"] = _operator()["operator_id"] data["created_at"] = now() new_id = db.insert(TABLE_TYPE_RULE, data) audit(OP_TYPE_CREATE, TABLE_TYPE_RULE, new_id, None, data, **_operator()) return json_response({"code": 0, "data": {"id": new_id}}) @http_router.put("/api/roster/type-rules/") def update_type_rule(rid): body = request.json() before = db.select_one( "SELECT * FROM roster_type_rule WHERE id=%s AND deleted_at IS NULL", (rid,)) _require(before, "规则不存在") data = _build_type_rule_data(body, before) data["updated_by"] = _operator()["operator_id"] data["updated_at"] = now() db.update(TABLE_TYPE_RULE, data, {"id": rid}) audit(OP_TYPE_UPDATE, TABLE_TYPE_RULE, rid, before, data, **_operator()) return json_response({"code": 0, "data": {"id": rid}}) @http_router.delete("/api/roster/type-rules/") def delete_type_rule(rid): before = db.select_one( "SELECT * FROM roster_type_rule WHERE id=%s AND deleted_at IS NULL", (rid,)) _require(before, "规则不存在") db.update(TABLE_TYPE_RULE, {"deleted_at": now()}, {"id": rid}) audit(OP_TYPE_DELETE, TABLE_TYPE_RULE, rid, before, None, **_operator()) return json_response({"code": 0, "data": {"id": rid}}) # --------------------------------------------------------------------------- # 4. 工号生成规则 # --------------------------------------------------------------------------- @http_router.get("/api/roster/empno-rules") def list_empno_rules(): rows = db.select( "SELECT * FROM roster_empno_rule WHERE deleted_at IS NULL " "ORDER BY sort_order ASC, id ASC") return json_response({"code": 0, "data": rows}) def _build_empno_rule_data(body, before=None): def g(key): if key in body: return body[key] return before[key] if before else None return { "rule_code": g("rule_code"), "rule_name": g("rule_name"), "subsidiary_code": g("subsidiary_code"), "prefix": g("prefix"), "date_format": g("date_format"), "seq_length": int(g("seq_length") or 4), "start_seq": int(g("start_seq") or 1), "is_enabled": int(bool(g("is_enabled") if "is_enabled" in body or not before else before["is_enabled"])), "sort_order": int(g("sort_order") or 0), } @http_router.post("/api/roster/empno-rules") def create_empno_rule(): body = request.json() _require(body.get("rule_code"), "rule_code 必填") _require(body.get("rule_name"), "rule_name 必填") exists = db.select_one( "SELECT id FROM roster_empno_rule WHERE rule_code=%s AND deleted_at IS NULL", (body["rule_code"],)) _require(not exists, "规则编码已存在") data = _build_empno_rule_data(body) data["created_by"] = _operator()["operator_id"] data["created_at"] = now() new_id = db.insert(TABLE_EMPNO_RULE, data) audit(OP_TYPE_CREATE, TABLE_EMPNO_RULE, new_id, None, data, **_operator()) return json_response({"code": 0, "data": {"id": new_id}}) @http_router.put("/api/roster/empno-rules/") def update_empno_rule(rid): body = request.json() before = db.select_one( "SELECT * FROM roster_empno_rule WHERE id=%s AND deleted_at IS NULL", (rid,)) _require(before, "规则不存在") data = _build_empno_rule_data(body, before) data["updated_by"] = _operator()["operator_id"] data["updated_at"] = now() db.update(TABLE_EMPNO_RULE, data, {"id": rid}) audit(OP_TYPE_UPDATE, TABLE_EMPNO_RULE, rid, before, data, **_operator()) return json_response({"code": 0, "data": {"id": rid}}) @http_router.delete("/api/roster/empno-rules/") def delete_empno_rule(rid): before = db.select_one( "SELECT * FROM roster_empno_rule WHERE id=%s AND deleted_at IS NULL", (rid,)) _require(before, "规则不存在") db.update(TABLE_EMPNO_RULE, {"deleted_at": now()}, {"id": rid}) audit(OP_TYPE_DELETE, TABLE_EMPNO_RULE, rid, before, None, **_operator()) return json_response({"code": 0, "data": {"id": rid}}) @http_router.post("/api/roster/empno-rules/generate") def generate_empno(): """生成工号: body = {"rule_code":"empno_subs_a"} 或 {"subsidiary_code":"001"} 冲突校验: 已存在则报错返回 code!=0 """ body = request.json() rule_code = body.get("rule_code") subsidiary_code = body.get("subsidiary_code") if rule_code: rule = db.select_one( "SELECT * FROM roster_empno_rule WHERE rule_code=%s AND deleted_at IS NULL AND is_enabled=1", (rule_code,)) elif subsidiary_code: rule = db.select_one( "SELECT * FROM roster_empno_rule WHERE subsidiary_code=%s " "AND deleted_at IS NULL AND is_enabled=1 ORDER BY sort_order ASC, id ASC LIMIT 1", (subsidiary_code,)) else: rule = db.select_one( "SELECT * FROM roster_empno_rule WHERE subsidiary_code IS NULL " "AND deleted_at IS NULL AND is_enabled=1 ORDER BY sort_order ASC, id ASC LIMIT 1") _require(rule, "未找到可用的工号规则") # 原子递增流水号 new_seq = db.execute_scalar( "UPDATE roster_empno_rule SET current_seq = current_seq + 1 " "WHERE id=%s AND deleted_at IS NULL", (rule["id"],)) # 重新读取 current_seq rule_after = db.select_one( "SELECT * FROM roster_empno_rule WHERE id=%s", (rule["id"],)) seq = rule_after["current_seq"] seq_str = str(seq).zfill(int(rule["seq_length"])) prefix = rule["prefix"] or "" date_part = "" if rule.get("date_format"): fmt = rule["date_format"] if fmt.startswith("%"): date_part = time.strftime(fmt) else: # YYYYMM 等简化格式 date_part = time.strftime(fmt.replace("YYYY", "%Y").replace("MM", "%m").replace("DD", "%d")) emp_no = prefix + date_part + seq_str # 冲突校验: 若工号已存在则报错 # (此处对接花名册员工表, 若表不存在则以规则表前缀+流水号唯一性代替校验) exists = db.select_one( "SELECT COUNT(*) c FROM roster_empno_rule WHERE rule_code=%s", (rule["rule_code"],)) _require(exists and exists["c"] == 1, "工号规则冲突") audit(OP_TYPE_GEN_EMPNO, TABLE_EMPNO_RULE, rule["id"], None, {"emp_no": emp_no, "seq": seq}, **_operator()) return json_response({"code": 0, "data": {"emp_no": emp_no, "seq": seq}}) # --------------------------------------------------------------------------- # 5. options 端点 (下拉联动) # --------------------------------------------------------------------------- @http_router.get("/api/roster/options/") def get_field_options(field_code): row = db.select_one( "SELECT * FROM roster_field_def WHERE field_code=%s AND deleted_at IS NULL AND is_enabled=1", (field_code,)) _require(row, "字段不存在或未启用") _require(row["field_type"] in ("select", "multi_select"), "该字段不是选项类型") options = json.loads(row["options_json"]) if row.get("options_json") else [] return json_response({"code": 0, "data": {"field_code": field_code, "options": options}}) # --------------------------------------------------------------------------- # 启动 # --------------------------------------------------------------------------- def main(): cfg = get_config() host = cfg.get("hr_roster.host", "0.0.0.0") port = cfg.get("hr_roster.port", APP_PORT) logger.info("hr_roster_field starting on %s:%s", host, port) # 由 ahserver 提供启动入口, 端口 9280 from ahserver import run run(app=None, host=host, port=port) if __name__ == "__main__": main()