LEKKALAGANESH/urja-meter-ops-api

# Flock Energy — Urja Meter Ops API 这是一个基于老旧的 **Urja Meter Ops** 门户构建的、干净且文档完善的 REST API，因为该门户本身并不提供 API。 该服务以普通用户身份登录门户，管理会话，调用我逆向工程出的内部 endpoint，将门户不一致的 payload 进行标准化处理，并将结果作为带类型的 JSON 提供，这样其他程序就可以在不直接接触门户的情况下消费数据。 | | | |---|---| | **门户工作原理** | [`PROTOCOL.md`](PROTOCOL.md) — 逆向工程说明文档 | | **API 规范** | [`openapi.json`](openapi.json) (OpenAPI 3.1，由代码生成) | | **交互式文档** | 运行后的 `/docs` (Swagger UI) 和 `/redoc` | | **反思** | [`REFLECTION.md`](REFLECTION.md) | | **设计记录** | [`docs/adr/`](docs/adr/) | | **运维** | [`docs/RUNBOOK.md`](docs/RUNBOOK.md) — 部署、健康语义、故障模式 | | **Web 控制台** | 运行后的 `/app` — 电表、层级、消耗、地图、数据质量 | ## 快速开始 需要 Python 3.11+。 ``` git clone && cd flock-energy-api python -m venv .venv && source .venv/bin/activate # Windows: .venv\Scripts\activate pip install -r requirements.txt # or: pip install -r requirements.lock (hash-pinned) cp .env.example .env # then fill in PORTAL_USERNAME / PORTAL_PASSWORD uvicorn app.main:app --reload ``` 打开 访问运维控制台，或 打开 查看 API 文档。在启动时，该服务会进行身份验证并拉取整个电表资产 — 包括 403 个电表和 40 个变压器， 耗时约 600 毫秒。

改用 Docker

``` cp .env.example .env # fill in credentials docker compose up --build ```

### 验证其是否正常工作 ``` curl -s localhost:8000/api/v1/health/live # {"status":"alive",...} curl -s localhost:8000/api/v1/system/snapshot # meter_count: 403, source: "export" ``` ## 示例请求 ``` curl -s 'localhost:8000/api/v1/meters?make=HPL&install_status=faulty&page_size=1' ``` ``` { "items": [ { "meter_id": "J100018", "serial_no": "GE50943", "make": "HPL", "phase_type": "single", "install_status": "faulty", "dt_code": "DT-019", "location": { "latitude": 27.023530279896317, "longitude": 75.74833874320397 } } ], "meta": { "page": 1, "page_size": 1, "total_items": 23, "total_pages": 23, "has_next": true, "has_previous": false } } ``` 针对该门户是无法执行此类查询的：它无法按品牌或状态进行过滤，并且永远不会在列表中暴露坐标。

更多示例

``` # 实际消耗量，按日汇总（portal 仅显示原始 register 读数） curl -s 'localhost:8000/api/v1/meters/J100001/consumption?granularity=daily' # 重建的网络树，深度为两层 curl -s 'localhost:8000/api/v1/hierarchy?depth=2' # 距离某点 2 公里内的 Meter curl -s 'localhost:8000/api/v1/meters/near?lat=26.9124&lng=75.7873&radius_km=2' # 上游数据中发现的所有不一致之处 curl -s 'localhost:8000/api/v1/data-quality' # Estate 汇总 curl -s 'localhost:8000/api/v1/stats' ``` 消耗量响应（节选）： ``` { "meter_id": "J100001", "window_start": "2026-06-23T23:30:00", "window_end": "2026-06-30T23:30:00", "granularity": "daily", "summary": { "total_kwh": 138.8, "total_kvah": 149.9, "average_power_factor": 0.926, "peak_interval_kwh": 19.83, "min_voltage_r": 220.0, "max_voltage_r": 240.0, "interval_minutes": 30.0, "reading_count": 337, "gap_count": 0 }, "intervals": [ { "start": "2026-06-24T00:00:00", "end": "2026-06-25T00:00:00", "duration_minutes": 1440.0, "kwh": 19.82, "kvah": 21.42, "average_voltage_r": 228.698, "estimated": false } ] } ```

