学会用Cron定时调度Python脚本

┌──────── 分钟 (0-59)│ ┌────── 小时 (0-23)│ │ ┌──── 日期 (1-31)│ │ │ ┌── 月份 (1-12)│ │ │ │ ┌ 星期 (0=周日,1=周一 … 6=周六)│ │ │ │ │* * * * *  执行命令

# 每天上午 10 点执行0 10 * * * python3 /脚本绝对路径/script.py# 每小时执行一次0 * * * * python3 /脚本绝对路径/script.py# 每 15 分钟执行一次*/15 * * * * python3 /脚本绝对路径/script.py# 每周一上午 9 点执行0 9 * * 1 python3 /脚本绝对路径/script.py# 每月 1 号凌晨 0 点执行0 0 1 * * python3 /脚本绝对路径/script.py

# 编辑当前用户的定时任务（核心命令）crontab -e# 查看已配置的所有定时任务crontab -l# 清空当前用户所有定时任务（谨慎使用，无法恢复）crontab -r

# 这样容易出问题，相对路径，Cron 运行报错withopen("state.json") as f:    data = json.load(f)# 正确写法：基于脚本自身位置拼接绝对路径import os# 获取脚本所在目录BASE_DIR = os.path.dirname(os.path.abspath(__file__))# 拼接文件完整路径STATE_FILE = os.path.join(BASE_DIR, "state.json")withopen(STATE_FILE) as f:    data = json.load(f)

#!/usr/bin/env python3"""脚本功能说明、Cron 配置备注"""import osimport json# 业务代码...

#!/usr/bin/env python3import loggingimport osBASE_DIR = os.path.dirname(os.path.abspath(__file__))# 日志文件绝对路径LOG_FILE = os.path.join(BASE_DIR, "publish.log")# 初始化日志logging.basicConfig(    filename=LOG_FILE,    level=logging.INFO,    format="%(asctime)s %(levelname)s %(message)s",    datefmt="%Y-%m-%d %H:%M:%S")log = logging.getLogger(__name__)def main():    log.info("脚本开始执行")    try:        # 你的业务逻辑        log.info("任务执行成功")    except Exception as e:        # 打印完整异常堆栈        log.error(f"任务执行失败：{e}", exc_info=True)        raiseif __name__ == "__main__":    main()

0 10 * * * /usr/bin/python3 /脚本路径/publish_queue.py >> /脚本路径/publish.log 2>&1

import osdef load_env(file_path: str):    """加载 .env 文件中的环境变量"""    with open(file_path, "r", encoding="utf-8") as f:        for line in f:            line = line.strip()            # 跳过空行、注释行            if not line or line.startswith("#") or "=" not in line:                continue            key, _, value = line.partition("=")            os.environ[key.strip()] = value.strip().strip("'\"")# 拼接 .env 绝对路径BASE_DIR = os.path.dirname(os.path.abspath(__file__))env_path = os.path.join(BASE_DIR, ".env")load_env(env_path)# 正常读取变量token = os.environ["DEVTO_TOKEN"]

# 顶部定义环境变量DEVTO_TOKEN=abc123yourtokenGUMROAD_TOKEN=xyz789yourtoken# 定时任务0 10 * * * python3 /脚本路径/publish_queue.py >> /脚本路径/publish.log 2>&1

#!/usr/bin/env python3import osimport jsonimport reimport loggingimport requestsfrom datetime import date# 基础路径配置BASE_DIR = os.path.dirname(os.path.abspath(__file__))QUEUE_FILE = os.path.join(BASE_DIR, "publish_queue.json")LOG_FILE = os.path.join(BASE_DIR, "queue.log")# 日志初始化logging.basicConfig(    filename=LOG_FILE,    level=logging.INFO,    format="%(asctime)s %(message)s")log = logging.getLogger(__name__)# 加载环境变量def load_env(file_path: str):    with open(file_path, "r", encoding="utf-8") as f:        for line in f:            line = line.strip()            if not line or line.startswith("#") or "=" not in line:                continue            key, _, value = line.partition("=")            os.environ[key.strip()] = value.strip()load_env(os.path.join(BASE_DIR, ".env"))TOKEN = os.environ.get("DEVTO_TOKEN", "")# 队列读写、发布业务逻辑def load_queue():    with open(QUEUE_FILE, "r", encoding="utf-8") as f:        return json.load(f)def save_queue(q):    with open(QUEUE_FILE, "w", encoding="utf-8") as f:        json.dump(q, f, indent=2)def publish(filepath):    file_full_path = os.path.join(BASE_DIR, filepath)    with open(file_full_path, "r", encoding="utf-8") as f:        content = f.read()    # 解析文章元信息    match = re.match(r'^---\n(.*?)\n---\n', content, re.DOTALL)    fm = {}    for line in match.group(1).splitlines():        if line.startswith('tags: '):            fm['tags'] = [t.strip() for t in line[6:].split(',')]        elif ': ' in line:            k, _, v = line.partition(': ')            fm[k.strip()] = v.strip()    # 调用接口发布文章    headers = {"api-key": TOKEN, "Content-Type": "application/json"}    resp = requests.post(        "https://dev.to/api/articles",        headers=headers,        json={            "article": {                "title": fm["title"],                "body_markdown": content,                "published": True,                "tags": fm.get("tags", [])            }        }    )    resp.raise_for_status()    return resp.json()def main():    q = load_queue()    if not q["pending"]:        log.info("任务队列为空，无需执行")        return    item = q["pending"][0]    log.info(f"开始发布：{item['title']}")    result = publish(item["filename"])    publish_url = f"https://dev.to{result['path']}"    log.info(f"发布成功：{publish_url}")    # 更新队列    q["pending"].pop(0)    q["published"].append({        "filename": item["filename"],        "title": item["title"],        "date": str(date.today()),        "url": publish_url,        "id": result["id"]    })    save_queue(q)if __name__ == "__main__":    main()

# shell# 全局环境变量DEVTO_TOKEN=your_token_here# 每天9点：推送RSS0 9 * * * /脚本目录/.venv/bin/python /脚本目录/daily_ping.py >> /脚本目录/ping.log 2>&1# 每天10点：发布队列文章0 10 * * * /脚本目录/.venv/bin/python /脚本目录/auto_publish_queue.py >> /脚本目录/queue.log 2>&1