如何利用Cron Job实现FTP CSV文件到Django视图的定时处理？

Hey there! Since you're new to Cron, let's break down the best ways to build this workflow—starting with the simplest, beginner-friendly approach, then moving to a more scalable option for future needs.

方案1：系统Cron + 自定义Python脚本（最适合新手） #

这是最容易上手的方式，不需要额外的Django依赖，只用系统自带的Cron调度器和简单的Python脚本就能搞定。

步骤1：编写Python脚本拉取FTP文件并发送到Django视图 #

这个脚本会处理FTP连接、拉取CSV文件，然后直接发送到你的Django视图（不需要把文件存到本地）：

import ftplib
import requests
from io import BytesIO
import os

# 从环境变量加载配置（比硬编码安全！）
FTP_HOST = os.environ.get("FTP_HOST")
FTP_USER = os.environ.get("FTP_USER")
FTP_PASS = os.environ.get("FTP_PASS")
CSV_FILENAME = "data.csv"  # 替换成你的文件名
DJANGO_VIEW_URL = "https://your-django-app.com/api/process-csv/"

def fetch_and_send_csv():
    try:
        # 连接FTP服务器
        ftp = ftplib.FTP(FTP_HOST)
        ftp.login(FTP_USER, FTP_PASS)
        
        # 将CSV内容读取到内存（避免本地存储）
        csv_content = BytesIO()
        ftp.retrbinary(f"RETR {CSV_FILENAME}", csv_content.write)
        csv_content.seek(0)  # 重置文件指针到开头
        
        # 通过POST请求发送CSV到Django视图
        files = {"file": (CSV_FILENAME, csv_content, "text/csv")}
        response = requests.post(DJANGO_VIEW_URL, files=files)
        
        # 验证请求是否成功
        if response.status_code == 200:
            print(f"✅ 成功发送CSV到Django，响应：{response.text}")
        else:
            print(f"❌ 发送CSV失败，状态码：{response.status_code}，消息：{response.text}")
            
        ftp.quit()
    except Exception as e:
        print(f"⚠️ 同步过程中出错：{str(e)}")

if __name__ == "__main__":
    fetch_and_send_csv()

步骤2：配置Cron定时任务 #

1. 先给脚本添加执行权限：

   chmod +x /path/to/your/sync_script.py
   

   （或者直接用Python路径运行，比如/usr/bin/python3）

2. 编辑当前用户的Cron任务表：

   crontab -e
   

3. 添加一行调度规则，比如每天凌晨2点执行：

   0 2 * * * /usr/bin/python3 /path/to/your/sync_script.py >> /var/log/csv_sync.log 2>&1
   

   解释一下规则：

   • 0 2 * * *：分钟、小时、日期、月份、星期（这里代表每天凌晨2点整）
   • /usr/bin/python3：你的Python3可执行文件路径（可以用which python3查看）
   • /path/to/your/sync_script.py：脚本的绝对路径
   • >> /var/log/csv_sync.log 2>&1：把所有输出（包括错误）记录到日志文件，方便调试

4. 保存退出后，Cron会自动加载新任务，你可以用以下命令确认：

   crontab -l
   

步骤3：Django视图处理CSV并保存到数据库 #

现在配置Django视图来接收CSV、解析内容，然后保存到数据库：

from django.http import JsonResponse
import csv
from .models import YourDataModel  # 替换成你的实际模型

def process_csv(request):
    if request.method == "POST" and request.FILES.get("file"):
        csv_file = request.FILES["file"]
        # 解码并拆分CSV内容为行
        decoded_file = csv_file.read().decode("utf-8").splitlines()
        reader = csv.DictReader(decoded_file)
        
        # 从CSV行创建数据库条目
        for row in reader:
            # 映射CSV列到模型字段（根据你的实际情况调整）
            YourDataModel.objects.create(
                field_one=row["csv_column_1"],
                field_two=row["csv_column_2"],
                field_three=row["csv_column_3"]
                # 添加其他字段
            )
        
        return JsonResponse({"status": "success", "message": "数据保存成功！"})
    
    return JsonResponse({"status": "error", "message": "无效请求或未提供文件"}, status=400)

别忘了在urls.py里添加路由：

from django.urls import path
from .views import process_csv

urlpatterns = [
    path("api/process-csv/", process_csv, name="process-csv"),
]

方案2：Django Celery + Celery Beat（适合复杂/多任务场景） #

如果你的Django项目已经在使用Celery，或者以后计划添加更多定时任务，这种集成式方案更优雅（不需要单独的系统脚本）。

步骤1：安装依赖 #

pip install celery redis  # Redis用作消息代理，用RabbitMQ也可以

步骤2：配置Celery和定时任务 #

在Django项目根目录创建celery.py文件：

import os
from celery import Celery
from celery.schedules import crontab

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "your_project.settings")

app = Celery("your_project")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()

# 定义定时任务
app.conf.beat_schedule = {
    "daily-csv-sync": {
        "task": "your_app.tasks.fetch_and_send_csv",
        "schedule": crontab(hour=2, minute=0),  # 每天凌晨2点执行
    },
}

步骤3：创建Celery任务 #

在你的Django应用中创建tasks.py文件：

from celery import shared_task
import ftplib
import requests
from io import BytesIO
from django.conf import settings

@shared_task(bind=True, max_retries=3)
def fetch_and_send_csv(self):
    try:
        # 使用Django配置中的FTP信息连接服务器
        ftp = ftplib.FTP(settings.FTP_HOST)
        ftp.login(settings.FTP_USER, settings.FTP_PASS)
        
        csv_content = BytesIO()
        ftp.retrbinary(f"RETR {settings.CSV_FILENAME}", csv_content.write)
        csv_content.seek(0)
        
        # 发送到Django视图
        files = {"file": (settings.CSV_FILENAME, csv_content, "text/csv")}
        response = requests.post(f"{settings.DJANGO_BASE_URL}/api/process-csv/", files=files)
        
        if response.status_code != 200:
            raise Exception(f"请求失败：{response.text}")
            
        ftp.quit()
        return "CSV同步完成"
    except Exception as e:
        # 如果失败，最多重试3次，每次间隔60秒
        self.retry(exc=e, countdown=60)

在settings.py中添加FTP和Celery配置：

# FTP配置
FTP_HOST = "your-ftp-host.com"
FTP_USER = "your-ftp-username"
FTP_PASS = "your-ftp-password"
CSV_FILENAME = "data.csv"
DJANGO_BASE_URL = "https://your-django-app.com"

# Celery配置
CELERY_BROKER_URL = "redis://localhost:6379/0"
CELERY_RESULT_BACKEND = "redis://localhost:6379/0"

步骤4：启动Celery服务 #

# 启动Celery worker（处理任务执行）
celery -A your_project worker --loglevel=info

# 启动Celery Beat（处理定时任务触发）
celery -A your_project beat --loglevel=info

关键最佳实践 #

• 安全优先：永远不要硬编码密码或敏感数据！用环境变量（Cron脚本）或Django的settings.py（Celery方案）存储机密信息。
• 日志记录：一定要记录任务的输出和错误——出问题时调试会轻松很多。
• 手动测试：在调度前，先手动运行脚本或Celery任务，确认端到端流程正常。
• 错误处理：添加重试机制（比如Celery示例中的重试）和告警（例如失败时发送邮件），让你的工作流更健壮。

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