## 我构建了什么 ### 塑造了设计的三个发现 下面的所有内容都源于该门户的实际情况。完整证据见 [`PROTOCOL.md`](PROTOCOL.md)。 **1. 存在批量导出 — 并且比其他任何内容都要丰富。** `GET /portal/export` (HMAC 签名) 会在**一次请求中返回所有 403 个电表**，其中包含了分页列表中完全忽略的完整层级路径和浮点坐标。这使得架构从“代理每一个调用”转变为“保留本地快照并对其进行查询”，这也是进行过滤、排序、层级和邻近搜索得以实现的唯一原因。 **2. `kwh`/`kvah` 是累积寄存器，而不是使用量。** 这些值只会不断增加 — 它们是里程表读数。将它们相加会得出一个家庭大约 1400 万 kWh 的读数。消耗量是连续读数之间的*差值*，因此 API 提供的是推导出的时间间隔，而不是原始寄存器数据。 **3. 层级代码不是全局唯一的。** `D-01` 出现在三个不同的 Circle 下。因此，节点是由其完整的祖先路径 (`Z-01/C-01/D-01`) 来标识的，而不是通过代码 — 按代码折叠会合并不相关的分支，并破坏它们上方的所有电表计数。 ### 架构 ``` HTTP client │ ┌───────────▼────────────┐ │ API layer │ routing · validation · pagination · error contract │ app/api/ │ ├────────────────────────┤ │ Snapshot store │ in-memory index · TTL freshness · filter/sort/geo queries │ app/store/ │ ├────────────────────────┤ │ Domain │ canonical models · normalisation · hierarchy · consumption │ app/domain/ │ ← pure, no I/O, no framework imports ├────────────────────────┤ │ Portal adapter │ auth · session · HMAC signing · devalue · retries │ app/portal/ │ ← the only layer that knows the portal exists └───────────▲────────────┘ │ Urja Meter Ops portal ``` 依赖关系向内指向。domain 层不导入任何 HTTP 库或框架，因此它的逻辑 — 所有真正棘手的推理都在这里 — 可以在没有网络或应用实例的情况下进行测试。 ``` app/ ├── main.py Composition root, lifespan, exception handlers ├── config.py Environment-driven settings ├── errors.py Public error contract ├── logging_setup.py Structured logs with request-scoped correlation IDs ├── portal/ │ ├── session.py Login, expiry tracking, single-flight re-auth, 429 backoff │ ├── client.py Endpoint calls, retries, three-way expiry detection │ ├── signing.py HMAC-SHA256 request signing │ ├── devalue.py SvelteKit __data.json decoder │ └── exceptions.py ├── domain/ │ ├── models.py Canonical pydantic models │ ├── normalize.py Messy upstream payloads → canonical objects │ ├── hierarchy.py Tree reconstruction, blank repair, registry reconciliation │ ├── consumption.py Register deltas, cadence inference, resampling │ └── quality.py Data-quality aggregation ├── store/snapshot.py Snapshot lifecycle + query layer └── api/v1/ meters · hierarchy · transformers · system ``` ### Web 控制台 `/app` 是一个基于相同 API 的小型运维控制台：资产概览、带有详情侧边栏和消耗图表的可过滤电表表、重建的网络树、地理散点图以及数据质量报告。电表表的过滤、排序和分页状态保存在 URL hash 中，因此过滤后的视图是可以共享的，并且在刷新后依然存在。 原生 ES modules，无构建步骤，无运行时依赖 — **总体积仅 19 KB (gzipped)**。它运行在严格的 `default-src 'self'` CSP 之下且不包含 `unsafe-inline`，这就是为什么样式是通过 CSSOM 而不是 style 属性应用的原因。 已在真实浏览器中验证：0 个控制台错误，两种主题均符合 WCAG 2.2 AA 对比度标准（测量了 20 个样本），对话框中的焦点陷阱支持完整的键盘操作，并且在 320/480/768/1024/1440/1920 以及 640×320 横屏下均无水平溢出。 ### Endpoints | 方法 | 路径 | 用途 | |---|---|---| | `GET` | `/api/v1/meters` | 带有过滤、排序、分页的列表 | | `GET` | `/api/v1/meters/{id}` | 包含层级路径的完整电表记录 | | `GET` | `/api/v1/meters/{id}/consumption` | 推导出的消耗量，`raw`/`hourly`/`daily` | | `GET` | `/api/v1/meters/near` | 邻近搜索 | | `GET` | `/api/v1/hierarchy` | 重建的网络树 | | `GET` | `/api/v1/hierarchy/nodes/{path}` | 按路径 id 获取的子树 | | `GET` | `/api/v1/hierarchy/nodes/{path}/meters` | 某节点下的电表 | | `GET` | `/api/v1/transformers` | 带有实时电表计数的 DT 注册表 | | `GET` | `/api/v1/transformers/{code}` · `/meters` | 单个 DT，及其电表 | | `GET` | `/api/v1/data-quality` | 上游不一致项，带有计数 | | `GET` | `/api/v1/stats` | 资产摘要 | | `GET` | `/api/v1/health/live` · `/health/ready` | 存活 · 就绪状态 | | `GET` | `/api/v1/system/snapshot` · `/system/session` | 快照和会话来源 | | `POST` | `/api/v1/system/snapshot/refresh` | 强制重新构建（有频率限制：在最小间隔内返回 `429` + `Retry-After`） | 每个错误都使用同一种格式，因此客户端可以根据 `code` 进行分支判断，而不是去解析散文式的文本： ``` { "error": { "code": "not_found", "message": "No meter with id 'NOPE'.", "details": { "hint": "Meter ids are case-sensitive upstream." }, "request_id": "7a6774a0558d4b57ac3510e1130d739a" } } ``` 门户故障被映射为 `502`/`503`/`504` — 绝不会是单纯的 `500`，并且上游的 HTML 或堆栈跟踪永远不会到达客户端。调用者始终可以区分“上游宕机”还是“此服务损坏”。 ## 测试 ``` make test # 286 tests, no network required make test-live # 18 contract tests against the real portal (needs credentials) ``` `make test` — 286 个测试，**91% 覆盖率**，无需网络。Payload fixtures 是**从实时门户录制的**，而不是手写的：编造的 fixtures 只能证明代码与我的假设一致，而录制的 fixtures 能证明它与必须与之通信的系统保持一致。 `make test-live` — 针对正在运行的门户，重新验证 `PROTOCOL.md` 中的每一项声明：签名方案及其 ±300 秒的窗口期、累积寄存器、两种仍然存在的电表构建版本、`LIKE` 通配符的怪癖、非唯一的层级代码。**可能会导致构建失败的文档就不会在不知不觉中失效。** 这些被排除在 CI 之外，因为第三方的可用性绝不应决定我们的构建是否通过。 测试以它们所保护的属性来命名，而不是它们调用的函数 — 例如 `test_merging_by_code_would_have_been_wrong`，`test_consumption_is_the_delta_not_the_reading`。 `tests/test_end_to_end.py` 弥补了单元测试在结构上无法覆盖的唯一缝隙：它**仅** mock 了门户的 HTTP 响应，并通过 `TestClient` 驱动真实的应用程序，因此依赖注入、生命周期排序、会话恢复、跨层错误映射以及快照经济学（多个请求 → 一次导出；刷新突发 → 一次导出）都进行了端到端的测试。 通过这种方式发现了两个真实的 bug，值得一提：`{node_id:path}` 路由转换器会默默地吞掉 `/meters`（见 `app/api/v1/hierarchy.py`），以及在上游故障期间，快照存储会将完整快照降级为降级快照。 ## 假设 1. **时间戳是 naive 的本地时间。** 门户发送的 `DD/MM/YYYY HH:MM` 没有时区。该公用事业单位位于斋浦尔，因此很可能是 IST — 但没有任何地方明确说明，所以我保留了值的 naive 状态，而不是在上面加盖 UTC 时间戳，从而默默地将每个读数偏移 5 个半小时。这被标记为一个悬而未决的问题，而不是私下里做出的决定。 2. **日份优先的日期解析。** 不是假设 — 而是已被证明：系列运行至 `30/06/2026`，而不存在第 30 个月。 3. **DT 注册表是 DT 名称的权威来源。** 如果电表的 DT 名称与 `/portal/dts` 不一致（`DT-007` 在三个电表上具有过期的别名），则以注册表为准，并且会报告冲突。 4. **寄存器是单调的；减少意味着溢出或更换电表。** 在此数据集中未观察到，但物理电表确实会发生回绕。负的 delta 会产生 `null`，而不是会产生负消耗量从而破坏任何下游总和的值。 5. **约 5 分钟的延迟窗口是可以接受的。** 电表属性和层级的变化周期约为几周。每个响应都带有 `X-Snapshot-Age-Seconds`，因此调用者可以应用更严格的策略；消耗量始终是实时获取的。 6. **层级代码的作用域仅限于其父节点**，而不是全局唯一的。数据无法解决此问题，因此我选择了无损读取 — 见 ADR-0003。 7. **门户是只读的**，并且可以从运行此服务的任何地方访问。 ## 设计决策与权衡 | 决策 | 原因 | 接受的权衡 | |---|---|---| | **对资产进行快照，而不是代理** | 一次签名导出即可返回所有 403 个带有层级和地理信息的电表。代理会继承所有限制并增加一跳网络请求。 | 高达 `TTL` 秒的延迟；内存随资产规模增长。 | | **内存存储，不使用数据库** | 403 个电表 ≈ 400 KB。Postgres+PostGIS 会增加容器、迁移和运维负担，而这个服务的整个数据集只占用了微不足道的 RAM。 | 重启时冷启动；无跨实例共享。见 *扩展性*。 | | **主动*且*被动的重新认证** | `expiresAt` 允许我们在过期前刷新；被动重试涵盖了时钟漂移和服务器端的撤销。单独使用任何一个都是一种赌博。 | 比简单的 401 重试代码稍微多一点。 | | **单次会话刷新** | 登录受频率限制（第 4 次尝试 → `429`）。N 个并发过期必须只产生一次登录。 | 在认证路径上加锁。 | | **在本地搜索，而不是上游** | 门户的 `q` 是未转义的 SQL `LIKE` — `_` 匹配所有 403 个电表。 | 我们无法使用任何服务器端索引；但在这个规模下无关紧要。 | | **基于路径的层级身份标识** | 代码不是全局唯一的；以代码为键会合并不相关的分支。 | 节点 ID 更长；调用者必须将它们视为不透明的。 | | **暴露数据质量** | 隐藏上游的混乱是导致下游团队最终调试*我们*服务的原因。 | 承认数据是不完美的 — 这是正确的做法。 | | **仅修复明确的空白** | 用一个候选值填补空白是推断；用两个则是捏造。 | 一些空白仍然保留 — 被报告，而不是被隐藏。 | | **在刷新失败时提供旧数据** | 带有真实年份的旧数据胜过 503。 | 如果新鲜度至关重要，调用者必须读取 `X-Snapshot-Age-Seconds`。 | | **从代码生成 `openapi.json`** | 手写的规范会产生偏差。如果提交的文件已过期，CI 将失败。 | 规范的格式受限于 FastAPI 输出的内容。 | ## 我有意省略的内容 * **我们自身 API 上的身份验证。** 在这个练习中它是一个内部服务，而构建了一半的认证模型比明确缺失的模型更糟糕。`POST /system/snapshot/refresh` 是未经身份验证的，在真正部署之前，它应该位于身份验证和基于客户端的频率限制之后。因为每次重新构建都需要消耗一次门户导出，所以它已经将强制刷新合并为每个 `SNAPSHOT_REFRESH_MIN_INTERVAL_SECONDS`一次（在该窗口内返回 `429` + `Retry-After`），因此它不能被用来放大对老旧门户的负载 — 但这是一种爆炸半径防护，不能替代身份验证。 * **持久化数据存储。** 在 403 条记录下是不合理的；关于阈值见 *扩展性*。 * **Prometheus 指标。** 带有请求 ID 和时间的结构化日志为这种规模的服务承载了相同的信息；配置一个导出器只是一种繁文缛节。 * **前端框架和构建步骤。** 位于 `/app` 的控制台是原生 ES modules，体积约 19 KB (gzipped)，零运行时依赖。React 加上一个打包工具会增加一个工具链、一个 lockfile，并且为五个视图和一个侧边栏增加约 45 KB 的体积。 * **切片底图。** 地图视图以 SVG 散点图的形式绘制真实的相对地理位置。真实的底图需要外部切片提供商，而此服务不应依赖它，并且控制台的 `default-src 'self'` CSP 也禁止这样做。 * **抓取每个电表的详细信息来丰富快照。** 导出已包含一个超集。SSR 路径已实现并作为*备选方案*进行了测试，而不是主要路由。 * **在 `POST` 时重试。** 没有上游写入操作，因此不存在重试安全性问题。 ## 下一步我会做什么 1. **与门户的所有者一起解决时区问题**，然后使时间戳具备时区感知能力。目前这是整个服务中最大的正确性风险。 2. **增量刷新。** 为了观察到少量变化而重建所有 403 条记录现在还可以，但在规模扩大时会很浪费；导出不提供增量机制，因此这需要一种变更检测策略。 3. **共享缓存 (Redis) + 持久化快照**，使得重启时是热启动并且各实例能保持数据一致。 4. **对消耗量序列进行异常检测** — `Installed` 状态电表出现的零消耗运行、电压偏移、寄存器数据停滞。数据支持这一点，这也是运维团队接下来真正会要求的功能。 5. **层级浏览器 UI**，一旦 API 表面稳定下来就会实现。 6. **CI 中的契约测试**，针对录制的 cassette 进行，这样就可以按照计划捕获上游偏差，而不是靠人工运行 `make test-live`。 ## 扩展性 当前设计的实际限制，因为任务书询问了它在哪里会遇到瓶颈： | 资产规模 | 行为表现 | |---|---| | **~400 (当前)** | 快照 ≈ 400 KB，构建时间约 600 毫秒，查询时间远低于 1 毫秒。游刃有余。 | | **~50,000** | 快照 ≈ 50 MB；仍然没问题。线性过滤扫描达到 ~10 毫秒 — 有点明显，但不痛苦。邻近搜索需要空间索引 (R-tree / geohash)。 | | **~500,000** | 完全在内存中重建变得不可行：高达数百 MB 的工作集，缓慢的冷启动，并且每个实例都持有一个重复的副本。迁移到 Postgres + PostGIS 并进行增量同步，仅在内存中保留热数据，并对层级进行分页，而不是提供整棵树。 | 首先崩溃的不是内存，而是**完整重建**：它在计时器上是 O(estate) 的，所以刷新成本会增加，而变化率却不变。增量同步是解决方案，但它受限于门户提供增量或时间戳变更的能力 — 而门户目前并不提供此功能。 ## 配置 所有设置都是环境变量；见 [`.env.example`](.env.example)。凭证 没有默认值，因此缺少凭证会引发明显的失败，而不是产生一个对每个请求都返回 401 的服务。 | 变量 | 默认值 | 用途 | |---|---|---| | `PORTAL_USERNAME` / `PORTAL_PASSWORD` | *(必填)* | 门户凭证 | | `PORTAL_BASE_URL` | `https://urja-ops.flockenergy.tech` | 上游基础 URL | | `SNAPSHOT_TTL_SECONDS` | `300` | 快照新鲜度窗口 | | `SNAPSHOT_REFRESH_MIN_INTERVAL_SECONDS` | `30` | 强制刷新之间的最小间隔（`0` 表示禁用） | | `CONSUMPTION_CACHE_TTL_SECONDS` | `120` | 单个电表的序列缓存 | | `SESSION_REFRESH_MARGIN_SECONDS` | `120` | 在过期前这么长时间进行刷新 | | `PORTAL_MAX_CONCURRENCY` | `6` | 并发上游请求的上限 | | `PORTAL_MAX_RETRIES` | `3` | 针对超时和 5xx 的重试次数 | | `LOG_FORMAT` | `json` | 生产环境中使用 `json`，本地使用 `console` | `.env.example` 记录了全部 24 个设置。`requirements.txt` 锁定了直接依赖项； `requirements.lock` 额外锁定了每个具有哈希值的传递依赖项，以实现可重现的安装。 **密钥：** `.env` 被添加到了 git-ignore，凭据作为 pydantic 的 `SecretStr` 保存，并且登录失败永远不会回显请求体。门户的 HMAC 签名密钥在运行时获取，并且永远不会写入磁盘或日志。 ## 开发 ``` make dev # install dev dependencies make test # tests + coverage make lint # ruff check + format check make fmt # auto-format make openapi # regenerate openapi.json make docker # build the image ``` CI (GitHub Actions) 会运行 lint，在 Python 3.11 和 3.12 上以 80% 的覆盖率底线进行测试，验证 `openapi.json` 是否已过期，并构建 Docker 镜像。

标签：OpenAPI, Python, REST API, Web 控制台, 无后门, 智能电表, 请求拦截, 逆向工具