API使用示例

用curl、Python 和JavaScript 编写的现成配方可用于查询公共数据、同步数据库并发送产卵器读数和阻止访问。

/zh-mo/api返回参考

第一步

所有调用都来自同一个数据库并返回 JSON。公共端点不需要身份验证;私人的需要您的访问密钥。

基址

/zh-mo/api

语言前缀是 URL 的一部分 (zh-mo), 但它不会改变返回的数据——记录来自银行,与现场注册的完全一样。

快速测试
curl -G -d "municipality=Ponta Pora" --data-urlencode "country=Brasil" \
  /zh-mo/api/lastcountingpublic

如果响应是 JSON 列表,则集成已经在运行。

关键在 URL 中。 在所有专用端点(包括 POST)上 key 从查询字符串中读取 (?key=SUA_CHAVE). 仅在表单返回的正文中发送 404 "Wrong key". 其他字段继续出现在请求正文中。

查阅公开数据

终点 /api/lastcountingpublic 返回最新的鸡蛋计数以及流行病学纬度、经度、周和年。它是地图、面板和研究的起点。

GET/api/lastcountingpublic民众
curl
curl -G \
  --data-urlencode "municipality=Ponta Pora" \
  --data-urlencode "date_start=2025-01-01" \
  --data-urlencode "date_end=2025-12-31" \
  /zh-mo/api/lastcountingpublic
Python
import requests

BASE = "/zh-mo/api"

resposta = requests.get(
    BASE + "/lastcountingpublic",
    params={
        "municipality": "Ponta Pora",
        "date_start": "2025-01-01",
        "date_end": "2025-12-31",
    },
    timeout=60,
)
resposta.raise_for_status()

for contagem in resposta.json():
    print(contagem["date"], contagem["ovitrap_id"], contagem["eggs"])
JavaScript
const BASE = "/zh-mo/api";

const params = new URLSearchParams({
    state: "MS",
    date_start: "2025-01-01",
});

const resposta = await fetch(BASE + "/lastcountingpublic?" + params);
const contagens = await resposta.json();

console.log(contagens.length, "contagens");

相同的位置参数 (country, state, municipality) 它们适用于其他公共端点:产卵器、街区、EDL、EDL 维护和战略点。仅更改端点路径就已经更改了数据集。

如果没有任何位置参数,终端节点将返回整个巴西的数据。按市或州进行过滤可以使响应速度更快。

浏览所有页面

分页端点支持 page 从 1 到 100。在此之上,API 会响应文本 "Maximum pagination is 100" 状态 200——也就是说,主体不再是一个列表。在迭代之前始终检查类型。

Python
import requests

BASE = "/zh-mo/api"


def baixar_tudo(endpoint, **filtros):
    """Percorre as paginas ate a API devolver uma pagina vazia."""
    registros = []

    for pagina in range(1, 101):
        resposta = requests.get(
            BASE + endpoint,
            params=dict(filtros, page=pagina),
            timeout=60,
        )
        resposta.raise_for_status()
        dados = resposta.json()

        # limite de paginacao atingido: a API devolve uma string, nao uma lista
        if not isinstance(dados, list):
            print("Aviso:", dados)
            break

        if not dados:
            break

        registros.extend(dados)

    return registros


ovitrampas = baixar_tudo("/getmunicipalityovitrapspublic", municipality="Ponta Pora")
print(len(ovitrampas), "ovitrampas")
你读到100页了吗? 缩小过滤器范围,而不是尝试第 101 页:逐个县查找,或使用 date_start e date_end 将时间段划分为更小的间隔。

增量同步

为了维护镜像基础,不要在每次运行时再次下载所有内容。终点 /api/lastcountingpublic 公认 id, 它仅返回该标识符的计数,并且 date (纳入日期) e date_collect (data de coleta).

Python
import json
import os
import requests

BASE = "/zh-mo/api"
ESTADO = "ultimo_id.json"


def ultimo_id_lido():
    if os.path.exists(ESTADO):
        with open(ESTADO) as arquivo:
            return json.load(arquivo)["counting_id"]
    return 0


def sincronizar():
    ultimo = ultimo_id_lido()
    novas = []

    for pagina in range(1, 101):
        resposta = requests.get(
            BASE + "/lastcountingpublic",
            params={"municipality": "Ponta Pora", "id": ultimo, "page": pagina},
            timeout=60,
        )
        resposta.raise_for_status()
        dados = resposta.json()

        if not isinstance(dados, list) or not dados:
            break

        novas.extend(dados)

    if novas:
        maior = max(item["counting_id"] for item in novas)
        with open(ESTADO, "w") as arquivo:
            json.dump({"counting_id": maior}, arquivo)

    return novas


