# AIS 多源采集、冲突记录与聚合接口计划

**状态**：规划中  
**创建日期**：2026-04-30  
**核心原则**：采集器只写原始观测；去重、合并、冲突解释放在聚合接口中完成

## 已确认决策

| 项目 | 决策 |
|-----|------|
| AISStream 接入方式 | 单独实现 WebSocket 采集器，不塞进现有 BarentsWatch HTTP collector |
| 采集器职责 | 连接上游、标准化字段、写入原始观测，不直接决定最终展示值 |
| 去重合并位置 | 放在聚合服务和聚合 API 中，而不是散落在每个 collector 的保存逻辑里 |
| 冲突处理 | 先记录冲突事实和当前选择原因，后续再开放用户规则配置 |
| 默认可信度 | 同类 AIS 数据源优先按 `delivery_mode` 评估：`realtime_stream` 优于 `batch_stream`，再优于 `polling` 和 `snapshot` |
| 过期保护 | 实时流源断流超过 freshness 窗口后，不能仅凭“实时源”身份压过更新的轮询数据 |

## 背景

当前 AIS 链路以 BarentsWatch 为主。它是 HTTP polling 模式，覆盖挪威附近海域，适合作为稳定的免费起点，但不适合承担全球实时船只数据的全部职责。后续接入 AISStream 后，会出现同一个 MMSI 被多个来源同时上报的情况：

- 位置、航速、航向可能在多个来源之间存在秒级差异。
- 船名、IMO、呼号、船型、尺寸等静态字段可能不完整，甚至互相冲突。
- WebSocket 或其他实时流通常更接近实时，但也可能断流或批量延迟。
- 如果每个 collector 自己做去重合并，规则会分散、不可审计，也很难让用户后续配置“某个字段信任哪个来源”。

因此 v1 不应让采集器直接覆盖最终船只表。更稳的方式是先保留观测事实，再由聚合接口统一给出当前展示视图。

## 目标架构

```mermaid
flowchart LR
    A[BarentsWatch HTTP collector] --> D[AIS raw observations]
    B[AISStream WebSocket collector] --> D
    C[Custom mapped vessel_ais sources] --> D
    D --> E[AIS aggregation service]
    E --> F[Conflict records]
    E --> G[GeoJSON vessels API]
    E --> H[Vessel detail API]
    I[Aggregation strategy config] --> E
```

### 原始观测层

原始观测层保存每个来源看到的事实。建议模型包含：

| 字段 | 用途 |
|-----|------|
| `target_schema` | 例如 `vessel_ais` |
| `source` | 例如 `barentswatch_vessels`、`aisstream_vessels` |
| `entity_key` | AIS 使用 MMSI |
| `delivery_mode` | `realtime_stream`、`batch_stream`、`polling`、`snapshot` |
| `transport` | `websocket`、`sse`、`http`、`file` 等 |
| `observed_at` | 上游数据时间，优先使用 AIS 消息时间 |
| `collected_at` | 本系统接收或采集时间 |
| `normalized_payload` | 标准化后的 AIS JSON |
| `raw_payload` | 可选，保存原始或裁剪后的上游记录 |

`delivery_mode` 和 `transport` 不应混为一谈。WebSocket 是传输方式；streaming 是交付模式。聚合可信度主要看 `delivery_mode`，`transport` 只作为辅助信息。

### 冲突记录层

聚合服务发现同一个实体、同一个字段存在多个非空不同值时，写入冲突记录。冲突记录不代表错误，只代表“有多个可用候选值”。

```json
{
  "target_schema": "vessel_ais",
  "entity_key": "257123000",
  "field": "name",
  "candidates": {
    "barentswatch_vessels": "OSLO TRADER",
    "aisstream_vessels": "OSLO TRADER II"
  },
  "selected_source": "aisstream_vessels",
  "selected_value": "OSLO TRADER II",
  "selected_reason": "delivery_mode_priority",
  "resolved_by": "system",
  "status": "open"
}
```

第一阶段只需要记录冲突和当前选择原因，不需要做人工逐条确认。后续 UI 的目标也不是让用户处理每条冲突，而是把冲突沉淀成字段级规则。

## 聚合规则

### 字段分类

| 类型 | 字段 | 默认策略 |
|-----|------|----------|
| 动态位置 | `lat`、`lon`、`sog`、`cog`、`heading`、`nav_status` | 优先最新 `observed_at`，同时间再按来源优先级 |
| 静态身份 | `name`、`callsign`、`imo`、`flag` | 非空优先，再按字段策略或来源优先级 |
| 静态规格 | `vessel_type`、`vessel_type_name`、`length`、`width`、`draught` | 非空优先；冲突时记录候选值 |
| 元信息 | `field_sources`、`conflict_count`、`selected_reasons` | 聚合接口生成，便于调试和后续 UI 展示 |

### 默认优先级

默认优先级应使用两个维度：

```yaml
delivery_mode_priority:
  - realtime_stream
  - batch_stream
  - polling
  - snapshot

transport_priority:
  - websocket
  - sse
  - http
  - file
```

`delivery_mode_priority` 是主判断。比如 AISStream 如果提供实时推送，应标记为 `realtime_stream + websocket`；BarentsWatch 当前是 `polling + http`。

### 断流保护

实时流不能永久凭身份占优。聚合时需要 freshness 窗口：

