diff --git a/AGENTS.md b/AGENTS.md index 321f1f7..685f740 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,4 +1,4 @@ -# cpa-ext +# billing 当前项目的本质是对 `CLIProxyAPI` 的核心扩展，功能边界会非常收敛 @@ -19,9 +19,9 @@ - 默认不检查这些目录的 Git 状态、提交历史、分支、远端或工作区改动。 - 不为这些仓库修复问题，不修改或提交其中的代码，也不处理其 Git 状态。 - 上述限制不影响运行时测试：可直接加载已经构建好的参考插件；测试 CPA 插件组合时，按需启用 `cpa-key-billing` 作为对照插件，但不得因此进入其仓库做源码或 Git 操作。 -- 只有在 cpa-ext 的兼容性验证、测试或实现判断确有必要时，才按最小范围只读查看相关文件。 +- 只有在 billing 的兼容性验证、测试或实现判断确有必要时，才按最小范围只读查看相关文件。 - 需要修改、构建、提交或维护任一参考仓库时，必须由用户明确提出。 -- 项目的实现、提交和工作区清洁度只以 cpa-ext 根仓库为准。 +- 项目的实现、提交和工作区清洁度只以 billing 根仓库为准。 ## DO diff --git a/README.md b/README.md index f79aa1d..58539a6 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ -# cpa-ext +# billing -`cpa-ext` 是 CLIProxyAPI 的收敛式核心扩展，提供持久化用量与价格展示、下游 Key 管理、模型准入和上游凭证定向路由。一个 Key 代表一个调用用户；管理员可以创建、禁用或永久归档 Key，并按 Key 查看统计。 +`billing` 是 CLIProxyAPI 的收敛式计费插件，提供持久化用量与价格展示、下游 Key 管理、模型准入和上游凭证定向路由。一个 Key 代表一个调用用户；管理员可以创建、禁用或永久归档 Key，并按 Key 查看统计。 每个 Key 同时拥有独立的美元额度账户。管理员可以分配本周期额度、设置日/周/月自动重置和并发上限，并查看不可变的扣费、额度调整与重置账目。额度采用请求结束后结算的软限制：余额为正时放行，实际 Usage 到达后扣费，最后一个或多个在途请求可能产生负余额，之后的新请求返回 429。 @@ -11,39 +11,39 @@ sequenceDiagram autonumber participant Client as 用户 / Codex participant CPA as CLIProxyAPI - participant Ext as cpa-ext + participant Billing as billing participant DB as SQLite participant Upstream as 实际模型服务 Client->>CPA: 携带下游 Key 发起模型请求 - CPA->>Ext: frontend_auth.authenticate - Ext->>DB: 查询 Key 与状态 - DB-->>Ext: 用户访问配置 + CPA->>Billing: frontend_auth.authenticate + Billing->>DB: 查询 Key 与状态 + DB-->>Billing: 用户访问配置 alt Key 无效或已停用 - Ext-->>CPA: 认证失败 + Billing-->>CPA: 认证失败 CPA-->>Client: 401 Unauthorized else 认证成功 - Ext-->>CPA: 返回用户身份（稳定 Key ID） - CPA->>Ext: request.intercept_before - Ext->>DB: 检查模型权限、余额与并发并占用名额 + Billing-->>CPA: 返回用户身份（稳定 Key ID） + CPA->>Billing: request.intercept_before + Billing->>DB: 检查模型权限、余额与并发并占用名额 alt 模型、额度或并发不允许 - Ext-->>CPA: 终止请求（403 / 429 / 503） + Billing-->>CPA: 终止请求（403 / 429 / 503） CPA-->>Client: 返回结构化错误 else 请求准入 - Ext-->>CPA: 允许继续 - CPA->>Ext: scheduler.pick（可用上游候选） + Billing-->>CPA: 允许继续 + CPA->>Billing: scheduler.pick（可用上游候选） alt 自动路由 - Ext-->>CPA: 委托 CPA 内置调度 + Billing-->>CPA: 委托 CPA 内置调度 else 严格路由 - Ext->>DB: 读取该用户绑定的上游账号 - Ext-->>CPA: 指定唯一 Auth ID + Billing->>DB: 读取该用户绑定的上游账号 + Billing-->>CPA: 指定唯一 Auth ID end - CPA->>Ext: request.intercept_after - Ext-->>CPA: 复核价格配置与实际上游 + CPA->>Billing: request.intercept_after + Billing-->>CPA: 复核价格配置与实际上游 alt 价格缺失或严格路由失效 CPA-->>Client: 503 Service Unavailable @@ -52,24 +52,25 @@ sequenceDiagram Upstream-->>CPA: 模型响应 / 流式输出 CPA-->>Client: 返回模型结果 - CPA->>Ext: request.complete（请求终态） - Ext->>DB: 释放并发名额，保存成功、失败或取消结果 - CPA->>Ext: usage.handle（如有最终 Token 用量） - Ext->>Ext: 按模型价格计算成本 - Ext->>DB: 保存用量、扣减余额并写入账目 + CPA->>Billing: request.complete（请求终态） + Billing->>DB: 释放并发名额，保存成功、失败或取消结果 + CPA->>Billing: usage.handle（如有最终 Token 用量） + Billing->>Billing: 按模型价格计算成本 + Billing->>DB: 保存用量、扣减余额并写入账目 end end end ``` -CLIProxyAPI 负责 HTTP 接入、协议转换、上游凭证和实际请求执行；`cpa-ext` 通过插件回调参与请求决策与记录，不直接连接模型服务。`request.complete` 与 `usage.handle` 是独立事实回调，实际到达顺序不作为数据正确性的前提。 +CLIProxyAPI 负责 HTTP 接入、协议转换、上游凭证和实际请求执行；`billing` 通过插件回调参与请求决策与记录，不直接连接模型服务。`request.complete` 与 `usage.handle` 是独立事实回调，实际到达顺序不作为数据正确性的前提。 ## 当前兼容目标 - CLIProxyAPI 源码：`.externals/CLIProxyAPI/`，检查时 revision 为 `f43aad7637ad813745bf7d341acb5663617570c5` +- 插件版本：`0.1.0` - Native ABI：`1` - RPC schema：最高 `3`，注册时按宿主版本向下协商 -- 插件 ID / 动态库文件名：`cpa-ext` +- 插件 ID / 动态库文件名：`billing` - 已声明能力：`frontend_auth_provider`（独占）、`scheduler`、`request_interceptor`、`request_lifecycle_plugin`、`usage_plugin`、`management_api` 契约来源以本仓库内 `.externals/CLIProxyAPI/sdk/pluginabi/types.go`、`.externals/CLIProxyAPI/sdk/pluginapi/types.go`、`.externals/CLIProxyAPI/internal/pluginhost/rpc_schema.go` 为准。 @@ -91,7 +92,15 @@ sudo apt-get install -y build-essential ./scripts/build.ps1 ``` -产物为 `bin/cpa-ext.so`，适用于运行在 WSL/Linux 的 CLIProxyAPI。SQLite 驱动依赖 CGO，普通开发测试也应在已经安装 GCC 的 WSL/Linux 环境执行： +生成 Linux/amd64 发布包和 SHA-256 校验文件： + +```bash +bash ./scripts/package.sh 0.1.0 +``` + +产物位于 `dist/billing_0.1.0_linux_amd64.tar.gz`，只包含动态库、版本元数据、README 和示例配置。 + +产物为 `bin/billing.so`，适用于运行在 WSL/Linux 的 CLIProxyAPI。SQLite 驱动依赖 CGO，普通开发测试也应在已经安装 GCC 的 WSL/Linux 环境执行： ```bash CGO_ENABLED=1 go test ./... @@ -99,9 +108,9 @@ CGO_ENABLED=1 go test ./... ## 加载到 CLIProxyAPI -1. 将 `bin/cpa-ext.so` 放进 Linux CLIProxyAPI 配置的插件目录（默认 `plugins/`）。 +1. 将 `bin/billing.so` 放进 Linux CLIProxyAPI 配置的插件目录（默认 `plugins/`）。 2. 合并 `config.example.yaml` 中的 `plugins` 配置。 -3. 重启宿主，确认日志中成功加载插件 `cpa-ext`。 +3. 重启宿主，确认日志中成功加载插件 `billing`。 4. 首次启动会创建名称为 `default`、值为 `000000` 的下游 Key，保持现有 Codex 配置可用。 5. 发起一个 Codex 请求，确认管理页面出现该 Key 的用量和实际上游 Auth ID。 @@ -122,7 +131,9 @@ CGO_ENABLED=1 go test ./... 现有和新建 Key 的初始额度都是 `$0`，默认不自动重置、并发上限为 4。管理员分配额度后才能发起模型请求。金额以微美元整数保存；修改额度不会清空本周期已用金额，手动或自动重置不会结转旧余额。允许模型没有价格配置时，请求会在触达上游前以 503 拒绝。 -`config_yaml` 中包含宿主补充的 `enabled` 和 `priority`；插件会解析 `codex_only`、数据库路径和首次导入 Key。完整下游 Key 只通过受 CPA Management Key 保护的 Key 管理接口返回，不写入普通日志或错误消息；上游 Token、Cookie 和原始凭证不会由插件读取或保存。 +价格页面可以手动从 `models.dev/catalog.json` 更新供应商级参考目录，并把任意参考模型的价格导入任意本地模型。例如本地的 `deepseek-v4-flash` 可以显式采用 `OpenAI / gpt-5.6-sol` 价格。目录和本地有效价格彼此独立；刷新只展示变化，管理员确认后才更新已关联价格，人工保存则解除目录关联。全零目录项不作为免费价格导入，下载失败继续保留上次缓存且不影响请求链路。 + +`config_yaml` 中包含宿主补充的 `enabled` 和 `priority`；插件会解析 `codex_only`、数据库路径、`models_dev_url`、目录缓存路径和首次导入 Key。目录缓存默认与 SQLite 数据库放在同一目录。完整下游 Key 只通过受 CPA Management Key 保护的 Key 管理接口返回，不写入普通日志或错误消息；上游 Token、Cookie 和原始凭证不会由插件读取或保存。 ## 请求明细与汇总 @@ -132,7 +143,8 @@ CGO_ENABLED=1 go test ./... ## 工程布局 -- `cmd/cpa-ext`：仅负责 C ABI、请求字节复制和 C 内存释放。 +- `cmd/billing`：仅负责 C ABI、请求字节复制和 C 内存释放。 - `internal/plugin`：RPC dispatcher、契约 DTO、原子配置与 Usage 接入。 +- `internal/modelcatalog`：models.dev 下载、规范化、搜索和最后可用缓存。 - `scripts`：环境检查与可复现构建。 - `.externals/CLIProxyAPI`：上游契约参考，不属于插件实现。 diff --git a/TODOS.md b/TODOS.md index 234aea3..a466e7f 100644 --- a/TODOS.md +++ b/TODOS.md @@ -5,3 +5,4 @@ - [x] 测试模式与真实数据隔离的紧凑管理台 - [x] 美元额度、周期重置、并发限制与不可变账目 - [x] 请求明细的服务端分页与筛选 +- [x] models.dev 供应商参考价目录、显式导入与确认更新 diff --git a/cmd/cpa-ext/main.go b/cmd/billing/main.go similarity index 97% rename from cmd/cpa-ext/main.go rename to cmd/billing/main.go index 6006ce5..1bec84a 100644 --- a/cmd/cpa-ext/main.go +++ b/cmd/billing/main.go @@ -1,6 +1,6 @@ //go:build cshared -// Command cpa-ext is the thin C ABI entry point for CLIProxyAPI. +// Command billing is the thin C ABI entry point for CLIProxyAPI. package main /* @@ -63,7 +63,7 @@ import ( "net/http" "unsafe" - "cpa-ext/internal/plugin" + "billing/internal/plugin" ) var app = plugin.NewApp() diff --git a/cmd/cpa-ext/main_stub.go b/cmd/billing/main_stub.go similarity index 100% rename from cmd/cpa-ext/main_stub.go rename to cmd/billing/main_stub.go diff --git a/config.example.yaml b/config.example.yaml index 548f352..240b1e0 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -2,10 +2,12 @@ plugins: enabled: true dir: plugins configs: - cpa-ext: + billing: enabled: true priority: 100 codex_only: true - database_path: data/cpa-ext.db + database_path: data/billing.db + models_dev_url: https://models.dev/catalog.json + models_dev_cache_path: data/models-dev-catalog.json bootstrap_name: default bootstrap_key: "000000" diff --git a/docs/modules/admin-console.md b/docs/modules/admin-console.md index 4d20142..e4e5072 100644 --- a/docs/modules/admin-console.md +++ b/docs/modules/admin-console.md @@ -2,7 +2,7 @@ ## 模块定位 -管理台是 cpa-ext 各业务模块的统一操作和查看入口。它不单独保存业务事实，而是通过 CLIProxyAPI 受保护的 Management API 读取或修改 SQLite 中的真实配置与记录。 +管理台是 billing 各业务模块的统一操作和查看入口。它不单独保存业务事实，而是通过 CLIProxyAPI 受保护的 Management API 读取或修改 SQLite 中的真实配置与记录。 当前管理台采用单页、紧凑布局，直接作为插件资源嵌入，不依赖 CDN、外部前端框架或独立构建服务。 @@ -12,7 +12,7 @@ | --- | --- | | 用户 Key | 管理用户、访问策略、路由、额度和并发，查看用户汇总与账目 | | 请求明细 | 分页查看请求结果、上游、Token、性能和成本，执行组合筛选 | -| 价格配置 | 维护模型基础价格、长上下文价格和 Fast 倍率 | +| 价格配置 | 维护本地价格，并搜索、导入和确认更新 models.dev 参考价格 | 页面顶部统一提供实时/测试模式切换和 CPA 管理密钥输入。 @@ -70,9 +70,15 @@ Key 列表展示： - 设置长上下文输入门槛及“大于/大于等于”比较方式； - 可选启用 Fast 价格并设置倍率； - 保存或删除模型价格。 +- 手动更新 models.dev 供应商价格目录； +- 按模型或供应商搜索参考价格，并导入到当前本地模型； +- 查看价格是手动配置还是关联到具体 models.dev 供应商与模型； +- 目录价格变化时先查看输入/输出差异，再确认是否更新本地价格。 价格单位统一显示为 `$ / 1M Token`。保存前由管理接口完成严格字段校验，页面不会自行推断缺失价格。 +点击普通“保存价格”始终把该模型切换为手动配置。目录更新或下载失败不会改变本地价格，也不会影响用户请求。 + ## 实时模式 实时模式连接当前 CLIProxyAPI 实例： @@ -89,8 +95,9 @@ Key 列表展示： 测试模式用于管理员熟悉页面和验证交互，与真实数据完全隔离： - 不需要管理密钥； -- 不访问任何真实 cpa-ext 管理接口； +- 不访问任何真实 billing 管理接口； - 使用浏览器本地的模拟 Key、上游、请求、价格和账目； +- 使用浏览器本地的参考价格样例测试搜索和导入，不访问 models.dev； - 创建、编辑、归档、重置和删除只修改本地模拟状态； - 提供放大的请求记录量，便于测试分页、筛选和表格布局； - 可以一键恢复初始测试数据。 @@ -102,7 +109,7 @@ Key 列表展示： 管理接口统一挂载在： ```text -/v0/management/plugins/cpa-ext +/v0/management/plugins/billing ``` | 路径 | 方法 | 用途 | @@ -116,11 +123,15 @@ Key 列表展示： | `/usage` | GET | 查询分页请求明细 | | `/usage-summary` | GET | 查询今日、用户和每日汇总 | | `/prices` | GET、PUT、DELETE | 查询、保存和删除模型价格 | +| `/prices/import` | POST | 将指定 models.dev 参考价格导入本地模型 | +| `/price-catalog` | GET | 查看目录状态并搜索供应商级参考价格 | +| `/price-catalog/refresh` | POST | 下载新目录并返回已关联价格差异，不修改本地价格 | +| `/price-catalog/apply` | POST | 按目录 revision 确认应用选中的价格变化 | 管理台 HTML 由以下插件资源提供： ```text -/v0/resource/plugins/cpa-ext/ui +/v0/resource/plugins/billing/ui ``` 管理接口返回结构化 JSON 和明确的 HTTP 状态码；未知路径返回 `not_found`，数据库不可用、请求字段错误等情况返回对应错误码和消息。 @@ -140,7 +151,7 @@ Key 列表展示： 本模块负责： - 提供统一管理页面； -- 调用并呈现 cpa-ext Management API； +- 调用并呈现 billing Management API； - 管理实时和测试数据模式； - 提供紧凑、可筛选、可分页的操作界面。 diff --git a/docs/modules/billing-and-pricing.md b/docs/modules/billing-and-pricing.md index a45a4db..b1f7d90 100644 --- a/docs/modules/billing-and-pricing.md +++ b/docs/modules/billing-and-pricing.md @@ -16,6 +16,7 @@ | 手动重置 | 管理员可立即开始新额度周期 | | 并发限制 | 每个 Key 独立限制同时执行的请求数 | | 价格规则 | 支持基础价格、长上下文价格和 Fast 倍率 | +| 参考价格 | 搜索并显式导入 models.dev 供应商级价格，刷新前展示差异 | | 计费账目 | 永久记录扣费、额度调整和周期重置 | | 请求拦截 | 余额、并发或价格不满足时在上游调用前拒绝 | @@ -118,6 +119,25 @@ 系统先选择基础或长上下文档位，再应用 Fast 倍率，最后对整次请求执行一次四舍五入。 +### models.dev 参考目录 + +`models.dev` 只提供候选参考价格，不直接参与在线计费。插件仅在管理员点击“更新目录”时下载 `catalog.json`，把能够由当前计费模型准确表达的供应商级 Token 价格规范化后保存到本地缓存。用户请求、认证、额度检查和最终扣费都不会访问外网。 + +管理员可以把任意目录价格导入任意本地模型，本地模型名与参考模型名不必一致。导入后，本地价格保存其 `models.dev` 供应商、模型、目录 revision 和获取时间；实际计费仍读取 SQLite 中已经确认的本地价格。 + +目录映射规则： + +- 输入和输出价格必须存在； +- 缓存价格缺失时使用输入价格，避免把缓存错误计为免费； +- 只支持一个可完整表示的长上下文档位； +- 独立音频或推理价格无法由当前四类 Token 表达时不提供导入； +- 输入、输出和缓存全部为零的条目不自动认定为免费； +- Fast 倍率是本地业务规则，导入和刷新不会改变它。 + +刷新目录会先返回所有已关联价格的变化和已下架来源。页面只有在管理员确认后才更新发生变化的本地价格；取消确认不会改动账单。人工保存某个模型价格会把来源切换为“手动配置”，以后的目录刷新不再跟随它。 + +下载或解析失败时继续使用最后一次成功缓存，本地有效价格保持不变。目录下载有超时和体积限制，解析、写缓存和发布新索引成功后才替换旧版本。 + ## 价格缺失 允许访问的模型必须存在价格配置。系统在 CPA 已经选择上游后、真正访问模型服务前再次检查实际模型价格。 @@ -154,6 +174,7 @@ 本模块负责： - 模型价格和确定性成本计算； +- 参考价格目录缓存、显式导入和确认更新； - 用户额度、余额和额度周期； - 并发请求准入； - 请求结束后的实际扣费； diff --git a/go.mod b/go.mod index 3715769..b33845b 100644 --- a/go.mod +++ b/go.mod @@ -1,4 +1,4 @@ -module cpa-ext +module billing go 1.24 diff --git a/internal/modelcatalog/catalog.go b/internal/modelcatalog/catalog.go new file mode 100644 index 0000000..e75e2f8 --- /dev/null +++ b/internal/modelcatalog/catalog.go @@ -0,0 +1,528 @@ +// Package modelcatalog downloads and indexes models.dev reference prices. +// The catalog is deliberately separate from the effective local price table: +// callers must explicitly import a row before it can affect billing. +package modelcatalog + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "math/big" + "net/http" + "os" + "path/filepath" + "sort" + "strings" + "sync" + "time" + + "billing/internal/pricing" +) + +const ( + DefaultSourceURL = "https://models.dev/catalog.json" + maxDownloadBytes = 32 << 20 + requestTimeout = 30 * time.Second +) + +// Info describes the currently cached, normalized catalog snapshot. +type Info struct { + SourceURL string `json:"source_url"` + FetchedAt time.Time `json:"fetched_at"` + Revision string `json:"revision"` + Models int `json:"models"` +} + +// Entry is one provider-specific reference price that can be represented by +// billing's deterministic token price model. +type Entry struct { + ID string `json:"id"` + Provider string `json:"provider"` + ProviderName string `json:"provider_name"` + Model string `json:"model"` + ModelName string `json:"model_name"` + Base pricing.Rates `json:"base"` + LongContext *pricing.LongContext `json:"long_context,omitempty"` +} + +// Policy creates a local price policy from the reference row. Fast pricing is +// local business policy and is therefore supplied by the caller, not models.dev. +func (e Entry) Policy(localModel string, fastEnabled bool, multiplier pricing.Ratio) pricing.Policy { + longContext := cloneLongContext(e.LongContext) + return pricing.Policy{ + Model: strings.TrimSpace(localModel), Base: e.Base, LongContext: longContext, + FastPricingEnabled: fastEnabled, FastMultiplier: multiplier, + } +} + +type cacheDocument struct { + Info Info `json:"info"` + Entries []Entry `json:"entries"` +} + +type index struct { + info Info + entries []Entry + byID map[string]Entry +} + +// Manager owns one immutable in-memory catalog index and a compact cache file. +// Refresh downloads and parses outside the read lock, so catalog searches and +// CPA request handling never wait on the network. +type Manager struct { + mu sync.RWMutex + refreshMu sync.Mutex + cachePath string + sourceURL string + client *http.Client + loaded *index + lastError string +} + +func NewManager(cachePath, sourceURL string) *Manager { + sourceURL = strings.TrimSpace(sourceURL) + if sourceURL == "" { + sourceURL = DefaultSourceURL + } + transport := &http.Transport{Proxy: http.ProxyFromEnvironment, DisableKeepAlives: true} + manager := &Manager{ + cachePath: strings.TrimSpace(cachePath), sourceURL: sourceURL, + client: &http.Client{Timeout: requestTimeout, Transport: transport}, + } + if err := manager.loadCache(); err != nil && !errors.Is(err, os.ErrNotExist) { + manager.lastError = err.Error() + } + return manager +} + +// Status reports cache availability without attempting network access. +func (m *Manager) Status() (Info, string, bool) { + m.mu.RLock() + defer m.mu.RUnlock() + if m.loaded == nil { + return Info{SourceURL: m.sourceURL}, m.lastError, false + } + return m.loaded.info, m.lastError, true +} + +// Refresh downloads and atomically publishes a new last-known-good snapshot. +// A failed refresh leaves both the in-memory index and cache file untouched. +func (m *Manager) Refresh(ctx context.Context) (Info, error) { + m.refreshMu.Lock() + defer m.refreshMu.Unlock() + + request, err := http.NewRequestWithContext(ctx, http.MethodGet, m.sourceURL, nil) + if err != nil { + return Info{}, m.rememberError(fmt.Errorf("创建 models.dev 请求: %w", err)) + } + request.Header.Set("Accept", "application/json") + request.Header.Set("User-Agent", "billing/models.dev") + response, err := m.client.Do(request) + if err != nil { + return Info{}, m.rememberError(fmt.Errorf("下载 models.dev 价格目录: %w", err)) + } + defer response.Body.Close() + if response.StatusCode != http.StatusOK { + return Info{}, m.rememberError(fmt.Errorf("下载 models.dev 价格目录: HTTP %s", response.Status)) + } + if response.ContentLength > maxDownloadBytes { + return Info{}, m.rememberError(fmt.Errorf("models.dev 价格目录超过 %d MiB 限制", maxDownloadBytes>>20)) + } + raw, err := io.ReadAll(io.LimitReader(response.Body, maxDownloadBytes+1)) + if err != nil { + return Info{}, m.rememberError(fmt.Errorf("读取 models.dev 价格目录: %w", err)) + } + if len(raw) > maxDownloadBytes { + return Info{}, m.rememberError(fmt.Errorf("models.dev 价格目录超过 %d MiB 限制", maxDownloadBytes>>20)) + } + loaded, err := parseSource(raw, m.sourceURL, time.Now().UTC()) + if err != nil { + return Info{}, m.rememberError(fmt.Errorf("解析 models.dev 价格目录: %w", err)) + } + if err := m.writeCache(loaded); err != nil { + return Info{}, m.rememberError(err) + } + m.mu.Lock() + m.loaded = loaded + m.lastError = "" + m.mu.Unlock() + return loaded.info, nil +} + +func (m *Manager) rememberError(err error) error { + m.mu.Lock() + m.lastError = err.Error() + m.mu.Unlock() + return err +} + +// Search ranks exact model/ID matches before prefixes and substrings. +func (m *Manager) Search(query string, limit int) ([]Entry, Info, bool) { + query = strings.ToLower(strings.TrimSpace(query)) + if limit <= 0 || limit > 50 { + limit = 20 + } + m.mu.RLock() + defer m.mu.RUnlock() + if m.loaded == nil { + return nil, Info{SourceURL: m.sourceURL}, false + } + if query == "" { + return []Entry{}, m.loaded.info, true + } + groups := [4][]Entry{} + for _, entry := range m.loaded.entries { + fields := []string{strings.ToLower(entry.ID), strings.ToLower(entry.Model), strings.ToLower(entry.ModelName), strings.ToLower(entry.Provider), strings.ToLower(entry.ProviderName)} + group := -1 + switch { + case fields[0] == query: + group = 0 + case fields[1] == query || fields[2] == query || fields[3] == query || fields[4] == query: + group = 1 + default: + for _, field := range fields { + if strings.HasPrefix(field, query) { + group = 2 + break + } + if group < 0 && strings.Contains(field, query) { + group = 3 + } + } + } + if group >= 0 { + groups[group] = append(groups[group], cloneEntry(entry)) + } + } + result := make([]Entry, 0, limit) + for _, group := range groups { + for _, entry := range group { + result = append(result, entry) + if len(result) == limit { + return result, m.loaded.info, true + } + } + } + return result, m.loaded.info, true +} + +func (m *Manager) Lookup(id string) (Entry, Info, bool) { + m.mu.RLock() + defer m.mu.RUnlock() + if m.loaded == nil { + return Entry{}, Info{SourceURL: m.sourceURL}, false + } + entry, ok := m.loaded.byID[strings.ToLower(strings.TrimSpace(id))] + return cloneEntry(entry), m.loaded.info, ok +} + +func (m *Manager) loadCache() error { + if m.cachePath == "" { + return os.ErrNotExist + } + raw, err := os.ReadFile(m.cachePath) + if err != nil { + return err + } + var document cacheDocument + if err := json.Unmarshal(raw, &document); err != nil { + return fmt.Errorf("读取 models.dev 本地缓存: %w", err) + } + if strings.TrimSpace(document.Info.SourceURL) != m.sourceURL { + return fmt.Errorf("读取 models.dev 本地缓存: 缓存来源与当前 models_dev_url 不一致") + } + loaded, err := indexDocument(document) + if err != nil { + return fmt.Errorf("读取 models.dev 本地缓存: %w", err) + } + m.loaded = loaded + return nil +} + +func (m *Manager) writeCache(loaded *index) error { + if m.cachePath == "" { + return nil + } + directory := filepath.Dir(m.cachePath) + if err := os.MkdirAll(directory, 0o755); err != nil { + return fmt.Errorf("创建 models.dev 缓存目录: %w", err) + } + raw, err := json.Marshal(cacheDocument{Info: loaded.info, Entries: loaded.entries}) + if err != nil { + return fmt.Errorf("编码 models.dev 本地缓存: %w", err) + } + temporary, err := os.CreateTemp(directory, ".billing-models-dev-*.tmp") + if err != nil { + return fmt.Errorf("创建 models.dev 临时缓存: %w", err) + } + temporaryPath := temporary.Name() + defer os.Remove(temporaryPath) + if chmodErr := temporary.Chmod(0o600); chmodErr != nil { + _ = temporary.Close() + return fmt.Errorf("设置 models.dev 本地缓存权限: %w", chmodErr) + } + _, err = temporary.Write(raw) + if err == nil { + err = temporary.Sync() + } + closeErr := temporary.Close() + if err != nil { + return fmt.Errorf("写入 models.dev 本地缓存: %w", err) + } + if closeErr != nil { + return fmt.Errorf("关闭 models.dev 本地缓存: %w", closeErr) + } + if err := os.Rename(temporaryPath, m.cachePath); err != nil { + return fmt.Errorf("替换 models.dev 本地缓存: %w", err) + } + return nil +} + +type sourceCatalog struct { + Providers map[string]sourceProvider `json:"providers"` +} + +type sourceProvider struct { + ID string `json:"id"` + Name string `json:"name"` + Models map[string]json.RawMessage `json:"models"` +} + +type sourceModel struct { + ID string `json:"id"` + Name string `json:"name"` + Cost *sourceCost `json:"cost"` +} + +type sourceCost struct { + Input *json.Number `json:"input"` + Output *json.Number `json:"output"` + Reasoning *json.Number `json:"reasoning"` + CacheRead *json.Number `json:"cache_read"` + CacheWrite *json.Number `json:"cache_write"` + InputAudio *json.Number `json:"input_audio"` + OutputAudio *json.Number `json:"output_audio"` + Tiers []sourceCostTier `json:"tiers"` +} + +type sourceCostTier struct { + Input *json.Number `json:"input"` + Output *json.Number `json:"output"` + Reasoning *json.Number `json:"reasoning"` + CacheRead *json.Number `json:"cache_read"` + CacheWrite *json.Number `json:"cache_write"` + InputAudio *json.Number `json:"input_audio"` + OutputAudio *json.Number `json:"output_audio"` + Tier struct { + Type string `json:"type"` + Size int64 `json:"size"` + } `json:"tier"` +} + +func parseSource(raw []byte, sourceURL string, fetchedAt time.Time) (*index, error) { + var source sourceCatalog + decoder := json.NewDecoder(bytes.NewReader(raw)) + decoder.UseNumber() + if err := decoder.Decode(&source); err != nil { + return nil, err + } + entryByID := make(map[string]Entry) + ambiguousIDs := make(map[string]struct{}) + for providerKey, provider := range source.Providers { + providerID := normalizedID(provider.ID, providerKey) + if providerID == "" { + continue + } + providerName := strings.TrimSpace(provider.Name) + if providerName == "" { + providerName = providerID + } + for modelKey, rawModel := range provider.Models { + var model sourceModel + modelDecoder := json.NewDecoder(bytes.NewReader(rawModel)) + modelDecoder.UseNumber() + if modelDecoder.Decode(&model) != nil || model.Cost == nil { + continue + } + modelID := strings.TrimSpace(model.ID) + if modelID == "" { + modelID = strings.TrimSpace(modelKey) + } + entry, ok := entryFromCost(providerID, providerName, modelID, model.Name, model.Cost) + if ok { + if _, ambiguous := ambiguousIDs[entry.ID]; ambiguous { + continue + } + if existing, duplicate := entryByID[entry.ID]; duplicate { + if sameEntryPrice(existing, entry) { + continue + } + delete(entryByID, entry.ID) + ambiguousIDs[entry.ID] = struct{}{} + continue + } + entryByID[entry.ID] = entry + } + } + } + entries := make([]Entry, 0, len(entryByID)) + for _, entry := range entryByID { + entries = append(entries, entry) + } + if len(entries) == 0 { + return nil, errors.New("目录中没有可安全表示的非零 Token 价格") + } + sort.Slice(entries, func(i, j int) bool { return entries[i].ID < entries[j].ID }) + digest := sha256.Sum256(raw) + document := cacheDocument{ + Info: Info{SourceURL: sourceURL, FetchedAt: fetchedAt, Revision: hex.EncodeToString(digest[:]), Models: len(entries)}, + Entries: entries, + } + return indexDocument(document) +} + +func indexDocument(document cacheDocument) (*index, error) { + if strings.TrimSpace(document.Info.Revision) == "" || document.Info.FetchedAt.IsZero() || len(document.Entries) == 0 { + return nil, errors.New("缓存元数据不完整") + } + loaded := &index{info: document.Info, entries: make([]Entry, 0, len(document.Entries)), byID: make(map[string]Entry, len(document.Entries))} + for _, entry := range document.Entries { + entry.ID = strings.ToLower(strings.TrimSpace(entry.ID)) + if entry.ID == "" { + return nil, errors.New("缓存包含空目录 ID") + } + if _, duplicate := loaded.byID[entry.ID]; duplicate { + return nil, fmt.Errorf("缓存包含重复目录 ID %q", entry.ID) + } + policy := entry.Policy("catalog-validation", false, pricing.Ratio{Numerator: 1, Denominator: 1}) + if err := policy.Validate(); err != nil { + return nil, fmt.Errorf("缓存价格 %q 无效: %w", entry.ID, err) + } + entry = cloneEntry(entry) + loaded.entries = append(loaded.entries, entry) + loaded.byID[entry.ID] = entry + } + loaded.info.Models = len(loaded.entries) + return loaded, nil +} + +func entryFromCost(providerID, providerName, modelID, modelName string, cost *sourceCost) (Entry, bool) { + if strings.TrimSpace(modelID) == "" || cost.Input == nil || cost.Output == nil { + return Entry{}, false + } + input, okInput := numberMicros(cost.Input) + output, okOutput := numberMicros(cost.Output) + if !okInput || !okOutput || !optionalMatches(cost.Reasoning, output) || !optionalMatches(cost.InputAudio, input) || !optionalMatches(cost.OutputAudio, output) { + return Entry{}, false + } + cacheRead, ok := optionalMicros(cost.CacheRead, input) + if !ok { + return Entry{}, false + } + cacheWrite, ok := optionalMicros(cost.CacheWrite, input) + if !ok || (input == 0 && output == 0 && cacheRead == 0 && cacheWrite == 0) { + return Entry{}, false + } + entry := Entry{ + ID: strings.ToLower(providerID + "/" + strings.TrimSpace(modelID)), Provider: providerID, + ProviderName: providerName, Model: strings.TrimSpace(modelID), ModelName: strings.TrimSpace(modelName), + Base: pricing.Rates{InputMicrosPer1M: input, CacheReadMicrosPer1M: cacheRead, CacheWriteMicrosPer1M: cacheWrite, OutputMicrosPer1M: output}, + } + if entry.ModelName == "" { + entry.ModelName = entry.Model + } + if len(cost.Tiers) > 1 { + return Entry{}, false + } + if len(cost.Tiers) == 1 { + tier := cost.Tiers[0] + if tier.Tier.Type != "context" || tier.Tier.Size <= 0 || tier.Input == nil || tier.Output == nil { + return Entry{}, false + } + tierInput, okInput := numberMicros(tier.Input) + tierOutput, okOutput := numberMicros(tier.Output) + if !okInput || !okOutput || !optionalMatches(tier.Reasoning, tierOutput) || !optionalMatches(tier.InputAudio, tierInput) || !optionalMatches(tier.OutputAudio, tierOutput) { + return Entry{}, false + } + tierCacheRead, okRead := optionalMicros(tier.CacheRead, tierInput) + tierCacheWrite, okWrite := optionalMicros(tier.CacheWrite, tierInput) + if !okRead || !okWrite { + return Entry{}, false + } + entry.LongContext = &pricing.LongContext{ + ThresholdInputTokens: tier.Tier.Size, Comparison: "gt", + Rates: pricing.Rates{InputMicrosPer1M: tierInput, CacheReadMicrosPer1M: tierCacheRead, CacheWriteMicrosPer1M: tierCacheWrite, OutputMicrosPer1M: tierOutput}, + } + } + return entry, true +} + +func normalizedID(id, fallback string) string { + id = strings.ToLower(strings.TrimSpace(id)) + if id == "" { + id = strings.ToLower(strings.TrimSpace(fallback)) + } + return id +} + +func sameEntryPrice(left, right Entry) bool { + if left.Base != right.Base { + return false + } + if left.LongContext == nil || right.LongContext == nil { + return left.LongContext == nil && right.LongContext == nil + } + return *left.LongContext == *right.LongContext +} + +func optionalMicros(number *json.Number, fallback int64) (int64, bool) { + if number == nil { + return fallback, true + } + return numberMicros(number) +} + +func optionalMatches(number *json.Number, expected int64) bool { + if number == nil { + return true + } + value, ok := numberMicros(number) + return ok && value == expected +} + +func numberMicros(number *json.Number) (int64, bool) { + if number == nil { + return 0, false + } + rational, ok := new(big.Rat).SetString(number.String()) + if !ok || rational.Sign() < 0 { + return 0, false + } + rational.Mul(rational, big.NewRat(1_000_000, 1)) + numerator := rational.Num() + denominator := rational.Denom() + quotient, remainder := new(big.Int), new(big.Int) + quotient.QuoRem(numerator, denominator, remainder) + if new(big.Int).Lsh(remainder, 1).Cmp(denominator) >= 0 { + quotient.Add(quotient, big.NewInt(1)) + } + return quotient.Int64(), quotient.IsInt64() +} + +func cloneLongContext(value *pricing.LongContext) *pricing.LongContext { + if value == nil { + return nil + } + cloned := *value + return &cloned +} + +func cloneEntry(value Entry) Entry { + value.LongContext = cloneLongContext(value.LongContext) + return value +} diff --git a/internal/modelcatalog/catalog_test.go b/internal/modelcatalog/catalog_test.go new file mode 100644 index 0000000..b0b2e1f --- /dev/null +++ b/internal/modelcatalog/catalog_test.go @@ -0,0 +1,140 @@ +package modelcatalog + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "path/filepath" + "strings" + "testing" + "time" +) + +const testCatalog = `{ + "providers": { + "openai": {"id":"openai","name":"OpenAI","models":{ + "gpt-5.6-sol":{"id":"gpt-5.6-sol","name":"GPT-5.6 Sol","cost":{"input":5,"output":30,"cache_read":0.5,"cache_write":6.25,"tiers":[{"input":10,"output":45,"cache_read":1,"cache_write":12.5,"tier":{"type":"context","size":272000}}]}}, + "gpt-sol-alias":{"id":"gpt-5.6-sol","name":"GPT Sol Alias","cost":{"input":5,"output":30,"cache_read":0.5,"cache_write":6.25,"tiers":[{"input":10,"output":45,"cache_read":1,"cache_write":12.5,"tier":{"type":"context","size":272000}}]}}, + "free-placeholder":{"id":"free-placeholder","cost":{"input":0,"output":0}}, + "audio-special":{"id":"audio-special","cost":{"input":1,"output":2,"input_audio":3}} + }}, + "google": {"id":"google","name":"Google","models":{ + "gemini-flash":{"id":"gemini-flash","name":"Gemini Flash","cost":{"input":0.1,"output":0.4}} + }} + } +}` + +func TestParseCatalogKeepsRepresentableNonZeroProviderPrices(t *testing.T) { + loaded, err := parseSource([]byte(testCatalog), DefaultSourceURL, time.Date(2026, 8, 15, 0, 0, 0, 0, time.UTC)) + if err != nil { + t.Fatal(err) + } + if loaded.info.Models != 2 { + t.Fatalf("models = %d, want 2", loaded.info.Models) + } + entry, ok := loaded.byID["openai/gpt-5.6-sol"] + if !ok || entry.Base.InputMicrosPer1M != 5_000_000 || entry.Base.CacheReadMicrosPer1M != 500_000 || entry.Base.CacheWriteMicrosPer1M != 6_250_000 || entry.Base.OutputMicrosPer1M != 30_000_000 { + t.Fatalf("openai entry = %+v", entry) + } + if entry.LongContext == nil || entry.LongContext.ThresholdInputTokens != 272_000 || entry.LongContext.Rates.OutputMicrosPer1M != 45_000_000 { + t.Fatalf("long context = %+v", entry.LongContext) + } + gemini := loaded.byID["google/gemini-flash"] + if gemini.Base.CacheReadMicrosPer1M != gemini.Base.InputMicrosPer1M || gemini.Base.CacheWriteMicrosPer1M != gemini.Base.InputMicrosPer1M { + t.Fatalf("missing cache prices did not fall back to input: %+v", gemini.Base) + } + if _, exists := loaded.byID["openai/free-placeholder"]; exists { + t.Fatal("zero/zero placeholder was treated as a usable free price") + } + if _, exists := loaded.byID["openai/audio-special"]; exists { + t.Fatal("unrepresentable audio price was imported") + } +} + +func TestParseCatalogDropsOnlyConflictingDuplicateID(t *testing.T) { + raw := []byte(`{"providers":{"provider":{"id":"provider","models":{"alias-a":{"id":"same","cost":{"input":1,"output":2}},"alias-b":{"id":"same","cost":{"input":3,"output":4}},"safe":{"id":"safe","cost":{"input":5,"output":6}}}}}}`) + loaded, err := parseSource(raw, DefaultSourceURL, time.Now().UTC()) + if err != nil { + t.Fatal(err) + } + if loaded.info.Models != 1 { + t.Fatalf("models = %d, want only safe row", loaded.info.Models) + } + if _, exists := loaded.byID["provider/same"]; exists { + t.Fatal("conflicting duplicate ID was retained") + } + if _, exists := loaded.byID["provider/safe"]; !exists { + t.Fatal("unrelated safe row was dropped") + } +} + +func TestManagerRefreshSearchCacheAndLastKnownGood(t *testing.T) { + body := testCatalog + status := http.StatusOK + server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, _ *http.Request) { + response.WriteHeader(status) + _, _ = response.Write([]byte(body)) + })) + defer server.Close() + + cachePath := filepath.Join(t.TempDir(), "models-dev.json") + manager := NewManager(cachePath, server.URL) + if _, _, loaded := manager.Status(); loaded { + t.Fatal("new manager unexpectedly had a catalog") + } + first, err := manager.Refresh(context.Background()) + if err != nil || first.Models != 2 || first.Revision == "" { + t.Fatalf("refresh info=%+v err=%v", first, err) + } + results, info, loaded := manager.Search("gpt-5.6", 20) + if !loaded || info.Revision != first.Revision || len(results) != 1 || results[0].ID != "openai/gpt-5.6-sol" { + t.Fatalf("search results=%+v info=%+v loaded=%v", results, info, loaded) + } + results, _, _ = manager.Search("openai/gpt-5.6-sol", 20) + if len(results) == 0 || results[0].ID != "openai/gpt-5.6-sol" { + t.Fatalf("exact catalog ID was not ranked first: %+v", results) + } + + reopened := NewManager(cachePath, server.URL) + results, cached, loaded := reopened.Search("OpenAI", 20) + if !loaded || cached.Revision != first.Revision || len(results) != 1 { + t.Fatalf("cached search results=%+v info=%+v loaded=%v", results, cached, loaded) + } + wrongSource := NewManager(cachePath, "https://unreachable.invalid/catalog.json") + if _, lastError, loaded := wrongSource.Status(); loaded || !strings.Contains(lastError, "来源") { + t.Fatalf("wrong-source cache loaded=%v error=%q", loaded, lastError) + } + + status = http.StatusBadGateway + if _, err := manager.Refresh(context.Background()); err == nil || !strings.Contains(err.Error(), "502") { + t.Fatalf("failed refresh error = %v", err) + } + entry, afterFailure, ok := manager.Lookup("openai/gpt-5.6-sol") + if !ok || entry.Model != "gpt-5.6-sol" || afterFailure.Revision != first.Revision { + t.Fatalf("last-known-good entry=%+v info=%+v ok=%v", entry, afterFailure, ok) + } + + body = strings.ReplaceAll(testCatalog, `"input":5`, `"input":6`) + status = http.StatusOK + second, err := manager.Refresh(context.Background()) + if err != nil || second.Revision == first.Revision { + t.Fatalf("second refresh info=%+v err=%v", second, err) + } + updated, _, _ := manager.Lookup("openai/gpt-5.6-sol") + if updated.Base.InputMicrosPer1M != 6_000_000 { + t.Fatalf("updated input = %d", updated.Base.InputMicrosPer1M) + } +} + +func TestManagerRejectsOversizedCatalogWithoutReadingPastLimit(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, _ *http.Request) { + response.Header().Set("Content-Length", fmt.Sprint(maxDownloadBytes+1)) + _, _ = response.Write([]byte("x")) + })) + defer server.Close() + manager := NewManager(filepath.Join(t.TempDir(), "catalog.json"), server.URL) + if _, err := manager.Refresh(context.Background()); err == nil || !strings.Contains(err.Error(), "超过") { + t.Fatalf("oversized refresh error = %v", err) + } +} diff --git a/internal/plugin/access.go b/internal/plugin/access.go index 9510b53..7ab036e 100644 --- a/internal/plugin/access.go +++ b/internal/plugin/access.go @@ -9,9 +9,9 @@ import ( "strings" "time" - managedaccess "cpa-ext/internal/access" - "cpa-ext/internal/pricing" - "cpa-ext/internal/repository" + managedaccess "billing/internal/access" + "billing/internal/pricing" + "billing/internal/repository" ) const ( diff --git a/internal/plugin/access_test.go b/internal/plugin/access_test.go index 1ad0a92..02e0158 100644 --- a/internal/plugin/access_test.go +++ b/internal/plugin/access_test.go @@ -8,7 +8,7 @@ import ( "testing" "time" - managedaccess "cpa-ext/internal/access" + managedaccess "billing/internal/access" ) func TestManagedKeyAuthenticationAndLifecycle(t *testing.T) { @@ -273,3 +273,32 @@ func TestBillingAdmissionConcurrencyAndUsageSettlement(t *testing.T) { t.Fatalf("settled state=%+v err=%v", state, err) } } + +func TestUsageSettlementResolvesCurrentCPAKeyPrincipal(t *testing.T) { + app := NewApp() + if _, err := app.HandleMethod(MethodPluginRegister, lifecycleRequest(t, SchemaVersion, testConfig(t, "enabled: true\ncodex_only: false\n"))); err != nil { + t.Fatal(err) + } + price := []byte(`{"model":"deepseek-v4-flash","base":{"input_per_1m":"1","cache_read_per_1m":"1","cache_write_per_1m":"1","output_per_1m":"2"},"fast_pricing_enabled":false,"fast_multiplier":"2.5"}`) + if response := managementCallBody(t, app, http.MethodPut, managementBase+routePrices, price); response.StatusCode != http.StatusOK { + t.Fatalf("put price=%d %s", response.StatusCode, response.Body) + } + if _, err := app.store.UpdateBilling(context.Background(), "key_default", managedaccess.BillingSettings{QuotaMicros: 1_000_000, ResetPeriod: managedaccess.ResetNone, MaxConcurrency: 4}, time.Now()); err != nil { + t.Fatal(err) + } + for _, reference := range []string{"key_default", managedaccess.CallerScope("key_default"), "000000"} { + key, err := app.resolveUsageManagedKey(app.store, reference) + if err != nil || key.ID != "key_default" { + t.Fatalf("resolve usage reference %q: key=%+v err=%v", reference, key, err) + } + } + usage := UsageRecord{APIKey: "key_default", Model: "deepseek-v4-flash", RequestedAt: time.Now(), Detail: UsageDetail{InputTokens: 10_000, OutputTokens: 10_000, TotalTokens: 20_000}} + raw, _ := json.Marshal(usage) + if _, err := app.HandleMethod(MethodUsageHandle, raw); err != nil { + t.Fatal(err) + } + state, err := app.store.BillingState(context.Background(), "key_default", time.Now()) + if err != nil || state.SpentMicros != 30_000 || state.BalanceMicros != 970_000 { + t.Fatalf("principal-settled state=%+v err=%v", state, err) + } +} diff --git a/internal/plugin/app.go b/internal/plugin/app.go index d000581..201be3c 100644 --- a/internal/plugin/app.go +++ b/internal/plugin/app.go @@ -10,9 +10,11 @@ import ( "sync/atomic" "time" - "cpa-ext/internal/collection" - "cpa-ext/internal/pricing" - "cpa-ext/internal/repository" + managedaccess "billing/internal/access" + "billing/internal/collection" + "billing/internal/modelcatalog" + "billing/internal/pricing" + "billing/internal/repository" ) type App struct { @@ -22,6 +24,7 @@ type App struct { usage *collection.Service store *repository.SQLiteUsageRepository prices map[string]pricing.Policy + catalog *modelcatalog.Manager host HostCaller closed bool seen atomic.Uint64 @@ -104,6 +107,7 @@ func (a *App) configure(raw []byte) ([]byte, error) { for _, policy := range policies { nextPrices[normalizeModelName(policy.Model)] = policy } + nextCatalog := modelcatalog.NewManager(cfg.CatalogPath, cfg.ModelsDevURL) a.mu.RLock() releaseStaleAdmissions := a.usage == nil || a.config.DatabasePath != cfg.DatabasePath a.mu.RUnlock() @@ -125,6 +129,7 @@ func (a *App) configure(raw []byte) ([]byte, error) { a.usage = nextUsage a.store = usageRepository a.prices = nextPrices + a.catalog = nextCatalog if previousUsage != nil { _ = previousUsage.Close() } @@ -137,7 +142,7 @@ func registration(schemaVersion uint32) Registration { Metadata: Metadata{ Name: PluginName, Version: Version, - Author: "cpa-ext", + Author: "billing", GitHubRepository: "https://git.pchuan.top/agent/cpa-plugin", ConfigFields: []ConfigField{ {Name: "enabled", Type: "boolean", Description: "启用 CPA 扩展。"}, @@ -145,6 +150,8 @@ func registration(schemaVersion uint32) Registration { {Name: "database_path", Type: "string", Description: "SQLite 数据库文件路径。"}, {Name: "bootstrap_name", Type: "string", Description: "首次启动时现有 Key 的名称。"}, {Name: "bootstrap_key", Type: "string", Description: "首次启动时导入的现有下游 Key。"}, + {Name: "models_dev_url", Type: "string", Description: "models.dev 合并价格目录地址。"}, + {Name: "models_dev_cache_path", Type: "string", Description: "规范化价格目录缓存文件；默认与数据库同目录。"}, }, }, Capabilities: Capabilities{ @@ -191,7 +198,7 @@ func (a *App) handleUsage(raw []byte) ([]byte, error) { TTFT: record.TTFT, Latency: record.Latency, } - if key, keyErr := a.store.ManagedKeyByCredential(context.Background(), record.APIKey); keyErr == nil { + if key, keyErr := a.resolveUsageManagedKey(a.store, record.APIKey); keyErr == nil { observed.ManagedKeyID = key.ID observed.KeyAlias = key.Name } @@ -218,6 +225,20 @@ func (a *App) handleUsage(raw []byte) ([]byte, error) { return OKEnvelope(struct{}{}) } +func (a *App) resolveUsageManagedKey(store *repository.SQLiteUsageRepository, reference string) (managedaccess.ManagedKey, error) { + if keyID, ok := a.scopes.Load(strings.TrimSpace(reference)); ok { + if id, valid := keyID.(string); valid { + if key, err := store.ManagedKeyByID(context.Background(), id); err == nil { + return key, nil + } + } + } + if key, err := store.ManagedKeyByReference(context.Background(), reference); err == nil { + return key, nil + } + return store.ManagedKeyByCredential(context.Background(), reference) +} + func (a *App) handleRequestComplete(raw []byte) ([]byte, error) { var completion RequestCompletion if err := json.Unmarshal(raw, &completion); err != nil { @@ -285,5 +306,6 @@ func (a *App) Shutdown() { _ = a.usage.Close() a.usage = nil a.store = nil + a.catalog = nil } } diff --git a/internal/plugin/app_test.go b/internal/plugin/app_test.go index ddd1205..feab213 100644 --- a/internal/plugin/app_test.go +++ b/internal/plugin/app_test.go @@ -10,8 +10,8 @@ import ( "testing" "time" - managedaccess "cpa-ext/internal/access" - "cpa-ext/internal/repository" + managedaccess "billing/internal/access" + "billing/internal/repository" ) func lifecycleRequest(t *testing.T, schema uint32, config string) []byte { @@ -71,6 +71,20 @@ func TestConfigRejectsShortBootstrapKey(t *testing.T) { } } +func TestConfigDerivesCatalogCacheBesideDatabaseAndValidatesURL(t *testing.T) { + databasePath := filepath.Join(t.TempDir(), "nested", "usage.db") + cfg, err := decodeConfig([]byte(fmt.Sprintf("database_path: %q\n", databasePath))) + if err != nil { + t.Fatal(err) + } + if cfg.CatalogPath != filepath.Join(filepath.Dir(databasePath), "models-dev-catalog.json") { + t.Fatalf("catalog path = %q", cfg.CatalogPath) + } + if _, err := decodeConfig([]byte("models_dev_url: file:///tmp/catalog.json\n")); err == nil { + t.Fatal("non-HTTP models_dev_url unexpectedly accepted") + } +} + func TestUsageRecordWireFieldsMatchTargetContract(t *testing.T) { raw, err := json.Marshal(UsageRecord{}) if err != nil { diff --git a/internal/plugin/catalog_management.go b/internal/plugin/catalog_management.go new file mode 100644 index 0000000..9bfc44f --- /dev/null +++ b/internal/plugin/catalog_management.go @@ -0,0 +1,299 @@ +package plugin + +import ( + "context" + "net/http" + "net/url" + "sort" + "strconv" + "strings" + "time" + + "billing/internal/modelcatalog" + "billing/internal/pricing" + "billing/internal/repository" +) + +type catalogEntryDTO struct { + ID string `json:"id"` + Provider string `json:"provider"` + ProviderName string `json:"provider_name"` + Model string `json:"model"` + ModelName string `json:"model_name"` + Base priceRatesDTO `json:"base"` + LongContext *longContextDTO `json:"long_context,omitempty"` +} + +type catalogChangeDTO struct { + Model string `json:"model"` + CatalogID string `json:"catalog_id"` + Status string `json:"status"` + Before priceDTO `json:"before"` + After *priceDTO `json:"after,omitempty"` + FetchedAt *time.Time `json:"fetched_at,omitempty"` +} + +type catalogImportRequest struct { + Model string `json:"model"` + CatalogID string `json:"catalog_id"` +} + +type catalogApplyRequest struct { + Revision string `json:"revision"` + Models []string `json:"models"` +} + +func (a *App) searchPriceCatalog(query url.Values) ManagementResponse { + a.mu.RLock() + catalog := a.catalog + a.mu.RUnlock() + if catalog == nil { + return catalogManagementError(http.StatusServiceUnavailable, "catalog_unavailable", "价格目录尚未初始化") + } + limit := 20 + if raw := strings.TrimSpace(query.Get("limit")); raw != "" { + parsed, err := strconv.Atoi(raw) + if err != nil || parsed < 1 || parsed > 50 { + return catalogManagementError(http.StatusBadRequest, "invalid_limit", "limit 必须是 1-50 的整数") + } + limit = parsed + } + entries, info, loaded := catalog.Search(query.Get("q"), limit) + _, lastError, _ := catalog.Status() + results := make([]catalogEntryDTO, 0, len(entries)) + for _, entry := range entries { + results = append(results, catalogEntryToDTO(entry)) + } + return jsonManagementResponse(http.StatusOK, map[string]any{ + "loaded": loaded, "catalog": info, "models": results, "last_error": lastError, + }) +} + +func (a *App) refreshPriceCatalog() ManagementResponse { + a.mu.RLock() + catalog := a.catalog + a.mu.RUnlock() + if catalog == nil { + return catalogManagementError(http.StatusServiceUnavailable, "catalog_unavailable", "价格目录尚未初始化") + } + ctx, cancel := context.WithTimeout(context.Background(), 35*time.Second) + defer cancel() + info, err := catalog.Refresh(ctx) + if err != nil { + return catalogManagementError(http.StatusBadGateway, "catalog_download_failed", err.Error()) + } + a.mu.RLock() + defer a.mu.RUnlock() + if a.catalog != catalog || a.store == nil { + return catalogManagementError(http.StatusServiceUnavailable, "catalog_reconfigured", "价格目录配置已经变化，请重新刷新") + } + changes, err := previewCatalogChanges(context.Background(), a.store, catalog) + if err != nil { + return catalogManagementError(http.StatusInternalServerError, "database_error", err.Error()) + } + changed, missing := catalogChangeCounts(changes) + return jsonManagementResponse(http.StatusOK, map[string]any{ + "catalog": info, "changes": changes, "changed": changed, "missing": missing, + }) +} + +func (a *App) importCatalogPrice(body []byte) ManagementResponse { + var request catalogImportRequest + if err := decodeJSONBody(body, &request); err != nil { + return catalogManagementError(http.StatusBadRequest, "invalid_catalog_import", err.Error()) + } + request.Model = strings.TrimSpace(request.Model) + request.CatalogID = strings.TrimSpace(request.CatalogID) + if request.Model == "" || request.CatalogID == "" { + return catalogManagementError(http.StatusBadRequest, "invalid_catalog_import", "model 和 catalog_id 不能为空") + } + a.priceMu.Lock() + defer a.priceMu.Unlock() + a.mu.RLock() + catalog := a.catalog + current, currentExists := a.prices[normalizeModelName(request.Model)] + a.mu.RUnlock() + if catalog == nil { + return catalogManagementError(http.StatusServiceUnavailable, "catalog_unavailable", "价格目录尚未初始化") + } + entry, info, found := catalog.Lookup(request.CatalogID) + if !found { + return catalogManagementError(http.StatusNotFound, "catalog_price_not_found", "参考价格不存在或当前目录尚未下载") + } + fastEnabled := false + multiplier := pricing.Ratio{Numerator: 5, Denominator: 2} + if currentExists { + fastEnabled = current.FastPricingEnabled + multiplier = current.FastMultiplier + } + policy := entry.Policy(request.Model, fastEnabled, multiplier) + if err := policy.Validate(); err != nil { + return catalogManagementError(http.StatusBadRequest, "invalid_catalog_price", err.Error()) + } + record := repository.PriceRecord{ + Policy: policy, + Source: repository.PriceSource{Kind: repository.PriceSourceModelsDev, CatalogID: entry.ID, Revision: info.Revision, FetchedAt: info.FetchedAt}, + } + return a.persistPriceRecordLocked(record, catalog) +} + +func (a *App) applyCatalogChanges(body []byte) ManagementResponse { + var request catalogApplyRequest + if err := decodeJSONBody(body, &request); err != nil { + return catalogManagementError(http.StatusBadRequest, "invalid_catalog_apply", err.Error()) + } + request.Revision = strings.TrimSpace(request.Revision) + requested := make(map[string]struct{}, len(request.Models)) + for _, model := range request.Models { + model = normalizeModelName(model) + if model != "" { + requested[model] = struct{}{} + } + } + if request.Revision == "" || len(requested) == 0 { + return catalogManagementError(http.StatusBadRequest, "invalid_catalog_apply", "revision 和 models 不能为空") + } + + a.priceMu.Lock() + defer a.priceMu.Unlock() + a.mu.Lock() + catalog, store := a.catalog, a.store + if catalog == nil || store == nil { + a.mu.Unlock() + return catalogManagementError(http.StatusServiceUnavailable, "catalog_unavailable", "价格目录或数据库尚未初始化") + } + info, _, loaded := catalog.Status() + if !loaded || info.Revision != request.Revision { + a.mu.Unlock() + return catalogManagementError(http.StatusConflict, "catalog_changed", "价格目录已经变化，请重新查看差异") + } + records, err := store.ListPriceRecords(context.Background()) + if err != nil { + a.mu.Unlock() + return catalogManagementError(http.StatusInternalServerError, "database_error", err.Error()) + } + updates := make([]repository.PriceRecord, 0, len(requested)) + seen := make(map[string]struct{}, len(requested)) + for _, record := range records { + normalized := normalizeModelName(record.Policy.Model) + if _, wanted := requested[normalized]; !wanted { + continue + } + if record.Source.Kind != repository.PriceSourceModelsDev { + a.mu.Unlock() + return catalogManagementError(http.StatusConflict, "price_source_changed", record.Policy.Model+" 已经改为手动价格") + } + entry, _, found := catalog.Lookup(record.Source.CatalogID) + if !found { + a.mu.Unlock() + return catalogManagementError(http.StatusConflict, "catalog_price_missing", record.Policy.Model+" 的参考价格已不存在") + } + policy := entry.Policy(record.Policy.Model, record.Policy.FastPricingEnabled, record.Policy.FastMultiplier) + updates = append(updates, repository.PriceRecord{ + Policy: policy, + Source: repository.PriceSource{Kind: repository.PriceSourceModelsDev, CatalogID: entry.ID, Revision: info.Revision, FetchedAt: info.FetchedAt}, + }) + seen[normalized] = struct{}{} + } + if len(seen) != len(requested) { + a.mu.Unlock() + return catalogManagementError(http.StatusConflict, "price_source_changed", "部分待更新价格已经被删除或改变") + } + if err := store.UpsertPriceRecords(context.Background(), updates); err != nil { + a.mu.Unlock() + return catalogManagementError(http.StatusInternalServerError, "database_error", err.Error()) + } + for _, update := range updates { + a.prices[normalizeModelName(update.Policy.Model)] = update.Policy + } + a.mu.Unlock() + + for _, update := range updates { + if _, err := store.BackfillMissingCosts(context.Background(), update.Policy); err != nil { + a.mu.RLock() + reconfigured := a.store != store + a.mu.RUnlock() + if reconfigured { + return catalogManagementError(http.StatusServiceUnavailable, "database_reconfigured", "价格已经更新，重新配置中止了历史补算") + } + return catalogManagementError(http.StatusInternalServerError, "database_error", err.Error()) + } + } + models := make([]string, 0, len(updates)) + for _, update := range updates { + models = append(models, update.Policy.Model) + } + sort.Strings(models) + return jsonManagementResponse(http.StatusOK, map[string]any{"updated": len(updates), "models": models, "catalog": info}) +} + +func previewCatalogChanges(ctx context.Context, store *repository.SQLiteUsageRepository, catalog *modelcatalog.Manager) ([]catalogChangeDTO, error) { + records, err := store.ListPriceRecords(ctx) + if err != nil { + return nil, err + } + changes := make([]catalogChangeDTO, 0) + for _, record := range records { + if record.Source.Kind != repository.PriceSourceModelsDev { + continue + } + entry, info, found := catalog.Lookup(record.Source.CatalogID) + before := priceRecordToDTO(record) + if !found { + changes = append(changes, catalogChangeDTO{Model: record.Policy.Model, CatalogID: record.Source.CatalogID, Status: "missing", Before: before}) + continue + } + policy := entry.Policy(record.Policy.Model, record.Policy.FastPricingEnabled, record.Policy.FastMultiplier) + if sameCatalogPolicy(record.Policy, policy) { + continue + } + afterRecord := repository.PriceRecord{ + Policy: policy, + Source: repository.PriceSource{Kind: repository.PriceSourceModelsDev, CatalogID: entry.ID, Revision: info.Revision, FetchedAt: info.FetchedAt}, + } + after := priceRecordToDTO(afterRecord) + fetchedAt := info.FetchedAt + changes = append(changes, catalogChangeDTO{Model: record.Policy.Model, CatalogID: entry.ID, Status: "changed", Before: before, After: &after, FetchedAt: &fetchedAt}) + } + sort.Slice(changes, func(i, j int) bool { return changes[i].Model < changes[j].Model }) + return changes, nil +} + +func catalogChangeCounts(changes []catalogChangeDTO) (changed, missing int) { + for _, change := range changes { + if change.Status == "changed" { + changed++ + } else if change.Status == "missing" { + missing++ + } + } + return changed, missing +} + +func sameCatalogPolicy(left, right pricing.Policy) bool { + if left.Model != right.Model || left.Base != right.Base || left.FastPricingEnabled != right.FastPricingEnabled || left.FastMultiplier != right.FastMultiplier { + return false + } + if left.LongContext == nil || right.LongContext == nil { + return left.LongContext == nil && right.LongContext == nil + } + return *left.LongContext == *right.LongContext +} + +func catalogEntryToDTO(entry modelcatalog.Entry) catalogEntryDTO { + dto := catalogEntryDTO{ + ID: entry.ID, Provider: entry.Provider, ProviderName: entry.ProviderName, + Model: entry.Model, ModelName: entry.ModelName, Base: ratesToDTO(entry.Base), + } + if entry.LongContext != nil { + dto.LongContext = &longContextDTO{ + ThresholdInputTokens: entry.LongContext.ThresholdInputTokens, + Comparison: entry.LongContext.Comparison, priceRatesDTO: ratesToDTO(entry.LongContext.Rates), + } + } + return dto +} + +func catalogManagementError(status int, code, message string) ManagementResponse { + return jsonManagementResponse(status, map[string]any{"error": map[string]string{"code": code, "message": message}}) +} diff --git a/internal/plugin/config.go b/internal/plugin/config.go index 60dfaf6..a5c04ac 100644 --- a/internal/plugin/config.go +++ b/internal/plugin/config.go @@ -2,8 +2,12 @@ package plugin import ( "fmt" + "net/url" + "path/filepath" "strings" + "billing/internal/modelcatalog" + "gopkg.in/yaml.v3" ) @@ -13,12 +17,15 @@ type Config struct { DatabasePath string `yaml:"database_path"` BootstrapName string `yaml:"bootstrap_name"` BootstrapKey string `yaml:"bootstrap_key"` + ModelsDevURL string `yaml:"models_dev_url"` + CatalogPath string `yaml:"models_dev_cache_path"` } func defaultConfig() Config { return Config{ - Enabled: true, CodexOnly: true, DatabasePath: "data/cpa-ext.db", + Enabled: true, CodexOnly: true, DatabasePath: "data/billing.db", BootstrapName: "default", BootstrapKey: "000000", + ModelsDevURL: modelcatalog.DefaultSourceURL, } } @@ -39,6 +46,14 @@ func decodeConfig(raw []byte) (Config, error) { if !validManagedSecret(cfg.BootstrapKey) { return Config{}, fmt.Errorf("bootstrap_key 必须为 6-256 个不含空白或控制字符的字符") } + catalogURL, err := url.Parse(strings.TrimSpace(cfg.ModelsDevURL)) + if err != nil || catalogURL.Host == "" || (catalogURL.Scheme != "http" && catalogURL.Scheme != "https") { + return Config{}, fmt.Errorf("models_dev_url 必须是有效的 HTTP(S) URL") + } + cfg.ModelsDevURL = catalogURL.String() + if strings.TrimSpace(cfg.CatalogPath) == "" && cfg.DatabasePath != ":memory:" { + cfg.CatalogPath = filepath.Join(filepath.Dir(cfg.DatabasePath), "models-dev-catalog.json") + } return cfg, nil } diff --git a/internal/plugin/key_management.go b/internal/plugin/key_management.go index 468d2e6..bdc8413 100644 --- a/internal/plugin/key_management.go +++ b/internal/plugin/key_management.go @@ -13,9 +13,9 @@ import ( "time" "unicode" - managedaccess "cpa-ext/internal/access" - "cpa-ext/internal/collection" - "cpa-ext/internal/repository" + managedaccess "billing/internal/access" + "billing/internal/collection" + "billing/internal/repository" ) type createManagedKeyRequest struct { diff --git a/internal/plugin/management.go b/internal/plugin/management.go index 9c227b7..3188c83 100644 --- a/internal/plugin/management.go +++ b/internal/plugin/management.go @@ -10,23 +10,27 @@ import ( "strings" "time" - "cpa-ext/internal/collection" - "cpa-ext/internal/web" + "billing/internal/collection" + "billing/internal/web" ) const ( - managementBase = "/v0/management/plugins/" + PluginName - resourceBase = "/v0/resource/plugins/" + PluginName - routeUsage = "/usage" - routeUsageSummary = "/usage-summary" - routePrices = "/prices" - routeKeys = "/keys" - routeKeyStats = "/key-stats" - routeUpstreams = "/upstreams" - routeModels = "/model-suggestions" - routeBillingReset = "/billing-reset" - routeBillingLedger = "/billing-ledger" - resourceUI = "/ui" + managementBase = "/v0/management/plugins/" + PluginName + resourceBase = "/v0/resource/plugins/" + PluginName + routeUsage = "/usage" + routeUsageSummary = "/usage-summary" + routePrices = "/prices" + routePriceImport = "/prices/import" + routeCatalog = "/price-catalog" + routeCatalogRefresh = "/price-catalog/refresh" + routeCatalogApply = "/price-catalog/apply" + routeKeys = "/keys" + routeKeyStats = "/key-stats" + routeUpstreams = "/upstreams" + routeModels = "/model-suggestions" + routeBillingReset = "/billing-reset" + routeBillingLedger = "/billing-ledger" + resourceUI = "/ui" ) func managementRegistration() ManagementRegistrationResponse { @@ -37,6 +41,10 @@ func managementRegistration() ManagementRegistrationResponse { {Method: http.MethodGet, Path: managementBase + routePrices, Description: "查看模型价格。"}, {Method: http.MethodPut, Path: managementBase + routePrices, Description: "保存模型价格。"}, {Method: http.MethodDelete, Path: managementBase + routePrices, Description: "删除模型价格。"}, + {Method: http.MethodPost, Path: managementBase + routePriceImport, Description: "从 models.dev 导入参考价格。"}, + {Method: http.MethodGet, Path: managementBase + routeCatalog, Description: "搜索 models.dev 参考价格。"}, + {Method: http.MethodPost, Path: managementBase + routeCatalogRefresh, Description: "刷新 models.dev 价格目录并预览变化。"}, + {Method: http.MethodPost, Path: managementBase + routeCatalogApply, Description: "确认应用参考价格变化。"}, {Method: http.MethodGet, Path: managementBase + routeKeys, Description: "查看下游 Key。"}, {Method: http.MethodPost, Path: managementBase + routeKeys, Description: "创建下游 Key。"}, {Method: http.MethodPatch, Path: managementBase + routeKeys, Description: "更新下游 Key。"}, @@ -85,6 +93,18 @@ func (a *App) handleManagement(raw []byte) ([]byte, error) { return OKEnvelope(a.deletePrice(req.Body)) } } + if req.Method == http.MethodPost && path == managementBase+routePriceImport { + return OKEnvelope(a.importCatalogPrice(req.Body)) + } + if req.Method == http.MethodGet && path == managementBase+routeCatalog { + return OKEnvelope(a.searchPriceCatalog(req.Query)) + } + if req.Method == http.MethodPost && path == managementBase+routeCatalogRefresh { + return OKEnvelope(a.refreshPriceCatalog()) + } + if req.Method == http.MethodPost && path == managementBase+routeCatalogApply { + return OKEnvelope(a.applyCatalogChanges(req.Body)) + } if path == managementBase+routeKeys { switch req.Method { case http.MethodGet: diff --git a/internal/plugin/management_test.go b/internal/plugin/management_test.go index 6fc0deb..08e3914 100644 --- a/internal/plugin/management_test.go +++ b/internal/plugin/management_test.go @@ -4,7 +4,9 @@ import ( "encoding/json" "fmt" "net/http" + "net/http/httptest" "net/url" + "path/filepath" "strings" "testing" "time" @@ -52,7 +54,7 @@ func TestManagementRegistrationDeclaresUsageAPIAndUI(t *testing.T) { if err := json.Unmarshal(envelope.Result, ®istration); err != nil { t.Fatal(err) } - if len(registration.Routes) != 14 || registration.Routes[0].Path != managementBase+routeUsage || registration.Routes[1].Path != managementBase+routeUsageSummary || registration.Routes[2].Path != managementBase+routePrices { + if len(registration.Routes) != 18 || registration.Routes[0].Path != managementBase+routeUsage || registration.Routes[1].Path != managementBase+routeUsageSummary || registration.Routes[2].Path != managementBase+routePrices { t.Fatalf("unexpected management routes: %+v", registration.Routes) } if len(registration.Resources) != 1 || registration.Resources[0].Path != resourceBase+resourceUI { @@ -60,6 +62,90 @@ func TestManagementRegistrationDeclaresUsageAPIAndUI(t *testing.T) { } } +func TestModelsDevCatalogRequiresExplicitImportAndConfirmedRefresh(t *testing.T) { + catalogBody := `{"providers":{"openai":{"id":"openai","name":"OpenAI","models":{"gpt-5.6-sol":{"id":"gpt-5.6-sol","name":"GPT-5.6 Sol","cost":{"input":5,"output":30,"cache_read":0.5,"cache_write":6.25}}}},"free":{"id":"free","name":"Free","models":{"placeholder":{"id":"placeholder","cost":{"input":0,"output":0}}}}}}` + serverStatus := http.StatusOK + server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, _ *http.Request) { + response.WriteHeader(serverStatus) + _, _ = response.Write([]byte(catalogBody)) + })) + defer server.Close() + + directory := t.TempDir() + config := fmt.Sprintf("database_path: %q\nmodels_dev_url: %q\nmodels_dev_cache_path: %q\nenabled: true\ncodex_only: false\n", + filepath.Join(directory, "usage.db"), server.URL, filepath.Join(directory, "catalog.json")) + app := NewApp() + defer app.Shutdown() + if _, err := app.HandleMethod(MethodPluginRegister, lifecycleRequest(t, SchemaVersion, config)); err != nil { + t.Fatal(err) + } + + status := managementCallRequest(t, app, ManagementRequest{Method: http.MethodGet, Path: managementBase + routeCatalog, Query: url.Values{"q": {"gpt"}}}) + if status.StatusCode != http.StatusOK || !strings.Contains(string(status.Body), `"loaded":false`) { + t.Fatalf("catalog status=%d body=%s", status.StatusCode, status.Body) + } + refresh := managementCall(t, app, http.MethodPost, managementBase+routeCatalogRefresh) + if refresh.StatusCode != http.StatusOK || !strings.Contains(string(refresh.Body), `"changed":0`) { + t.Fatalf("initial refresh status=%d body=%s", refresh.StatusCode, refresh.Body) + } + search := managementCallRequest(t, app, ManagementRequest{Method: http.MethodGet, Path: managementBase + routeCatalog, Query: url.Values{"q": {"gpt-5.6"}}}) + if search.StatusCode != http.StatusOK || !strings.Contains(string(search.Body), `"id":"openai/gpt-5.6-sol"`) || strings.Contains(string(search.Body), "placeholder") { + t.Fatalf("catalog search status=%d body=%s", search.StatusCode, search.Body) + } + + imported := managementCallBody(t, app, http.MethodPost, managementBase+routePriceImport, []byte(`{"model":"deepseek-v4-flash","catalog_id":"openai/gpt-5.6-sol"}`)) + if imported.StatusCode != http.StatusOK || !strings.Contains(string(imported.Body), `"input_per_1m":"5"`) || !strings.Contains(string(imported.Body), `"kind":"models.dev"`) { + t.Fatalf("catalog import status=%d body=%s", imported.StatusCode, imported.Body) + } + + catalogBody = strings.Replace(catalogBody, `"input":5`, `"input":6`, 1) + preview := managementCall(t, app, http.MethodPost, managementBase+routeCatalogRefresh) + if preview.StatusCode != http.StatusOK || !strings.Contains(string(preview.Body), `"changed":1`) || !strings.Contains(string(preview.Body), `"input_per_1m":"6"`) { + t.Fatalf("refresh preview status=%d body=%s", preview.StatusCode, preview.Body) + } + pricesBefore := managementCall(t, app, http.MethodGet, managementBase+routePrices) + if !strings.Contains(string(pricesBefore.Body), `"input_per_1m":"5"`) { + t.Fatalf("preview silently changed local price: %s", pricesBefore.Body) + } + var previewPayload struct { + Catalog struct { + Revision string `json:"revision"` + } `json:"catalog"` + } + if err := json.Unmarshal(preview.Body, &previewPayload); err != nil { + t.Fatal(err) + } + applyBody, _ := json.Marshal(map[string]any{"revision": previewPayload.Catalog.Revision, "models": []string{"deepseek-v4-flash"}}) + applied := managementCallBody(t, app, http.MethodPost, managementBase+routeCatalogApply, applyBody) + if applied.StatusCode != http.StatusOK || !strings.Contains(string(applied.Body), `"updated":1`) { + t.Fatalf("catalog apply status=%d body=%s", applied.StatusCode, applied.Body) + } + pricesAfter := managementCall(t, app, http.MethodGet, managementBase+routePrices) + if !strings.Contains(string(pricesAfter.Body), `"input_per_1m":"6"`) { + t.Fatalf("confirmed price was not applied: %s", pricesAfter.Body) + } + + manual := []byte(`{"model":"deepseek-v4-flash","base":{"input_per_1m":"7","cache_read_per_1m":"0.5","cache_write_per_1m":"6.25","output_per_1m":"30"},"fast_pricing_enabled":false,"fast_multiplier":"2.5"}`) + if response := managementCallBody(t, app, http.MethodPut, managementBase+routePrices, manual); response.StatusCode != http.StatusOK || !strings.Contains(string(response.Body), `"kind":"manual"`) { + t.Fatalf("manual save status=%d body=%s", response.StatusCode, response.Body) + } + catalogBody = strings.Replace(catalogBody, `"input":6`, `"input":8`, 1) + manualPreview := managementCall(t, app, http.MethodPost, managementBase+routeCatalogRefresh) + if manualPreview.StatusCode != http.StatusOK || !strings.Contains(string(manualPreview.Body), `"changed":0`) { + t.Fatalf("manual price followed catalog: status=%d body=%s", manualPreview.StatusCode, manualPreview.Body) + } + + serverStatus = http.StatusBadGateway + failed := managementCall(t, app, http.MethodPost, managementBase+routeCatalogRefresh) + if failed.StatusCode != http.StatusBadGateway { + t.Fatalf("failed refresh status=%d body=%s", failed.StatusCode, failed.Body) + } + lastGood := managementCallRequest(t, app, ManagementRequest{Method: http.MethodGet, Path: managementBase + routeCatalog, Query: url.Values{"q": {"gpt"}}}) + if lastGood.StatusCode != http.StatusOK || !strings.Contains(string(lastGood.Body), `"loaded":true`) || !strings.Contains(string(lastGood.Body), `"input_per_1m":"8"`) { + t.Fatalf("last-known-good catalog unavailable: status=%d body=%s", lastGood.StatusCode, lastGood.Body) + } +} + func TestUsageManagementSupportsServerPaginationFiltersAndSummary(t *testing.T) { app := NewApp() if _, err := app.HandleMethod(MethodPluginRegister, lifecycleRequest(t, SchemaVersion, testConfig(t, "enabled: true\ncodex_only: false\n"))); err != nil { @@ -467,7 +553,7 @@ func TestUsageResourceServesTablePage(t *testing.T) { if strings.Index(page, `id="page-buttons"`) > strings.Index(page, `id="headers"`) { t.Fatal("UI pagination controls are not above the usage table") } - for _, feature := range []string{`class="surface price-panel price-layout"`, `id="new-price"`, "基础价格 · $ / 1M Token", "长上下文价格", "Fast 价格", "删除 ${model} 的真实价格配置"} { + for _, feature := range []string{`class="surface price-panel price-layout"`, `id="new-price"`, "基础价格 · $ / 1M Token", "长上下文价格", "Fast 价格", "删除 ${model} 的真实价格配置", `id="catalog-query"`, `id="refresh-catalog"`, "PRICE_IMPORT_API", "发现 ${changed.length} 个已关联价格发生变化", "测试目录 · 不访问 models.dev"} { if !strings.Contains(page, feature) { t.Fatalf("UI does not contain redesigned pricing feature %q", feature) } diff --git a/internal/plugin/pricing_management.go b/internal/plugin/pricing_management.go index 686deb8..bf59f22 100644 --- a/internal/plugin/pricing_management.go +++ b/internal/plugin/pricing_management.go @@ -8,8 +8,11 @@ import ( "net/http" "strconv" "strings" + "time" - "cpa-ext/internal/pricing" + "billing/internal/modelcatalog" + "billing/internal/pricing" + "billing/internal/repository" ) type priceRatesDTO struct { @@ -31,6 +34,16 @@ type priceDTO struct { LongContext *longContextDTO `json:"long_context,omitempty"` FastPricingEnabled bool `json:"fast_pricing_enabled"` FastMultiplier string `json:"fast_multiplier"` + Source priceSourceDTO `json:"source"` +} + +type priceSourceDTO struct { + Kind string `json:"kind"` + CatalogID string `json:"catalog_id,omitempty"` + Provider string `json:"provider,omitempty"` + Model string `json:"model,omitempty"` + Revision string `json:"revision,omitempty"` + FetchedAt time.Time `json:"fetched_at,omitzero"` } type priceDeleteRequest struct { @@ -43,13 +56,13 @@ func (a *App) listPrices() ManagementResponse { if a.store == nil { return jsonManagementResponse(http.StatusServiceUnavailable, map[string]any{"error": map[string]string{"code": "database_unavailable", "message": "价格数据库尚未初始化"}}) } - policies, err := a.store.ListPrices(context.Background()) + records, err := a.store.ListPriceRecords(context.Background()) if err != nil { return jsonManagementResponse(http.StatusInternalServerError, map[string]any{"error": map[string]string{"code": "database_error", "message": err.Error()}}) } - prices := make([]priceDTO, 0, len(policies)) - for _, policy := range policies { - prices = append(prices, policyToDTO(policy)) + prices := make([]priceDTO, 0, len(records)) + for _, record := range records { + prices = append(prices, priceRecordToDTO(record)) } return jsonManagementResponse(http.StatusOK, map[string]any{"prices": prices}) } @@ -63,19 +76,35 @@ func (a *App) putPrice(body []byte) ManagementResponse { if err != nil { return jsonManagementResponse(http.StatusBadRequest, map[string]any{"error": map[string]string{"code": "invalid_price", "message": err.Error()}}) } + return a.persistPriceRecord(repository.PriceRecord{Policy: policy, Source: repository.PriceSource{Kind: repository.PriceSourceManual}}) +} + +func (a *App) persistPriceRecord(record repository.PriceRecord) ManagementResponse { + return a.persistPriceRecordFromCatalog(record, nil) +} + +func (a *App) persistPriceRecordFromCatalog(record repository.PriceRecord, expectedCatalog *modelcatalog.Manager) ManagementResponse { a.priceMu.Lock() defer a.priceMu.Unlock() + return a.persistPriceRecordLocked(record, expectedCatalog) +} + +func (a *App) persistPriceRecordLocked(record repository.PriceRecord, expectedCatalog *modelcatalog.Manager) ManagementResponse { a.mu.Lock() store := a.store if store == nil { a.mu.Unlock() return jsonManagementResponse(http.StatusServiceUnavailable, map[string]any{"error": map[string]string{"code": "database_unavailable", "message": "价格数据库尚未初始化"}}) } - if err := store.UpsertPrice(context.Background(), policy); err != nil { + if expectedCatalog != nil && a.catalog != expectedCatalog { + a.mu.Unlock() + return jsonManagementResponse(http.StatusServiceUnavailable, map[string]any{"error": map[string]string{"code": "catalog_reconfigured", "message": "价格目录配置已经变化，请重新搜索"}}) + } + if err := store.UpsertPriceRecord(context.Background(), record); err != nil { a.mu.Unlock() return jsonManagementResponse(http.StatusInternalServerError, map[string]any{"error": map[string]string{"code": "database_error", "message": err.Error()}}) } - a.prices[normalizeModelName(policy.Model)] = policy + a.prices[normalizeModelName(record.Policy.Model)] = record.Policy a.mu.Unlock() a.mu.RLock() currentStore := a.store == store @@ -83,7 +112,7 @@ func (a *App) putPrice(body []byte) ManagementResponse { if !currentStore { return jsonManagementResponse(http.StatusServiceUnavailable, map[string]any{"error": map[string]string{"code": "database_reconfigured", "message": "价格已经保存，请在重新配置后重试历史补算"}}) } - if _, err := store.BackfillMissingCosts(context.Background(), policy); err != nil { + if _, err := store.BackfillMissingCosts(context.Background(), record.Policy); err != nil { a.mu.RLock() reconfigured := a.store != store a.mu.RUnlock() @@ -92,7 +121,7 @@ func (a *App) putPrice(body []byte) ManagementResponse { } return jsonManagementResponse(http.StatusInternalServerError, map[string]any{"error": map[string]string{"code": "database_error", "message": err.Error()}}) } - return jsonManagementResponse(http.StatusOK, policyToDTO(policy)) + return jsonManagementResponse(http.StatusOK, priceRecordToDTO(record)) } func (a *App) deletePrice(body []byte) ManagementResponse { @@ -162,13 +191,30 @@ func dtoRates(dto priceRatesDTO) (pricing.Rates, error) { } func policyToDTO(policy pricing.Policy) priceDTO { - dto := priceDTO{Model: policy.Model, Base: ratesToDTO(policy.Base), FastPricingEnabled: policy.FastPricingEnabled, FastMultiplier: formatRatio(policy.FastMultiplier)} + dto := priceDTO{Model: policy.Model, Base: ratesToDTO(policy.Base), FastPricingEnabled: policy.FastPricingEnabled, FastMultiplier: formatRatio(policy.FastMultiplier), Source: priceSourceDTO{Kind: repository.PriceSourceManual}} if policy.LongContext != nil { dto.LongContext = &longContextDTO{ThresholdInputTokens: policy.LongContext.ThresholdInputTokens, Comparison: policy.LongContext.Comparison, priceRatesDTO: ratesToDTO(policy.LongContext.Rates)} } return dto } +func priceRecordToDTO(record repository.PriceRecord) priceDTO { + dto := policyToDTO(record.Policy) + source := record.Source + if source.Kind == "" { + source.Kind = repository.PriceSourceManual + } + dto.Source = priceSourceDTO{Kind: source.Kind, CatalogID: source.CatalogID, Revision: source.Revision, FetchedAt: source.FetchedAt} + if source.CatalogID != "" { + parts := strings.SplitN(source.CatalogID, "/", 2) + dto.Source.Provider = parts[0] + if len(parts) == 2 { + dto.Source.Model = parts[1] + } + } + return dto +} + func ratesToDTO(rates pricing.Rates) priceRatesDTO { return priceRatesDTO{InputPer1M: formatMicros(rates.InputMicrosPer1M), CacheReadPer1M: formatMicros(rates.CacheReadMicrosPer1M), CacheWritePer1M: formatMicros(rates.CacheWriteMicrosPer1M), OutputPer1M: formatMicros(rates.OutputMicrosPer1M)} } diff --git a/internal/plugin/types.go b/internal/plugin/types.go index 012d715..0d38281 100644 --- a/internal/plugin/types.go +++ b/internal/plugin/types.go @@ -10,8 +10,8 @@ import ( const ( ABIVersion uint32 = 1 SchemaVersion uint32 = 3 - PluginName = "cpa-ext" - Version = "0.1.0-dev" + PluginName = "billing" + Version = "0.1.0" ) const ( diff --git a/internal/pricing/pricing_test.go b/internal/pricing/pricing_test.go index eb96242..7400306 100644 --- a/internal/pricing/pricing_test.go +++ b/internal/pricing/pricing_test.go @@ -3,7 +3,7 @@ package pricing_test import ( "testing" - "cpa-ext/internal/pricing" + "billing/internal/pricing" ) func testPolicy() pricing.Policy { diff --git a/internal/repository/sqlite_access.go b/internal/repository/sqlite_access.go index 4eee825..03c0fff 100644 --- a/internal/repository/sqlite_access.go +++ b/internal/repository/sqlite_access.go @@ -9,7 +9,7 @@ import ( "strings" "time" - managedaccess "cpa-ext/internal/access" + managedaccess "billing/internal/access" ) var ErrManagedKeyNotFound = errors.New("managed key not found") diff --git a/internal/repository/sqlite_access_test.go b/internal/repository/sqlite_access_test.go index fdf5d47..6221cf4 100644 --- a/internal/repository/sqlite_access_test.go +++ b/internal/repository/sqlite_access_test.go @@ -6,9 +6,9 @@ import ( "testing" "time" - managedaccess "cpa-ext/internal/access" - "cpa-ext/internal/collection" - "cpa-ext/internal/repository" + managedaccess "billing/internal/access" + "billing/internal/collection" + "billing/internal/repository" ) func TestManagedKeysBootstrapLifecycleAndHistoricalUsage(t *testing.T) { diff --git a/internal/repository/sqlite_billing.go b/internal/repository/sqlite_billing.go index 66659da..0a1d048 100644 --- a/internal/repository/sqlite_billing.go +++ b/internal/repository/sqlite_billing.go @@ -8,7 +8,7 @@ import ( "strings" "time" - managedaccess "cpa-ext/internal/access" + managedaccess "billing/internal/access" ) var ( diff --git a/internal/repository/sqlite_billing_test.go b/internal/repository/sqlite_billing_test.go index e24c468..a36dce3 100644 --- a/internal/repository/sqlite_billing_test.go +++ b/internal/repository/sqlite_billing_test.go @@ -7,8 +7,8 @@ import ( "testing" "time" - managedaccess "cpa-ext/internal/access" - "cpa-ext/internal/collection" + managedaccess "billing/internal/access" + "billing/internal/collection" ) func TestBillingQuotaConcurrencySettlementAndReset(t *testing.T) { diff --git a/internal/repository/sqlite_performance_test.go b/internal/repository/sqlite_performance_test.go index 25c7b99..5609a73 100644 --- a/internal/repository/sqlite_performance_test.go +++ b/internal/repository/sqlite_performance_test.go @@ -7,8 +7,8 @@ import ( "testing" "time" - "cpa-ext/internal/collection" - "cpa-ext/internal/pricing" + "billing/internal/collection" + "billing/internal/pricing" ) func TestSQLiteSeparatesReaderPoolFromWriter(t *testing.T) { diff --git a/internal/repository/sqlite_pricing.go b/internal/repository/sqlite_pricing.go index e73086d..342b8d2 100644 --- a/internal/repository/sqlite_pricing.go +++ b/internal/repository/sqlite_pricing.go @@ -2,69 +2,163 @@ package repository import ( "context" + "database/sql" + "errors" "fmt" + "strings" "time" - "cpa-ext/internal/pricing" + "billing/internal/pricing" ) +const ( + PriceSourceManual = "manual" + PriceSourceModelsDev = "models.dev" +) + +// PriceSource records where the current effective local price was imported +// from. It is bookkeeping only; billing always reads the embedded Policy. +type PriceSource struct { + Kind string + CatalogID string + Revision string + FetchedAt time.Time +} + +type PriceRecord struct { + Policy pricing.Policy + Source PriceSource +} + // ListPrices returns every configured exact-model policy. func (r *SQLiteUsageRepository) ListPrices(ctx context.Context) ([]pricing.Policy, error) { + records, err := r.ListPriceRecords(ctx) + if err != nil { + return nil, err + } + policies := make([]pricing.Policy, 0, len(records)) + for _, record := range records { + policies = append(policies, record.Policy) + } + return policies, nil +} + +// ListPriceRecords returns effective prices together with their optional +// models.dev import link. +func (r *SQLiteUsageRepository) ListPriceRecords(ctx context.Context) ([]PriceRecord, error) { rows, err := r.readDB.QueryContext(ctx, ` SELECT model, input_rate_micros, cache_read_rate_micros, cache_write_rate_micros, output_rate_micros, long_context_enabled, long_context_threshold, long_context_comparison, long_input_rate_micros, long_cache_read_rate_micros, long_cache_write_rate_micros, long_output_rate_micros, - fast_pricing_enabled, fast_multiplier_numerator, fast_multiplier_denominator + fast_pricing_enabled, fast_multiplier_numerator, fast_multiplier_denominator, + source_kind, source_catalog_id, source_revision, source_fetched_at FROM model_prices ORDER BY model`) if err != nil { return nil, fmt.Errorf("查询模型价格: %w", err) } defer rows.Close() - var policies []pricing.Policy + var records []PriceRecord for rows.Next() { - var policy pricing.Policy + var record PriceRecord var longEnabled bool var long pricing.LongContext + var fetchedAt string if err := rows.Scan( - &policy.Model, - &policy.Base.InputMicrosPer1M, &policy.Base.CacheReadMicrosPer1M, - &policy.Base.CacheWriteMicrosPer1M, &policy.Base.OutputMicrosPer1M, + &record.Policy.Model, + &record.Policy.Base.InputMicrosPer1M, &record.Policy.Base.CacheReadMicrosPer1M, + &record.Policy.Base.CacheWriteMicrosPer1M, &record.Policy.Base.OutputMicrosPer1M, &longEnabled, &long.ThresholdInputTokens, &long.Comparison, &long.Rates.InputMicrosPer1M, &long.Rates.CacheReadMicrosPer1M, &long.Rates.CacheWriteMicrosPer1M, &long.Rates.OutputMicrosPer1M, - &policy.FastPricingEnabled, &policy.FastMultiplier.Numerator, &policy.FastMultiplier.Denominator, + &record.Policy.FastPricingEnabled, &record.Policy.FastMultiplier.Numerator, &record.Policy.FastMultiplier.Denominator, + &record.Source.Kind, &record.Source.CatalogID, &record.Source.Revision, &fetchedAt, ); err != nil { return nil, fmt.Errorf("读取模型价格: %w", err) } if longEnabled { - policy.LongContext = &long + record.Policy.LongContext = &long } - policies = append(policies, policy) + if strings.TrimSpace(record.Source.Kind) == "" { + record.Source.Kind = PriceSourceManual + } + if fetchedAt != "" { + parsed, parseErr := time.Parse(time.RFC3339Nano, fetchedAt) + if parseErr != nil { + return nil, fmt.Errorf("解析模型价格来源时间: %w", parseErr) + } + record.Source.FetchedAt = parsed + } + records = append(records, record) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("遍历模型价格: %w", err) } - return policies, nil + return records, nil } // UpsertPrice atomically replaces one model policy. func (r *SQLiteUsageRepository) UpsertPrice(ctx context.Context, policy pricing.Policy) error { - if err := policy.Validate(); err != nil { - return err + return r.UpsertPriceRecord(ctx, PriceRecord{Policy: policy, Source: PriceSource{Kind: PriceSourceManual}}) +} + +// UpsertPriceRecord atomically replaces one effective price and its source. +func (r *SQLiteUsageRepository) UpsertPriceRecord(ctx context.Context, record PriceRecord) error { + return r.UpsertPriceRecords(ctx, []PriceRecord{record}) +} + +// UpsertPriceRecords updates a confirmed catalog diff in one transaction. +func (r *SQLiteUsageRepository) UpsertPriceRecords(ctx context.Context, records []PriceRecord) error { + if len(records) == 0 { + return nil } + for index := range records { + if strings.TrimSpace(records[index].Source.Kind) == "" { + records[index].Source.Kind = PriceSourceManual + } + if err := validatePriceRecord(records[index]); err != nil { + return err + } + } + tx, err := r.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("开始保存模型价格: %w", err) + } + for _, record := range records { + if err := upsertPriceRecord(ctx, tx, record); err != nil { + _ = tx.Rollback() + return err + } + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("提交模型价格: %w", err) + } + return nil +} + +type priceExecer interface { + ExecContext(context.Context, string, ...any) (sql.Result, error) +} + +func upsertPriceRecord(ctx context.Context, executor priceExecer, record PriceRecord) error { + policy := record.Policy longEnabled := policy.LongContext != nil long := pricing.LongContext{Comparison: "gt"} if policy.LongContext != nil { long = *policy.LongContext } - _, err := r.db.ExecContext(ctx, ` + fetchedAt := "" + if !record.Source.FetchedAt.IsZero() { + fetchedAt = record.Source.FetchedAt.UTC().Format(time.RFC3339Nano) + } + _, err := executor.ExecContext(ctx, ` INSERT INTO model_prices ( model, input_rate_micros, cache_read_rate_micros, cache_write_rate_micros, output_rate_micros, long_context_enabled, long_context_threshold, long_context_comparison, long_input_rate_micros, long_cache_read_rate_micros, long_cache_write_rate_micros, long_output_rate_micros, - fast_pricing_enabled, fast_multiplier_numerator, fast_multiplier_denominator, updated_at -) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + fast_pricing_enabled, fast_multiplier_numerator, fast_multiplier_denominator, + source_kind, source_catalog_id, source_revision, source_fetched_at, updated_at +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(model) DO UPDATE SET input_rate_micros=excluded.input_rate_micros, cache_read_rate_micros=excluded.cache_read_rate_micros, @@ -80,6 +174,10 @@ ON CONFLICT(model) DO UPDATE SET fast_pricing_enabled=excluded.fast_pricing_enabled, fast_multiplier_numerator=excluded.fast_multiplier_numerator, fast_multiplier_denominator=excluded.fast_multiplier_denominator, + source_kind=excluded.source_kind, + source_catalog_id=excluded.source_catalog_id, + source_revision=excluded.source_revision, + source_fetched_at=excluded.source_fetched_at, updated_at=excluded.updated_at`, policy.Model, policy.Base.InputMicrosPer1M, policy.Base.CacheReadMicrosPer1M, @@ -88,6 +186,7 @@ ON CONFLICT(model) DO UPDATE SET long.Rates.InputMicrosPer1M, long.Rates.CacheReadMicrosPer1M, long.Rates.CacheWriteMicrosPer1M, long.Rates.OutputMicrosPer1M, policy.FastPricingEnabled, policy.FastMultiplier.Numerator, policy.FastMultiplier.Denominator, + record.Source.Kind, record.Source.CatalogID, record.Source.Revision, fetchedAt, time.Now().UTC().Format(time.RFC3339Nano), ) if err != nil { @@ -96,6 +195,26 @@ ON CONFLICT(model) DO UPDATE SET return nil } +func validatePriceRecord(record PriceRecord) error { + if err := record.Policy.Validate(); err != nil { + return err + } + record.Source.Kind = strings.TrimSpace(record.Source.Kind) + switch record.Source.Kind { + case "", PriceSourceManual: + if record.Source.CatalogID != "" || record.Source.Revision != "" || !record.Source.FetchedAt.IsZero() { + return errors.New("手动价格不能包含目录来源") + } + case PriceSourceModelsDev: + if strings.TrimSpace(record.Source.CatalogID) == "" || strings.TrimSpace(record.Source.Revision) == "" || record.Source.FetchedAt.IsZero() { + return errors.New("models.dev 价格来源不完整") + } + default: + return fmt.Errorf("不支持的价格来源 %q", record.Source.Kind) + } + return nil +} + const costBackfillBatchSize = 1000 // BackfillMissingCosts 只补算尚未定价的历史记录，已经保存的账单金额不会随价格修改而变化。 diff --git a/internal/repository/sqlite_pricing_test.go b/internal/repository/sqlite_pricing_test.go index 95f50a3..6f3e036 100644 --- a/internal/repository/sqlite_pricing_test.go +++ b/internal/repository/sqlite_pricing_test.go @@ -6,9 +6,9 @@ import ( "testing" "time" - "cpa-ext/internal/collection" - "cpa-ext/internal/pricing" - "cpa-ext/internal/repository" + "billing/internal/collection" + "billing/internal/pricing" + "billing/internal/repository" ) func TestSQLitePricingRoundTripAndDelete(t *testing.T) { @@ -44,6 +44,40 @@ func TestSQLitePricingRoundTripAndDelete(t *testing.T) { } } +func TestSQLitePricingPersistsCatalogSourceAndManualSaveClearsIt(t *testing.T) { + store, err := repository.OpenSQLiteUsage(filepath.Join(t.TempDir(), "usage.db")) + if err != nil { + t.Fatal(err) + } + defer store.Close() + fetchedAt := time.Date(2026, 8, 15, 10, 0, 0, 0, time.UTC) + policy := pricing.Policy{ + Model: "deepseek-v4-flash", Base: pricing.Rates{InputMicrosPer1M: 5_000_000, CacheReadMicrosPer1M: 500_000, CacheWriteMicrosPer1M: 6_250_000, OutputMicrosPer1M: 30_000_000}, + FastMultiplier: pricing.Ratio{Numerator: 5, Denominator: 2}, + } + if err := store.UpsertPriceRecord(context.Background(), repository.PriceRecord{ + Policy: policy, + Source: repository.PriceSource{Kind: repository.PriceSourceModelsDev, CatalogID: "openai/gpt-5.6-sol", Revision: "revision-one", FetchedAt: fetchedAt}, + }); err != nil { + t.Fatal(err) + } + records, err := store.ListPriceRecords(context.Background()) + if err != nil || len(records) != 1 { + t.Fatalf("records=%+v err=%v", records, err) + } + if records[0].Source.Kind != repository.PriceSourceModelsDev || records[0].Source.CatalogID != "openai/gpt-5.6-sol" || !records[0].Source.FetchedAt.Equal(fetchedAt) { + t.Fatalf("catalog source = %+v", records[0].Source) + } + policy.Base.InputMicrosPer1M = 7_000_000 + if err := store.UpsertPrice(context.Background(), policy); err != nil { + t.Fatal(err) + } + records, err = store.ListPriceRecords(context.Background()) + if err != nil || records[0].Source.Kind != repository.PriceSourceManual || records[0].Source.CatalogID != "" || records[0].Policy.Base.InputMicrosPer1M != 7_000_000 { + t.Fatalf("manual record=%+v err=%v", records, err) + } +} + func TestBackfillMissingCostsDoesNotRewriteExistingCost(t *testing.T) { store, err := repository.OpenSQLiteUsage(filepath.Join(t.TempDir(), "usage.db")) if err != nil { diff --git a/internal/repository/sqlite_projection.go b/internal/repository/sqlite_projection.go index cfc4b57..3cd8f4e 100644 --- a/internal/repository/sqlite_projection.go +++ b/internal/repository/sqlite_projection.go @@ -8,7 +8,7 @@ import ( "strings" "time" - "cpa-ext/internal/collection" + "billing/internal/collection" ) const requestDetailProjectionMigration = "request-detail-index-v1" diff --git a/internal/repository/sqlite_query.go b/internal/repository/sqlite_query.go index 9eaa41e..e05e063 100644 --- a/internal/repository/sqlite_query.go +++ b/internal/repository/sqlite_query.go @@ -13,7 +13,7 @@ import ( "strings" "time" - "cpa-ext/internal/collection" + "billing/internal/collection" ) type usageCursor struct { diff --git a/internal/repository/sqlite_query_migration_test.go b/internal/repository/sqlite_query_migration_test.go index befa61c..062cdd7 100644 --- a/internal/repository/sqlite_query_migration_test.go +++ b/internal/repository/sqlite_query_migration_test.go @@ -6,7 +6,7 @@ import ( "testing" "time" - "cpa-ext/internal/collection" + "billing/internal/collection" ) func TestRequestDetailProjectionBackfillsExistingFacts(t *testing.T) { diff --git a/internal/repository/sqlite_query_scale_test.go b/internal/repository/sqlite_query_scale_test.go index 5400f51..f1e1e39 100644 --- a/internal/repository/sqlite_query_scale_test.go +++ b/internal/repository/sqlite_query_scale_test.go @@ -8,7 +8,7 @@ import ( "testing" "time" - "cpa-ext/internal/collection" + "billing/internal/collection" ) func TestSQLiteUsageQueryAtMillionRows(t *testing.T) { diff --git a/internal/repository/sqlite_query_test.go b/internal/repository/sqlite_query_test.go index 036f309..ab9c0fa 100644 --- a/internal/repository/sqlite_query_test.go +++ b/internal/repository/sqlite_query_test.go @@ -7,8 +7,8 @@ import ( "testing" "time" - "cpa-ext/internal/collection" - "cpa-ext/internal/repository" + "billing/internal/collection" + "billing/internal/repository" ) func TestSQLiteUsageQueryPaginatesAndFilters(t *testing.T) { diff --git a/internal/repository/sqlite_summary.go b/internal/repository/sqlite_summary.go index 7047437..ed9d656 100644 --- a/internal/repository/sqlite_summary.go +++ b/internal/repository/sqlite_summary.go @@ -6,7 +6,7 @@ import ( "fmt" "time" - "cpa-ext/internal/collection" + "billing/internal/collection" ) const usageDashboardUsersSQL = ` diff --git a/internal/repository/sqlite_usage.go b/internal/repository/sqlite_usage.go index 9e6d3b4..4e2139e 100644 --- a/internal/repository/sqlite_usage.go +++ b/internal/repository/sqlite_usage.go @@ -1,4 +1,4 @@ -// Package repository 实现 cpa-ext 的本地持久化。 +// Package repository 实现 billing 的本地持久化。 package repository import ( @@ -14,7 +14,7 @@ import ( "strings" "time" - "cpa-ext/internal/collection" + "billing/internal/collection" _ "github.com/mattn/go-sqlite3" ) @@ -113,6 +113,7 @@ ON request_detail_index(endpoint_kind, requested_at DESC, id DESC); CREATE INDEX IF NOT EXISTS idx_request_detail_request ON request_detail_index(request_id, requested_at DESC, id DESC); +-- Keep the legacy table name so existing databases retain migration history. CREATE TABLE IF NOT EXISTS cpa_ext_migrations ( name TEXT PRIMARY KEY, completed_at TEXT NOT NULL @@ -168,6 +169,10 @@ CREATE TABLE IF NOT EXISTS model_prices ( fast_pricing_enabled INTEGER NOT NULL DEFAULT 0, fast_multiplier_numerator INTEGER NOT NULL DEFAULT 5, fast_multiplier_denominator INTEGER NOT NULL DEFAULT 2, + source_kind TEXT NOT NULL DEFAULT 'manual', + source_catalog_id TEXT NOT NULL DEFAULT '', + source_revision TEXT NOT NULL DEFAULT '', + source_fetched_at TEXT NOT NULL DEFAULT '', updated_at TEXT NOT NULL ); @@ -268,6 +273,10 @@ func OpenSQLiteUsage(databasePath string) (*SQLiteUsageRepository, error) { `ALTER TABLE usage_records ADD COLUMN auth_index TEXT NOT NULL DEFAULT ''`, `ALTER TABLE usage_records ADD COLUMN auth_type TEXT NOT NULL DEFAULT ''`, `ALTER TABLE usage_records ADD COLUMN billing_event_key TEXT NOT NULL DEFAULT ''`, + `ALTER TABLE model_prices ADD COLUMN source_kind TEXT NOT NULL DEFAULT 'manual'`, + `ALTER TABLE model_prices ADD COLUMN source_catalog_id TEXT NOT NULL DEFAULT ''`, + `ALTER TABLE model_prices ADD COLUMN source_revision TEXT NOT NULL DEFAULT ''`, + `ALTER TABLE model_prices ADD COLUMN source_fetched_at TEXT NOT NULL DEFAULT ''`, } { if _, err := db.Exec(migration); err != nil && !strings.Contains(strings.ToLower(err.Error()), "duplicate column") { _ = db.Close() @@ -326,7 +335,7 @@ func sqliteDSN(databasePath string) (string, error) { return "", errors.New("database_path 不能为空") } if databasePath == ":memory:" { - return "file:cpa-ext-memory?mode=memory&cache=shared&_busy_timeout=5000&_foreign_keys=on", nil + return "file:billing-memory?mode=memory&cache=shared&_busy_timeout=5000&_foreign_keys=on", nil } if strings.HasPrefix(databasePath, "file:") { separator := "?" diff --git a/internal/repository/sqlite_usage_test.go b/internal/repository/sqlite_usage_test.go index ded5b3b..8d71515 100644 --- a/internal/repository/sqlite_usage_test.go +++ b/internal/repository/sqlite_usage_test.go @@ -7,8 +7,8 @@ import ( "testing" "time" - "cpa-ext/internal/collection" - "cpa-ext/internal/repository" + "billing/internal/collection" + "billing/internal/repository" ) func TestSQLiteUsagePersistsAllRecordsAndLimitsQueries(t *testing.T) { diff --git a/internal/web/ui.html b/internal/web/ui.html index 87c1f94..be12c68 100644 --- a/internal/web/ui.html +++ b/internal/web/ui.html @@ -3,7 +3,7 @@ - CPA Ext 管理台 + Billing 管理台
-
CX
CPA Ext · 管理控制台
+
B
Billing · 管理控制台
@@ -245,7 +255,8 @@