print(len(sincronizar()), "contagens novas")

始终保留最大的 counting_id 收到的,而不是执行的日期:今天在字段中输入的计数可能指的是上周,并且按 id 进行过滤不会让这些记录逃脱。

导出为 CSV

公共计数已经带有坐标,因此您可以生成电子表格或提供地图,而无需任何额外的连接。

Python
import csv
import requests

BASE = "/zh-mo/api"

COLUNAS = [
    "counting_id",
    "municipality",
    "state_code",
    "ovitrap_id",
    "latitude",
    "longitude",
    "week",
    "year",
    "eggs",
    "date",
    "date_collect",
]

resposta = requests.get(
    BASE + "/lastcountingpublic",
    params={"state": "MS", "date_start": "2025-01-01"},
    timeout=60,
)
contagens = resposta.json()

with open("contagens.csv", "w", newline="", encoding="utf-8") as arquivo:
    escritor = csv.DictWriter(arquivo, fieldnames=COLUNAS, extrasaction="ignore")
    escritor.writeheader()
    escritor.writerows(contagens)

print("Arquivo contagens.csv gerado com", len(contagens), "linhas")

发送数据

以下示例使用私有 API 并需要密钥。密钥的地理范围定义了它可以记录的内容——市政密钥仅记录该市本身。

POST/api/postcounting私人的

发送现有产卵器的读数

发送已注册的产卵器的卵数。产卵器位于 ovitrap_group_id 重点市辖区内,流行病学周按实地计算 date.

curl
curl -X POST \
  -d "ovitrap_group_id=97" \
  -d "ovitrap_lat=-7.000000" \
  -d "ovitrap_lng=-8.000000" \
  -d "date=2025-01-20" \
  -d "counting_observation_id=1" \
  -d "counting_eggs=5" \
  "/zh-mo/api/postcounting?key=SUA_CHAVE"
Python
import requests

BASE = "/zh-mo/api"
CHAVE = "SUA_CHAVE"

resposta = requests.post(
    BASE + "/postcounting",
    params={"key": CHAVE},          # a chave vai na URL
    data={                          # os dados vao no corpo
        "ovitrap_group_id": 97,
        "ovitrap_lat": -7.000000,
        "ovitrap_lng": -8.000000,
        "date": "2025-01-20",
        "counting_observation_id": 1,
        "counting_eggs": 5,
    },
    timeout=60,
)

print(resposta.status_code, resposta.json())
# 200 "Contagem registrada"
注意力: 如果发送的坐标与注册的坐标不同,则产卵器将更新为新位置。始终发送该点的真实坐标,而不是读取时代理的坐标。
POST/api/postcounting私人的

在读数旁边安装一个新的产卵器

如果 ovitrap_group_id 该市尚不存在产卵器,因此在同一请求中创建了产卵器。同时发送地址和类型(如果适用) (ovitrap_type_id: 1 城市, 模式; 2 rural).

Python
import requests

BASE = "/zh-mo/api"
CHAVE = "SUA_CHAVE"

resposta = requests.post(
    BASE + "/postcounting",
    params={"key": CHAVE},
    data={
        "ovitrap_group_id": 96,
        "ovitrap_address_district": "Centro",
        "ovitrap_address_street": "Rua das Flores",
        "ovitrap_address_number": "123",
        "ovitrap_address_complement": "",
        "ovitrap_address_sector": "Setor 04",
        "ovitrap_responsable": "Maria Souza",
        "ovitrap_block_id": "12",
        "ovitrap_type_id": 2,          # 1 = urbana (padrao), 2 = rural
        "ovitrap_lat": -7.000000,
        "ovitrap_lng": -8.000000,
        "date": "2025-01-20",
        "counting_date_collect": "2025-01-27",
        "counting_observation_id": 1,
        "counting_eggs": 5,
    },
    timeout=60,
)

print(resposta.status_code, resposta.json())
具体答复:
  • 400 — 缺少 ovitrap_lat, ovitrap_lng ou ovitrap_group_id.
  • 400ovitrap_type_id 与 1 或 2 不同。
  • 404 — 已经对这个诱产蛋器进行了年份和星期的计数。
  • 500 — 发送的日期属于未来或无效的流行病学周。
POST/api/postaction私人的

登记一次街区访问

使用 block_id 在已注册的街区中启动访问。如果该块尚不存在,则发送 block_group_id 与块数据一起,它是在主要城市内自动创建的。

Python — 现有街区
import requests

BASE = "/zh-mo/api"
CHAVE = "SUA_CHAVE"

