实战SOP:CodeArts操作FunctionGraph,云端处理本地文件

背景:本地处理小文件,环境比算法更麻烦

手上有个很普通的活:一批 CSV 和 Excel,要转成 JSON 落库,几百个小文件,每个几 KB 到几 MB。单机跑不难,难的是环境——依赖装在本机能跑,换到同事电脑上就报编码错、缺库;跑完还得把结果拷回去。三台机器同步一次环境,时间比处理文件本身还长。

换成另一种思路就顺了:把处理逻辑放到云端函数,本地只做三件事——读文件、调函数、写结果。函数的环境是固定的,依赖是打进包的,谁来跑结果都一样。这就是"云端操作本地小文件":文件不出门,处理逻辑统一在云上。

关键是规模要匹配:小文件用同步调用足够;大文件就该走对象存储和异步触发,别硬塞进请求体。

方案:五步,一条命令跑完整批

1. 定契约。 我先约定三件事:请求体是 {op, name, data},data 是文件内容的 base64;响应统一是 {ok, op, data, req};失败也返回 200 但 ok=false,把错误信息带回来。这样本地客户端只有一条判断路径。

2. 写 handler:

import base64, csv, io, json

def handler(event, context):
    try:
        body = json.loads(event.get("body") or "{}") if isinstance(event, dict) else {}
        op = body.get("op", "csv2json")
        raw = base64.b64decode(body.get("data", ""))

        if op == "csv2json":
            # utf-8-sig 吃掉 BOM,否则第一个列名会带上不可见字符
            rows = list(csv.DictReader(io.StringIO(raw.decode("utf-8-sig"))))
            out = json.dumps(rows, ensure_ascii=False).encode("utf-8")
        else:
            out = raw

        return {
            "statusCode": 200,
            "headers": {"Content-Type": "application/json"},
            "body": json.dumps({
                "ok": True, "op": op,
                "data": base64.b64encode(out).decode(),
                "req": context.request_id,
            }),
        }
    except Exception as e:
        return {"statusCode": 200,
                "body": json.dumps({"ok": False, "err": str(e),
                                    "req": context.request_id})}

3. 写本地客户端(遍历、调用、写回):

import base64, json, pathlib, subprocess

FUNC_URN = "<你的函数URN>"

def call_func(path, op="csv2json"):
    payload = json.dumps({
        "op": op, "name": path.name,
        "data": base64.b64encode(path.read_bytes()).decode(),
    })
    r = subprocess.run(
        ["hcloud", "FunctionGraph", "InvokeFunction",
         f"--func_urn={FUNC_URN}", "--body", payload],
        capture_output=True, check=True,
    )
    return json.loads(r.stdout)

out_dir = pathlib.Path("outbox"); out_dir.mkdir(exist_ok=True)
for p in pathlib.Path("inbox").glob("*.csv"):
    res = call_func(p)["body"]          # 外层是 API 响应,内层是函数返回
    inner = json.loads(res)
    if inner["ok"]:
        out_dir.joinpath(p.stem + ".json").write_text(
            base64.b64decode(inner["data"]).decode("utf-8"),
            encoding="utf-8")
    else:
        print("FAIL", p.name, inner["err"], inner["req"])

4. 部署与验证。 函数侧和普通函数一样:建函数、上传 zip、设超时与内存。验证要盯"逐份一致"——拿十份样本同时跑本地实现与云端实现,比对输出差异;第一次跑最容易暴露的就是编码与时区问题。

实践要点

1. 同步调用有体积上限,小文件才合适。 请求体与响应体都有大小限制,base64 还会让体积再膨胀三分之一。几 MB 以内走同步最省事;更大的文件改成"上传对象存储 → 异步触发函数 → 回写结果",别硬塞。
2. 编码是第一批必踩的坑。 CSV 带 BOM 时,列名会带一个不可见字符,读出来键名对不上。统一用 utf-8-sig 解码文本文件;二进制文件全程按字节处理,不要中途 decode。
3. 批量调用要控并发。 本地串行慢,一口气几十个并发又容易撞函数并发配额,收到限流错误。稳妥做法是并发数固定为 4~8,遇到限流退避重试(等 1 秒、2 秒、4 秒)。
4. 失败要有清单,不能只打印。 一批几百份总有几份失败,把失败的"文件名 + 错误 + request_id"写进单独的清单,处理后单独重跑。request_id 是事后对账的唯一线索。
5. 输出按文件名幂等。 重试、重复执行是常态,输出命名不要带时间戳,直接用原文件名的 stem,重复跑只会覆盖不会堆出一堆副本。
6. 敏感文件先想清楚传输路径。 内容 base64 进请求体、进日志,等于明文到处流。脱敏字段在函数里就删掉,日志只记文件名与大小,不记内容。
7. 算清收益边界。 云端处理的价值是"环境一致、换机器不改代码",不是"更快"。单个文件本就几十毫秒的任务,本地跑更省;只有当依赖重、环境多、批量大时,这套流程才真正划算。

小结

这个方案没有什么高级技巧,它解决的是一个很朴素的问题:同一段处理逻辑,不应该因为换了台电脑就失效。把逻辑放进函数,把文件留在本地,批处理就变成了"读—调—写"三步。

对个人开发者,它还有个额外好处:一次部署,以后所有机器、所有同事都共用同一份处理能力,连环境说明文档都省了。小文件走同步调用,大文件走对象存储,想清楚这条分界线,函数工作流就是随手可用的云端工具台。