```yaml
freshness:
  realtime_stream_seconds: 900
  polling_seconds: 3600
```

如果 `aisstream_vessels` 最近 15 分钟没有该 MMSI 的新观测，而 BarentsWatch 轮询源有更新位置，则位置类字段应采用 BarentsWatch 的更新观测，并记录选择原因 `newest_observation` 或 `freshness_fallback`。

## 聚合接口

现有展示接口应逐步改为消费聚合服务，而不是自己直接拼 `VesselPosition + VesselStatic`。

```text
GET /api/v1/visualization/geo/vessels
GET /api/v1/visualization/vessels/{mmsi}
GET /api/v1/visualization/vessels/{mmsi}/track
GET /api/v1/visualization/vessels/{mmsi}/conflicts
```

GeoJSON properties 建议增加：

```json
{
  "mmsi": 257123000,
  "name": "OSLO TRADER",
  "lat": 59.91,
  "lon": 10.73,
  "received_at": "2026-04-30T10:00:00Z",
  "field_sources": {
    "name": "aisstream_vessels",
    "lat": "aisstream_vessels",
    "lon": "aisstream_vessels",
    "vessel_type": "barentswatch_vessels"
  },
  "selected_reasons": {
    "name": "delivery_mode_priority",
    "lat": "newest_observation",
    "vessel_type": "non_empty_priority"
  },
  "conflict_count": 2
}
```

## 开放配置计划

### Phase 1 — 内置默认策略和只读解释

- 实现后端默认策略。
- 聚合接口返回 `field_sources`、`selected_reasons`、`conflict_count`。
- 冲突记录可查询，但不允许用户修改。
- 保持现有前端船只图层接口形状基本兼容，新增字段只作为调试和后续 UI 输入。

### Phase 2 — 系统设置中的 JSON/YAML 策略配置

新增系统设置项，例如：

```yaml
collector_aggregation:
  vessel_ais:
    source_priority:
      - aisstream_vessels
      - barentswatch_vessels
    field_rules:
      name:
        mode: source_priority
      vessel_type:
        mode: source_priority
        source_priority:
          - barentswatch_vessels
          - aisstream_vessels
      lat:
        mode: newest
      lon:
        mode: newest
```

配置校验要求：

- 未知 source 只警告，不阻断保存，便于先配置后启用。
- 未知 field 必须拒绝，避免拼写错误悄悄失效。
- 动态位置字段默认不允许被固定来源永久锁死，除非显式开启高级选项。
- 空值不覆盖非空值是全局保护，不建议开放关闭。

### Phase 3 — 冲突治理 UI

基于冲突记录提供页面或 drawer：

- 查看某个 MMSI 的冲突字段。
- 查看每个字段的候选来源和值。
- 查看当前选择原因。
- 将一次人工选择保存成字段规则，而不是只处理单条冲突。
- 支持恢复默认策略。

## AISStream 采集器计划

AISStream 采集器单独实现，建议命名为 `aisstream_vessels`。它的职责是：

- 维护 WebSocket 连接、订阅范围和重连。
- 将上游 AIS 消息标准化为 `vessel_ais` payload。
- 标记 `delivery_mode = realtime_stream`，`transport = websocket`。
- 写入原始观测层。
- 不直接 upsert 最终展示数据。

配置应放入采集器设置，而不是硬编码：

```yaml
aisstream_vessels:
  api_key: "${AISSTREAM_API_KEY}"
  bounding_boxes:
    - [[-180, -90], [180, 90]]
  message_types:
    - PositionReport
    - ShipStaticData
```

## 实施顺序

1. 新增原始观测模型和冲突记录模型。
2. 实现 AIS 聚合服务，先从现有 `vessel_position` / `vessel_static` 兼容读取，再逐步切换到原始观测层。
3. 将 `/geo/vessels` 和 `/vessels/{mmsi}` 改为走聚合服务。
4. 改造 BarentsWatch 保存逻辑，让它写入原始观测，同时保留现有表作为兼容缓存。
5. 实现 AISStream WebSocket collector。
6. 接入系统设置中的聚合策略配置。
7. 做冲突治理 UI。

## 测试计划

- 同一来源同一 `mmsi + observed_at + lat + lon` 重复记录只聚合一次。
- 多来源同一 MMSI 的位置字段优先选择最新观测。
- 实时流和轮询源同时间冲突时，实时流优先。
- 实时流过期后，更新的轮询源可以接管动态字段。
- 静态字段不会被空值覆盖。
- 静态字段冲突会写入冲突记录。
- 字段级配置可以覆盖默认来源优先级。
- 聚合接口在没有冲突表时仍可返回兼容 GeoJSON。

## 相关文件

- [实时船只监控系统计划](/home/ray/dev/linkong/planet/docs/plans/earth-vessel-tracking-plan.md)
- [自定义 API 数据源与 LLM 映射系统计划](/home/ray/dev/linkong/planet/docs/plans/datasource-custom-api-mapping-plan.md)
- [BarentsWatch AIS collector](/home/ray/dev/linkong/planet/backend/app/services/collectors/vessel_ais.py)
- [船只模型](/home/ray/dev/linkong/planet/backend/app/models/vessel.py)
- [可视化 API](/home/ray/dev/linkong/planet/backend/app/api/v1/visualization.py)