PRA-only-demo/packages/rpa-core/src/capabilities/ftp.ts

import { Client } from "basic-ftp";
import path from "node:path";
import fs from "node:fs";
import { register } from "../engine/registry";
import { RuntimeError } from "../engine/types";
import { defaultEngine } from "../engine/engine";

async function requireClient(nodeId: string): Promise<Client> {
  if (defaultEngine.ftpClient) return defaultEngine.ftpClient;
  throw new RuntimeError("validation_error", "尚未连接 FTP，请先使用 FTP_Connect", { nodeId, componentId: nodeId, recommendation: "先运行 FTP_Connect 建立连接" });
}

function requireText(args: Record<string, unknown>, keys: string[], label: string, nodeId: string): string {
  for (const key of keys) {
    const v = String(args[key] ?? "").trim();
    if (v) return v;
  }
  throw new RuntimeError("validation_error", label + "为空", { nodeId, componentId: nodeId });
}

register("FTP_Connect", async (ctx, node, args) => {
  const host = requireText(args, ["address"], "FTP 地址", node.id);
  const port = Number(args.port ?? 21) || 21;
  const user = String(args.user ?? "anonymous");
  const password = String(args.pass_word ?? args.pass_word_file ?? "guest");
  const secure = String(args.ftp_type ?? "").toLowerCase().includes("ftps");
  const client = new Client(10_000);
  await client.access({ host, port, secure, user, password });
  defaultEngine.ftpClient = client;
  ctx.log("已连接 FTP：" + host + ":" + port, "info", node.id);
  return { connected: true, host, port };
});

register("FTP_Disconnect", async (ctx, node) => {
  defaultEngine.ftpClient?.close();
  defaultEngine.ftpClient = undefined;
  ctx.log("已断开 FTP 连接", "info", node.id);
  return { disconnected: true };
});

register("FTP_GetFileList", async (ctx, node, args) => {
  const client = await requireClient(node.id);
  const list = await client.list("");
  const entries = list.map(item => ({ name: item.name, size: item.size, isDirectory: item.isDirectory, modifiedAt: item.modifiedAt instanceof Date ? item.modifiedAt.toISOString() : String(item.modifiedAt ?? "") }));
  ctx.log("已获取文件列表：" + entries.length + " 项", "info", node.id);
  return entries;
});

register("FTP_Cwd", async (ctx, node, args) => {
  const client = await requireClient(node.id);
  const dir = requireText(args, ["dir"], "目标目录", node.id);
  await client.cd(dir);
  ctx.log("已切换工作目录：" + dir, "info", node.id);
  return { cwd: dir };
});

register("FTP_Download", async (ctx, node, args) => {
  const client = await requireClient(node.id);
  const remote = requireText(args, ["file_path"], "远程文件路径", node.id);
  const localDir = String(args.folder_path ?? "").trim();
  const target = requireText(args, ["target_path"], "本地保存路径", node.id) || (localDir ? path.join(localDir, path.basename(remote)) : path.basename(remote));
  fs.mkdirSync(path.dirname(path.resolve(target)), { recursive: true });
  if (String(args.file_type ?? "0") === "1") {
    fs.mkdirSync(path.resolve(target), { recursive: true });
    await client.ensureDir(remote);
    await client.downloadToDir(target, remote);
  } else {
    await client.downloadTo(target, remote);
  }
  ctx.log("已下载到：" + target, "info", node.id);
  return { downloaded: true, local: target };
});

register("FTP_Upload", async (ctx, node, args) => {
  const client = await requireClient(node.id);
  const local = requireText(args, ["file_path"], "本地文件路径", node.id);
  const remote = requireText(args, ["remote_path"], "远程目标路径", node.id);
  if (String(args.file_type ?? "0") === "1" || fs.statSync(local).isDirectory()) {
    await client.uploadFromDir(local, remote);
  } else {
    await client.uploadFrom(local, remote);
  }
  ctx.log("已上传：" + local + " → " + remote, "info", node.id);
  return { uploaded: true, remote };
});

register("FTP_Rename", async (ctx, node, args) => {
  const client = await requireClient(node.id);
  const from = requireText(args, ["file_path"], "原路径", node.id);
  const to = requireText(args, ["toname"], "新名称", node.id);
  const parent = path.posix.dirname(from);
  await client.rename(from, parent === "." ? to : path.posix.join(parent, to));
  ctx.log("已重命名：" + from + " → " + to, "info", node.id);
  return { renamed: true };
});

register("FTP_Delete", async (ctx, node, args) => {
  const client = await requireClient(node.id);
  const remote = requireText(args, ["file_path"], "远程路径", node.id);
  if (String(args.file_type ?? "0") === "1") {
    await client.removeDir(remote);
  } else {
    await client.remove(remote);
  }
  ctx.log("已删除：" + remote, "info", node.id);
  return { deleted: true };
});

register("FTP_CreateDir", async (ctx, node, args) => {
  const client = await requireClient(node.id);
  const dir = requireText(args, ["dirname"], "目录名", node.id);
  await client.ensureDir(dir);
  // 创建后进入该目录；必须 await，否则下个节点的命令与本次 cd 并发，
  // basic-ftp 会以 "task while another one is still running" 关闭客户端
  await client.cd(dir).catch(() => {});
  ctx.log("已创建目录：" + dir, "info", node.id);
  return { created: true, dir };
});

register("FTP_Exist", async (ctx, node, args) => {
  const client = await requireClient(node.id);
  const target = requireText(args, ["path"], "远程路径", node.id);
  const parent = path.posix.dirname(target);
  const name = path.posix.basename(target);
  const list = await client.list(parent === "." ? "" : parent);
  const exists = list.some(item => item.name === name);
  return { exists, path: target };
});