第一步
所有调用都来自同一个数据库并返回 JSON。公共端点不需要身份验证;私人的需要您的访问密钥。
基址
/zh-mo/api 语言前缀是 URL 的一部分 (zh-mo), 但它不会改变返回的数据——记录来自银行,与现场注册的完全一样。
curl -G -d "municipality=Ponta Pora" --data-urlencode "country=Brasil" \
/zh-mo/api/lastcountingpublic 如果响应是 JSON 列表,则集成已经在运行。
key 从查询字符串中读取 (?key=SUA_CHAVE). 仅在表单返回的正文中发送 404 "Wrong key". 其他字段继续出现在请求正文中。 查阅公开数据
终点 /api/lastcountingpublic 返回最新的鸡蛋计数以及流行病学纬度、经度、周和年。它是地图、面板和研究的起点。
/api/lastcountingpublic民众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 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"]) 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——也就是说,主体不再是一个列表。在迭代之前始终检查类型。
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") date_start e date_end 将时间段划分为更小的间隔。 增量同步
为了维护镜像基础,不要在每次运行时再次下载所有内容。终点 /api/lastcountingpublic 公认 id, 它仅返回该标识符的计数,并且 date (纳入日期) e date_collect (data de coleta).
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
公共计数已经带有坐标,因此您可以生成电子表格或提供地图,而无需任何额外的连接。
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 并需要密钥。密钥的地理范围定义了它可以记录的内容——市政密钥仅记录该市本身。
/api/postcounting私人的发送现有产卵器的读数
发送已注册的产卵器的卵数。产卵器位于 ovitrap_group_id 重点市辖区内,流行病学周按实地计算 date.
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" 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" /api/postcounting私人的在读数旁边安装一个新的产卵器
如果 ovitrap_group_id 该市尚不存在产卵器,因此在同一请求中创建了产卵器。同时发送地址和类型(如果适用) (ovitrap_type_id: 1 城市, 模式; 2 rural).
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_lngouovitrap_group_id.400—ovitrap_type_id与 1 或 2 不同。404— 已经对这个诱产蛋器进行了年份和星期的计数。500— 发送的日期属于未来或无效的流行病学周。
/api/postaction私人的登记一次街区访问
使用 block_id 在已注册的街区中启动访问。如果该块尚不存在,则发送 block_group_id 与块数据一起,它是在主要城市内自动创建的。
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()) 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。
删除记录
拆除是明确的,并且始终仅限于主要城市。
# 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. |
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_id和block_id收到 - 它们允许您将 Conta Ovos 记录与系统中的记录进行匹配。 - 将流行病学周作为决定字段:Conta Ovos 仅接受每个产卵器每周一次的读数,以及每周每个街区一次的访问。
- 关注页面 mudanças 在更新您的集成之前。