# dabot_api

**Repository Path**: shawmon/dabot_api

## Basic Information

- **Project Name**: dabot_api
- **Description**: No description available
- **Primary Language**: Unknown
- **License**: GPL-3.0
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-08-07
- **Last Updated**: 2026-08-07

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# 数博数据采集平台（DaBOT API）

以**表名作为参数**的统一数据接入网关：外部系统通过 `app_key + app_secret` 认证后，向任意已注册的逻辑表提交**单条或批量**数据；平台对每条数据执行**字段级校验**（动态 Pydantic 模型）与**业务级校验**（插件式规则引擎），全部通过的数据写入对应物理表，未通过的按记录索引返回精确错误信息。

## 1. 技术栈

| 层次 | 选型 |
|------|------|
| Web 框架 | FastAPI |
| 数据校验 | Pydantic v2（运行时动态建模）|
| 数据库 | PostgreSQL（SQLAlchemy 2.0 ORM + Core）|
| 迁移 | Alembic |
| 缓存 / 限流 | Redis |
| 异步队列 | Celery（Redis 作 broker/backend）|
| 表达式求值 | simpleeval（安全沙箱）|
| 密钥存储 | PBKDF2-HMAC-SHA256（全局盐 + 每用户独立盐）|

## 2. 系统架构

```mermaid
flowchart LR
    Client[外部系统] -->|app_key + app_secret| GW[FastAPI 网关]

    subgraph MW[中间件链]
        RL[限流<br/>Redis 固定窗口] --> IPW[IP 白名单] --> LOG[访问日志]
    end
    GW --> MW

    MW --> AUTH[用户认证<br/>sys_user 校验]
    AUTH --> VS[ValidationService]

    subgraph VS[校验引擎]
        FM[动态 Pydantic 模型<br/>字段级校验] -->|字段级通过| RE[规则引擎<br/>业务级校验]
    end

    META[(meta_table /<br/>meta_field_def)] -.->|Redis 缓存| FM
    RULE[(rule_validation)] -.->|Redis 缓存| RE
    CODE[(meta_code)] -.->|Redis 缓存| FM

    VS -->|通过| DBW[动态 Insert<br/>写入目标物理表]
    VS -->|失败| ERR[按索引返回错误<br/>field_errors / business_errors 分离]
    VS --> TRACE[(validation_trace<br/>逐条校验轨迹)]

    GW -->|批量超阈值 / 强制异步| CELERY[Celery Worker]
    CELERY --> VS
    CELERY --> TASK[(import_task<br/>任务状态)]
```

### 目录结构

```
data-validation-platform/
├── app/
│   ├── main.py                  # 应用入口：中间件、异常处理、路由
│   ├── config.py                # 配置（环境变量驱动）
│   ├── database.py              # SQLAlchemy 引擎/会话/Base
│   ├── redis_client.py          # Redis 客户端与缓存读写（故障自动降级）
│   ├── celery_app.py            # Celery 应用
│   ├── core/
│   │   ├── errors.py            # 6 位分段错误码体系 + AppException
│   │   ├── security.py          # 密钥加盐哈希/校验
│   │   └── model_factory.py     # 动态 Pydantic 模型工厂（带编译缓存）
│   ├── models/                  # sys_user / meta_table / meta_field_def /
│   │                            # meta_code / rule_validation / import_task / validation_trace
│   ├── schemas/common.py        # 统一响应模型（字段/业务错误分离）
│   ├── validators/
│   │   ├── base.py              # BaseValidator + ValidationContext
│   │   ├── registry.py          # @register 注册器
│   │   └── builtin.py           # 8 个内置业务校验器
│   ├── services/
│   │   ├── meta_service.py      # 元数据/规则/代码加载（Redis 缓存 + 热更新）
│   │   └── validation_service.py# 校验主流程 + 动态入库 + 轨迹落库
│   ├── api/
│   │   ├── deps.py              # app_key/app_secret 认证依赖
│   │   └── v1/
│   │       ├── submit.py        # 单条/批量提交、任务查询
│   │       ├── rules.py         # 规则配置 CRUD + 缓存刷新 + 校验器清单
│   │       └── meta.py          # 表定义查询、代码表 CRUD
│   ├── middleware/              # 限流 / IP 白名单 / 访问日志
│   └── tasks/import_tasks.py    # Celery 批量导入任务
├── alembic/                     # 迁移环境 + 0001 初始版本
├── scripts/seed.py              # 演示数据（幂等）
├── docker-compose.yml           # postgres + redis + api + worker 一键启动
├── Dockerfile / alembic.ini / requirements.txt / .env.example
```

## 3. 数据流（一次批量提交）

