RobotDaily/arxiv-digest/scripts/install_system_cron.py

#!/usr/bin/env python3
"""Install or remove a system cron entry for RobotDaily."""

from __future__ import annotations

import argparse
import subprocess
from pathlib import Path

from utils import SKILL_DIR, log

MARKER = "# robotdaily-arxiv-digest"


def read_crontab() -> str:
    result = subprocess.run(["crontab", "-l"], capture_output=True, text=True, check=False)
    if result.returncode != 0:
        return ""
    return result.stdout


def write_crontab(content: str) -> None:
    result = subprocess.run(["crontab", "-"], input=content, text=True, check=False)
    if result.returncode != 0:
        raise SystemExit("Failed to write crontab")


def build_line(python_bin: str) -> str:
    skill_dir = SKILL_DIR
    log_dir = skill_dir / "logs"
    log_file = log_dir / "robotdaily.log"
    return (
        f"30 10 * * * mkdir -p {log_dir} && cd {skill_dir} && "
        f"{python_bin} scripts/run_daily.py --publish-discord >> {log_file} 2>&1 {MARKER}"
    )


def main() -> None:
    parser = argparse.ArgumentParser(description="Install RobotDaily daily cron")
    parser.add_argument("--python-bin", default="python3")
    parser.add_argument("--remove", action="store_true")
    args = parser.parse_args()

    existing_lines = [line for line in read_crontab().splitlines() if MARKER not in line]
    if args.remove:
        write_crontab("\n".join(existing_lines) + ("\n" if existing_lines else ""))
        log("Removed RobotDaily cron entry")
        return

    existing_lines.append(build_line(args.python_bin))
    write_crontab("\n".join(existing_lines) + "\n")
    log("Installed RobotDaily cron entry for 10:30 Asia/Shanghai")


if __name__ == "__main__":
    main()