钉钉加班审批自动转入调休余额：连接流 + FaaS 实战

加班审批通过后，需要把加班时长自动折算成调休余额（假期额度）转入员工的假期账户。但钉钉自带的「调休」假期类型只有 1 个，无法按不同有效期、不同部门拆分配置。于是这次二开通过 API 新建了 1 个自定义假期类型，并在此基础上做自动化——API 新建的假期类型只能通过 API 修改，管理后台无法操作。技术栈是钉钉连接流（iPaaS）+ FaaS 脚本 + 假期余额 API，过程中踩了时区、取整、API 语义等一堆坑，全部记录在这里。

一、需求与方案设计

业务规则很简单，但有三个硬性约束：

• 加班时长按扣除午休（12:00~13:00）后的实际时长计算
• 余额发放向下按半小时取整（1.6h 只发 1.5h）
• 发入的调休额度有效期 6 个月

一开始用钉钉连接流可视化编排，3 个节点：审批触发 → FaaS 计算 → 批量更新假期余额。测试时连接器明明返回成功，余额数量却对不上，一度怀疑连接器是黑盒；后来用本地 Python 直调 API 验证才发现真相是 quota/update 的语义是「设目标值」而非「增量追加」。最终收敛为 2 节点方案：审批触发 → FaaS 脚本直接调 API（先查余额 → 算新目标值 → 写回），逻辑与本地验证一致。

二、核心坑 1：休息时间没扣掉（quota=1000 而不是 900）

测试数据是 08:00 ~ 18:00，跨了 12:00~13:00 午休，期望扣 1 小时得 9 小时（quota=900），但 FaaS 输出 quota_num_per_hour: 1000（10 小时）。

重叠区间计算逻辑本身没问题：

# 扣 12:00-13:00 午休
total_seconds = (dt_end - dt_start).total_seconds()
rest_start = dt_start.replace(hour=12, minute=0, second=0)
rest_end   = dt_start.replace(hour=13, minute=0, second=0)

overlap = 0
if dt_start < rest_end and dt_end > rest_start:
    a = max(dt_start, rest_start)
    b = min(dt_end, rest_end)
    overlap = (b - a).total_seconds()

actual_hours = (total_seconds - overlap) / 3600.0

问题出在时间戳解析：钉钉连接流传过来的是毫秒时间戳，而 FaaS 环境跑在 UTC 时区。直接 datetime.fromtimestamp(ts/1000) 解析出来比北京时间少 8 小时，午休区间判断全部错位，休息自然没扣掉。

def parse_time(ts):
    ts = int(float(ts))
    if ts > 1000000000000:      # 毫秒戳
        ts = ts // 1000
    ts = ts + 28800              # 强制转北京时间 UTC+8
    return datetime.utcfromtimestamp(ts)

三、核心坑 2：FaaS 编辑器缩进地狱

钉钉 FaaS 编辑器对缩进极其敏感，从聊天窗口直接复制代码过去，容易混入不可见字符或 Tab，反复报 SyntaxError: invalid syntax，行号还指向无辜的 if dt_end <= dt_start:。

两个实用解法：

• 代码先粘到记事本过一遍，再从记事本复制粘贴到钉钉，过滤不可见字符
• 尽量写扁平结构，减少 if/else 嵌套层级，缩进只有一层最安全

四、核心坑 3：API 是「设目标值」，不是「增量追加」

这是整个项目最大的坑。文档示例看起来是增量，实际验证结果完全相反：

quota_num_per_hour 代表该员工、该假期类型、该周期的目标总余额，每次更新都是整值覆盖。所以正确姿势必须是：

def query_quota(user_id):
    body = {
        "op_userid": OP_USER_ID,
        "leave_code": LEAVE_CODE,
        "userids": user_id,
        "offset": 0, "size": 1
    }
    result = api_call("https://oapi.dingtalk.com/topapi/attendance/vacation/quota/list", body)
    total = 0
    if result.get("errcode") == 0:
        for q in result.get("result", {}).get("list", []):
            total += q.get("quota", 0) / 100.0   # 百分之一小时 → 小时
    return total

# 新目标值 = 当前 + 本次
new_total_hours = query_quota(user_id) + actual_hours
new_quota = int(new_total_hours * 100)

五、核心坑 4：连接器成功但数量对不上，FaaS 直调反而失败