1. 请求经过中间件链：限流 → IP 白名单 → 访问日志；
2. `deps.get_current_user` 校验用户：**存在、有效（status=1）、密钥哈希匹配**，任一失败返回 700001/700002/700003；
3. `meta_service` 从 Redis（未命中则查库并回填）加载**表定义**，不存在返回 200002、停用返回 200003；
4. `model_factory` 按字段定义**动态构建 Pydantic 模型**（按定义快照 LRU 缓存），对每条记录做字段级校验；
5. **字段级失败的记录跳过业务级校验**；通过的记录进入规则引擎，按 `priority` 顺序执行该用户（含全局 `*`）的全部启用规则；
6. 每条记录的结果独立：字段错误进 `field_errors`，业务错误进 `business_errors`，两者完全分离；
7. 全部通过的记录由 SQLAlchemy Core **动态构建的 Table 对象**批量 insert 到目标物理表（列名来自元数据，杜绝注入）；
8. 每条记录的校验轨迹（耗时、错误明细）写入 `validation_trace`，独立事务，失败不影响主流程；
9. 批量条数超过 `BATCH_SYNC_THRESHOLD` 或 `force_async=true` 时，转入 Celery 任务，返回 `task_id`，通过 `GET /api/v1/tasks/{task_id}` 查询进度与逐条结果。

## 4. 统一错误码（6 位分段）

> 分段规则：第 1 位 = 大类，2-3 位 = 子模块，4-6 位 = 序号。

| 错误码 | 含义 | 错误码 | 含义 |
|--------|------|--------|------|
| **1xxxxx 系统** | | **4xxxxx 业务** | |
| 100001 | 系统内部错误 | 400001 | 条件必填校验失败 |
| 100002 | 缓存服务不可用 | 400002 | 数值范围校验失败 |
| 100003 | 任务队列不可用 | 400003 | 正则匹配校验失败 |
| **2xxxxx 请求** | | 400004 | 跨字段比较校验失败 |
| 200001 | 请求参数错误 | 400005 | 自定义表达式校验失败 |
| 200002 | 表不存在 | 400006 | 允许值列表校验失败 |
| 200003 | 表已停用 | 400007 | 日期范围校验失败 |
| 200004 | 请求体不能为空 | 400008 | 唯一性校验失败 |
| **3xxxxx 字段** | | **5xxxxx 批量** | |
| 300001 | 必填字段缺失 | 500001 | 批量大小超过上限 |
| 300002 | 字段类型错误 | 500002 | 批量数据为空 |
| 300003 | 字段长度超限 | 500003 | 任务不存在 |
| 300004 | 代码值非法 | 500004 | 任务状态异常 |
| 300005 | 字段格式错误 | **6xxxxx 数据库** | |
| **7xxxxx 权限** | | 600001 | 数据库连接失败 |
| 700001 | 用户不存在 | 600002 | 数据唯一约束冲突 |
| 700002 | 用户已禁用 | 600003 | 数据写入失败 |
| 700003 | 密钥校验失败 | | |
| 700004 | IP 不在白名单 | | |
| 700005 | 请求过于频繁，已限流 | | |

## 5. 表设计

| 表 | 用途 | 关键字段 |
|----|------|----------|
| `sys_user` | 接入方身份 | `app_key`、`app_secret_hash`（加盐哈希）、`salt`、`status` |
| `sys_meta_table` | 逻辑表定义 | `table_name`（接口参数）→ `target_table`（物理表）、`status` |
| `sys_meta_field` | 字段定义 | `data_type`、`max_length`、`nullable`、`code_category`、`regex_pattern` |
| `sys_meta_code` | 代码表 | `category`、`code_value`、`code_label` |
| `sys_validation_rule` | 校验规则 | `user_code`（`*`=全局）、`table_name`、`field_name`、`rule_type`、`rule_params`(JSONB)、`priority` |
| `sys_task` | 异步任务 | `task_id`、`status`、`result`(JSONB 逐条结果) |
| `sys_validation_trace` | 校验轨迹 | `request_id`、`record_index`、`field_errors`、`business_errors`、`duration_ms` |

支持的字段类型：`string` / `integer` / `decimal` / `boolean` / `date` / `datetime`。

## 6. 两级校验

### 字段级（动态 Pydantic 模型，自动完成）

| 元数据配置 | 校验行为 | 错误码 |
|-----------|---------|--------|
| `nullable=false` | 必填 | 300001 |
| `data_type` | 类型转换与校验 | 300002 |
| `max_length` | 字符串长度 | 300003 |
| `code_category` | 枚举值域（查代码表） | 300004 |
| `regex_pattern` | 正则格式 | 300005 |

### 业务级（8 个内置插件校验器）

| rule_type | 说明 | 关键参数 | 错误码 |
|-----------|------|----------|--------|
| `conditional_required` | 条件必填 | `when_field`, `when_value(s)` | 400001 |
| `number_range` | 数值范围 | `min`, `max` | 400002 |
| `regex_match` | 正则匹配 | `pattern` | 400003 |
| `cross_field_compare` | 跨字段比较 | `other_field`, `operator`(eq/ne/gt/gte/lt/lte) | 400004 |
| `expression` | 自定义表达式 | `expression`（变量=字段名与 `value`） | 400005 |
| `allowed_values` | 允许值列表 | `values` | 400006 |
| `date_range` | 日期范围 | `min_date`, `max_date`（支持 `today`） | 400007 |
| `unique` | 唯一性（查目标表） | `scope_fields`（联合唯一，可空） | 400008 |

