diff --git a/src/cli/CHANGELOG.md b/src/cli/CHANGELOG.md index c1f250e..b8514f1 100644 --- a/src/cli/CHANGELOG.md +++ b/src/cli/CHANGELOG.md @@ -2,15 +2,20 @@ ## [Unreleased] +## [0.3.0] - 2026-08-24 + ### Added - `Specification` envelope 新增可选 `manifest` 输入契约（raw / map / config_tables / review_decisions）及确定性 YAML 校验。 - `catalog` 新增 `VolumeArtifactType` 与 `catalog set-status`，用于区分预审核产物、审核决策文件、最终交付产物并推进 volume 状态。 +- 新增 `runtime/` 模块（`Runtime` trait + 注册表），支持 Python、R、Stata、Matlab、Bash 和 builtin 运行时。 +- `implement` 增加 R、Stata、Matlab codegen；`process` 通过运行时注册表执行脚本。 +- 全局 `--json` 命令错误输出与 `CliError` 错误码结构化；`spec validate` 成功结果支持结构化输出。 ### Changed - `blueprint list/show` 与 `pipeline list/show` 改为文件直读优先，cue 降为可选增强；`doctor` 中 cue 检查同步降为 warning。 -- v0.2.2 计划文档收口：baidu/sftp 覆盖率补测后移至 v0.3.0，存储凭证常量表去重后移至未分配低优先事项。 -- 新增 `runtime/` 模块（`Runtime` trait + 注册表）：codegen（implement 用）+ execute（process 用），`from_name`/`from_ext` 注册表驱动；Python codegen 逻辑从 `stage/implement.rs` 分离，`process` 执行分发改注册表（`.py`→python / `.sh`→bash）。旧 `stage::implement::{implement_step_prompt, implement_assemble_prompt, to_snake}` 保留为 deprecated 转发（随 v0.3 移除）。 -- `storage/` 统一概念命名：`StorageProvider` → `Storage`、`DropboxProvider` → `DropboxStorage`（等 6 平台）。旧名保留为 deprecated re-export（随 v0.3 移除）。 +- `doctor` 的运行时检查改由运行时注册表驱动。 +- `transfer` provider 改为枚举分发，`process` pipeline 改为结构化 Blueprint state 解析。 +- `storage/` 统一概念命名为 `Storage` 与具体 `*Storage` 类型，旧名保留为 deprecated re-export 供迁移使用。 ## [0.2.2] - 2026-08-02 diff --git a/src/cli/CONTRIBUTING.md b/src/cli/CONTRIBUTING.md index 2d1ebb2..8fc8eb6 100644 --- a/src/cli/CONTRIBUTING.md +++ b/src/cli/CONTRIBUTING.md @@ -19,7 +19,7 @@ | [TODO.md](TODO.md) | — | 按模块拆解的执行任务 | | [CHANGELOG.md](CHANGELOG.md) | — | 版本变更记录（发布事实源） | | [docs/index.md](docs/index.md) | main/lib/error/registry/util | 命令结构总览 + 文档映射表 + 横切基础（错误模型 + 注册表/工具机制） | -| [docs/transfer.md](docs/transfer.md) | transfer/storage | 传输服务与 StorageProvider | +| [docs/transfer.md](docs/transfer.md) | transfer/storage | 传输服务与 Storage | | [docs/catalog.md](docs/catalog.md) | catalog | 数据格式（registry/jobs/delivery-links 字段级） | | [docs/process.md](docs/process.md) | process | StepExecutor 编排 | | [docs/llm.md](docs/llm.md) | clarify/design/implement/review | LLM 命令与 Handler 注入 | @@ -118,7 +118,7 @@ qtcloud-devops plan audit --scope cli **2. Code**：在 feature 分支上完成实现与文档改动 ```bash -git switch -c codex/cli-v0.2.X-release +git switch -c codex/cli-v0.3.0-release qtcloud-devops code audit src/cli ``` @@ -135,9 +135,9 @@ cargo test --locked cargo clippy --locked -- -A warnings ``` -**4. 更新发布记录**：`Cargo.toml` 版本号 + `CHANGELOG.md` 发布条目 + `ROADMAP.md`/`TODO.md` 勾选 +**4. 更新发布记录**：`Cargo.toml` 版本号 + `CHANGELOG.md` 发布条目 + `ROADMAP.md`/`TODO.md` 收口 -**5. 提交与审查**：release-prep 提交（`chore(cli): prepare v0.2.X release`）， +**5. 提交与审查**：release-prep 提交（`chore(cli): prepare v0.3.0 release`）， feature 分支推远端 → Pull Request → review → 合并 `main`。release tag 必须指向 `main` 可达提交。 若变更已在 `main` 且 CI 通过（`push: [main]` 触发），可直接 `main` 发布。 @@ -145,8 +145,8 @@ feature 分支推远端 → Pull Request → review → 合并 `main`。release ```bash qtcloud-devops release status -qtcloud-devops release audit -v cli/v0.2.X --scope cli -qtcloud-devops release publish -v cli/v0.2.X --registry crates --dry-run +qtcloud-devops release audit -v cli/v0.3.0 --scope cli +qtcloud-devops release publish -v cli/v0.3.0 --registry crates --dry-run ``` dry-run 不得创建 tag、GitHub Release 或 crates.io 版本。 @@ -154,7 +154,7 @@ dry-run 不得创建 tag、GitHub Release 或 crates.io 版本。 **7. 发布**（maintainer 确认后） ```bash -qtcloud-devops release publish -v cli/v0.2.X --registry crates -y +qtcloud-devops release publish -v cli/v0.3.0 --registry crates -y ``` 命令创建并推送 `cli/v0.2.X` tag，随后 `release-cli.yml` GitHub Actions 完成： diff --git a/src/cli/Cargo.lock b/src/cli/Cargo.lock index e959c77..00b8672 100644 --- a/src/cli/Cargo.lock +++ b/src/cli/Cargo.lock @@ -1734,7 +1734,7 @@ dependencies = [ [[package]] name = "qtcloud-data-cli" -version = "0.2.2" +version = "0.3.0" dependencies = [ "async-trait", "aws-config", diff --git a/src/cli/Cargo.toml b/src/cli/Cargo.toml index c64b2e5..b0c745f 100644 --- a/src/cli/Cargo.toml +++ b/src/cli/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "qtcloud-data-cli" -version = "0.2.2" +version = "0.3.0" edition = "2024" authors = ["QuantTide Inc. "] description = "量潮数据云 CLI" diff --git a/src/cli/README.md b/src/cli/README.md index 5a24d93..f0901c9 100644 --- a/src/cli/README.md +++ b/src/cli/README.md @@ -4,6 +4,7 @@ 当前 crates.io 发布版本：`qtcloud-data-cli` v0.2.1（GitHub Release `cli/v0.2.1`，含 Linux / Windows 二进制）。 v0.2.1 重点：统一错误处理模型（CliError）、store 模块收敛路径与 JSON 读写、cue 输出结构化解析、全链路 e2e，测试覆盖率提升至 83.7%。 +当前开发版本为 v0.3.0，新增多语言 Runtime 注册表和命令错误结构化输出；正式发布状态以 `qtcloud-devops release status` 为准。 ## 命令一览 @@ -12,7 +13,7 @@ v0.2.1 重点：统一错误处理模型（CliError）、store 模块收敛路 | `clarify` | 从客户聊天记录或上下文生成 DRD 数据需求文档 | | `design` | 从 DRD 生成 Contract / Blueprint Specification（YAML + MD + HTML） | | `spec` | 固化 Specification YAML envelope（wrap / validate） | -| `implement` | 从 Blueprint YAML 生成 Python 代码实现 | +| `implement` | 从 Blueprint YAML 生成 Python / R / Stata / Matlab 代码实现 | | `review` | 审计 DRD 或 Specification 的完整性和一致性 | | `version` | 查看和比较规格版本 | | `transfer` | 数据传输（send / receive），支持 6 个平台 | @@ -76,6 +77,8 @@ qtcloud-data design blueprint .quanttide/data/drd/context.md # 从规格书生成 Python 实现 qtcloud-data implement .quanttide/data/spec/context-blueprint.yaml --lang python +# 也支持 R / Stata / Matlab +qtcloud-data implement .quanttide/data/spec/context-blueprint.yaml --lang r # 将旧 Blueprint YAML 包装成稳定 Specification envelope qtcloud-data spec wrap .quanttide/data/spec/context-blueprint.yaml @@ -100,6 +103,8 @@ qtcloud-data process ABC "https://..." --blueprint csv-standardization 成功交付时，`process` 还会把最终产物登记到 `CATALOG_DIR/registry.json`，provider 为 `process`，source 为 `process:`，status 为 `delivered`。 +使用全局 `--json` 时，命令错误会输出包含稳定 `code` 和用户可读 `message` 的 JSON；`spec validate` 也支持结构化成功结果。 + ## DataOps 目录 默认目录根是 `.quanttide/data`，也可以用 `DATA_ROOT` 调整。单个目录仍可用对应环境变量覆盖。 diff --git a/src/cli/ROADMAP.md b/src/cli/ROADMAP.md index 0e109be..d4c74fb 100644 --- a/src/cli/ROADMAP.md +++ b/src/cli/ROADMAP.md @@ -22,19 +22,14 @@ > **待排期（不阻塞 v0.2.2）**：CLI 发起 Provider run 的执行入口（`PROVIDER_URL` 配置 + run 请求参数校验），以及业务 e2e（raw + map.dta → review_master，依赖 Provider merge_review / export）——待 Provider ROADMAP [0.0.3] 的 merge_review/export 落地后启动。 -## [0.3.0] +## [0.3.1] + +> v0.3.0 已完成项已迁移至 [CHANGELOG.md](CHANGELOG.md)。 ### Added -- [ ] `src/storage/` baidu/sftp 补测（需真实服务或本地模拟：CI 起 sshd 或本地模拟，0% → ≥50%） -- [ ] `Cargo.toml` 构建 Linux、Windows、macOS 二进制包 -- [ ] `Cargo.toml` 自动上传各平台 Release 制品 -- [ ] `src/lib.rs` 新增 runtimes 模块：RuntimeAdapter trait 与注册表（python / r / stata / matlab / bash / builtin） -- [ ] `src/process.rs` run_pipeline 改为注册表查表，替代扩展名 if-else 分发 -- [ ] `src/implement.rs` implement 支持 --lang r / --lang stata -- [ ] `src/blueprint_core.rs` 新增 R / Stata codegen prompt 模板 -- [ ] `src/doctor.rs` 检查表由 RuntimeAdapter 注册表驱动 -- [ ] `src/main.rs` 全局 --json 结构化输出，供 Studio/CI 消费 +- [ ] `Cargo.toml` / `.github/workflows/release-cli.yml` 构建并上传 macOS 二进制包 +- [ ] 其余命令成功结果逐步迁移为结构化输出，供 Studio/CI 消费 ### Changed diff --git a/src/cli/TODO.md b/src/cli/TODO.md index f6c188f..8fb9f06 100644 --- a/src/cli/TODO.md +++ b/src/cli/TODO.md @@ -26,21 +26,11 @@ - [x] coverage：更新覆盖率基线（当前 83.7%，见 `CONTRIBUTING.md` 测试分层） - [x] storage：凭证环境变量名集中为常量表迁移至 99. 后续（`DROPBOX_ACCESS_TOKEN` / `GOOGLE_DRIVE_ACCESS_TOKEN` 等魔法字符串去重，低优先，可选） -## [0.3.0] +## [0.3.1] -- [ ] coverage：`src/storage/` baidu/sftp 补测（需真实服务或本地模拟，0% → ≥50%，需 CI 起 sshd 或本地模拟） -- [ ] distribution：增加 macOS 二进制构建（`../../.github/workflows/release-cli.yml`） +- [ ] distribution：增加 macOS 二进制构建和 Release 上传（`../../.github/workflows/release-cli.yml`） - [ ] distribution：增加发布后的 deploy、operate、monitor 记录（`../../.github/workflows/`、`docs/`） -- [ ] runtimes：`src/lib.rs` 注册 runtimes 模块（新增 src/runtimes/） -- [ ] runtimes：`src/process.rs` run_pipeline 改注册表查表，替代扩展名 if-else -- [ ] runtimes：`src/implement.rs` implement 支持 --lang r / --lang stata -- [ ] runtimes：`src/blueprint_core.rs` 新增 R / Stata codegen prompt 模板 -- [ ] runtimes：`src/doctor.rs` 检查表由 RuntimeAdapter 注册表驱动 -- [ ] structured-output：`src/main.rs` 全局 --json 结构化输出 -- [ ] structured-output：`CliError` 携带结构化错误码（`--json` 前置，`src/error.rs`） -- [ ] structured-output：`run_command` 分发移入 lib（`Commands` 可单测，llm 参数注入，`src/main.rs` / `src/lib.rs`） -- [ ] structured-output：`src/transfer.rs` provider 枚举化，替代字符串匹配 -- [ ] structured-output：`src/process.rs` pipeline 引用结构化（Blueprint states），替代逗号分隔字符串 +- [ ] structured-output：其余命令成功结果逐步迁移为结构化输出（`src/`） ## [0.5.0] diff --git a/src/cli/docs/index.md b/src/cli/docs/index.md index c5fe05d..cedcd7f 100644 --- a/src/cli/docs/index.md +++ b/src/cli/docs/index.md @@ -8,7 +8,7 @@ | 文档 | 对应模块 | 内容 | |------|---------|------| | [index.md](index.md)（本文件） | `main.rs`、`lib.rs`、`error.rs`、`registry.rs`、`util.rs` | 命令结构、文档映射、横切基础（错误模型 + 注册表/工具机制） | -| [transfer.md](transfer.md) | `transfer.rs`、`storage/` | 传输服务与 StorageProvider trait、添加新平台 | +| [transfer.md](transfer.md) | `transfer.rs`、`storage/` | 传输服务与 Storage trait、添加新平台 | | [catalog.md](catalog.md) | `implementation/catalog.rs` | 数据格式（registry/jobs/delivery-links 字段级定义） | | [process.md](process.md) | `stage/process.rs` | StepExecutor 编排（receive → pipeline → send） | | [llm.md](llm.md) | `stage/clarify.rs`、`stage/design.rs`、`stage/implement.rs`、`review.rs` | LLM 命令与 Handler 注入模式 | @@ -18,6 +18,7 @@ | [pipeline.md](pipeline.md) | `implementation/pipeline.rs` | 管道定义查看 | | [doctor.md](doctor.md) | `doctor.rs` | 环境检查 | | [version.md](version.md) | `spec/version.rs` | Specification 版本管理（spec version） | +| — | `runtime/` | Python / R / Stata / Matlab / Bash / builtin 运行时注册表 | 贡献与发布流程见 [CONTRIBUTING.md](../CONTRIBUTING.md)。 @@ -29,7 +30,7 @@ qtcloud-data ├── 生命周期命令（纵向流程，按数据流顺序） │ ├── clarify 需求澄清：聊天记录 → DRD │ ├── design 规格设计：DRD → Contract / Blueprint（contract / blueprint / formalize / preview） -│ ├── implement 代码实现：Blueprint → Python +│ ├── implement 代码实现：Blueprint → Python / R / Stata / Matlab │ ├── process 流程编排：receive → pipeline → send（StepExecutor） │ └── transfer 数据传输：send / receive（6 平台） │ @@ -38,7 +39,7 @@ qtcloud-data │ ├── review 质量审查：审计任意阶段产物（需求 / 设计 / 实现 / 交付） │ ├── spec Specification 工具：wrap（包装 envelope）/ validate（结构校验） │ ├── catalog 数据目录：volume 登记（list / show / add / rm） -│ └── version 规格版本管理：git 历史（list / show / diff） +│ └── spec version 规格版本管理：git 历史（list / show / diff） │ └── 查看命令（定义查看） ├── blueprint 蓝图定义查看（list / show） @@ -46,6 +47,10 @@ qtcloud-data └── pipeline 管道定义查看（list / show） ``` +全局选项 `--json` 当前统一命令错误输出为 `{"error":{"code","message"}}`； +`spec validate`、`pipeline list/show`、`blueprint list/show` 和 `catalog` 已提供结构化成功输出， +其余成功结果按命令逐步迁移。 + ## 命令分类原则 命令按**用途定位**分三类，而非按名称平铺： @@ -110,7 +115,8 @@ fn show(name: &str) -> Result<(), CliError> { - 错误路径通过 `Result` 传播，**不直接 `std::process::exit(1)`**（仅 `main` 保留 bin 入口 exit） - `Result<_, String>` 的公开函数已收敛为 `CliError`（`From/String/&str`） - 错误路径因此可测试：`cmd_xxx(...).unwrap_err()` -- `CliError` 只携带用户可读消息（`Display` 即消息本体），不携带结构化错误码 +- `CliError` 携带稳定错误码和用户可读消息；`Display` 仍只输出消息本体， + `to_json_value()` 提供 `--json` 可复用的 `{code, message}` 错误对象 ### 注册表与工具机制（registry.rs + util.rs） diff --git a/src/cli/docs/transfer.md b/src/cli/docs/transfer.md index db4d701..77cc160 100644 --- a/src/cli/docs/transfer.md +++ b/src/cli/docs/transfer.md @@ -18,14 +18,14 @@ pub fn send(file: &str, remote: Option<&str>, output: Option<&Path>, provider: & ``` - **委派逃生舱**：`QTDATA_CLI` 环境变量设置时，委派给外部 CLI（`transfer receive/send` 子命令）——测试与部署场景使用 -- **进程内路径**：默认走进程内 provider（tokio runtime + StorageProvider） +- **进程内路径**：默认走进程内 provider（tokio runtime + Storage） - 错误类型收敛为 `CliError` -## StorageProvider trait +## Storage trait ```rust #[async_trait] -pub trait StorageProvider: Send + Sync { +pub trait Storage: Send + Sync { fn name(&self) -> &'static str; async fn send(&self, local_path: &str, remote_path: &str) -> Result; async fn receive(&self, url: &str, local_path: &str) -> Result<(), String>; @@ -61,7 +61,7 @@ pub trait StorageProvider: Send + Sync { ## 添加新平台 -1. 新建 `storage/.rs`，实现 `StorageProvider` trait +1. 新建 `storage/.rs`，实现 `Storage` trait 2. 在 `storage/mod.rs` 注册（`from_name` / `detect`） 3. 认证环境变量约定见上表 4. 需要 mock 的平台遵循 `*_with_base` 注入约定并补 wiremock 测试（`tests/storage_test.rs` 参考） diff --git a/src/cli/docs/version.md b/src/cli/docs/version.md index 1c21a61..494a148 100644 --- a/src/cli/docs/version.md +++ b/src/cli/docs/version.md @@ -17,7 +17,7 @@ qtcloud-data version diff # 比较两个版本 - 优先 `spec/`（`-blueprint.cue`），回退旧 `blueprint/`（`.cue`） - 依赖 git 可用 -## 命令降级（v0.3 移除顶层） +## 命令降级 顶层 `qtcloud-data version` 已废弃，主入口改为 `qtcloud-data spec version`： @@ -27,4 +27,4 @@ qtcloud-data spec version show qtcloud-data spec version diff ``` -顶层 `version` 命令在 v0.3 移除前保留（帮助已标注废弃与替代入口）。 +顶层 `version` 命令在兼容迁移期间保留（帮助已标注废弃与替代入口）；新代码应使用 `spec version`。 diff --git a/src/cli/src/doctor.rs b/src/cli/src/doctor.rs index 03366c4..f4679ce 100644 --- a/src/cli/src/doctor.rs +++ b/src/cli/src/doctor.rs @@ -7,6 +7,7 @@ use std::io::{self, Write}; use std::path::{Path, PathBuf}; use crate::error::CliError; +use crate::runtime; #[derive(Args)] pub struct DoctorArgs { @@ -98,15 +99,24 @@ fn checks_with_dirs(dirs: &[DataDir]) -> Vec { check_command("git", true, "版本记录和协作事实源需要 git"), check_command("cargo", true, "CLI 开发和发布需要 cargo"), check_command("rustc", true, "CLI 编译需要 rustc"), - check_command( - "python3", - false, - "process 执行 Python pipeline 时会用到 python3", - ), - check_command("bash", false, "process 执行 shell pipeline 时会用到 bash"), check_command("cue", false, "cue 可选增强：CUE 模块化目录查看时使用"), ]; + for adapter in runtime::registered() { + if let Some(command) = adapter.doctor_command() { + let mut check = check_command( + command, + false, + format!( + "process 执行 {} pipeline 时会用到 {command}", + adapter.name() + ), + ); + check.name = format!("runtime:{}", adapter.name()); + checks.push(check); + } + } + for dir in dirs { checks.push(check_directory(&dir.path, &dir.name)); } @@ -306,7 +316,8 @@ fn create_data_dirs(dirs: &[DataDir]) -> Vec { } // ── 检查函数 ── -fn check_command(command: &str, required: bool, purpose: &str) -> Check { +fn check_command(command: &str, required: bool, purpose: impl Into) -> Check { + let purpose = purpose.into(); if command_exists(command) { Check::pass(command, format!("{purpose}: found")) } else if required { @@ -629,10 +640,19 @@ mod tests { let checks = checks_with_dirs(&dirs); let names: Vec<&str> = checks.iter().map(|c| c.name.as_str()).collect(); - // 6 个工具检查 - for tool in ["git", "cargo", "rustc", "python3", "bash", "cue"] { + // 基础工具检查 + for tool in ["git", "cargo", "rustc", "cue"] { assert!(names.contains(&tool), "缺工具检查: {tool}"); } + // 运行时检查由 runtime 注册表驱动，builtin 不需要外部命令 + for runtime in ["python", "r", "stata", "matlab", "bash"] { + assert!( + names + .iter() + .any(|name| name == &format!("runtime:{runtime}")), + "缺 runtime 检查: {runtime}" + ); + } // 2 个目录检查 assert!(names.contains(&"DRD")); assert!(names.contains(&"SPEC")); @@ -647,7 +667,7 @@ mod tests { ] { assert!(names.contains(&env), "缺 env 检查: {env}"); } - assert_eq!(checks.len(), 6 + 2 + 6); + assert_eq!(checks.len(), 4 + 5 + 2 + 6); std::fs::remove_dir_all(&root).ok(); } diff --git a/src/cli/src/error.rs b/src/cli/src/error.rs index 415b7fb..9734563 100644 --- a/src/cli/src/error.rs +++ b/src/cli/src/error.rs @@ -3,18 +3,50 @@ use std::fmt; use std::io; -/// CLI 统一错误，携带用户可读消息。 +/// 未显式分类时使用的通用错误码。 +pub const DEFAULT_ERROR_CODE: &str = "cli_error"; + +/// `io::Error` 转换使用的错误码。 +pub const IO_ERROR_CODE: &str = "io_error"; + +/// CLI 统一错误，携带稳定错误码和用户可读消息。 #[derive(Debug)] pub struct CliError { + code: String, message: String, } impl CliError { + /// 创建通用 CLI 错误，保持旧有文本错误行为。 pub fn new(message: impl Into) -> Self { + Self::with_code(DEFAULT_ERROR_CODE, message) + } + + /// 使用调用方指定的稳定错误码创建错误。 + pub fn with_code(code: impl Into, message: impl Into) -> Self { Self { + code: code.into(), message: message.into(), } } + + /// 返回机器可读错误码。 + pub fn code(&self) -> &str { + &self.code + } + + /// 返回用户可读错误消息。 + pub fn message(&self) -> &str { + &self.message + } + + /// 返回供 `--json` 输出复用的错误对象。 + pub fn to_json_value(&self) -> serde_json::Value { + serde_json::json!({ + "code": self.code, + "message": self.message, + }) + } } impl fmt::Display for CliError { @@ -27,7 +59,7 @@ impl std::error::Error for CliError {} impl From for CliError { fn from(err: io::Error) -> Self { - Self::new(err.to_string()) + Self::with_code(IO_ERROR_CODE, err.to_string()) } } @@ -83,4 +115,37 @@ mod tests { assert_eq!(b.to_string(), "borrowed"); assert_eq!(c.to_string(), "fmt 42"); } + + #[test] + fn cli_error_exposes_default_code_and_json_value() { + let err = CliError::new("pipeline failed"); + + assert_eq!(err.code(), "cli_error"); + assert_eq!(err.message(), "pipeline failed"); + assert_eq!( + err.to_json_value(), + serde_json::json!({ + "code": "cli_error", + "message": "pipeline failed" + }) + ); + } + + #[test] + fn cli_error_with_code_preserves_text_display() { + let err = CliError::with_code("not_found", "missing blueprint"); + + assert_eq!(err.code(), "not_found"); + assert_eq!(err.to_string(), "missing blueprint"); + assert_eq!(err.to_json_value()["code"], "not_found"); + } + + #[test] + fn io_error_uses_io_code() { + let io_err = io::Error::new(io::ErrorKind::NotFound, "no such file"); + let err: CliError = io_err.into(); + + assert_eq!(err.code(), "io_error"); + assert_eq!(err.to_json_value()["message"], "no such file"); + } } diff --git a/src/cli/src/implementation/catalog.rs b/src/cli/src/implementation/catalog.rs index ddd39bc..03e71a4 100644 --- a/src/cli/src/implementation/catalog.rs +++ b/src/cli/src/implementation/catalog.rs @@ -6,6 +6,7 @@ use std::fmt; use std::path::{Path, PathBuf}; use std::str::FromStr; +use crate::OutputMode; use crate::error::CliError; use crate::registry; use crate::util; @@ -249,9 +250,13 @@ pub fn register_volume_in( // ── 命令（run / list / show / add / rm） ── /// catalog 命令入口（list / show / add / rm）。 pub fn run(args: &CatalogArgs) -> Result<(), CliError> { + run_with_mode(args, OutputMode::Text) +} + +pub fn run_with_mode(args: &CatalogArgs, mode: OutputMode) -> Result<(), CliError> { match &args.action { - CatalogAction::List => list(), - CatalogAction::Show { name } => show(name), + CatalogAction::List => list(mode), + CatalogAction::Show { name } => show(name, mode), CatalogAction::Add { path, name, @@ -270,41 +275,79 @@ pub fn run(args: &CatalogArgs) -> Result<(), CliError> { } } -fn list() -> Result<(), CliError> { +fn list(mode: OutputMode) -> Result<(), CliError> { let registry = open_registry(); if registry.is_empty() { - println!("catalog 为空"); + if mode == OutputMode::Text { + println!("catalog 为空"); + } else { + println!( + "{}", + serde_json::json!({ + "ok": true, + "command": "catalog list", + "items": [], + }) + ); + } return Ok(()); } - println!("Volume:"); - for v in registry.entries().values() { - let status_icon = match v.status { - VolumeStatus::Received => "📥", - VolumeStatus::Processing => "⏳", - VolumeStatus::Processed => "✅", - VolumeStatus::Delivered => "📤", - VolumeStatus::Unknown => "📄", - }; - println!(" {status_icon} {} ({})", v.name, v.path); + match mode { + OutputMode::Text => { + println!("Volume:"); + for v in registry.entries().values() { + let status_icon = match v.status { + VolumeStatus::Received => "📥", + VolumeStatus::Processing => "⏳", + VolumeStatus::Processed => "✅", + VolumeStatus::Delivered => "📤", + VolumeStatus::Unknown => "📄", + }; + println!(" {status_icon} {} ({})", v.name, v.path); + } + } + OutputMode::Json => { + let items: Vec<_> = registry.entries().values().collect(); + println!( + "{}", + serde_json::json!({ + "ok": true, + "command": "catalog list", + "items": items, + }) + ); + } } Ok(()) } -fn show(name: &str) -> Result<(), CliError> { +fn show(name: &str, mode: OutputMode) -> Result<(), CliError> { let registry = open_registry(); match registry.get(name) { Some(v) => { - println!("名称: {}", v.name); - println!("路径: {}", v.path); - println!("大小: {}", format_size(v.size)); - println!("接收时间: {}", v.received_at); - println!("状态: {}", v.status); - println!("产物类型: {}", v.artifact_type); - if let Some(p) = &v.provider { - println!("Provider: {p}"); - } - if let Some(s) = &v.source { - println!("来源: {s}"); + match mode { + OutputMode::Text => { + println!("名称: {}", v.name); + println!("路径: {}", v.path); + println!("大小: {}", format_size(v.size)); + println!("接收时间: {}", v.received_at); + println!("状态: {}", v.status); + println!("产物类型: {}", v.artifact_type); + if let Some(p) = &v.provider { + println!("Provider: {p}"); + } + if let Some(s) = &v.source { + println!("来源: {s}"); + } + } + OutputMode::Json => println!( + "{}", + serde_json::json!({ + "ok": true, + "command": "catalog show", + "volume": v, + }) + ), } Ok(()) } @@ -560,7 +603,7 @@ mod tests { unsafe { std::env::set_var("CATALOG_DIR", &catalog_dir); } - let err = show("ghost").unwrap_err(); + let err = show("ghost", OutputMode::Text).unwrap_err(); unsafe { std::env::remove_var("CATALOG_DIR"); } @@ -627,7 +670,7 @@ mod tests { std::env::set_var("CATALOG_DIR", &catalog_dir); } seed_volume(&catalog_dir); - let result = list(); + let result = list(OutputMode::Text); unsafe { std::env::remove_var("CATALOG_DIR"); } @@ -646,7 +689,7 @@ mod tests { unsafe { std::env::set_var("CATALOG_DIR", &catalog_dir); } - let result = list(); + let result = list(OutputMode::Text); unsafe { std::env::remove_var("CATALOG_DIR"); } @@ -666,7 +709,7 @@ mod tests { std::env::set_var("CATALOG_DIR", &catalog_dir); } let name = seed_volume(&catalog_dir); - let result = show(&name); + let result = show(&name, OutputMode::Text); unsafe { std::env::remove_var("CATALOG_DIR"); } diff --git a/src/cli/src/implementation/pipeline.rs b/src/cli/src/implementation/pipeline.rs index 296d526..888ffa9 100644 --- a/src/cli/src/implementation/pipeline.rs +++ b/src/cli/src/implementation/pipeline.rs @@ -5,6 +5,7 @@ use serde_json::Value; use std::path::{Path, PathBuf}; use std::process::Command; +use crate::OutputMode; use crate::error::CliError; use crate::util::collect_defined_names; @@ -29,23 +30,24 @@ pub enum PipelineAction { /// 管道查看命令入口（list / show）。 pub fn run(args: &PipelineArgs) -> Result<(), CliError> { + run_with_mode(args, OutputMode::Text) +} + +pub fn run_with_mode(args: &PipelineArgs, mode: OutputMode) -> Result<(), CliError> { let dir = std::env::var("PIPELINE_DIR").unwrap_or_else(|_| ".quanttide/data/pipeline".to_string()); match &args.action { - PipelineAction::List => cmd_list(&dir), - PipelineAction::Show { name } => cmd_show(&dir, name), + PipelineAction::List => cmd_list(&dir, mode), + PipelineAction::Show { name } => cmd_show(&dir, name, mode), } } -fn cmd_list(dir: &str) -> Result<(), CliError> { +fn cmd_list(dir: &str, mode: OutputMode) -> Result<(), CliError> { let dir_path = Path::new(dir); if dir_path.is_dir() { let names = definition_names(dir_path); - println!("可用的 Pipeline:"); - for name in names { - println!(" - {name}"); - } + render_list(&names, mode); return Ok(()); } @@ -59,19 +61,16 @@ fn cmd_list(dir: &str) -> Result<(), CliError> { let value: Value = serde_json::from_slice(&output.stdout) .map_err(|e| CliError::new(format!("cue 输出不是合法 JSON: {e}")))?; let names = collect_defined_names(&value); - println!("可用的 Pipeline:"); - for name in names { - println!(" - {name}"); - } + render_list(&names, mode); Ok(()) } -fn cmd_show(dir: &str, name: &str) -> Result<(), CliError> { +fn cmd_show(dir: &str, name: &str, mode: OutputMode) -> Result<(), CliError> { let dir_path = Path::new(dir); if let Some(path) = find_definition(dir_path, name) { let content = std::fs::read_to_string(&path) .map_err(|err| CliError::new(format!("读取 Pipeline 失败: {err}")))?; - println!("{content}"); + render_show(name, &content, mode)?; return Ok(()); } @@ -83,11 +82,49 @@ fn cmd_show(dir: &str, name: &str) -> Result<(), CliError> { } let value: Value = serde_json::from_slice(&output.stdout) .map_err(|e| CliError::new(format!("cue 输出不是合法 JSON: {e}")))?; - println!( - "{}", - serde_json::to_string_pretty(&value) - .map_err(|e| CliError::new(format!("序列化失败: {e}")))? - ); + let content = serde_json::to_string_pretty(&value) + .map_err(|e| CliError::new(format!("序列化失败: {e}")))?; + render_show(name, &content, mode)?; + Ok(()) +} + +fn render_list(names: &[String], mode: OutputMode) { + match mode { + OutputMode::Text => { + println!("可用的 Pipeline:"); + for name in names { + println!(" - {name}"); + } + } + OutputMode::Json => println!( + "{}", + serde_json::json!({ + "ok": true, + "command": "pipeline list", + "items": names, + }) + ), + } +} + +fn render_show(name: &str, content: &str, mode: OutputMode) -> Result<(), CliError> { + match mode { + OutputMode::Text => println!("{content}"), + OutputMode::Json => { + let definition: serde_json::Value = serde_yaml::from_str(content) + .or_else(|_| serde_json::from_str(content)) + .map_err(|err| CliError::new(format!("Pipeline 不是合法结构化数据: {err}")))?; + println!( + "{}", + serde_json::json!({ + "ok": true, + "command": "pipeline show", + "name": name, + "pipeline": definition, + }) + ); + } + } Ok(()) } @@ -174,7 +211,7 @@ mod tests { let _guard = ENV_LOCK.lock().unwrap(); let (root, old_path) = fake_cue_env(); - let result = cmd_list(root.to_str().unwrap()); + let result = cmd_list(root.to_str().unwrap(), OutputMode::Text); restore_path(old_path); assert!(result.is_ok(), "{result:?}"); @@ -186,7 +223,7 @@ mod tests { let _guard = ENV_LOCK.lock().unwrap(); let (root, old_path) = fake_cue_env(); - let result = cmd_show(root.to_str().unwrap(), "pipe1"); + let result = cmd_show(root.to_str().unwrap(), "pipe1", OutputMode::Text); restore_path(old_path); assert!(result.is_ok(), "{result:?}"); @@ -206,7 +243,7 @@ mod tests { unsafe { std::env::set_var("PATH", &empty_bin); } - let result = cmd_list(root.to_str().unwrap()); + let result = cmd_list(root.to_str().unwrap(), OutputMode::Text); restore_path(old_path); assert!(result.is_ok(), "{result:?}"); @@ -225,7 +262,7 @@ mod tests { unsafe { std::env::set_var("PATH", &empty_bin); } - let result = cmd_show(root.to_str().unwrap(), "normalize"); + let result = cmd_show(root.to_str().unwrap(), "normalize", OutputMode::Text); restore_path(old_path); assert!(result.is_ok(), "{result:?}"); diff --git a/src/cli/src/lib.rs b/src/cli/src/lib.rs index 6927600..d3faa79 100644 --- a/src/cli/src/lib.rs +++ b/src/cli/src/lib.rs @@ -1,5 +1,7 @@ //! qtcloud-data-cli 库入口：模块注册与测试共享支持。 +use clap::Subcommand; + pub mod doctor; pub mod error; pub mod implementation; @@ -11,6 +13,93 @@ pub mod stage; pub mod storage; pub mod util; +/// 命令输出模式；JSON 模式逐个命令迁移，文本模式保持现有行为。 +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub enum OutputMode { + #[default] + Text, + Json, +} + +/// CLI 子命令集合，供二进制入口解析，也供库级分发测试复用。 +#[derive(Subcommand)] +pub enum Commands { + /// 从客户上下文澄清需求 → 生成 DRD（数据需求文档） + Clarify(stage::clarify::ClarifyArgs), + /// 设计 Specification（Contract + Blueprint）← 从 DRD + Design(stage::design::DesignArgs), + /// 审计 Specification 完整性和一致性 + Review(review::ReviewArgs), + /// Specification YAML 契约工具 + Spec(spec::SpecArgs), + /// 规格版本管理（已废弃：v0.3 移除，改用 `spec version`） + Version(spec::version::SpecVersionArgs), + /// 检查本机 DataOps 环境 + Doctor(doctor::DoctorArgs), + /// 蓝图管理（list / show） + Blueprint(spec::blueprint::BlueprintArgs), + /// 契约查看 + Contract(spec::contract::ContractArgs), + /// 管道管理 + Pipeline(implementation::pipeline::PipelineArgs), + /// 数据目录 + Catalog(implementation::catalog::CatalogArgs), + /// 从 Specification 生成代码实现 + Implement(stage::implement::ImplementArgs), + /// 编排流程（receive → pipeline → send） + Process(stage::process::ProcessArgs), + /// 数据传输（send / receive） + Transfer(stage::transfer::TransferArgs), +} + +/// 使用生产环境默认 LLM 执行一个 CLI 子命令。 +pub fn run_command(command: &Commands) -> Result<(), error::CliError> { + run_command_with_mode(command, OutputMode::Text) +} + +/// 使用调用方注入的 LLM 执行一个 CLI 子命令，便于测试和集成。 +pub fn run_command_with_llm( + command: &Commands, + llm: quanttide_agent::LLM, +) -> Result<(), error::CliError> { + dispatch_command(command, Some(llm), OutputMode::Text) +} + +/// 使用指定输出模式执行一个 CLI 子命令。 +pub fn run_command_with_mode(command: &Commands, mode: OutputMode) -> Result<(), error::CliError> { + dispatch_command(command, None, mode) +} + +fn dispatch_command( + command: &Commands, + injected_llm: Option, + mode: OutputMode, +) -> Result<(), error::CliError> { + match command { + Commands::Clarify(args) => { + stage::clarify::ClarifyHandler::new(injected_llm.unwrap_or_default()).run(args) + } + Commands::Design(args) => { + stage::design::DesignHandler::new(injected_llm.unwrap_or_default()).run(args) + } + Commands::Review(args) => { + review::ReviewHandler::new(injected_llm.unwrap_or_default()).run(args) + } + Commands::Spec(args) => spec::run_with_mode(args, mode), + Commands::Version(args) => spec::version::run(args), + Commands::Doctor(args) => doctor::run(args), + Commands::Blueprint(args) => spec::blueprint::run_with_mode(args, mode), + Commands::Contract(args) => spec::contract::run(args), + Commands::Pipeline(args) => implementation::pipeline::run_with_mode(args, mode), + Commands::Catalog(args) => implementation::catalog::run_with_mode(args, mode), + Commands::Implement(args) => { + stage::implement::ImplementHandler::new(injected_llm.unwrap_or_default()).run(args) + } + Commands::Process(args) => stage::process::run(args), + Commands::Transfer(args) => stage::transfer::run(args), + } +} + /// 测试共享的全局环境变量锁：各模块测试直接 `std::env::set_var` 时统一互斥， /// 避免并行执行互相污染进程级环境变量。 #[cfg(test)] @@ -104,3 +193,53 @@ pub mod test_support { } } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::spec::{SpecAction, SpecArgs}; + use crate::stage::clarify::{ClarifyAction, ClarifyArgs}; + + #[test] + fn command_dispatch_accepts_injected_llm() { + let _guard = ENV_LOCK.lock().unwrap(); + let root = test_support::temp_dir("qtcloud-command-dispatch"); + let chat = root.join("chat.md"); + let drd_dir = root.join("drd"); + std::fs::write(&chat, "客户需要清洗订单数据\n").unwrap(); + + unsafe { + std::env::set_var("DRD_DIR", &drd_dir); + } + + let command = Commands::Clarify(ClarifyArgs { + action: ClarifyAction::FromChat { + input: chat.to_string_lossy().into_owned(), + }, + }); + run_command_with_llm(&command, test_support::fake_llm("# 订单数据 DRD\n")).unwrap(); + + unsafe { + std::env::remove_var("DRD_DIR"); + } + + assert_eq!( + std::fs::read_to_string(drd_dir.join("chat.md")).unwrap(), + "# 订单数据 DRD\n" + ); + } + + #[test] + fn command_dispatch_uses_default_path_for_non_llm_command() { + let root = test_support::temp_dir("qtcloud-command-dispatch-default"); + let missing = root.join("missing.yaml"); + let command = Commands::Spec(SpecArgs { + action: SpecAction::Validate { + input: missing.to_string_lossy().into_owned(), + }, + }); + + let error = run_command(&command).unwrap_err(); + assert!(error.to_string().contains("无法读取 YAML")); + } +} diff --git a/src/cli/src/main.rs b/src/cli/src/main.rs index a0ebda3..457f4b2 100644 --- a/src/cli/src/main.rs +++ b/src/cli/src/main.rs @@ -1,81 +1,39 @@ -//! 二进制入口：CLI 参数解析与命令分发（run_command）。 +//! 二进制入口：CLI 参数解析，命令分发由库层 `run_command` 负责。 -use clap::{Parser, Subcommand}; -use qtcloud_data_cli::error::CliError; -use qtcloud_data_cli::implementation::{catalog, pipeline}; -use qtcloud_data_cli::stage::{clarify, design, implement, process, transfer}; -use qtcloud_data_cli::{doctor, review, spec}; +use clap::Parser; +use qtcloud_data_cli::{Commands, OutputMode, run_command_with_mode}; #[derive(Parser)] #[command(name = "qtcloud-data", about = "量潮数据云 CLI")] struct Cli { + /// 以机器可读 JSON 输出命令错误 + #[arg(long, global = true)] + json: bool, + #[command(subcommand)] command: Commands, } -#[derive(Subcommand)] -enum Commands { - /// 从客户上下文澄清需求 → 生成 DRD（数据需求文档） - Clarify(clarify::ClarifyArgs), - /// 设计 Specification（Contract + Blueprint）← 从 DRD - Design(design::DesignArgs), - /// 审计 Specification 完整性和一致性 - Review(review::ReviewArgs), - /// Specification YAML 契约工具 - Spec(spec::SpecArgs), - /// 规格版本管理（已废弃：v0.3 移除，改用 `spec version`） - Version(spec::version::SpecVersionArgs), - /// 检查本机 DataOps 环境 - Doctor(doctor::DoctorArgs), - /// 蓝图管理（list / show） - Blueprint(spec::blueprint::BlueprintArgs), - /// 契约查看 - Contract(spec::contract::ContractArgs), - /// 管道管理 - Pipeline(pipeline::PipelineArgs), - /// 数据目录 - Catalog(catalog::CatalogArgs), - /// 从 Specification 生成代码实现 - Implement(implement::ImplementArgs), - /// 编排流程（receive → pipeline → send） - Process(process::ProcessArgs), - /// 数据传输（send / receive） - Transfer(transfer::TransferArgs), -} - fn main() { let cli = Cli::parse(); - if let Err(err) = run_command(&cli.command) { - eprintln!("错误: {err}"); - std::process::exit(1); - } -} - -/// 命令分发：返回 `Result<(), CliError>` 的命令由顶层统一格式化； -/// 其余命令保持内部 exit 处理（逐步迁移中）。 -/// LLM 命令通过 Handler 注入：生产路径构造 `LLM::default()`，测试替换为 fake。 -fn run_command(command: &Commands) -> Result<(), CliError> { - match command { - Commands::Clarify(args) => { - clarify::ClarifyHandler::new(quanttide_agent::LLM::default()).run(args) - } - Commands::Design(args) => { - design::DesignHandler::new(quanttide_agent::LLM::default()).run(args) - } - Commands::Review(args) => { - review::ReviewHandler::new(quanttide_agent::LLM::default()).run(args) + if let Err(err) = run_command_with_mode( + &cli.command, + if cli.json { + OutputMode::Json + } else { + OutputMode::Text + }, + ) { + if cli.json { + println!( + "{}", + serde_json::json!({ + "error": err.to_json_value(), + }) + ); + } else { + eprintln!("错误: {err}"); } - Commands::Spec(args) => spec::run(args), - Commands::Version(args) => spec::version::run(args), - Commands::Doctor(args) => doctor::run(args), - Commands::Blueprint(args) => spec::blueprint::run(args), - Commands::Contract(args) => spec::contract::run(args), - Commands::Pipeline(args) => pipeline::run(args), - Commands::Catalog(args) => catalog::run(args), - Commands::Implement(args) => { - implement::ImplementHandler::new(quanttide_agent::LLM::default()).run(args) - } - Commands::Process(args) => process::run(args), - Commands::Transfer(args) => transfer::run(args), + std::process::exit(1); } } diff --git a/src/cli/src/review.rs b/src/cli/src/review.rs index 317623d..a3a10f2 100644 --- a/src/cli/src/review.rs +++ b/src/cli/src/review.rs @@ -79,6 +79,7 @@ impl ReviewHandler { } #[cfg(test)] +#[allow(clippy::items_after_test_module)] mod tests { use super::*; diff --git a/src/cli/src/runtime/builtin.rs b/src/cli/src/runtime/builtin.rs new file mode 100644 index 0000000..992a717 --- /dev/null +++ b/src/cli/src/runtime/builtin.rs @@ -0,0 +1,43 @@ +//! 内置运行时：实现 `builtin:copy` 等无需外部解释器的资源。 + +use std::fs; +use std::path::Path; + +use super::Runtime; + +pub struct BuiltinRuntime; + +impl Runtime for BuiltinRuntime { + fn name(&self) -> &'static str { + "builtin" + } + + fn extension(&self) -> &'static str { + "builtin" + } + + fn command(&self) -> &'static str { + "" + } + + fn doctor_command(&self) -> Option<&'static str> { + None + } + + fn execute_with( + &self, + script: &Path, + input: &str, + output: &str, + _work_dir: &str, + _cmd: &str, + ) -> Result { + if script.file_name().and_then(|name| name.to_str()) != Some("copy") { + return Err(format!("不支持的 builtin 资源: {}", script.display())); + } + + fs::copy(input, output) + .map(|_| String::new()) + .map_err(|err| format!("builtin:copy 执行失败: {err}")) + } +} diff --git a/src/cli/src/runtime/matlab.rs b/src/cli/src/runtime/matlab.rs new file mode 100644 index 0000000..39bbc7e --- /dev/null +++ b/src/cli/src/runtime/matlab.rs @@ -0,0 +1,138 @@ +//! Matlab 运行时：函数 codegen prompt 与脚本执行。 + +use std::path::Path; + +use super::Runtime; + +pub struct MatlabRuntime; + +impl Runtime for MatlabRuntime { + fn name(&self) -> &'static str { + "matlab" + } + + fn extension(&self) -> &'static str { + "m" + } + + fn command(&self) -> &'static str { + "matlab" + } + + fn step_prompt( + &self, + step_name: &str, + from_desc: &str, + to_desc: &str, + step_desc: &str, + prev_functions: &str, + ) -> String { + format!( + r#"你是一个 Matlab 数据处理工程师。请根据以下步骤生成一个 Matlab 函数。 + +函数名: {step_name} +输入: {from_desc} +输出: {to_desc} +处理逻辑: {step_desc} + +前置函数: +{prev} + +要求: +1. 函数名使用 snake_case: {func_name} +2. 函数签名应接收输入数据并返回输出数据 +3. 代码可以由 Matlab 执行 +4. 只输出 Matlab 代码，不要解释 + +代码: +"#, + prev = if prev_functions.is_empty() { + "无（这是第一步）" + } else { + prev_functions + }, + func_name = self.to_snake(step_name), + ) + } + + fn assemble_prompt( + &self, + project_name: &str, + all_functions: &str, + pipeline_desc: &str, + ) -> String { + format!( + r#"你是一个 Matlab 数据处理工程师。请将以下函数组装成完整可执行的 Matlab 脚本。 + +项目: {project_name} +管道: {pipeline_desc} + +函数: +{all_functions} + +要求: +1. 按管道顺序调用函数 +2. 通过命令行参数接收输入和输出路径 +3. 保存最终输出 +4. 只输出 Matlab 代码，不要解释 + +完整脚本: +"#, + ) + } + + fn extract(&self, response: &str) -> String { + extract_code(response) + } + + fn extract_signature(&self, code: &str, step_name: &str) -> String { + let snake = self.to_snake(step_name); + code.lines() + .map(str::trim) + .find(|line| line.starts_with("function")) + .map(str::to_string) + .unwrap_or_else(|| format!("function data = {snake}(data)")) + } + + fn to_snake(&self, value: &str) -> String { + value + .to_lowercase() + .replace([' ', '-', '.'], "_") + .replace("__", "_") + } + + fn execute_with( + &self, + script: &Path, + input: &str, + output: &str, + work_dir: &str, + cmd: &str, + ) -> Result { + let status = std::process::Command::new(cmd) + .args(["-batch"]) + .arg(format!("run('{}','{}')", input, output)) + .arg(script) + .current_dir(work_dir) + .status() + .map_err(|err| format!("执行 Matlab 脚本失败: {err}"))?; + if !status.success() { + return Err("Matlab 脚本执行失败".to_string()); + } + Ok(String::new()) + } +} + +fn extract_code(response: &str) -> String { + for marker in ["```matlab", "```m", "```"] { + if let Some(start) = response.find(marker) { + let code_start = start + marker.len(); + let code_end = response[code_start..] + .find("```") + .map(|offset| code_start + offset) + .unwrap_or(response.len()); + return response[code_start..code_end].trim().to_string(); + } + } + response.to_string() +} diff --git a/src/cli/src/runtime/mod.rs b/src/cli/src/runtime/mod.rs index 90da5e5..5d3a1a6 100644 --- a/src/cli/src/runtime/mod.rs +++ b/src/cli/src/runtime/mod.rs @@ -4,7 +4,11 @@ //! 模块名 = 概念名：`runtime::Runtime`、`runtime::PythonRuntime`（↔ `storage::Storage`、`storage::DropboxStorage`）。 pub mod bash; +pub mod builtin; +pub mod matlab; pub mod python; +pub mod r; +pub mod stata; use std::path::Path; @@ -23,6 +27,11 @@ pub trait Runtime: Send + Sync { /// 默认执行命令 fn command(&self) -> &'static str; + /// 外部解释器名称；builtin 等内置运行时不需要 PATH 检查。 + fn doctor_command(&self) -> Option<&'static str> { + Some(self.command()) + } + // ── codegen（默认不支持，codegen 语言覆盖）── /// 单步实现 prompt @@ -90,6 +99,9 @@ pub trait Runtime: Send + Sync { pub fn from_name(name: &str) -> Option> { match name { "python" => Some(Box::new(python::PythonRuntime)), + "r" => Some(Box::new(r::RRuntime)), + "stata" => Some(Box::new(stata::StataRuntime)), + "matlab" => Some(Box::new(matlab::MatlabRuntime)), _ => None, } } @@ -98,11 +110,26 @@ pub fn from_name(name: &str) -> Option> { pub fn from_ext(ext: &str) -> Option> { match ext { "py" => Some(Box::new(python::PythonRuntime)), + "r" => Some(Box::new(r::RRuntime)), + "do" => Some(Box::new(stata::StataRuntime)), + "m" => Some(Box::new(matlab::MatlabRuntime)), "sh" => Some(Box::new(bash::BashRuntime)), _ => None, } } +/// 返回所有已注册运行时，供 doctor 和集成层消费。 +pub fn registered() -> Vec> { + vec![ + Box::new(python::PythonRuntime), + Box::new(r::RRuntime), + Box::new(stata::StataRuntime), + Box::new(matlab::MatlabRuntime), + Box::new(bash::BashRuntime), + Box::new(builtin::BuiltinRuntime), + ] +} + #[cfg(test)] mod tests { use super::*; @@ -110,15 +137,38 @@ mod tests { #[test] fn from_name_registers_codegen_runtimes() { assert!(from_name("python").is_some()); + assert!(from_name("r").is_some()); + assert!(from_name("stata").is_some()); + assert!(from_name("matlab").is_some()); assert!(from_name("bash").is_none(), "bash 仅执行，不做 codegen"); - assert!(from_name("r").is_none(), "r 尚未实现"); } #[test] fn from_ext_registers_execution_runtimes() { assert!(from_ext("py").is_some()); + assert!(from_ext("r").is_some()); + assert!(from_ext("do").is_some()); + assert!(from_ext("m").is_some()); assert!(from_ext("sh").is_some()); - assert!(from_ext("r").is_none()); assert!(from_ext("csv").is_none()); } + + #[test] + fn registered_includes_builtin_without_exposing_it_as_codegen() { + let names: Vec<_> = registered().iter().map(|runtime| runtime.name()).collect(); + + assert_eq!( + names, + vec!["python", "r", "stata", "matlab", "bash", "builtin"] + ); + assert!(from_name("builtin").is_none()); + assert!( + registered() + .iter() + .find(|runtime| runtime.name() == "builtin") + .unwrap() + .doctor_command() + .is_none() + ); + } } diff --git a/src/cli/src/runtime/r.rs b/src/cli/src/runtime/r.rs new file mode 100644 index 0000000..4b5510e --- /dev/null +++ b/src/cli/src/runtime/r.rs @@ -0,0 +1,182 @@ +//! R 运行时：R codegen prompt 与脚本执行。 + +use std::path::Path; + +use super::Runtime; + +pub struct RRuntime; + +impl Runtime for RRuntime { + fn name(&self) -> &'static str { + "r" + } + + fn extension(&self) -> &'static str { + "r" + } + + fn command(&self) -> &'static str { + "Rscript" + } + + fn step_prompt( + &self, + step_name: &str, + from_desc: &str, + to_desc: &str, + step_desc: &str, + prev_functions: &str, + ) -> String { + format!( + r#"你是一个 R 数据处理工程师。请根据以下步骤描述，生成一个 R 函数。 + +函数名: {step_name} +输入: {from_desc} +输出: {to_desc} +处理逻辑: {step_desc} + +已生成的前置函数: +{prev} + +要求: +1. 函数名使用 snake_case: {func_name} +2. 函数接收上一步的数据并返回处理后的数据 +3. 使用清晰的 R 语法，可在 Rscript 中执行 +4. 添加注释说明输入输出 +5. 只输出 R 代码，不要解释 + +生成的函数: +"#, + prev = if prev_functions.is_empty() { + "无（这是第一步）" + } else { + prev_functions + }, + func_name = self.to_snake(step_name), + ) + } + + fn assemble_prompt( + &self, + project_name: &str, + all_functions: &str, + pipeline_desc: &str, + ) -> String { + format!( + r#"你是一个 R 数据处理工程师。请将以下函数组装成一个完整的可执行 R 脚本。 + +项目: {project_name} +管道: {pipeline_desc} + +函数列表: +{all_functions} + +要求: +1. 添加必要的 library/import 语句 +2. 使用 commandArgs(trailingOnly = TRUE) 接收输入和输出路径 +3. 按管道顺序调用函数 +4. 只输出 R 代码，不要解释 + +完整脚本: +"#, + ) + } + + fn extract(&self, response: &str) -> String { + extract_code(response, &["```r", "```R", "```"]) + } + + fn extract_signature(&self, code: &str, step_name: &str) -> String { + let snake = self.to_snake(step_name); + code.lines() + .map(str::trim) + .find(|line| line.starts_with("function ") || line.contains(" <- function")) + .map(str::to_string) + .unwrap_or_else(|| format!("{snake} <- function(data) {{ # {step_name}")) + } + + fn to_snake(&self, value: &str) -> String { + value + .to_lowercase() + .replace([' ', '-', '.'], "_") + .replace("__", "_") + } + + fn execute_with( + &self, + script: &Path, + input: &str, + output: &str, + work_dir: &str, + cmd: &str, + ) -> Result { + run_script(cmd, script, input, output, work_dir, "R") + } +} + +fn extract_code(response: &str, markers: &[&str]) -> String { + for marker in markers { + if let Some(start) = response.find(marker) { + let code_start = start + marker.len(); + let code_end = response[code_start..] + .find("```") + .map(|offset| code_start + offset) + .unwrap_or(response.len()); + return response[code_start..code_end].trim().to_string(); + } + } + response.to_string() +} + +fn run_script( + cmd: &str, + script: &Path, + input: &str, + output: &str, + work_dir: &str, + language: &str, +) -> Result { + let status = std::process::Command::new(cmd) + .arg(script) + .arg(input) + .arg(output) + .current_dir(work_dir) + .status() + .map_err(|err| format!("执行 {language} 脚本失败: {err}"))?; + if !status.success() { + return Err(format!("{language} 脚本执行失败")); + } + Ok(String::new()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn r_prompt_and_extract_are_language_specific() { + let runtime = RRuntime; + assert!( + runtime + .step_prompt("清洗 数据", "raw", "clean", "去空格", "") + .contains("R") + ); + assert!( + runtime + .step_prompt("清洗 数据", "raw", "clean", "去空格", "") + .contains("qing_") + || runtime + .step_prompt("clean data", "raw", "clean", "trim", "") + .contains("clean_data") + ); + assert_eq!( + runtime.extract("```r\nclean <- function(x) x\n```").trim(), + "clean <- function(x) x" + ); + assert!( + runtime + .extract_signature("x <- 1", "Clean Data") + .contains("clean_data") + ); + } +} diff --git a/src/cli/src/runtime/stata.rs b/src/cli/src/runtime/stata.rs new file mode 100644 index 0000000..f19af6d --- /dev/null +++ b/src/cli/src/runtime/stata.rs @@ -0,0 +1,147 @@ +//! Stata 运行时：do-file codegen prompt 与脚本执行。 + +use std::path::Path; + +use super::Runtime; + +pub struct StataRuntime; + +impl Runtime for StataRuntime { + fn name(&self) -> &'static str { + "stata" + } + + fn extension(&self) -> &'static str { + "do" + } + + fn command(&self) -> &'static str { + "stata" + } + + fn step_prompt( + &self, + step_name: &str, + from_desc: &str, + to_desc: &str, + step_desc: &str, + prev_functions: &str, + ) -> String { + format!( + r#"你是一个 Stata 数据处理工程师。请根据以下步骤描述生成一个 Stata do-file 片段。 + +步骤名: {step_name} +输入: {from_desc} +输出: {to_desc} +处理逻辑: {step_desc} + +前置步骤: +{prev} + +要求: +1. 使用 Stata 合法命令完成数据处理 +2. 使用 `args input output` 接收输入和输出路径 +3. 使用清晰的注释说明处理逻辑 +4. 只输出 Stata 代码，不要解释 + +代码: +"#, + prev = if prev_functions.is_empty() { + "无（这是第一步）" + } else { + prev_functions + }, + ) + } + + fn assemble_prompt( + &self, + project_name: &str, + all_functions: &str, + pipeline_desc: &str, + ) -> String { + format!( + r#"你是一个 Stata 数据处理工程师。请将以下 Stata 代码组装成完整可执行的 do-file。 + +项目: {project_name} +管道: {pipeline_desc} + +代码: +{all_functions} + +要求: +1. 文件开头使用 `args input output` +2. 按管道顺序执行每个步骤 +3. 最后保存输出数据 +4. 只输出 Stata 代码，不要解释 + +完整 do-file: +"#, + ) + } + + fn extract(&self, response: &str) -> String { + extract_code(response, &["```stata", "```do", "```"]) + } + + fn extract_signature(&self, code: &str, step_name: &str) -> String { + code.lines() + .map(str::trim) + .find(|line| line.starts_with("program define")) + .map(str::to_string) + .unwrap_or_else(|| format!("* {step_name}")) + } + + fn to_snake(&self, value: &str) -> String { + value + .to_lowercase() + .replace([' ', '-', '.'], "_") + .replace("__", "_") + } + + fn execute_with( + &self, + script: &Path, + input: &str, + output: &str, + work_dir: &str, + cmd: &str, + ) -> Result { + run_script(cmd, script, input, output, work_dir) + } +} + +fn extract_code(response: &str, markers: &[&str]) -> String { + for marker in markers { + if let Some(start) = response.find(marker) { + let code_start = start + marker.len(); + let code_end = response[code_start..] + .find("```") + .map(|offset| code_start + offset) + .unwrap_or(response.len()); + return response[code_start..code_end].trim().to_string(); + } + } + response.to_string() +} + +fn run_script( + cmd: &str, + script: &Path, + input: &str, + output: &str, + work_dir: &str, +) -> Result { + let status = std::process::Command::new(cmd) + .args(["-b", "do"]) + .arg(script) + .arg(input) + .arg(output) + .current_dir(work_dir) + .status() + .map_err(|err| format!("执行 Stata 脚本失败: {err}"))?; + if !status.success() { + return Err("Stata 脚本执行失败".to_string()); + } + Ok(String::new()) +} diff --git a/src/cli/src/spec/blueprint.rs b/src/cli/src/spec/blueprint.rs index 4980f36..eca923a 100644 --- a/src/cli/src/spec/blueprint.rs +++ b/src/cli/src/spec/blueprint.rs @@ -5,6 +5,7 @@ use serde_json::Value; use std::path::{Path, PathBuf}; use std::process::Command; +use crate::OutputMode; use crate::error::CliError; use crate::util::collect_defined_names; @@ -29,22 +30,23 @@ pub enum BlueprintAction { /// 蓝图查看命令入口（list / show），返回 `Result<(), CliError>`。 pub fn run(args: &BlueprintArgs) -> Result<(), CliError> { + run_with_mode(args, OutputMode::Text) +} + +pub fn run_with_mode(args: &BlueprintArgs, mode: OutputMode) -> Result<(), CliError> { let dir = crate::util::blueprint_dir(); match &args.action { - BlueprintAction::List => cmd_list(&dir), - BlueprintAction::Show { name } => cmd_show(&dir, name), + BlueprintAction::List => cmd_list(&dir, mode), + BlueprintAction::Show { name } => cmd_show(&dir, name, mode), } } -fn cmd_list(dir: &str) -> Result<(), CliError> { +fn cmd_list(dir: &str, mode: OutputMode) -> Result<(), CliError> { let dir_path = Path::new(dir); if dir_path.is_dir() { let names = definition_names(dir_path); - println!("可用的 Blueprint:"); - for name in names { - println!(" - {name}"); - } + render_list(&names, mode); return Ok(()); } @@ -59,19 +61,16 @@ fn cmd_list(dir: &str) -> Result<(), CliError> { let value: Value = serde_json::from_slice(&output.stdout) .map_err(|e| CliError::new(format!("cue 输出不是合法 JSON: {e}")))?; let names = collect_defined_names(&value); - println!("可用的 Blueprint:"); - for name in names { - println!(" - {name}"); - } + render_list(&names, mode); Ok(()) } -fn cmd_show(dir: &str, name: &str) -> Result<(), CliError> { +fn cmd_show(dir: &str, name: &str, mode: OutputMode) -> Result<(), CliError> { let dir_path = Path::new(dir); if let Some(path) = find_definition(dir_path, name) { let content = std::fs::read_to_string(&path) .map_err(|err| CliError::new(format!("读取 Blueprint 失败: {err}")))?; - println!("{content}"); + render_show(name, &content, mode)?; return Ok(()); } @@ -83,11 +82,49 @@ fn cmd_show(dir: &str, name: &str) -> Result<(), CliError> { } let value: Value = serde_json::from_slice(&output.stdout) .map_err(|e| CliError::new(format!("cue 输出不是合法 JSON: {e}")))?; - println!( - "{}", - serde_json::to_string_pretty(&value) - .map_err(|e| CliError::new(format!("序列化失败: {e}")))? - ); + let content = serde_json::to_string_pretty(&value) + .map_err(|e| CliError::new(format!("序列化失败: {e}")))?; + render_show(name, &content, mode)?; + Ok(()) +} + +fn render_list(names: &[String], mode: OutputMode) { + match mode { + OutputMode::Text => { + println!("可用的 Blueprint:"); + for name in names { + println!(" - {name}"); + } + } + OutputMode::Json => println!( + "{}", + serde_json::json!({ + "ok": true, + "command": "blueprint list", + "items": names, + }) + ), + } +} + +fn render_show(name: &str, content: &str, mode: OutputMode) -> Result<(), CliError> { + match mode { + OutputMode::Text => println!("{content}"), + OutputMode::Json => { + let definition: serde_json::Value = serde_yaml::from_str(content) + .or_else(|_| serde_json::from_str(content)) + .map_err(|err| CliError::new(format!("Blueprint 不是合法结构化数据: {err}")))?; + println!( + "{}", + serde_json::json!({ + "ok": true, + "command": "blueprint show", + "name": name, + "blueprint": definition, + }) + ); + } + } Ok(()) } @@ -175,7 +212,7 @@ mod tests { let _guard = ENV_LOCK.lock().unwrap(); let (root, old_path) = fake_cue_env(); - let result = cmd_list(root.to_str().unwrap()); + let result = cmd_list(root.to_str().unwrap(), OutputMode::Text); restore_path(old_path); assert!(result.is_ok(), "{result:?}"); @@ -187,7 +224,7 @@ mod tests { let _guard = ENV_LOCK.lock().unwrap(); let (root, old_path) = fake_cue_env(); - let result = cmd_show(root.to_str().unwrap(), "demo"); + let result = cmd_show(root.to_str().unwrap(), "demo", OutputMode::Text); restore_path(old_path); assert!(result.is_ok(), "{result:?}"); @@ -207,7 +244,7 @@ mod tests { unsafe { std::env::set_var("PATH", &empty_bin); } - let result = cmd_list(root.to_str().unwrap()); + let result = cmd_list(root.to_str().unwrap(), OutputMode::Text); restore_path(old_path); assert!(result.is_ok(), "{result:?}"); @@ -226,7 +263,7 @@ mod tests { unsafe { std::env::set_var("PATH", &empty_bin); } - let result = cmd_show(root.to_str().unwrap(), "demo"); + let result = cmd_show(root.to_str().unwrap(), "demo", OutputMode::Text); restore_path(old_path); assert!(result.is_ok(), "{result:?}"); diff --git a/src/cli/src/spec/mod.rs b/src/cli/src/spec/mod.rs index 24dd192..aced7e2 100644 --- a/src/cli/src/spec/mod.rs +++ b/src/cli/src/spec/mod.rs @@ -8,6 +8,7 @@ use clap::{Args, Subcommand}; use serde::{Deserialize, Serialize}; use std::path::{Path, PathBuf}; +use crate::OutputMode; use crate::error::CliError; pub const SPEC_API_VERSION: &str = "qtcloud.quanttide.com/v1alpha1"; @@ -106,9 +107,14 @@ impl Specification { // ── 命令（wrap / validate） ── /// Specification 工具命令入口（wrap / validate）。 pub fn run(args: &SpecArgs) -> Result<(), CliError> { + run_with_mode(args, OutputMode::Text) +} + +/// Specification 工具命令入口，按输出模式执行已迁移的子命令。 +pub fn run_with_mode(args: &SpecArgs, mode: OutputMode) -> Result<(), CliError> { match &args.action { - SpecAction::Wrap { input, output } => wrap_file(input, output), - SpecAction::Validate { input } => validate_file(input), + SpecAction::Wrap { input, output } => wrap_file(input, output, mode), + SpecAction::Validate { input } => validate_file(input, mode), SpecAction::Version(args) => version::run(args), } } @@ -160,7 +166,7 @@ pub fn parse_specification_yaml(yaml: &str) -> Result { Ok(spec) } -fn wrap_file(input: &str, output: &Option) -> Result<(), CliError> { +fn wrap_file(input: &str, output: &Option, mode: OutputMode) -> Result<(), CliError> { let content = std::fs::read_to_string(input) .map_err(|err| CliError::new(format!("无法读取 YAML: {err}")))?; @@ -180,11 +186,21 @@ fn wrap_file(input: &str, output: &Option) -> Result<(), CliError> { std::fs::write(&output_path, wrapped) .map_err(|err| CliError::new(format!("写入 Specification YAML 失败: {err}")))?; - println!("已生成: {}", output_path.display()); + match mode { + OutputMode::Text => println!("已生成: {}", output_path.display()), + OutputMode::Json => println!( + "{}", + serde_json::json!({ + "ok": true, + "command": "spec wrap", + "output": output_path, + }) + ), + } Ok(()) } -fn validate_file(input: &str) -> Result<(), CliError> { +fn validate_file(input: &str, mode: OutputMode) -> Result<(), CliError> { let content = std::fs::read_to_string(input) .map_err(|err| CliError::new(format!("无法读取 YAML: {err}")))?; @@ -201,7 +217,17 @@ fn validate_file(input: &str) -> Result<(), CliError> { ))); } - println!("Specification OK: {}", blueprint.name); + match mode { + OutputMode::Text => println!("Specification OK: {}", blueprint.name), + OutputMode::Json => println!( + "{}", + serde_json::json!({ + "ok": true, + "command": "spec validate", + "name": blueprint.name, + }) + ), + } Ok(()) } diff --git a/src/cli/src/stage/clarify.rs b/src/cli/src/stage/clarify.rs index 6702354..25c005d 100644 --- a/src/cli/src/stage/clarify.rs +++ b/src/cli/src/stage/clarify.rs @@ -69,6 +69,7 @@ impl ClarifyHandler { } #[cfg(test)] +#[allow(clippy::items_after_test_module)] mod tests { use super::*; diff --git a/src/cli/src/stage/design.rs b/src/cli/src/stage/design.rs index c4b2368..797d268 100644 --- a/src/cli/src/stage/design.rs +++ b/src/cli/src/stage/design.rs @@ -247,6 +247,7 @@ fn write_spec_files(stem: &str, kind: &str, yaml: &str, md: &str) -> Result<(), } #[cfg(test)] +#[allow(clippy::items_after_test_module)] mod tests { use super::*; diff --git a/src/cli/src/stage/implement.rs b/src/cli/src/stage/implement.rs index 8179b9a..9f09690 100644 --- a/src/cli/src/stage/implement.rs +++ b/src/cli/src/stage/implement.rs @@ -35,7 +35,10 @@ impl ImplementHandler { pub fn run(&self, args: &ImplementArgs) -> Result<(), CliError> { let rt = runtime::from_name(&args.lang).ok_or_else(|| { - CliError::new(format!("不支持的语言: {}（目前只支持 python）", args.lang)) + CliError::new(format!( + "不支持的语言: {}（目前支持 python / r / stata / matlab）", + args.lang + )) })?; self.cmd_implement(rt.as_ref(), &args.input, &args.output, &args.lang) } @@ -227,10 +230,43 @@ def step1(data): let handler = ImplementHandler::new(fake_llm("")); let args = ImplementArgs { input: "x.yaml".to_string(), - lang: "r".to_string(), + lang: "bash".to_string(), output: None, }; let err = handler.run(&args).unwrap_err(); - assert!(err.to_string().contains("不支持的语言")); + assert!( + err.to_string() + .contains("目前支持 python / r / stata / matlab") + ); + } + + #[test] + fn implement_non_python_runtimes_generate_language_files() { + let _guard = ENV_LOCK.lock().unwrap(); + let root = crate::test_support::temp_dir("qtcloud-implement-runtimes"); + let bp = root.join("demo.yaml"); + write_blueprint(&bp); + + for (lang, extension, response) in [ + ("r", "r", "```r\nclean <- function(data) data\n```"), + ("stata", "do", "```stata\nuse `input', clear\n```"), + ( + "matlab", + "m", + "```matlab\nfunction data = clean_data(data)\nend\n```", + ), + ] { + let output = root.join(format!("out.{extension}")); + let handler = ImplementHandler::new(fake_llm(response)); + handler + .run(&ImplementArgs { + input: bp.to_string_lossy().into_owned(), + lang: lang.to_string(), + output: Some(output.to_string_lossy().into_owned()), + }) + .unwrap(); + assert!(output.is_file(), "{lang} 输出文件不存在"); + assert!(!std::fs::read_to_string(output).unwrap().trim().is_empty()); + } } } diff --git a/src/cli/src/stage/process.rs b/src/cli/src/stage/process.rs index 0821dbf..8b91300 100644 --- a/src/cli/src/stage/process.rs +++ b/src/cli/src/stage/process.rs @@ -2,6 +2,7 @@ use clap::Args; use serde::{Deserialize, Serialize}; +use std::collections::HashSet; use std::io; use std::path::{Path, PathBuf}; use std::process::Command; @@ -27,6 +28,59 @@ pub struct ProcessArgs { pub pipeline: Option, } +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct PipelineStep { + pub name: String, + pub resource: String, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct PipelineSpec { + pub steps: Vec, +} + +impl PipelineSpec { + fn from_legacy(value: &str) -> Result { + let steps = value + .split(',') + .map(str::trim) + .filter(|step| !step.is_empty()) + .enumerate() + .map(|(index, resource)| PipelineStep { + name: resource + .strip_prefix("builtin:") + .unwrap_or(resource) + .rsplit(['/', '\\']) + .next() + .and_then(|name| name.split('.').next()) + .filter(|name| !name.is_empty()) + .map(str::to_string) + .unwrap_or_else(|| format!("step_{index}")), + resource: resource.to_string(), + }) + .collect::>(); + + if steps.is_empty() { + return Err(CliError::new("Pipeline 至少需要一个执行步骤")); + } + + Ok(Self { steps }) + } + + fn display(&self) -> String { + self.steps + .iter() + .map(|step| step.resource.as_str()) + .collect::>() + .join(",") + } +} + +struct ResolvedPipeline { + display: String, + spec: PipelineSpec, +} + #[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)] // ── 数据模型（ProcessJobRecord / ProcessArgs） ── pub struct ProcessJobRecord { @@ -89,7 +143,8 @@ impl ProcessJobRecord { // ── 命令与 StepExecutor（receive → pipeline → send） ── /// 编排命令入口：按 blueprint 执行 receive → pipeline → send。 pub fn run(args: &ProcessArgs) -> Result<(), CliError> { - let pipeline = resolve_pipeline(args)?; + let resolved_pipeline = resolve_pipeline(args)?; + let pipeline = resolved_pipeline.display; let started_at = util::now_utc(); let job_id = new_job_id(&args.customer_id); let work_dir = work_dir(); @@ -131,6 +186,7 @@ pub fn run(args: &ProcessArgs) -> Result<(), CliError> { args, job_id, pipeline, + pipeline_spec: resolved_pipeline.spec, customer_dir, raw_path, expected_output_path, @@ -143,13 +199,21 @@ pub fn run(args: &ProcessArgs) -> Result<(), CliError> { } // ── pipeline 解析 ── -fn resolve_pipeline(args: &ProcessArgs) -> Result { +fn resolve_pipeline(args: &ProcessArgs) -> Result { if let Some(bp) = &args.blueprint { - resolve_blueprint_pipeline(bp) + let spec = resolve_blueprint_pipeline(bp)?; + Ok(ResolvedPipeline { + display: spec.display(), + spec, + }) } else { - Ok(args.pipeline.clone().unwrap_or_else(|| { + let display = args.pipeline.clone().unwrap_or_else(|| { std::env::var("PIPELINE").unwrap_or_else(|_| "csv-standard".to_string()) - })) + }); + Ok(ResolvedPipeline { + spec: PipelineSpec::from_legacy(&display)?, + display, + }) } } @@ -158,6 +222,7 @@ struct StepExecutor<'a> { args: &'a ProcessArgs, job_id: String, pipeline: String, + pipeline_spec: PipelineSpec, customer_dir: PathBuf, raw_path: PathBuf, expected_output_path: PathBuf, @@ -190,7 +255,7 @@ impl StepExecutor<'_> { let result_path = run_pipeline( &path_string(&self.raw_path), &path_string(&self.customer_dir), - &self.pipeline, + &self.pipeline_spec, ) .map_err(|err| self.fail(format!("pipeline failed: {err}")))?; self.log_lines @@ -276,9 +341,19 @@ fn register_process_output(job_id: &str, result_path: &str) { } } -fn resolve_blueprint_pipeline(name: &str) -> Result { +fn resolve_blueprint_pipeline(name: &str) -> Result { let dir = std::env::var("BLUEPRINT_DIR").unwrap_or_else(|_| ".quanttide/data/blueprint".to_string()); + + for ext in ["yaml", "yml", "json"] { + let path = Path::new(&dir).join(format!("{name}.{ext}")); + if path.is_file() { + let content = std::fs::read_to_string(&path) + .map_err(|err| CliError::new(format!("读取 Blueprint 失败: {err}")))?; + return pipeline_spec_from_yaml(&content); + } + } + let key = crate::util::to_camel(name); let output = Command::new("cue") .args([ @@ -299,10 +374,97 @@ fn resolve_blueprint_pipeline(name: &str) -> Result { } let pipe: String = serde_json::from_slice(&output.stdout) .map_err(|err| CliError::new(format!("解析 Blueprint pipeline 失败: {err}")))?; - if pipe.trim().is_empty() { - return Err(CliError::new(format!("Blueprint {name} 中未定义 pipeline"))); + PipelineSpec::from_legacy(&pipe) +} + +fn pipeline_spec_from_yaml(yaml: &str) -> Result { + let root: serde_yaml::Value = serde_yaml::from_str(yaml) + .map_err(|err| CliError::new(format!("解析 Blueprint YAML 失败: {err}")))?; + let blueprint = root + .get("spec") + .and_then(|spec| spec.get("blueprint")) + .unwrap_or(&root); + let pipeline = blueprint + .get("pipeline") + .ok_or_else(|| CliError::new("Blueprint 中未定义 pipeline"))?; + + if let Some(states) = pipeline + .get("states") + .and_then(serde_yaml::Value::as_mapping) + { + return pipeline_spec_from_states(pipeline, states); + } + + let steps = pipeline + .get("steps") + .and_then(serde_yaml::Value::as_sequence) + .ok_or_else(|| CliError::new("Blueprint pipeline 中未定义 states 或 steps"))?; + let resources = steps + .iter() + .enumerate() + .map(|(index, step)| { + let name = yaml_string(step, "name").unwrap_or_else(|| format!("step_{index}")); + let resource = yaml_string(step, "resource").unwrap_or_else(|| name.clone()); + PipelineStep { name, resource } + }) + .collect::>(); + if resources.is_empty() { + return Err(CliError::new("Blueprint pipeline 至少需要一个步骤")); + } + Ok(PipelineSpec { steps: resources }) +} + +fn pipeline_spec_from_states( + pipeline: &serde_yaml::Value, + states: &serde_yaml::Mapping, +) -> Result { + let start_at = yaml_string(pipeline, "start_at") + .or_else(|| { + states + .keys() + .find_map(|key| key.as_str().map(str::to_string)) + }) + .ok_or_else(|| CliError::new("Blueprint pipeline.states 不能为空"))?; + let mut steps = Vec::new(); + let mut current = start_at; + let mut visited = HashSet::new(); + + loop { + if !visited.insert(current.clone()) { + return Err(CliError::new(format!( + "Blueprint pipeline 状态存在循环: {current}" + ))); + } + + let state = states + .get(serde_yaml::Value::String(current.clone())) + .ok_or_else(|| CliError::new(format!("Blueprint pipeline 缺少状态: {current}")))?; + let resource = yaml_string(state, "resource") + .ok_or_else(|| CliError::new(format!("Blueprint 状态缺少 resource: {current}")))?; + steps.push(PipelineStep { + name: current.clone(), + resource, + }); + + let is_end = state + .get("end") + .and_then(serde_yaml::Value::as_bool) + .unwrap_or(false); + let next = yaml_string(state, "next"); + if is_end || next.is_none() { + break; + } + current = next.unwrap(); } - Ok(pipe) + + Ok(PipelineSpec { steps }) +} + +fn yaml_string(value: &serde_yaml::Value, key: &str) -> Option { + value + .get(key) + .and_then(serde_yaml::Value::as_str) + .map(str::to_string) } // ── 脱敏工具 ── @@ -376,34 +538,52 @@ fn path_string(path: &Path) -> String { path.to_string_lossy().to_string() } -fn run_pipeline(input: &str, work_dir: &str, pipeline_spec: &str) -> Result { +fn run_pipeline( + input: &str, + work_dir: &str, + pipeline_spec: &PipelineSpec, +) -> Result { let mut prev = input.to_string(); - let steps: Vec<&str> = pipeline_spec.split(',').collect(); - - for (i, step) in steps.iter().enumerate() { - let step_name = std::path::Path::new(step) - .file_stem() - .and_then(|s| s.to_str()) - .unwrap_or(step); - let step_output = if i == steps.len() - 1 { - format!("{work_dir}/final.csv") + + for (i, step) in pipeline_spec.steps.iter().enumerate() { + let step_name = step.name.as_str(); + let step_output = if i == pipeline_spec.steps.len() - 1 { + Path::new(work_dir) + .join("final.csv") + .to_string_lossy() + .to_string() } else { - format!("{work_dir}/step_{i}_{step_name}.csv") + Path::new(work_dir) + .join(format!("step_{i}_{step_name}.csv")) + .to_string_lossy() + .to_string() }; - println!(" ▶ Step {}/{}: {step_name}", i + 1, steps.len()); + println!( + " ▶ Step {}/{}: {step_name}", + i + 1, + pipeline_spec.steps.len() + ); + + let (step_path, runtime) = if let Some(resource) = step.resource.strip_prefix("builtin:") { + ( + PathBuf::from(resource), + Some(Box::new(runtime::builtin::BuiltinRuntime) as Box), + ) + } else { + let path = Path::new(&step.resource); + let ext = path.extension().and_then(|e| e.to_str()).unwrap_or(""); + (path.to_path_buf(), runtime::from_ext(ext)) + }; - let step_path = std::path::Path::new(step); - let ext = step_path.extension().and_then(|e| e.to_str()).unwrap_or(""); - match runtime::from_ext(ext) { - // 注册表驱动：.py → python / .sh → bash + match runtime { Some(rt) => { - rt.execute(step_path, &prev, &step_output, work_dir) + rt.execute(&step_path, &prev, &step_output, work_dir) .map_err(|err| format!("执行 pipeline 步骤 {step_name} 失败: {err}"))?; } // 其他：直接执行（builtin 可执行脚本） None => { - let status = Command::new(step) + let status = Command::new(&step_path) .arg(&prev) .arg(&step_output) .status() @@ -635,4 +815,54 @@ mod tests { std::fs::remove_dir_all(&catalog_dir).ok(); } + + #[test] + fn blueprint_pipeline_spec_follows_state_transitions() { + let yaml = r#" +pipeline: + start_at: load + states: + load: + resource: builtin:copy + next: normalize + normalize: + resource: normalize.py + end: true +"#; + + let spec = pipeline_spec_from_yaml(yaml).unwrap(); + + assert_eq!( + spec.steps, + vec![ + PipelineStep { + name: "load".to_string(), + resource: "builtin:copy".to_string(), + }, + PipelineStep { + name: "normalize".to_string(), + resource: "normalize.py".to_string(), + }, + ] + ); + } + + #[test] + fn builtin_copy_pipeline_step_copies_input_to_output() { + let root = temp_catalog_dir("qtcloud-process-builtin-copy"); + std::fs::create_dir_all(&root).unwrap(); + let input = root.join("input.csv"); + std::fs::write(&input, "a,b\n1,2\n").unwrap(); + + let spec = PipelineSpec { + steps: vec![PipelineStep { + name: "copy".to_string(), + resource: "builtin:copy".to_string(), + }], + }; + let output = run_pipeline(input.to_str().unwrap(), root.to_str().unwrap(), &spec).unwrap(); + + assert_eq!(std::fs::read_to_string(output).unwrap(), "a,b\n1,2\n"); + std::fs::remove_dir_all(&root).ok(); + } } diff --git a/src/cli/src/stage/transfer.rs b/src/cli/src/stage/transfer.rs index c252554..0c80904 100644 --- a/src/cli/src/stage/transfer.rs +++ b/src/cli/src/stage/transfer.rs @@ -1,6 +1,6 @@ //! 传输命令与服务函数：send / receive（6 平台，进程内 + QTDATA_CLI 委派）。 -use clap::{Args, Subcommand}; +use clap::{Args, Subcommand, ValueEnum}; use serde::{Deserialize, Serialize}; use std::io; use std::path::{Path, PathBuf}; @@ -13,13 +13,59 @@ use crate::util; #[derive(Args)] pub struct TransferArgs { /// 网盘提供商: dropbox（默认）| baidu | google | onedrive | s3 | sftp - #[arg(long, default_value = "dropbox")] - pub provider: String, + #[arg(long, value_enum, default_value_t = TransferProvider::Dropbox)] + pub provider: TransferProvider, #[command(subcommand)] pub action: TransferAction, } +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, ValueEnum)] +pub enum TransferProvider { + #[default] + Dropbox, + Baidu, + Google, + Onedrive, + S3, + Sftp, +} + +impl TransferProvider { + pub fn parse(value: &str) -> Result { + match value.trim().to_ascii_lowercase().as_str() { + "dropbox" => Ok(Self::Dropbox), + "baidu" | "baidudrive" => Ok(Self::Baidu), + "google" | "googledrive" => Ok(Self::Google), + "onedrive" => Ok(Self::Onedrive), + "s3" => Ok(Self::S3), + "sftp" => Ok(Self::Sftp), + other => Err(CliError::new(format!( + "不支持的提供商: {other}，可选: dropbox / baidu / google / onedrive / s3 / sftp" + ))), + } + } + + pub const fn as_str(self) -> &'static str { + match self { + Self::Dropbox => "dropbox", + Self::Baidu => "baidu", + Self::Google => "google", + Self::Onedrive => "onedrive", + Self::S3 => "s3", + Self::Sftp => "sftp", + } + } + + pub const fn storage_name(self) -> &'static str { + self.as_str() + } + + pub fn storage(self) -> Option> { + storage::from_name(self.storage_name()) + } +} + #[derive(Subcommand)] pub enum TransferAction { /// 发送文件：上传到网盘并生成分享链接，把链接给对方 @@ -55,7 +101,7 @@ pub fn run(args: &TransferArgs) -> Result<(), CliError> { output, } => { let output_path = output.as_deref().map(Path::new); - send(file, remote.as_deref(), output_path, &args.provider) + send_with_provider(file, remote.as_deref(), output_path, args.provider) .map_err(|err| CliError::new(format!("发送失败: {err}")))?; Ok(()) } @@ -64,7 +110,7 @@ pub fn run(args: &TransferArgs) -> Result<(), CliError> { .as_deref() .map(Path::new) .unwrap_or_else(|| Path::new(source.rsplit('/').next().unwrap_or("received"))); - receive(source, output_path, &args.provider)?; + receive_with_provider(source, output_path, args.provider)?; Ok(()) } } @@ -77,6 +123,15 @@ pub fn run(args: &TransferArgs) -> Result<(), CliError> { // ── 服务函数（receive / send / 委派） ── /// 进程内接收服务：从 URL 或远程路径下载到本地文件。 pub fn receive(source: &str, output: &Path, provider: &str) -> Result<(), CliError> { + let provider = TransferProvider::parse(provider)?; + receive_with_provider(source, output, provider) +} + +pub fn receive_with_provider( + source: &str, + output: &Path, + provider: TransferProvider, +) -> Result<(), CliError> { let output_str = output.to_string_lossy().to_string(); if let Ok(bin) = std::env::var("QTDATA_CLI") { @@ -92,15 +147,16 @@ pub fn receive(source: &str, output: &Path, provider: &str) -> Result<(), CliErr if is_url { // 手动模式：从 URL 自动识别提供商 let p = storage::detect(source) - .or_else(|| storage::from_name(provider)) - .ok_or_else(|| CliError::new(format!("不支持的提供商: {provider}")))?; + .or_else(|| provider.storage()) + .ok_or_else(|| CliError::new(format!("不支持的提供商: {}", provider.as_str())))?; return rt .block_on(p.receive(source, &output_str)) .map_err(|e| CliError::new(format!("接收失败: {e}"))); } // 自动模式：使用指定提供商直接拉取 - let p = storage::from_name(provider) - .ok_or_else(|| CliError::new(format!("不支持的提供商: {provider}")))?; + let p = provider + .storage() + .ok_or_else(|| CliError::new(format!("不支持的提供商: {}", provider.as_str())))?; rt.block_on(p.receive_path(source, &output_str)) .map_err(|e| CliError::new(format!("自动接收失败: {e}"))) } @@ -113,6 +169,16 @@ pub fn send( remote: Option<&str>, output: Option<&Path>, provider: &str, +) -> Result { + let provider = TransferProvider::parse(provider)?; + send_with_provider(file, remote, output, provider) +} + +pub fn send_with_provider( + file: &str, + remote: Option<&str>, + output: Option<&Path>, + provider: TransferProvider, ) -> Result { let remote_path = remote .map(str::to_string) @@ -124,12 +190,14 @@ pub fn send( let rt = tokio::runtime::Runtime::new() .map_err(|e| CliError::new(format!("创建运行时失败: {e}")))?; - let p = storage::from_name(provider).ok_or_else(|| format!("不支持的提供商: {provider}"))?; + let p = provider + .storage() + .ok_or_else(|| format!("不支持的提供商: {}", provider.as_str()))?; let link = rt .block_on(p.send(file, &remote_path)) .map_err(|e| format!("发送失败: {e}"))?; handle_sent_link(SentLinkInput { - provider, + provider: provider.as_str(), file, remote_path: &remote_path, link: &link, @@ -550,4 +618,37 @@ mod tests { std::fs::remove_dir_all(&root).ok(); } + + #[test] + fn provider_enum_maps_cli_values_and_storage_aliases() { + assert_eq!( + TransferProvider::parse("dropbox").unwrap(), + TransferProvider::Dropbox + ); + assert_eq!( + TransferProvider::parse("google").unwrap(), + TransferProvider::Google + ); + assert_eq!( + TransferProvider::parse("sftp").unwrap(), + TransferProvider::Sftp + ); + assert_eq!(TransferProvider::Google.as_str(), "google"); + assert_eq!(TransferProvider::Google.storage_name(), "google"); + assert!(TransferProvider::parse("unknown").is_err()); + } + + #[test] + fn provider_enum_resolves_registered_storage() { + for provider in [ + TransferProvider::Dropbox, + TransferProvider::Baidu, + TransferProvider::Google, + TransferProvider::Onedrive, + TransferProvider::S3, + TransferProvider::Sftp, + ] { + assert!(provider.storage().is_some(), "{} 未注册", provider.as_str()); + } + } } diff --git a/src/cli/src/storage/baidu_drive.rs b/src/cli/src/storage/baidu_drive.rs index a36cdac..d14fddb 100644 --- a/src/cli/src/storage/baidu_drive.rs +++ b/src/cli/src/storage/baidu_drive.rs @@ -21,21 +21,21 @@ impl BaiduDriveStorage { } } -#[async_trait] -impl Storage for BaiduDriveStorage { - fn name(&self) -> &'static str { - "baidudrive" - } - - async fn send(&self, local_path: &str, remote_path: &str) -> Result { - let token = self.token()?; +impl BaiduDriveStorage { + pub async fn send_with_base( + &self, + token: &str, + local_path: &str, + remote_path: &str, + api_base: &str, + share_base: &str, + ) -> Result { let data = fs::read(local_path).map_err(|e| format!("读取文件失败: {e}"))?; let size = data.len(); let client = self.client(); - let _file_name = local_path.rsplit('/').next().unwrap_or("file"); // 1. 预创建文件 - let precreate_url = format!("{API_BASE}?method=precreate&access_token={token}"); + let precreate_url = format!("{api_base}?method=precreate&access_token={token}"); let precreate_body = serde_json::json!({ "path": remote_path, "size": size, @@ -62,7 +62,7 @@ impl Storage for BaiduDriveStorage { // 2. 上传文件内容 let upload_url = format!( - "{API_BASE}?method=upload&access_token={token}&type=tmpfile&path={remote_path}&uploadid={uploadid}&partseq=0" + "{api_base}?method=upload&access_token={token}&type=tmpfile&path={remote_path}&uploadid={uploadid}&partseq=0" ); client @@ -73,7 +73,7 @@ impl Storage for BaiduDriveStorage { .map_err(|e| format!("上传请求失败: {e}"))?; // 3. 创建文件 - let create_url = format!("{API_BASE}?method=create&access_token={token}"); + let create_url = format!("{api_base}?method=create&access_token={token}"); let create_body = serde_json::json!({ "path": remote_path, "size": size, @@ -99,7 +99,7 @@ impl Storage for BaiduDriveStorage { .ok_or_else(|| format!("创建文件失败: {create_json}"))?; // 4. 创建分享链接 - let share_url = format!("{SHARE_API}?method=create&access_token={token}"); + let share_url = format!("{share_base}?method=create&access_token={token}"); let share_body = serde_json::json!({ "path": [remote_path], "period": 7, // 7天有效期 @@ -133,9 +133,15 @@ impl Storage for BaiduDriveStorage { } } - async fn receive(&self, url: &str, local_path: &str) -> Result<(), String> { + pub async fn receive_with_base( + &self, + token: &str, + url: &str, + local_path: &str, + _api_base: &str, + share_base: &str, + ) -> Result<(), String> { let client = self.client(); - let token = self.token()?; // 从分享链接提取 surl let surl = url @@ -146,7 +152,7 @@ impl Storage for BaiduDriveStorage { let surl = surl.trim_end_matches('/'); // 解析分享信息：获取文件列表 - let info_url = format!("{SHARE_API}?method=list&access_token={token}"); + let info_url = format!("{share_base}?method=list&access_token={token}"); let info_body = serde_json::json!({ "shorturl": surl, @@ -202,3 +208,22 @@ impl Storage for BaiduDriveStorage { Ok(()) } } + +#[async_trait] +impl Storage for BaiduDriveStorage { + fn name(&self) -> &'static str { + "baidudrive" + } + + async fn send(&self, local_path: &str, remote_path: &str) -> Result { + let token = self.token()?; + self.send_with_base(&token, local_path, remote_path, API_BASE, SHARE_API) + .await + } + + async fn receive(&self, url: &str, local_path: &str) -> Result<(), String> { + let token = self.token()?; + self.receive_with_base(&token, url, local_path, API_BASE, SHARE_API) + .await + } +} diff --git a/src/cli/src/storage/sftp.rs b/src/cli/src/storage/sftp.rs index 044b49c..a169e9f 100644 --- a/src/cli/src/storage/sftp.rs +++ b/src/cli/src/storage/sftp.rs @@ -60,6 +60,67 @@ impl SftpStorage { } } +#[cfg(test)] +#[allow(clippy::items_after_test_module)] +#[allow(clippy::await_holding_lock)] +mod tests { + use super::*; + + fn clear_sftp_host() -> std::sync::MutexGuard<'static, ()> { + let guard = crate::ENV_LOCK.lock().unwrap(); + unsafe { + std::env::remove_var("SFTP_HOST"); + } + guard + } + + #[tokio::test] + async fn receive_rejects_malformed_sftp_url_before_connecting() { + let _guard = clear_sftp_host(); + let error = SftpStorage + .receive("https://example.com/file.csv", "out.csv") + .await; + + assert!(error.unwrap_err().contains("不支持的 URL 格式")); + } + + #[tokio::test] + async fn receive_rejects_url_without_user_before_connecting() { + let _guard = clear_sftp_host(); + let error = SftpStorage + .receive("sftp://host:22/file.csv", "out.csv") + .await; + + assert!(error.unwrap_err().contains("缺少 user@host")); + } + + #[tokio::test] + async fn receive_rejects_invalid_port_before_connecting() { + let _guard = clear_sftp_host(); + let error = SftpStorage + .receive("sftp://user@host:not-a-port/file.csv", "out.csv") + .await; + + assert!(error.unwrap_err().contains("端口格式错误")); + } + + #[tokio::test] + async fn send_requires_sftp_host_configuration() { + let _guard = clear_sftp_host(); + let error = SftpStorage.send("missing.csv", "/remote.csv").await; + + assert!(error.unwrap_err().contains("SFTP_HOST")); + } + + #[tokio::test] + async fn receive_path_requires_sftp_host_configuration() { + let _guard = clear_sftp_host(); + let error = SftpStorage.receive_path("/remote.csv", "out.csv").await; + + assert!(error.unwrap_err().contains("SFTP_HOST")); + } +} + #[async_trait] impl Storage for SftpStorage { fn name(&self) -> &'static str { diff --git a/src/cli/src/util.rs b/src/cli/src/util.rs index bfea6f1..7c23687 100644 --- a/src/cli/src/util.rs +++ b/src/cli/src/util.rs @@ -46,6 +46,7 @@ fn days_to_date(mut days: i64) -> (i64, u32, u32) { } #[cfg(test)] +#[allow(clippy::items_after_test_module)] mod tests { use super::*; @@ -88,6 +89,10 @@ mod tests { #[test] fn test_drd_dir_default() { + let _guard = ENV_LOCK.lock().unwrap(); + unsafe { + std::env::remove_var("DRD_DIR"); + } assert_eq!(drd_dir(), ".quanttide/data/drd"); } diff --git a/src/cli/tests/blueprint_test.rs b/src/cli/tests/blueprint_test.rs index c0f2af6..02fe981 100644 --- a/src/cli/tests/blueprint_test.rs +++ b/src/cli/tests/blueprint_test.rs @@ -34,3 +34,27 @@ fn test_blueprint_list_runs() { std::fs::remove_dir_all(&tmp).ok(); } + +#[test] +fn test_blueprint_list_json_returns_items() { + let tmp = std::env::temp_dir().join(format!("bp-json-test-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&tmp); + std::fs::create_dir_all(&tmp).unwrap(); + std::fs::write(tmp.join("customer.yaml"), "name: customer\n").unwrap(); + + let output = cli() + .env("BLUEPRINT_DIR", &tmp) + .arg("--json") + .arg("blueprint") + .arg("list") + .output() + .unwrap(); + + assert!(output.status.success()); + assert!(output.stderr.is_empty()); + let report: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap(); + assert_eq!(report["command"], "blueprint list"); + assert_eq!(report["items"][0], "customer"); + + std::fs::remove_dir_all(&tmp).ok(); +} diff --git a/src/cli/tests/cli_test.rs b/src/cli/tests/cli_test.rs index 085c1c8..05ae54c 100644 --- a/src/cli/tests/cli_test.rs +++ b/src/cli/tests/cli_test.rs @@ -2,13 +2,14 @@ mod common; -use common::cli; +use common::{cli, sample_blueprint_yaml}; #[test] fn test_cli_help_shows_all_commands() { let output = cli().arg("--help").output().unwrap(); assert!(output.status.success()); let stdout = String::from_utf8_lossy(&output.stdout); + assert!(stdout.contains("--json")); assert!(stdout.contains("clarify")); assert!(stdout.contains("design")); assert!(stdout.contains("review")); @@ -17,3 +18,134 @@ fn test_cli_help_shows_all_commands() { assert!(stdout.contains("blueprint")); assert!(stdout.contains("spec")); } + +#[test] +fn test_global_json_formats_command_errors() { + let missing = std::env::temp_dir().join(format!( + "qtcloud-global-json-missing-{}.yaml", + std::process::id() + )); + let output = cli() + .arg("--json") + .arg("spec") + .arg("validate") + .arg(&missing) + .output() + .unwrap(); + + assert!(!output.status.success()); + assert!( + output.stderr.is_empty(), + "JSON errors should not use stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + + let report: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap(); + assert_eq!(report["error"]["code"], "cli_error"); + assert!( + report["error"]["message"] + .as_str() + .unwrap() + .contains("无法读取 YAML") + ); +} + +#[test] +fn test_global_json_formats_pipeline_list_success() { + let root = std::env::temp_dir().join(format!( + "qtcloud-global-json-pipeline-{}", + std::process::id() + )); + let _ = std::fs::remove_dir_all(&root); + std::fs::create_dir_all(&root).unwrap(); + std::fs::write(root.join("normalize.yaml"), "name: normalize\n").unwrap(); + + let output = cli() + .env("PIPELINE_DIR", &root) + .arg("--json") + .arg("pipeline") + .arg("list") + .output() + .unwrap(); + + assert!(output.status.success()); + assert!(output.stderr.is_empty()); + let report: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap(); + assert_eq!(report["command"], "pipeline list"); + assert_eq!(report["items"][0], "normalize"); + + std::fs::remove_dir_all(&root).ok(); +} + +#[test] +fn test_global_json_formats_spec_wrap_success() { + let root = + std::env::temp_dir().join(format!("qtcloud-global-json-wrap-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&root); + std::fs::create_dir_all(&root).unwrap(); + let input = root.join("sample.yaml"); + let output_path = root.join("sample-spec.yaml"); + std::fs::write(&input, sample_blueprint_yaml()).unwrap(); + + let output = cli() + .arg("--json") + .arg("spec") + .arg("wrap") + .arg(&input) + .arg("--output") + .arg(&output_path) + .output() + .unwrap(); + + assert!(output.status.success()); + assert!(output.stderr.is_empty()); + let report: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap(); + assert_eq!(report["ok"], true); + assert_eq!(report["command"], "spec wrap"); + assert!(output_path.is_file()); + + std::fs::remove_dir_all(&root).ok(); +} + +#[test] +fn test_global_json_formats_catalog_list_success() { + let root = std::env::temp_dir().join(format!( + "qtcloud-global-json-catalog-{}", + std::process::id() + )); + let _ = std::fs::remove_dir_all(&root); + std::fs::create_dir_all(&root).unwrap(); + let data = root.join("report.csv"); + std::fs::write(&data, "a,b\n1,2\n").unwrap(); + let registry = serde_json::json!({ + "report": { + "name": "report", + "path": data, + "size": 8, + "received_at": "2026-08-24 00:00:00", + "status": "received", + "artifact_type": "pre_review" + } + }); + std::fs::write( + root.join("registry.json"), + serde_json::to_vec_pretty(®istry).unwrap(), + ) + .unwrap(); + + let output = cli() + .env("CATALOG_DIR", &root) + .arg("--json") + .arg("catalog") + .arg("list") + .output() + .unwrap(); + + assert!(output.status.success()); + assert!(output.stderr.is_empty()); + let report: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap(); + assert_eq!(report["command"], "catalog list"); + assert_eq!(report["items"][0]["name"], "report"); + + std::fs::remove_dir_all(&root).ok(); +} diff --git a/src/cli/tests/spec_test.rs b/src/cli/tests/spec_test.rs index 95b9632..c5a7c11 100644 --- a/src/cli/tests/spec_test.rs +++ b/src/cli/tests/spec_test.rs @@ -92,3 +92,37 @@ fn test_spec_validate_accepts_enveloped_yaml() { std::fs::remove_dir_all(&root).ok(); } + +#[test] +fn test_spec_validate_json_outputs_success_object() { + let root = + std::env::temp_dir().join(format!("qtcloud-spec-validate-json-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&root); + std::fs::create_dir_all(&root).unwrap(); + + let input = root.join("sample-blueprint.yaml"); + std::fs::write(&input, sample_blueprint_yaml()).unwrap(); + + let validate = cli() + .arg("--json") + .arg("spec") + .arg("validate") + .arg(&input) + .output() + .unwrap(); + + assert!( + validate.status.success(), + "spec validate --json failed: {}\n{}", + String::from_utf8_lossy(&validate.stdout), + String::from_utf8_lossy(&validate.stderr) + ); + assert!(validate.stderr.is_empty()); + + let report: serde_json::Value = serde_json::from_slice(&validate.stdout).unwrap(); + assert_eq!(report["ok"], true); + assert_eq!(report["command"], "spec validate"); + assert_eq!(report["name"], "sample"); + + std::fs::remove_dir_all(&root).ok(); +} diff --git a/src/cli/tests/storage_test.rs b/src/cli/tests/storage_test.rs index aab7612..c0f78f1 100644 --- a/src/cli/tests/storage_test.rs +++ b/src/cli/tests/storage_test.rs @@ -1,4 +1,7 @@ +#![allow(clippy::await_holding_lock)] + use qtcloud_data_cli::storage::Storage; +use qtcloud_data_cli::storage::baidu_drive::BaiduDriveStorage; use qtcloud_data_cli::storage::dropbox; use qtcloud_data_cli::storage::google_drive::{receive_with_base, send_with_base}; use qtcloud_data_cli::storage::onedrive; @@ -10,6 +13,7 @@ use wiremock::{Mock, MockServer, ResponseTemplate}; // s3 测试通过进程级 AWS_* 环境变量指向 wiremock，并行线程会互相覆盖， // 因此用静态锁串行化这两个测试（仅限本测试进程内）。 static AWS_ENV_LOCK: Mutex<()> = Mutex::new(()); +static BAIDU_ENV_LOCK: Mutex<()> = Mutex::new(()); // ── 辅助函数 ── @@ -125,6 +129,99 @@ async fn test_dropbox_upload_500() { std::fs::remove_file(&tmp).ok(); } +#[tokio::test] +async fn baidu_send_runs_precreate_upload_create_and_share_flow() { + let server = MockServer::start().await; + for method_name in ["precreate", "upload", "create"] { + let response = if method_name == "precreate" { + ResponseTemplate::new(200).set_body_json(serde_json::json!({"uploadid": "upload-1"})) + } else if method_name == "create" { + ResponseTemplate::new(200).set_body_json(serde_json::json!({"fs_id": 42})) + } else { + ResponseTemplate::new(200) + }; + Mock::given(method("POST")) + .and(path("/file")) + .and(query_param("method", method_name)) + .respond_with(response) + .mount(&server) + .await; + } + Mock::given(method("POST")) + .and(path("/share")) + .and(query_param("method", "create")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "link": "https://pan.baidu.com/s/abc", + "pwd": "1234" + }))) + .mount(&server) + .await; + + let file = tmp_file("baidu-send.csv", "a,b\n1,2\n"); + let link = BaiduDriveStorage + .send_with_base( + "fake-token", + &file, + "/apps/report.csv", + &format!("{}/file", server.uri()), + &format!("{}/share", server.uri()), + ) + .await + .unwrap(); + + assert_eq!(link, "https://pan.baidu.com/s/abc?pwd=1234"); + std::fs::remove_file(&file).ok(); +} + +#[tokio::test] +async fn baidu_receive_reads_share_listing_and_downloads_file() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/share")) + .and(query_param("method", "list")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "list": [{ + "fs_id": 42, + "dlink": format!("{}/download?x=1", server.uri()) + }] + }))) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path("/download")) + .respond_with(ResponseTemplate::new(200).set_body_string("id,value\n1,x\n")) + .mount(&server) + .await; + + let out = std::env::temp_dir().join("baidu-receive.csv"); + BaiduDriveStorage + .receive_with_base( + "fake-token", + &format!("{}/s/abc", server.uri()), + out.to_str().unwrap(), + &format!("{}/file", server.uri()), + &format!("{}/share", server.uri()), + ) + .await + .unwrap(); + + assert_eq!(std::fs::read_to_string(&out).unwrap(), "id,value\n1,x\n"); + std::fs::remove_file(&out).ok(); +} + +#[tokio::test] +async fn baidu_provider_reports_missing_token_without_network_call() { + let _guard = BAIDU_ENV_LOCK.lock().unwrap(); + unsafe { + std::env::remove_var("BAIDU_ACCESS_TOKEN"); + std::env::remove_var("BAIDUDRIVE_ACCESS_TOKEN"); + } + + let result = BaiduDriveStorage.send("missing.csv", "/report.csv").await; + + assert!(result.unwrap_err().contains("BAIDU_ACCESS_TOKEN")); +} + // ── 网盘类 provider receive_path 测试 ── #[tokio::test] @@ -279,13 +376,13 @@ async fn mock_gdrive_upload_flow(server: &MockServer, file_id: &str) { .await; // 2. 上传内容 Mock::given(method("PUT")) - .and(path(&format!("/upload/drive/v3/files/{file_id}"))) + .and(path(format!("/upload/drive/v3/files/{file_id}"))) .respond_with(ResponseTemplate::new(200)) .mount(server) .await; // 3. 设置权限 Mock::given(method("POST")) - .and(path(&format!("/drive/v3/files/{file_id}/permissions"))) + .and(path(format!("/drive/v3/files/{file_id}/permissions"))) .respond_with(ResponseTemplate::new(200)) .mount(server) .await;