Digital Ocean App Platform如何每日运行脚本更新Django外部内容

最推荐的实现方式是使用平台原生的定时触发无服务器函数，不需要额外维护第三方服务或常驻实例，成本最低、稳定性最好，和现有Django栈的适配成本也最低。

具体实现步骤 #

• 直接在现有Django项目代码仓库中新增函数目录，不需要拆分独立项目，函数逻辑可以直接复用Django的模型、存储、配置能力，只需要在入口处先初始化Django运行环境即可，参考代码如下：

import os
import sys
import requests
from django.core.wsgi import get_wsgi_application

# 把项目根目录加入系统路径，确保能正确导入Django模块
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
os.environ.setdefault('DJANGO_SETTINGS_MODULE', '替换为你的Django项目settings模块路径')
# 初始化Django环境
get_wsgi_application()

from django.core.files.base import ContentFile
from 你的业务app.models import 存储图片资源的模型

def run_crawl_task(args):
    # 拉取外部资源，加超时和异常捕获
    try:
        # 替换为你要抓取的目标图片URL
        source_url = "https://目标外部资源地址"
        resp = requests.get(source_url, timeout=30)
        resp.raise_for_status()
    except Exception as e:
        return {"status": "fail", "reason": f"资源下载失败: {str(e)}"}

    # 校验返回内容确实是图片，避免把错误页、拦截响应存到库里
    content_type = resp.headers.get("Content-Type", "")
    if not content_type.startswith("image/"):
        return {"status": "fail", "reason": "下载内容非合法图片格式"}

    # 更新Django模型数据，和你在视图里写的存图逻辑完全一致，自动兼容你配置的本地存储/DO Spaces存储
    img_record, is_new = 存储图片资源的模型.objects.update_or_create(
        source_url=source_url,
        defaults={
            "image_field": ContentFile(resp.content, name="自定义保存的图片文件名.jpg")
        }
    )
    return {"status": "success", "is_new_record": is_new, "record_id": img_record.id}

• 登录DigitalOcean App Platform控制台，给你已部署的Django应用新增组件，选择「无服务器函数」类型，关联你刚才写的函数代码路径。
• 给函数配置触发规则：选择定时触发，频率设置为每日，指定你需要运行的具体时间，注意选对和你业务匹配的时区。
• 把函数的环境变量配置成和主Django Web服务完全一致即可，同应用下的组件默认内网互通，不需要额外配置数据库、对象存储的访问白名单。

方案优势 #

• 成本极低：无服务器函数只有实际触发运行时才会产生计费，每日执行一次的任务量基本可以覆盖在平台免费额度内，不需要额外付费购买常驻实例
• 维护成本低：所有代码和主应用在同一个仓库，跟着主应用一起版本管理、部署，不会出现调度服务和业务代码版本不一致的问题，任务运行日志直接在平台控制台就能查看排查
• 稳定性高：完全由平台托管定时触发逻辑，不会出现Web服务扩缩容、重启导致任务丢失或重复执行的问题

备选方案（适合长耗时任务场景） #

如果你的单次爬取+处理逻辑耗时很长，超过无服务器函数的最长执行时限，可以选择给应用新增常驻Worker组件配合轻量调度库实现：

• 在项目中新增Worker入口文件crawl_worker.py，参考代码：

import time
import schedule
# 把上面写的爬取更新逻辑封装成函数导入
from crawl_func import run_crawl_task

# 配置每日指定时间执行
schedule.every().day.at("02:00").do(run_crawl_task)

if __name__ == "__main__":
    while True:
        schedule.run_pending()
        # 每分钟轮询一次待执行任务，资源占用极低
        time.sleep(60)

• 在App Platform控制台新增Worker组件，设置启动命令为python crawl_worker.py，环境变量和主Web服务保持一致即可。

• 注意这个方案的Worker是常驻运行的，会持续占用实例资源，成本比定时函数高，仅推荐任务耗时超过函数执行上限的场景使用。

避坑提醒 #

• 不要在主Web服务进程里开后台线程跑定时任务：App Platform的Web服务会根据负载自动扩缩容、重启迁移，会导致任务重复执行、或者中途被中断丢失，稳定性完全没有保障
• 爬取逻辑一定要加全异常捕获和内容校验，避免外部接口异常时把垃圾数据写入业务库
• 不需要额外引入Celery这类重型分布式任务队列，对于单节点每日执行的简单任务来说过度设计，徒增维护成本

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