目标
方便协同,在线excel的形式,后端数据库可以通过REST api的形式操作处理。
满足产运和RD之间数据交付的需求,支持多种格式,包括附件
开源方案:nocodb
- 增
import requests
import json
# 配置信息
TOKEN = "n31XQu11z22KThmbcJzwBGlAV8fPms_tJiR5MeTu"
HEADERS = {
"xc-token": TOKEN,
"Content-Type": "application/json"
}
# 源表和目标表 ID
SOURCE_TABLE_ID = "mpt1n1867woemc8"
NEW_TABLE_ID = "m44vmdj7t9l6sxc" # 👈 请在此处填写新表 ID
# 1. 从源表获取数据
get_url = f"http://localhost:8080/api/v2/tables/{SOURCE_TABLE_ID}/records?offset=0&limit=25"
response = requests.get(get_url, headers=HEADERS)
if response.status_code == 200:
raw_data = response.json().get("list", [])
# 2. 过滤数据
post_url = f"http://localhost:8080/api/v2/tables/{NEW_TABLE_ID}/records"
cleaned_data = []
for item in raw_data:
# 移除自增 ID 和系统时间(让新表自动生成,避免冲突)
# 如果新表结构完全一致,可以保留,但通常建议删除旧 ID
item.pop('Id', None)
item.pop('CreatedAt', None)
item.pop('UpdatedAt', None)
# 处理图片逻辑:如果是 =DISPIMG 开头,则设为 None
img_content = item.get("成分图片")
if isinstance(img_content, str) and img_content.startswith("=DISPIMG"):
continue
# 3. 将清洗后的数据导入新表
# NocoDB 支持直接发送数组进行批量创建
post_response = requests.post(post_url, headers=HEADERS, json=item)
cleaned_data.append(item)
# print(cleaned_data[0].keys())
if post_response.status_code in [200, 201]:
print(f"成功导入 {len(cleaned_data)} 条数据到新表!")
else:
print(f"导入失败: {post_response.text}")
else:
print(f"获取源数据失败: {response.status_code}")
- 删
import requests
import json
# 配置信息
TOKEN = "n31XQu11z22KThmbcJzwBGlAV8fPms_tJiR5MeTu"
HEADERS = {
"xc-token": TOKEN,
"Content-Type": "application/json"
}
TABLE_ID = "m44vmdj7t9l6sxc"
url = f"http://localhost:8080/api/v2/tables/{TABLE_ID}/records"
# 1. 构造要删除的 ID 列表 (假设 ID 是从 1 到 7)
# NocoDB 的批量删除接口通常接受对象数组或单个对象
ids_to_delete = [{"Id": i} for i in range(1, 7)]
# 2. 发送 DELETE 请求
# 注意:在 requests 中发送带 body 的 DELETE 请求使用 json 参数
try:
response = requests.delete(url, headers=HEADERS, json=ids_to_delete)
if response.status_code == 200:
print(f"成功删除了 {len(ids_to_delete)} 条记录。")
print("响应结果:", response.text)
else:
print(f"删除失败,状态码: {response.status_code}")
print("错误详情:", response.text)
except Exception as e:
print(f"发生错误: {e}")
- 查
import requests
import json
url = "http://localhost:8080/api/v2/tables/mpt1n1867woemc8/records?offset=0&limit=25"
headers = {"xc-token": "n31XQu11z22KThmbcJzwBGlAV8fPms_tJiR5MeTu"}
response = requests.get(url, headers=headers)
if response.status_code == 200:
data = response.json()
new_data = {"list": []}
# 遍历列表中的每一项记录
for item in data.get("list", []):
img_content = item.get("成分图片")
# 判断是否为字符串类型且以 '=DISPIMG' 开头
if isinstance(img_content, str) and img_content.startswith("=DISPIMG"):
continue
new_data["list"].append(item)
# 打印处理后的结果(格式化输出)
print(json.dumps(new_data, indent=2, ensure_ascii=False))
else:
print(f"请求失败,状态码:{response.status_code}")