模型价格

-
+
手动配置
+

models.dev 参考价格

基础价格 · $ / 1M Token

@@ -288,23 +299,31 @@ diff --git a/scripts/build.sh b/scripts/build.sh index 2a38f73..16472a7 100644 --- a/scripts/build.sh +++ b/scripts/build.sh @@ -14,6 +14,6 @@ if [[ -n "$unformatted" ]]; then fi go list -mod=readonly ./... >/dev/null CGO_ENABLED=1 go test ./... -CGO_ENABLED=1 go build -tags cshared -buildmode=c-shared -o bin/cpa-ext.so ./cmd/cpa-ext +CGO_ENABLED=1 go build -tags cshared -buildmode=c-shared -o bin/billing.so ./cmd/billing -echo "Built: $root/bin/cpa-ext.so" +echo "Built: $root/bin/billing.so" diff --git a/scripts/package.sh b/scripts/package.sh new file mode 100755 index 0000000..6a10804 --- /dev/null +++ b/scripts/package.sh @@ -0,0 +1,41 @@ +#!/usr/bin/env bash +set -euo pipefail + +root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +version="${1:-0.1.0}" +target="linux_amd64" +name="billing_${version}_${target}" +artifact="$root/bin/billing.so" + +if [[ ! "$version" =~ ^[0-9]+\.[0-9]+\.[0-9]+([.-][0-9A-Za-z.-]+)?$ ]]; then + echo "Invalid version: $version" >&2 + exit 1 +fi +if [[ ! -f "$artifact" ]]; then + echo "Missing $artifact; run scripts/build.sh first." >&2 + exit 1 +fi + +stage="$(mktemp -d)" +trap 'rm -rf "$stage"' EXIT +package_dir="$stage/$name" +mkdir -p "$package_dir" "$root/dist" + +install -m 0755 "$artifact" "$package_dir/billing.so" +install -m 0644 "$root/README.md" "$package_dir/README.md" +install -m 0644 "$root/config.example.yaml" "$package_dir/config.example.yaml" +printf '%s\n' \ + "name=billing" \ + "version=$version" \ + "target=linux/amd64" \ + "native_abi=1" \ + "rpc_schema=3" \ + "cliproxyapi_revision=f43aad7637ad813745bf7d341acb5663617570c5" \ + >"$package_dir/VERSION.txt" + +archive="$root/dist/$name.tar.gz" +tar -czf "$archive" -C "$stage" "$name" +sha256sum "$archive" >"$archive.sha256" + +echo "Packaged: $archive" +echo "Checksum: $archive.sha256"