### 扩展新校验器（两步）

```python
# app/validators/my_validator.py
from app.validators import BaseValidator, register

@register
class MyValidator(BaseValidator):
    rule_type = "my_rule"
    description = "我的自定义校验"
    error_code = "400100"
    params_schema = {"threshold": "int"}

    def validate(self, value, record, params, ctx):
        if bad(value):
            return "错误提示"
        return None
```

在 `app/validators/__init__.py` 中 import 该模块即完成注册，随后即可在规则配置接口中按 `rule_type="my_rule"` 配置使用。

## 7. API 一览

认证方式：所有业务接口需携带请求头 `X-App-Key` / `X-App-Secret`。

| 方法 | 路径 | 说明 |
|------|------|------|
| POST | `/api/v1/submit/{table_name}` | 单条提交 `{"data": {...}}` |
| POST | `/api/v1/submit/{table_name}/batch` | 批量提交 `{"data": [...], "force_async": false}` |
| GET | `/api/v1/tasks/{task_id}` | 异步任务状态与逐条结果 |
| GET | `/api/v1/rules/validators` | 可用校验器清单（含参数说明） |
| GET/POST | `/api/v1/rules` | 规则查询 / 新建 |
| PUT/DELETE | `/api/v1/rules/{id}` | 规则更新 / 删除（自动热更新缓存） |
| POST | `/api/v1/rules/cache/refresh?table_name=xx` | 手动刷新缓存 |
| GET | `/api/v1/meta/tables` | 已注册表及字段定义 |
| GET/POST/DELETE | `/api/v1/meta/codes...` | 代码表查询 / 新增 / 删除 |

### 调用示例

* linux

  ```shell
  # 单条
  curl -X POST http://localhost:8000/api/v1/submit/customer_info \
    -H "X-App-Key: demo-key" -H "X-App-Secret: demo-secret" \
    -H "Content-Type: application/json" \
    -d '{"data": {"name": "张三", "gender": "M", "age": 30, "email": "zs@example.com", "birth_date": "1995-01-01", "amount": 100.5}}'
  ```

  ```shell
  # 批量（逐条独立校验，按索引返回错误）
  curl -X POST http://localhost:8000/api/v1/submit/customer_info/batch \
    -H "X-App-Key: demo-key" -H "X-App-Secret: demo-secret" \
    -H "Content-Type: application/json" \
    -d '{"data": [
          {"name": "张三", "gender": "M", "age": 30, "email": "a@x.com", "amount": 100},
          {"name": "", "gender": "X", "age": 200}
        ]}'
  ```

* windows

  ```
  
  ```

批量响应示例（第 2 条同时命中字段级与业务级错误，完全分离）：

```json
{
  "code": "000000", "message": "success",
  "data": {
    "table_name": "customer_info", "total": 2, "success_count": 1, "failed_count": 1, "inserted_count": 1,
    "results": [
      {"index": 0, "success": true, "field_errors": [], "business_errors": []},
      {"index": 1, "success": false,
       "field_errors": [{"field": "name", "code": "300001", "message": "必填字段缺失"},
                         {"field": "gender", "code": "300004", "message": "代码值非法，允许值: ['F', 'M', 'U']"}],
       "business_errors": []}
    ]
  }
}
```

## 8. 快速开始

```bash
cp .env.example .env
docker compose up --build
# API: http://localhost:8000/docs
# 演示账号：X-App-Key: demo-key / X-App-Secret: demo-secret
```

本地开发：

```bash
pip install -r requirements.txt
alembic upgrade head
# python scripts/seed.py
python -m scripts.seed
uvicorn app.main:app --reload
celery -A app.celery_app.celery_app worker --loglevel=info
```

数据库版本管理：

```bash
alembic revision --autogenerate -m "add xxx"   # 基于模型自动生成迁移
alembic upgrade head                            # 升级
alembic downgrade -1                            # 回滚一个版本
```

## 9. 关键设计说明

- **字段/业务错误分离**：`RecordResult` 中 `field_errors` 与 `business_errors` 为两个独立数组；字段级失败的记录不再执行业务规则，避免基于脏数据的误判。
- **缓存热更新**：表元数据、规则、代码字典分别按 `meta:table:*`、`rule:{user}:{table}`、`code:{category}` 缓存；规则/代码的 CRUD 接口写库后主动失效对应 key，等效热更新；Redis 故障时自动降级直查数据库。
- **入库安全**：目标表名与列名仅来源于管理员维护的元数据，通过 SQLAlchemy Core `Table` 对象参数化写入，无 SQL 拼接注入面。
- **批量策略**：`BATCH_SYNC_THRESHOLD`（默认 200）以内同步返回完整结果；超过则转 Celery，结果（含逐条明细，最多保留 1000 条）落 `import_task.result`。
- **表达式安全**：自定义表达式经 simpleeval 沙箱求值，仅可访问当前记录字段与 `value`，无内建函数与 IO 能力。
- **轨迹审计**：每条记录的校验结果、耗时随 `request_id` 落 `validation_trace`，同时输出结构化访问日志。