3 节点方案里，FaaS 算好的数据交给「批量更新假期余额」连接器去写。测试时连接器返回成功（errcode 0），余额也确实变了，但数量对不上：测试数据是 08:00~18:00 应该加 9 小时，结果只加了 6.5 小时。而反过来，试过让 FaaS 直接调 API 的 2 节点方案，反而报错发不出去，又回滚到 3 节点。当时的对比实验：

数量对不上的原因，正是核心坑 3 的 API 语义：当时余额是 2.5 小时，传入目标值 9 小时，结果余额变成 9 小时——只增加了 6.5 小时。不是连接器的问题，也不是 FaaS 的问题，而是 quota/update 本来就是「设目标值」：传入的是该员工该假期该周期的目标总余额，每次都是整值覆盖。

六、核心坑 5：取整规则（连 API 都无法设置）

想让员工请假时长向上取整，可以在假期规则里传 leaveTimeCeil / leaveTimeCeilMinUnit，但实测发现这两个字段连 API 都无法设置——API 新建的假期类型虽然只能通过 API 修改，但取整相关字段 API 也写不进去，传了也被忽略。只有 paidLeave（是否带薪）这类普通字段能正常更新。

也就是说：自定义假期类型一旦创建，取整规则完全不可配置，只能靠 FaaS 在发放侧自己实现取整。发放时向下按半小时取整：

import math

def floor_half_hour(hours):
    """向下按半小时取整：1.6h → 1.5h，1.1h → 1.0h"""
    minutes = hours * 60
    units = math.floor(minutes / 30)
    return units * 30 / 60

actual_hours = floor_half_hour(actual_hours)

七、核心坑 6：有效期与 quota_cycle

另一个大坑：quota_cycle 是必填的「额度归属年度」（格式 yyyy，如 2026），真正控制有效期的是 start_time / end_time。同一员工同一假期类型下，同一周期的额度被系统视为一个整体，每次更新都会整体覆盖——想实现「每笔加班单独 6 个月有效期」在单条额度模型下做不到，只能按「统一有效期」处理（行业通用做法）：

# 6 个月有效期（月末 23:59:59）
now = datetime.now()
y, m = now.year, now.month + 6
while m > 12:
    m -= 12; y += 1
last = calendar.monthrange(y, m)[1]
ed = datetime(y, m, last, 23, 59, 59)

body = {
    "op_userid": OP_USER_ID,
    "leave_quotas": [{
        "userid": user_id,
        "leave_code": LEAVE_CODE,
        "quota_num_per_hour": new_quota,
        "start_time": int(time.time() * 1000),
        "end_time": int(ed.timestamp() * 1000),
        "quota_cycle": str(now.year),
        "reason": "加班审批自动转入"
    }]
}
result = api_call("https://oapi.dingtalk.com/topapi/attendance/vacation/quota/update", body)

八、核心坑 7：中文乱码与 charset

FaaS 里调 API 写中文（reason 字段）必须注意两点，否则静默失败或乱码：

• 请求头必须带 Content-Type: application/json; charset=utf-8
• json.dumps(body, ensure_ascii=False).encode("utf-8")，确保中文原样发送

body_bytes = json.dumps(body, ensure_ascii=False).encode("utf-8")
req = urllib.request.Request(
    url, data=body_bytes,
    headers={"Content-Type": "application/json; charset=utf-8"},
    method="POST")

九、最终完整 FaaS 脚本

2 节点方案的最终脚本（含查余额、扣休息、向下取整、6 个月有效期），把 APP_KEY / APP_SECRET / OP_USER_ID / LEAVE_CODE 四个参数填成自己的即可：

import urllib.request
import json
import time
from datetime import datetime
import calendar
import math

# ==================== 【必填参数】 ====================
APP_KEY = "你的AppKey"
APP_SECRET = "***"
OP_USER_ID = "你的管理员UserId"       # 操作者（管理员）
LEAVE_CODE = "你的调休假期类型ID"  # 调休假期类型ID
# ==================== 【以下不用改】 ====================

def floor_half_hour(hours):
    """向下按半小时取整"""
    minutes = hours * 60
    units = math.floor(minutes / 30)
    return units * 30 / 60

