如何使用Python下载GitHub仓库内的每日更新CSV文件？

1. 依赖安装 #

首先安装需要的第三方库：
pip install requests psycopg2-binary
如果需要做数据预处理，可以额外安装pandas和sqlalchemy：
pip install pandas sqlalchemy

2. 自动下载CSV文件 #

Python下载逻辑如下，支持跳过已更新的文件避免重复下载：

import requests
import os

# 配置项
CSV_REMOTE_URL = "https://raw.githubusercontent.com/owid/covid-19-data/master/public/data/owid-covid-data.csv"
LOCAL_SAVE_PATH = "./owid_covid_latest.csv"

def download_latest_csv():
    request_headers = {}
    # 本地文件存在时，用最后修改时间校验是否需要更新
    if os.path.exists(LOCAL_SAVE_PATH):
        local_modify_time = os.path.getmtime(LOCAL_SAVE_PATH)
        request_headers["If-Modified-Since"] = requests.utils.formatdate(local_modify_time, usegmt=True)
    
    response = requests.get(CSV_REMOTE_URL, headers=request_headers, timeout=180)
    if response.status_code == 304:
        print("本地文件为最新版本，无需重新下载")
        return True
    elif response.status_code == 200:
        with open(LOCAL_SAVE_PATH, "wb") as f:
            f.write(response.content)
        print(f"最新CSV已下载至：{LOCAL_SAVE_PATH}")
        return True
    else:
        print(f"下载失败，响应状态码：{response.status_code}")
        return False

3. CSV数据导入PostgreSQL #

3.1 前置准备 #

提前在PostgreSQL中创建和CSV列字段一一对应的表结构，字段类型需要和数据格式匹配

3.2 高效导入代码（适合大数据量） #

使用psycopg2的copy_from方法导入，性能远高于逐行插入：

import psycopg2
from psycopg2 import sql

# 数据库连接配置，生产环境建议从环境变量读取敏感信息
PG_CONF = {
    "host": "127.0.0.1",
    "port": 5432,
    "user": "your_username",
    "password": "your_password",
    "dbname": "your_dbname"
}
TARGET_TABLE = "owid_covid_data"

def import_csv_to_postgres():
    conn = None
    try:
        conn = psycopg2.connect(**PG_CONF)
        cursor = conn.cursor()
        # 全量更新场景先清空目标表，增量更新可删除此行，自行添加去重逻辑
        cursor.execute(sql.SQL("TRUNCATE TABLE {}").format(sql.Identifier(TARGET_TABLE)))
        # 导入CSV数据
        with open(LOCAL_SAVE_PATH, "r", encoding="utf-8") as f:
            # 跳过CSV表头行
            next(f)
            cursor.copy_from(f, TARGET_TABLE, sep=",", null="")
        conn.commit()
        print("数据导入PostgreSQL成功")
    except Exception as e:
        if conn:
            conn.rollback()
        print(f"数据导入失败：{str(e)}")
        raise
    finally:
        if conn:
            cursor.close()
            conn.close()

3.3 需预处理场景导入代码（适合小数据量） #

如果需要对CSV数据做清洗、转换后再导入，可以用pandas处理：

import pandas as pd
from sqlalchemy import create_engine

# 构造数据库连接引擎
engine = create_engine(f'postgresql+psycopg2://{PG_CONF["user"]}:{PG_CONF["password"]}@{PG_CONF["host"]}:{PG_CONF["port"]}/{PG_CONF["dbname"]}')

def import_with_pandas():
    # 读取CSV，可在此处添加数据清洗、过滤逻辑
    df = pd.read_csv(LOCAL_SAVE_PATH)
    # 导入数据库，if_exists可选'replace'/'append'，对应全量/增量更新
    df.to_sql(TARGET_TABLE, engine, if_exists='replace', index=False)
    print("pandas导入数据完成")

4. 调度配置 #

如果需要每日自动执行，可以根据运行环境配置定时任务：

• Linux环境：配置crontab定时任务，每天固定时间执行脚本
• Windows环境：使用系统自带的任务计划程序，设置每日触发执行脚本

注意事项 #

• 下载和导入逻辑建议添加异常捕获和告警通知，任务失败时及时处理
• 增量更新场景不要直接清空表，可按日期字段做对比后插入新增数据
• 生产环境不要硬编码数据库账号密码，建议存在环境变量或配置中心中

内容的提问来源于stack exchange，提问作者blackwings15