如何每晚定时发送CSV报告到不支持API的系统？

2024-07-20 03:36:32

摆脱数据孤岛：如何每晚定时发送 CSV 报告到不支持 API 的系统？

在当今信息化的时代，企业往往依赖各种不同的软件系统来管理运营。然而，这些系统之间的数据传输却常常成为令人头疼的问题。你是否也正面临这样的困境：系统 A 每晚生成一份 CSV 格式的预约数据报告，你需要将其发送到系统 B，但系统 A 却没有提供任何 API 接口？更麻烦的是，系统 B 只接受 JSON 格式的数据，并要求使用 POST 请求进行数据传输。

你或许尝试过手动下载 CSV 文件，将其转换为 JSON 格式，然后登录系统 B 手动上传。但这种方式不仅费时费力，还容易出错，尤其是需要每天执行的时候。

本文将为你提供一个一劳永逸的解决方案：利用 Python 脚本实现 CSV 文件的下载、转换和发送，彻底解放你的双手，告别手动操作的烦恼。

Python 自动化：你的数据传输利器

我们的目标是创建一个 Python 脚本，实现以下功能：

自动从系统 A 下载最新的 CSV 报告。
将 CSV 文件转换为系统 B 要求的 JSON 格式。
使用系统 B 提供的 POST 接口，将 JSON 数据发送到目标系统。
设置定时任务，每晚自动执行该脚本。

接下来，我们将逐步拆解这个方案，并展示如何用代码实现。

下载 CSV 文件：精准获取你的数据

首先，我们需要使用 Python 下载系统 A 生成的 CSV 文件。

场景一：系统 A 提供了固定的 CSV 文件下载链接

这种情况下，我们可以直接使用 Python 的 requests 库下载文件，就像从网页下载文件一样简单。
```
import requests

url = 'https://www.example.com/report.csv'  # 替换为实际的下载链接
response = requests.get(url)

with open('report.csv', 'wb') as f:
    f.write(response.content)
```
这段代码首先使用 requests.get 方法获取指定 URL 的内容，然后将内容写入到名为 report.csv 的文件中。

场景二：系统 A 需要登录才能下载文件，或者下载链接是动态生成的

这种情况下，我们需要借助 selenium 或 requests-html 等库模拟浏览器操作，自动登录系统并下载文件。

from selenium import webdriver
from selenium.webdriver.common.by import By

driver = webdriver.Chrome()  # 选择你的浏览器驱动
driver.get('https://www.example.com/login')  # 替换为实际的登录页面地址

# 输入用户名和密码
driver.find_element(By.ID, 'username').send_keys('your_username')
driver.find_element(By.ID, 'password').send_keys('your_password')
driver.find_element(By.ID, 'login-button').click()

# 下载文件
driver.find_element(By.ID, 'download-button').click()

driver.quit()

这段代码模拟了用户登录网站并点击下载按钮的过程。你需要根据实际情况修改登录信息、按钮 ID 等参数。

转换 CSV 为 JSON：让数据格式不再成为障碍

下载 CSV 文件后，我们需要将其转换为系统 B 要求的 JSON 格式。Python 的 csv 和 json 库为我们提供了便捷的工具。

  import csv
  import json

  # 读取 CSV 文件
  with open('report.csv', 'r', encoding='utf-8') as csvfile:  # 根据实际编码方式修改 encoding 参数
      reader = csv.DictReader(csvfile)
      data = list(reader)

  # 将数据转换为 JSON 格式
  json_data = json.dumps(data, ensure_ascii=False, indent=4)  # 使用 ensure_ascii=False 保留中文，indent=4 格式化输出

  # 可选：将 JSON 数据保存到文件
  with open('report.json', 'w', encoding='utf-8') as jsonfile:  # 根据实际编码方式修改 encoding 参数
      jsonfile.write(json_data)

这段代码首先使用 csv.DictReader 将 CSV 文件读取为一个字典列表，每个字典代表 CSV 文件中的一行数据。然后，使用 json.dumps 将字典列表转换为 JSON 字符串，并使用 ensure_ascii=False 保留中文，indent=4 格式化输出。

发送 JSON 数据：将数据送达目的地

现在我们已经将数据转换为 JSON 格式，可以使用 requests 库将其发送到系统 B。

  import requests

  url = 'https://api.example.com/appointments'  # 替换为系统 B 提供的 POST 接口地址
  headers = {
      'Content-Type': 'application/json',
      'Authorization': f'Bearer {token}'  # 替换为你的授权 token
  }
  data = {'appointments': json_data}  # 根据系统 B 的要求调整数据格式

  response = requests.post(url, headers=headers, data=data.encode('utf-8'))

  # 检查请求是否成功
  if response.status_code == 200:
      print('数据发送成功！')
  else:
      print(f'数据发送失败：{response.text}')

这段代码使用 requests.post 方法发送 POST 请求，并设置了请求头和数据。其中，Content-Type 告诉系统 B 我们发送的数据是 JSON 格式，Authorization 用于身份验证。

设置定时任务：让脚本自动运行

最后一步是设置定时任务，让脚本每晚自动运行。我们可以使用操作系统的定时任务功能，或者使用 Python 的 schedule 库。

使用操作系统定时任务 (以 Windows 为例):

打开“任务计划程序”。
创建一个新的基本任务。
设置任务名称和触发时间 (例如，每天晚上 11 点)。
选择“启动程序”操作，并将程序或脚本设置为你的 Python 脚本路径，参数设置为脚本所需的参数 (如果需要)。

使用 schedule 库:

import schedule
import time

def job():
    # 你的数据传输脚本代码

schedule.every().day.at("23:00").do(job)

while True:
    schedule.run_pending()
    time.sleep(1)

这段代码将每晚 11 点执行一次 job 函数，你只需要将数据传输脚本代码放入 job 函数即可。