def api_call(url, body=None):
    """通用 API 调用（自动取 token）"""
    token_url = "https://oapi.dingtalk.com/gettoken?appkey=" + APP_KEY \
              + "&appsecret=" + APP_SECRET
    req1 = urllib.request.Request(token_url, method="GET")
    resp1 = urllib.request.urlopen(req1, timeout=10)
    data1 = json.loads(resp1.read().decode("utf-8"))
    token = data1.get("access_token", "")

    full_url = url + "?access_token=" + token
    if body is not None:
        body_bytes = json.dumps(body, ensure_ascii=False).encode("utf-8")
        req = urllib.request.Request(full_url, data=body_bytes,
            headers={"Content-Type": "application/json; charset=utf-8"},
            method="POST")
    else:
        req = urllib.request.Request(full_url, method="POST")
    resp = urllib.request.urlopen(req, timeout=15)
    return json.loads(resp.read().decode("utf-8"))

def query_quota(user_id):
    """查询当前调休余额（小时）"""
    body = {
        "op_userid": OP_USER_ID,
        "leave_code": LEAVE_CODE,
        "userids": user_id,
        "offset": 0, "size": 1
    }
    result = api_call("https://oapi.dingtalk.com/topapi/attendance/vacation/quota/list", body)
    total = 0
    if result.get("errcode") == 0:
        for q in result.get("result", {}).get("list", []):
            total += q.get("quota", 0) / 100.0
    return total

def parse_time(ts):
    """毫秒时间戳 → 北京时间 datetime"""
    ts = int(float(ts))
    if ts > 1000000000000:
        ts = ts // 1000
    ts = ts + 28800
    return datetime.utcfromtimestamp(ts)

# 读取入参
raw_userid = input.get("userid", "")
user_id = str(raw_userid[0]) if isinstance(raw_userid, list) and len(raw_userid) > 0 else str(raw_userid)
dt_start = parse_time(input.get("start_time", ""))
dt_end   = parse_time(input.get("end_time", ""))

# 时长 = 总时长 - 午休重叠（12:00-13:00）
total_seconds = (dt_end - dt_start).total_seconds()
rest_start = dt_start.replace(hour=12, minute=0, second=0)
rest_end   = dt_start.replace(hour=13, minute=0, second=0)
overlap = 0
if dt_start < rest_end and dt_end > rest_start:
    a = dt_start if dt_start > rest_start else rest_start
    b = dt_end if dt_end < rest_end else rest_end
    overlap = (b - a).total_seconds()
actual_hours = (total_seconds - overlap) / 3600.0
actual_hours = floor_half_hour(actual_hours)   # 向下按半小时取整

# 先查当前余额，算新目标值（API 是设目标值，不是增量）
current_hours = query_quota(user_id)
new_total_hours = current_hours + actual_hours
new_quota = int(new_total_hours * 100)

# 6 个月有效期
now = datetime.now()
y, m = now.year, now.month + 6
while m > 12:
    m -= 12; y += 1
last = calendar.monthrange(y, m)[1]
ed = datetime(y, m, last, 23, 59, 59)

# 更新余额（设目标值）
body = {
    "op_userid": OP_USER_ID,
    "leave_quotas": [{
        "userid": user_id,
        "leave_code": LEAVE_CODE,
        "quota_num_per_hour": new_quota,
        "start_time": int(time.time() * 1000),
        "end_time": int(ed.timestamp() * 1000),
        "quota_cycle": str(now.year),
        "reason": "加班审批自动转入（扣休息后" + str(actual_hours) + "h）"
    }]
}
result = api_call("https://oapi.dingtalk.com/topapi/attendance/vacation/quota/update", body)

output = {
    "success": result.get("errcode") == 0,
    "current": current_hours,
    "added": actual_hours,
    "new_total": new_total_hours,
    "api_result": result
}

十、踩坑清单（可直接复用）

1. 时区：FaaS 环境是 UTC，毫秒时间戳 +28800 秒再解析，否则扣休息判断全部错位
2. 缩进：钉钉 FaaS 编辑器对缩进极其敏感，代码先过记事本再粘贴，写扁平结构
3. API 语义：quota/update 是设目标值（整值覆盖），必须先查余额再算「当前+本次」
4. 排查方法：节点行为诡异先别怀疑平台，用本地 Python 直调 API 复现，把接口语义验证清楚再下结论
5. 取整：leaveTimeCeil 系列字段连 API 都无法设置（自定义假期类型取整规则不可配置），取整必须在 FaaS 里自己实现
6. 自定义类型：API 新建的假期类型只能通过 API 修改，管理后台无法操作，所有规则维护都要走接口
7. 有效期：quota_cycle 是年度归属（yyyy），真正的有效期由 start/end_time 控制，同周期额度整体覆盖
8. 中文：请求头必须 charset=utf-8，ensure_ascii=False，否则中文字段静默出错