visita = {
    "block_id": 97,
    "date": "2025-01-20",

    "block_land_residence": 10,
    "action_land_residence_out": 2,
    "action_land_residence_breedings": 1,
    "action_land_residence_treated": 1,

    "action_deposit_a1_quantity": 4,
    "action_deposit_a1_eliminated": 2,
    "action_deposit_a1_treated": 1,
    "action_deposit_a1_larvicid": 10,

    "action_observation": "Foco encontrado em pneus nos fundos do imovel.",
}

resposta = requests.post(
    BASE + "/postaction", params={"key": CHAVE}, data=visita, timeout=60
)

if resposta.status_code == 409:
    print("Ja existe visita para esse quarteirao nessa semana")
else:
    print(resposta.status_code, resposta.json())
Python — 创建街区
visita = {
    "block_group_id": 97,
    "date": "2025-01-20",
    "block_address_district": "Centro Historico",
    "block_address_sector": "Setor 04 - Norte",
    "block_coordinates": "[[-7.123, -34.845], [-7.124, -34.846]]",
    "block_lat": -7.123456,
    "block_lng": -34.845678,
    "block_responsable": "Joao da Silva",

    "block_land_residence": 10,
    "action_land_residence_out": 2,
    "action_land_residence_breedings": 1,
    "action_land_residence_treated": 1,

    "action_observation": "Area com alta densidade de recipientes descartaveis.",
}

resposta = requests.post(
    BASE + "/postaction", params={"key": CHAVE}, data=visita, timeout=60
)
print(resposta.status_code, resposta.json())

房地产和存款领域的完整列表 (a1, a2, b, c, d1, d2, e) 是在 端点参考. 未发送的数字字段假定为零,但任何非数字的值都会返回 400。

删除记录

拆除是明确的,并且始终仅限于主要城市。

curl
# leitura de uma ovitrampa (identificada pela ovitrampa + data)
curl -X POST -d "ovitrap_group_id=97" -d "date=2025-01-20" \
  "/zh-mo/api/postdeletecounting?key=SUA_CHAVE"

# a ovitrampa inteira
curl -X POST -d "ovitrap_group_id=97" \
  "/zh-mo/api/postdeleteovitrap?key=SUA_CHAVE"

# visita de um quarteirao (quarteirao + data)
curl -X POST -d "block_group_id=97" -d "date=2025-01-20" \
  "/zh-mo/api/postdeleteaction?key=SUA_CHAVE"

# o quarteirao inteiro
curl -X POST -d "block_group_id=97" \
  "/zh-mo/api/postdeleteblock?key=SUA_CHAVE"
要更正使用错误值发布的计数,请使用以下命令删除读数 /api/postdeletecounting 并再次发送 /api/postcounting — 重新发送会返回重复错误。

错误处理

响应正文始终是带有消息的 JSON 字符串。每种情况都值得处理,因为并非每个错误都值得再次尝试:

情况代码该怎么办
"Wrong key"404检查键是否在查询字符串中而不是在正文中。
重复计算404本周已经有关于产卵器的阅读了;请在重新发送之前删除。
重复访问409本周已有人对该街区进行了访问。
超出范围403该街区不属于查韦市。不要重复。
缺少必填字段400更正提交内容;重复返回相同的错误。
周或年无效500该日期超出了公认的流行病学周。更正字段 date.
Python
import time
import requests

NAO_REPETIR = {400, 403, 404, 409}


def enviar(url, chave, dados, tentativas=3):
    for tentativa in range(tentativas):
        try:
            resposta = requests.post(
                url, params={"key": chave}, data=dados, timeout=60
            )
        except requests.RequestException as erro:
            print("Falha de rede:", erro)
            time.sleep(5 * (tentativa + 1))
            continue

        if resposta.status_code in NAO_REPETIR:
            print("Erro definitivo:", resposta.status_code, resposta.json())
            return None

        if resposta.ok:
            return resposta.json()

        # 500 e demais erros de servidor: espera progressiva e tenta de novo
        time.sleep(5 * (tentativa + 1))

    return None

良好做法

  • 尽可能按市或州进行过滤 - 如果不进行过滤,查询将扫描整个国家。
  • 首选增量同步 id 每天降低整个底座。
  • 串行发送请求。该 API 不适用于每个客户端同时进行数十个调用。
  • 在迭代之前检查响应是否是一个列表:寻呼限制消息的状态为 200。
  • 保存 counting_idblock_id 收到 - 它们允许您将 Conta Ovos 记录与系统中的记录进行匹配。
  • 将流行病学周作为决定字段:Conta Ovos 仅接受每个产卵器每周一次的读数,以及每周每个街区一次的访问。
  • 关注页面 mudanças 在更新您的集成之前。