使用 Cron 任务，像专业人士一样自动发布你的 Instagram 帖子！🚀

2025-06-07

使用 Cron 任务，像专业人士一样自动发布你的 Instagram 帖子！🚀

TL;DR✨

在这个简单易懂的教程中，您将学习如何使用 cron 作业从头开始构建自己的 Instagram 自动化工具。😎

你将学到：👀

  • 了解如何在 Python 项目中设置日志记录。
  • 学习使用python-crontab模块在基于 Unix 的操作系统中添加 cron 作业。
  • 了解如何使用instagrapi模块在 Instagram 上发帖。

那么，你准备好打造最酷的 Instagram 自动化工具了吗？😉

准备好 GIF

设置环境⚙️

在深入构建项目之前，请先查看一下项目架构，以简要了解布局。

项目架构

💁 我们将从头开始构建这个项目，使其具备日志记录支持以及类和函数中结构化的所有内容，可以投入生产。

初始化项目🛠️

创建一个文件夹来保存项目的所有源代码：

mkdir insta-cron-post-automation
cd insta-cron-post-automation
Enter fullscreen mode Exit fullscreen mode

创建一些新的子文件夹，用于存储帖子数据、日志和 Shell 脚本：

mkdir -p data logs src/scripts
Enter fullscreen mode Exit fullscreen mode

现在已经设置了初始文件夹结构，是时候创建一个新的虚拟环境并安装我们将在项目中使用的所有模块了。

运行以下命令在项目根目录中创建并激活一个新的虚拟环境：

python3 -m venv .venv
source .venv/bin/activate # If you are using fish shell, change the activate binary to activate.fish
Enter fullscreen mode Exit fullscreen mode

运行此命令来安装我们将在项目中使用的所有必要模块：

pip3 install instagrapi python-crontab python-dotenv lorem numpy pillow
Enter fullscreen mode Exit fullscreen mode

每个模块的用途如下：

  • instagrapi：登录并发布到 Instagram。
  • python-crontab：创建和编辑用户的 crontab。
  • python-dotenv：从文件中读取环境变量.env。

可选模块

  • lorem：生成用于创建示例帖子的虚拟描述。
  • numpy：生成随机像素数据来为我们的示例帖子创建图像。
  • pillow：使用来自 NumPy 的像素数据创建示例图像。

让我们开始编码吧💻

火 GIF

设置日志记录

💡 由于我们的工具是在用户通过 cron 任务设置的特定时间运行的，因此我们不能依赖打印语句来记录输出。一切都在后台进行，所以我们需要一个中心位置来查看程序的日志，例如日志文件。

为了支持日志记录，我们将使用我们的老 Python 朋友，logging模块。

在目录中src，添加一个名为的文件logger_config.py，其代码如下：

💡 请注意，我正在使用typing模块来设置变量的类型。使用 TypeScript 这么久之后，我忍不住要使用类型定义 🫠。

# 👇 insta-cron-post-automation/src/logger_config.py
import logging


def get_logger(log_file: str) -> logging.Logger:
    """
    Creates and configures a logger to log messages to a specified file.

    This function sets up a logger with an INFO logging level, adds a file handler
    to direct log messages to the specified log file, and applies a specific log
    message format.

    Args:
        log_file (str): The path to the log file where log messages will be saved.

    Returns:
        logging.Logger: Configured logger instance.
    """

    # Create a logger instance
    logger = logging.getLogger()

    # Set the logging level to INFO
    logger.setLevel(logging.INFO)

    # Create a file handler to write log messages to the specified file
    file_handler = logging.FileHandler(log_file)

    # Define the format for log messages
    formatter = logging.Formatter(
        "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
    )
    file_handler.setFormatter(formatter)

    # Add the file handler to the logger
    logger.addHandler(file_handler)

    return logger
Enter fullscreen mode Exit fullscreen mode

因此，此get_logger()函数接收一个日志文件的路径，该文件需要存储所有日志，而不是将它们记录到控制台。然后，它创建一个记录器实例并返回它。

现在，通过设置此功能，我们可以在项目中的任何位置调用它，并且它将保持相同的日志记录配置。😎

实现 Instagram 登录 🔑

使用以下代码在目录setup.py内创建一个名为的新文件：src

# 👇 insta-cron-post-automation/src/setup.py
import logging
import os
import sys
from typing import NoReturn, Tuple

from dotenv import load_dotenv
from instagrapi import Client


def log_and_exit(logger: logging.Logger, message: str) -> NoReturn:
    """
    Log an error message and exit the program.

    Args:
    - logger (logging.Logger): The logger to use.
    - message (str): The error message to log.
    """
    logger.error(message)
    sys.exit(1)


def get_credentials(logger: logging.Logger) -> Tuple[str, str]:
    """
    Retrieve the username and password from environment variables.

    This function loads the environment variables from a .env file using dotenv,
    then retrieves the username and password from the environment variables.

    Args:
    - logger (logging.Logger): The logger instance to use for logging.

    Returns:
    - Tuple[str, str]: A tuple containing the username and password retrieved from the environment variables.

    Raises:
    - SystemExit: If the username or password environment variable is missing.
    """

    load_dotenv()

    # Get the username and password from the environment variables:
    username: str | None = os.getenv("INSTA_USERNAME")
    password: str | None = os.getenv("INSTA_PASSWORD")

    # Check if username or password is None, and raise an exception if so.
    if username is None or password is None:
        log_and_exit(
            logger=logger,
            message="Username or password environment variable is missing",
        )

    return username, password


def setup_instagrapi(logger: logging.Logger) -> Client:
    """
    Set up the instagrapi client with the provided username and password.

    This function uses the get_credentials() function to retrieve the username and password,
    then initializes the instagrapi client with the credentials.

    Args:
    - logger (logging.Logger): The logger instance to use for logging.

    Returns:
    - client (instagrapi.Client): The instagrapi client with the provided credentials.

    Raises:
    - SystemExit: If an error occurs while logging in to Instagram.
    """
    username, password = get_credentials(logger=logger)
    client = Client()

    try:
        login_success = client.login(username=username, password=password)

        if not login_success:
            log_and_exit(logger=logger, message="Instagram Login failed")

        logger.info("Instagram Login successful")

    except Exception as e:
        log_and_exit(
            logger=logger, message=f"An error occurred while trying to login: {e}"
        )

    return client
Enter fullscreen mode Exit fullscreen mode

该get_credentials()函数读取用户环境变量并返回它们。正如你可能已经猜到的，该函数要求你设置INSTA_USERNAME环境INSTA_PASSWORD变量。

.env在项目根目录中创建一个新文件，并定义两个变量。

INSTA_USERNAME=<your-insta-username>
INSTA_PASSWORD=<your-insta-password>
Enter fullscreen mode Exit fullscreen mode

该setup_instagrapi()函数创建一个新的 Instagrapi 客户端，登录 Instagram，并返回该客户端。

定义类🧩

我们将设置两个不同的类：Post和PostList。该类PostList将包含多个Post对象。

使用以下代码在目录post.py内创建一个名为的新文件：src

# 👇 insta-cron-post-automation/src/post.py
from typing import Any, Dict, Optional


class Post:
    """
    Initializes a new instance of the Post class.

    Args:
    - description (str): The description for the post.
    - image_path (str): The path to the image file.
    - post_date (str): The date and time of the post.
    - extra_data (Optional[Dict[str, Any]]): Additional data for the post. Defaults to None.
    """

    ALLOWED_EXTRA_DATA_FIELDS = {
        "custom_accessibility_caption",
        "like_and_view_counts_disabled",
        "disable_comments",
    }

    def __init__(
        self,
        description: str,
        image_path: str,
        post_date: str,
        extra_data: Optional[Dict[str, Any]] = None,
    ):
        self.image_path = image_path
        self.description = description
        self.post_date = post_date
        self.extra_data = self.validate_extra_data(extra_data=extra_data)

    def validate_extra_data(
        self, extra_data: Optional[Dict[str, Any]]
    ) -> Optional[Dict[str, Any]]:
        """
        Validates and filters the extra_data dictionary to ensure it contains only allowed fields.

        Args:
        - extra_data (Optional[Dict[str, Any]]): The extra data dictionary to validate.

        Returns:
        - Optional[Dict[str, Any]]: The validated extra data dictionary, or None if input is None or invalid.
        """
        if extra_data is None:
            return None

        validated_data = {
            key: extra_data[key]
            for key in extra_data
            if key in self.ALLOWED_EXTRA_DATA_FIELDS
        }

        return validated_data if validated_data else None

    def serialize(self) -> Dict[str, Any]:
        """
        Serialize the object into a dictionary representation.

        Returns:
        - dict: A dictionary containing the serialized data of the object.
                The dictionary has the following keys:
                - "image_path" (str): The path to the image file.
                - "description" (str): The description for the post.
                - "post_date" (str): The date and time of the post.
                If the object has extra data, it is added to the dictionary under the key "extra_data".
        """
        data: Dict[str, Any] = {
            "image_path": self.image_path,
            "description": self.description,
            "post_date": self.post_date,
        }

        if self.extra_data is not None:
            data["extra_data"] = self.extra_data

        return data
Enter fullscreen mode Exit fullscreen mode

该类采用一些参数，例如帖子的描述，图像路径，发布日期和可选的额外字段属性，可用于传递帖子的附加元数据，如下所示：

  "extra_data": {
    "custom_accessibility_caption": "An astronaut in the ocean!",
    "like_and_view_counts_disabled": 0,
    "disable_comments": 1
  },
Enter fullscreen mode Exit fullscreen mode

这里，二进制值 1 和 0 分别代表True和False。

该validate_extra_data()方法检查提供的extra_data字段是否仅包含有效键，并删除用户提供的任何其他键。

该serialize()方法检查extra_data参数是否已传递给构造函数。如果是，则将其添加到字典中并返回该字典；否则，返回不包含extra_data键的字典。

现在Post类已经准备好了，让我们创建另一个类，PostList来保存Post对象。

post_list.py在目录内创建一个名为的新文件src并添加以下代码行：

# 👇 insta-cron-post-automation/src/post_list.py
import json
import sys
from datetime import datetime
from typing import List, NoReturn, Optional

from logger_config import get_logger
from post import Post


class PostList:
    """
    A class to manage/represent a list of posts.
    """

    def __init__(self, log_path: str):
        self.posts = []
        self.logger = get_logger(log_path)

    def _log_and_exit(self, message: str) -> NoReturn:
        """
        Log an error message and exit the program.

        Args:
        - message (str): The error message to log.
        """
        self.logger.error(message)
        sys.exit(1)

    def to_json(self) -> str:
        """
        Serialize the list of posts into a JSON string.
        Use this method to write the content in the `self.posts` array to a JSON file.

        Returns:
        - str: JSON string representing the serialized posts.
        """
        serialized_posts = [post.serialize() for post in self.posts]
        return json.dumps({"posts": serialized_posts}, default=str)

    # Custom function to parse the date without seconds
    def parse_post_date(self, post_date: str) -> str:
        """
        Custom function to parse the date without seconds.

        Args:
        - post_date (str): The date string to parse.

        Returns:
        - str: The parsed date string without seconds.
        """
        date_format = "%Y-%m-%d %H:%M"

        # Parse the date
        parsed_date = datetime.strptime(post_date, date_format)

        # Return the date formatted without seconds
        return parsed_date.strftime("%Y-%m-%d %H:%M")

    def get_posts_from_json_file(self, posts_file_path: str) -> List[Post]:
        """
        Load posts from a JSON file and populate the list.

        Args:
        - posts_file_path (str): The path to the JSON file containing post data.

        Returns:
        - List[Post]: List of Post objects loaded from the JSON file.

        Raises:
        - FileNotFoundError: If the JSON file is not found.
        - PermissionError: If the JSON file cannot be accessed.
        - json.JSONDecodeError: If the JSON file is not valid JSON.
        """
        try:
            with open(posts_file_path, "r") as posts_json_file:
                data = json.load(posts_json_file)

                if "posts" not in data:
                    self._log_and_exit(message="No 'posts' key found in the json file")

                for post in data["posts"]:
                    if not all(
                        key in post
                        for key in ["image_path", "description", "post_date"]
                    ):
                        self._log_and_exit(
                            message="Missing required keys in the post object"
                        )

                    extra_data: Optional[dict] = post.get("extra_data")

                    post_obj = Post(
                        image_path=post["image_path"],
                        description=post["description"],
                        post_date=self.parse_post_date(post_date=post["post_date"]),
                        extra_data=extra_data,
                    )
                    self.posts.append(post_obj)

        except FileNotFoundError:
            self._log_and_exit(message=f"File not found: {posts_file_path}")

        except PermissionError:
            self._log_and_exit(message=f"Permission denied: {posts_file_path}")

        except json.JSONDecodeError:
            self._log_and_exit(message=f"Invalid JSON file: {posts_file_path}")

        except ValueError as ve:
            self._log_and_exit(
                message=f"Invalid date format provided in the post object: {ve}"
            )

        except Exception as e:
            self._log_and_exit(message=f"Unexpected error: {e}")

        return self.posts
Enter fullscreen mode Exit fullscreen mode

该_log_and_exit()方法，顾名思义，是一个私有方法，将消息记录到文件并退出程序。

该to_json()方法，顾名思义，以 JSON 字符串的形式返回帖子列表。

该parse_post_date()方法采用一个post_date变量并以字符串格式返回不带秒部分的日期，因为我们在 cron 作业中不需要秒数。

该get_posts_from_json_file()方法读取 JSON 文件，将每篇文章作为Post对象填充到文章数组中，并处理读取文件内容时可能发生的各种异常。

编写媒体后记

现在我们已经设置好了所有的类，是时候编写负责在 Instagram 上发布内容的主要 Python 脚本了。

media_post.py在目录中创建一个名为 的新文件src。这个文件会很长，所以我们会把代码拆分成每个函数，我会逐步解释。

# 👇 insta-cron-post-automation/src/media_post.py
import json
import logging
import os
import sys
from datetime import datetime
from typing import Any, Dict, List, NoReturn, Optional

from instagrapi import Client

from logger_config import get_logger
from setup import setup_instagrapi


def log_and_exit(logger: logging.Logger, message: str) -> NoReturn:
    """
    Log an error message and exit the program.

    Args:
    - logger (logging.Logger): The logger to use.
    - message (str): The error message to log.
    """
    logger.error(message)
    sys.exit(1)


def is_valid_image_extension(file_name: str) -> bool:
    """
    Check if the given file name has a valid image extension.

    Valid extensions are: .jpg, .jpeg, .png.

    Args:
    - file_name (str): The name of the file to check.

    Returns:
    - bool: True if the file has a valid image extension, False otherwise.
    """
    valid_extensions = {".jpg", ".jpeg", ".png"}
    return any(file_name.endswith(ext) for ext in valid_extensions)
Enter fullscreen mode Exit fullscreen mode

这些函数相当简单。该log_and_exit()函数将消息记录到文件中并退出程序。

该is_valid_image_extension()函数检查图像是否具有允许在 Instagram 上发布的有效扩展名。

💁 我不确定是否还有其他扩展名可以使用，但这些似乎是标准扩展名。如果有其他扩展名，请随时进行相应的更新。

一旦尝试将帖子上传到 Instagram，我们需要将其从目录to-post.json中的文件中删除data，该目录包含了我们想要安排发布的所有帖子。无论上传是否成功，我们都会将帖子添加到目录中的error.json或success.json文件中data。

创建一个处理此过程的新功能。

# 👇 insta-cron-post-automation/src/media_post.py

# Rest of the code...

def handle_post_update(
    success: bool, json_post_content: Dict[str, Any], logger: logging.Logger
) -> None:
    """
    Update the post error file based on the success of the upload.

    Args:
    - success (bool): True if the upload was successful, False otherwise.
    - json_post_content (dict): The content of the post.

    Returns:
    - Return the content of the post file if the read is successful; otherwise, return the default value if provided, or None.
    """

    def load_json_file(file_path: str, default: Optional[Any] = None) -> Any:
        """Helper function to load JSON data from a file."""
        if os.path.exists(file_path):
            try:
                with open(file_path, "r") as file:
                    return json.load(file)
            except Exception:
                log_and_exit(
                    logger=logger, message=f"Failed to load post file: {file_path}"
                )
        else:
            # Create the file with default content if it does not exist
            write_json_file(file_path, default if default is not None else [])
            return default if default is not None else []

    def write_json_file(file_path: str, posts: List[Dict[str, Any]]) -> None:
        """Helper function to save JSON data to a file."""
        for post in posts:
            if "post_date" in post:
                try:
                    post_date = datetime.strptime(
                        post["post_date"], "%Y-%m-%d %H:%M:%S"
                    )
                    post["post_date"] = post_date.strftime("%Y-%m-%d %H:%M")
                except ValueError:
                    post_date = datetime.strptime(post["post_date"], "%Y-%m-%d %H:%M")
                    post["post_date"] = post_date.strftime("%Y-%m-%d %H:%M")
                except Exception as e:
                    log_and_exit(
                        logger=logger, message=f"Failed to parse post date: {e}"
                    )

        try:
            with open(file_path, "w") as file:
                json.dump(posts, file, indent=2)
            logger.info(f"Post file updated: {file_path}")

        except (IOError, json.JSONDecodeError) as e:
            log_and_exit(logger=logger, message=f"Failed to write post file: {e}")

    # Get the directory of the current script
    current_dir = os.path.dirname(os.path.abspath(__file__))

    # Define the directory where the data files are located
    data_dir = os.path.join(current_dir, "..", "data")

    # Define paths to the success, error, and to-post files
    success_file = os.path.join(data_dir, "success.json")
    error_file = os.path.join(data_dir, "error.json")
    to_post_file = os.path.join(data_dir, "to-post.json")

    # Ensure the success and error files exist
    if not os.path.exists(success_file):
        write_json_file(success_file, [])

    if not os.path.exists(error_file):
        write_json_file(error_file, [])

    # Load the current 'to-post' data if it exists, otherwise initialize an empty list
    to_post_data = load_json_file(file_path=to_post_file, default={"posts": []})

    # Determine which file to write to based on the success of the upload
    target_file = success_file if success else error_file

    # Load the current content of the target file if it exists, otherwise initialize an empty list
    target_data = load_json_file(file_path=target_file, default=[])

    # Append the current post content to the target data
    target_data.append(json_post_content)

    # Write the updated target data back to the target file
    write_json_file(file_path=target_file, posts=target_data)

    user_posts = to_post_data["posts"]

    # Filter the posted post from the 'to-post' data
    if any(post == json_post_content for post in user_posts):
        user_posts = [item for item in user_posts if item != json_post_content]
        to_post_data["posts"] = user_posts
        write_json_file(file_path=to_post_file, posts=to_post_data)
Enter fullscreen mode Exit fullscreen mode

该handle_post_update()函数管理更新跟踪帖子上传成功或失败的文件的过程。根据帖子上传是否成功，该函数会使用帖子内容更新成功文件或错误文件。

该函数使用嵌套的辅助函数load_json_file()和write_json_file()来处理 JSON 数据的加载和保存。load_json_file()从文件读取数据，同时write_json_file()将数据保存回文件，确保数据格式正确。

最后，该函数通过将新帖子内容附加到data/success.json或来更新相关文件data/error.json，并从文件中删除已发布的内容data/to-post.json。

现在，我们需要一个函数将文件内容解析为 JSON。如果解析失败，我们还需要一种方法来处理错误。

# 👇 insta-cron-post-automation/src/media_post.py

# Rest of the code...

def parse_post_file_to_json(post_path: str, logger: logging.Logger) -> Dict[str, Any]:
    """
    Parses the content of a post file into a JSON dictionary.

    Args:
    - post_path (str): The path to the post file.
    - logger (logging.Logger): The logger instance to use for logging errors.

    Returns:
    - Dict[str, Any]: The content of the post file parsed as a JSON dictionary.

    Raises:
    - SystemExit: Exits the program with an error status if the file does not exist,
                  if permission is denied, if JSON decoding fails, or if any other
                  exception occurs during file reading.
    """
    try:
        with open(post_path, "r") as post_file:
            content = post_file.read()
        return json.loads(content)

    except FileNotFoundError:
        log_and_exit(logger=logger, message=f"Post file '{post_path}' does not exist")

    except PermissionError:
        log_and_exit(
            logger=logger,
            message=f"Permission denied when trying to access post file '{post_path}'",
        )

    except json.JSONDecodeError:
        log_and_exit(
            logger=logger, message=f"Failed to decode JSON from post file '{post_path}'"
        )

    except Exception as e:
        log_and_exit(
            logger=logger, message=f"Failed to read post file '{post_path}': {e}"
        )


def handle_post_error(
    error_message: str, json_post_content: Dict[str, Any], logger: logging.Logger
) -> None:
    """
    This function logs an error message, updates the post files to indicate failure,
    and terminates the program with an exit status of 1.

    Args:
    - error_message (str): The error message to be logged.
    - json_post_content (Dict[str, Any]): The content of the post file in JSON format.
    - logger (logging.Logger): The logger instance to use for logging the error.

    Returns:
    - None

    Raises:
    - SystemExit: The program will exit with an exit status of 1.
    """
    handle_post_update(
        success=False, json_post_content=json_post_content, logger=logger
    )
    log_and_exit(logger=logger, message=error_message)
Enter fullscreen mode Exit fullscreen mode

该parse_post_file_to_json()函数接收 JSON 文件的路径，并尝试将其内容解析为 JSON。如果解析失败，handle_invalid_post_file()则使用该函数处理失败。它会将成功布尔值设置为false，更新data/error.json文件，并从文件中删除特定的帖子data/to-post.json。

现在所有这些都已完成，我们终于可以计算最终的上传参数并将帖子上传到 Instagram 了。

添加这两个函数：

# 👇 insta-cron-post-automation/src/media_post.py

# Rest of the code...

def prepare_upload_params(
    json_post_content: Dict[str, Any], logger: logging.Logger
) -> Dict[str, Any]:
    # Initial needed upload parameters
    upload_params = {
        "path": json_post_content.get("image_path"),
        "caption": json_post_content.get("description"),
    }

    # If the optional field is provided
    if "extra_data" in json_post_content:
        extra_data = json_post_content["extra_data"]
        try:
            extra_data["custom_accessibility_caption"] = str(
                extra_data.get("custom_accessibility_caption", "")
            )
            extra_data["like_and_view_counts_disabled"] = int(
                extra_data.get("like_and_view_counts_disabled", 0)
            )
            extra_data["disable_comments"] = int(extra_data.get("disable_comments", 0))

        except (ValueError, TypeError):
            handle_post_error(
                error_message=f"Failed to parse 'extra_data' field: {json_post_content}",
                json_post_content=json_post_content,
                logger=logger,
            )

        extra_data["like_and_view_counts_disabled"] = max(
            0, min(1, extra_data["like_and_view_counts_disabled"])
        )
        extra_data["disable_comments"] = max(0, min(1, extra_data["disable_comments"]))
        upload_params["extra_data"] = extra_data

    return upload_params


def upload_to_instagram(
    client: Client,
    upload_params: Dict[str, Any],
    json_post_content: Dict[str, Any],
    logger: logging.Logger,
) -> None:
    """
    Uploads media to Instagram and handles logging and updating post files based on the result.

    Args:
    - client: The Instagram client used for uploading media.
    - upload_params (Dict[str, Any]): The parameters for the media upload.
    - json_post_content (Dict[str, Any]): The content of the post file in JSON format.
    - logger (logging.Logger): The logger instance to use for logging errors and success messages.

    Returns:
    - None

    Raises:
    - SystemExit: Exits the program with an error status if the upload fails.
    """
    try:
        # Upload the media to Instagram
        upload_media = client.photo_upload(**upload_params)

        # Get the uploaded post ID
        uploaded_post_id = upload_media.model_dump().get("id", None)
        logger.info(
            f"Successfully uploaded the post on Instagram. ID: {uploaded_post_id}"
        )
        handle_post_update(
            success=True, json_post_content=json_post_content, logger=logger
        )
    except Exception as e:
        handle_post_error(
            error_message=f"Failed to upload the post: {e}",
            json_post_content=json_post_content,
            logger=logger,
        )
Enter fullscreen mode Exit fullscreen mode

该prepare_upload_params()函数接收帖子内容并准备上传参数。它对字段进行了显式验证，extra_data以确保所有键都属于预期类型，并最终返回整套上传参数。

该upload_to_instagram()函数使用提供的客户端和将媒体上传到 Instagram upload_params。如果上传成功，它会记录帖子 ID 并使用该handle_post_update()函数更新帖子状态。

如果上传过程中发生错误，它会记录错误并调用handle_post_error()来处理失败。

现在，最后为src/media_post.py文件编写主要函数：

# 👇 insta-cron-post-automation/src/media_post.py

# Rest of the code...

def main() -> None:
    """
    Main function to handle the posting process.

    - Sets up logging.
    - Checks if a post file path is provided and valid.
    - Reads and parses the post file.
    - Validates the image file extension.
    - Prepares upload parameters.
    - Logs the upload parameters and response.
    """

    # Get the current directory of this script
    current_dir = os.path.dirname(os.path.abspath(__file__))

    # Path to the log file, assuming 'logs' is one level up from the current directory
    log_path = os.path.join(current_dir, "..", "logs", "post-activity.log")
    logger = get_logger(log_file=log_path)

    if len(sys.argv) > 1:
        post_path = sys.argv[1]

        # Set up the instagrapi client
        client = setup_instagrapi(logger=logger)

        json_post_content: Dict[str, Any] = parse_post_file_to_json(
            post_path=post_path, logger=logger
        )

        # If the path does not exist or the path is not a file
        if (not os.path.exists(post_path)) or (not os.path.isfile(post_path)):
            return handle_post_error(
                error_message=f"'{post_path}' does not exist or is not a file",
                json_post_content=json_post_content,
                logger=logger,
            )

        image_path = json_post_content["image_path"]

        # Validate image file extension
        if not is_valid_image_extension(image_path):
            return handle_post_error(
                error_message=f"'{image_path}' is not a valid image",
                json_post_content=json_post_content,
                logger=logger,
            )

        upload_params: Dict[str, Any] = prepare_upload_params(
            json_post_content=json_post_content, logger=logger
        )

        # Log the final upload parameters
        logger.info(f"Posting to Instagram with the following details: {upload_params}")

        upload_to_instagram(
            client=client,
            upload_params=upload_params,
            json_post_content=json_post_content,
            logger=logger,
        )

    else:
        log_and_exit(logger=logger, message="Please provide the path to the post file")


if __name__ == "__main__":
  main()
Enter fullscreen mode Exit fullscreen mode

我们首先设置日志记录并验证帖子文件路径是否存在。然后，我们初始化 Instagrapi 客户端并读取帖子文件的内容，检查文件路径和图片扩展名的有效性。

如果检测到任何问题，例如无效的文件路径或不支持的图像类型，我们会将其记录到日志文件中。

验证完成后，该函数将准备上传参数并将其上传到 Instagram。✨

构建 Shell 脚本

🤔为什么需要编写 shell 脚本？

我们将在 Cron 作业中使用一个 shell 脚本来执行media_post.py，因为所有模块都安装在那里，所以我们需要在运行 Python 脚本之前先在虚拟环境中执行 source 命令。如果不需要在虚拟环境中执行 source 命令，我们可以直接将 Python 脚本作为 Cron 作业命令运行，而无需编写这个 shell 脚本。

使用以下代码行在目录run_media_post.sh内创建一个名为的新文件：src/scripts

💁 如果您使用的是 fish shell，您可以在这里run_media_post.fish找到使用 fish 语法的相同代码。在目录中创建一个名为 的新文件src/scripts，然后从链接添加代码。

#!/usr/bin/env bash
# Using this above way of writing shebang can have some security concerns.
# See this stackoverflow thread: https://stackoverflow.com/a/21614603
# Since, I want this script to be portable for most of the users, instead of hardcoding like '#!/usr/bin/bash', I am using this way.

# 👇 insta-cron-post-automation/src/scripts/run_media_post.sh

# Constants for error messages
ERROR_USAGE="ERROR: Usage: bash {media_post_path} {post_file_path}"
ERROR_FILE_NOT_FOUND="ERROR: One or both of the files do not exist or are not valid files."
ERROR_PYTHON_NOT_FOUND="ERROR: No suitable Python executable found."
ERROR_BASH_NOT_INSTALLED="ERROR: Bash shell is not installed. Please install Bash."
ERROR_ACTIVATE_NOT_FOUND="ERROR: activate file not found in '$VENV_DIR/bin'"
ERROR_UNSUPPORTED_SHELL="ERROR: Unsupported shell: '$SHELL'"

# Determine the script directory and virtual environment directory
SCRIPT_DIR="$(dirname "$(realpath "$0")")"
VENV_DIR="$(realpath "$SCRIPT_DIR/../../.venv")"
LOG_FILE="$(realpath "$SCRIPT_DIR/../../logs/shell-error.log")"

log_and_exit() {
  local message="$1"

  echo "[$(date +"%Y-%m-%d %H:%M:%S")] $message" | tee -a $LOG_FILE
  exit 1
}

# Check if both arguments are provided
if [ $# -ne 2 ]; then
  log_and_exit "$ERROR_USAGE"
fi

# Function to check if a file exists and has the correct extension
check_file() {
    local file_path="$1"
    local expected_extension="$2"

    if [ ! -f "$file_path" ]; then
        log_and_exit "$ERROR_FILE_NOT_FOUND"
    fi

    if ! [[ "$file_path" == *".$expected_extension" ]]; then
        log_and_exit "The file '$file_path' must be a '.$expected_extension' file."
    fi
}

# Validate the provided files
check_file "$1" "py"
check_file "$2" "json"

# Extract and validate arguments
MEDIA_POST_PATH="$(realpath "$1")"
POST_FILE_PATH="$(realpath "$2")"

# Find the appropriate Python executable
PYTHON_EXEC="$(command -v python3 || command -v python)"

# Ensure that the Python executable is available before creating the virtual environment
if [ ! -d "$VENV_DIR" ]; then
    if [ -z "$PYTHON_EXEC" ]; then
        log_and_exit "$ERROR_PYTHON_NOT_FOUND"
    fi
    "$PYTHON_EXEC" -m venv "$VENV_DIR"
fi

if ! command -v bash &> /dev/null; then
    log_and_exit "$ERROR_BASH_NOT_INSTALLED"
fi

# Activate the virtual environment based on the shell type
if [[ "$SHELL" == *"/bash" ]]; then
    # Check if the activate file exists before sourcing it
    if [ -f "$VENV_DIR/bin/activate" ]; then
        source "$VENV_DIR/bin/activate"
    else
        log_and_exit "$ERROR_ACTIVATE_NOT_FOUND"
    fi
else
    log_and_exit "$ERROR_UNSUPPORTED_SHELL"
fi

# Set the python executable to the one from the virtual environment
PYTHON_EXEC="$(command -v python)"

"$PYTHON_EXEC" "$MEDIA_POST_PATH" "$POST_FILE_PATH"

# Remove the cron job after running the script
crontab -l | grep -v "$POST_FILE_PATH" | crontab -
Enter fullscreen mode Exit fullscreen mode

该脚本旨在自动执行 Python 脚本media_post.py，该脚本负责使用指定的参数将内容上传到 Instagram，同时确保事先正确设置环境。

它首先检查是否提供了正确数量的参数（两个文件路径），然后验证这些文件是否存在并具有正确的扩展名（Python 脚本为.py ，帖子数据文件为.json ）。

该脚本还会检查系统上是否安装了Python和Bash ，并设置虚拟环境。它仅支持 Bash shell，并会在运行 Python 脚本之前激活虚拟环境。

执行后，脚本通过与命令反向匹配来删除触发其执行的 cron 作业grep。

写入main.py文件

这是我们在填充文件后需要手动运行的唯一 Python 脚本data/to-post.json。

我们将分块编写此文件，并逐步进行解释。main.py在项目根目录中创建一个名为 的新文件，并添加以下代码行：

# 👇 insta-cron-post-automation/main.py
import json
import logging
import os
import secrets
import string
import sys
from datetime import datetime
from os import environ
from typing import Dict, NoReturn

from dateutil import tz

# Add the src directory to the module search path
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "src"))

from crontab import CronTab

from src import logger_config, post_list


def log_and_exit(logger: logging.Logger, message: str) -> NoReturn:
    """
    Log an error message and exit the program.

    Args:
    - logger (logging.Logger): The logger to use.
    - message (str): The error message to log.
    """
    logger.error(message)
    sys.exit(1)


def get_shell_script_to_run(
    user_shell: str, current_dir: str, logger: logging.Logger
) -> str:
    """
    Determine the script to run based on the user's shell.

    Args:
    - user_shell (str): The user's shell.
    - current_dir (str): The current directory of the script.
    - logger (logging.Logger): The logger to use.

    Returns:
    - str: The path to the appropriate shell script for the user's shell.

    Raises:
    - SystemExit: If the user's shell is unsupported.
    """

    shell_script_map: Dict[str, str] = {
        "bash": os.path.join(current_dir, "src", "scripts", "run_media_post.sh"),
        "fish": os.path.join(current_dir, "src", "scripts", "run_media_post.fish"),
    }

    run_media_post_path = shell_script_map.get(user_shell, None)
    if run_media_post_path is None:
        log_and_exit(logger=logger, message=f"Unsupported shell: {user_shell}")

    return run_media_post_path
Enter fullscreen mode Exit fullscreen mode

👀 请注意，我们正在使用该方法将路径插入到我们的src目录中sys.path.insert()，以确保 Python 可以从src目录中找到并导入模块。

该log_and_exit()函数与之前相同——如果出现错误，它会记录错误并退出程序。该get_shell_script_to_run()函数会根据用户的 shell 是Bash还是Fish返回应在 cron 作业中运行的 shell 脚本的路径。如果用户的 shell 不是其中之一，程序将退出。

现在，让我们添加一个辅助函数来验证发布日期并使用提供的参数添加一个 cron 作业。

# 👇 insta-cron-post-automation/main.py

# Rest of the code...

def validate_post_date(post_date: str, logger: logging.Logger) -> datetime:
    """
    Validate the post date to ensure it is in the future.

    Args:
    - post_date (string): The date and time of the post.
    - logger (logging.Logger): The logger to use.

    Returns:
    - datetime: The validated and parsed datetime object.

    Raises:
    - SystemExit: If the post date is not valid or not in the future.
    """

    # Define the expected format for parsing
    date_format = "%Y-%m-%d %H:%M"

    try:
        # Attempt to parse the post_date string into a datetime object
        parsed_date = datetime.strptime(post_date, date_format)
    except ValueError:
        log_and_exit(
            logger=logger,
            message=f"The post_date is not in the correct format: {post_date}",
        )

    # Check if the parsed date is in the future
    if parsed_date.astimezone(tz.UTC) <= datetime.now(tz=tz.UTC):
        log_and_exit(
            logger=logger, message=f"The post_date `{post_date}` is in the past."
        )

    return parsed_date


def create_cron_job(
    cron: CronTab,
    user_shell: str,
    run_media_post_path: str,
    media_post_path: str,
    scheduled_post_file_path: str,
    post_date: datetime,
    logger: logging.Logger,
) -> None:
    """
    Create a cron job for a scheduled post.

    Args:
    - cron (CronTab): The crontab object for the current user.
    - user_shell (str): The user's shell.
    - run_media_post_path (str): The path to the shell script to run.
    - media_post_path (str): The path to the media post script.
    - scheduled_post_file_path (str): The path to the scheduled post file.
    - post_date (datetime): The date and time to run the job.
    - logger (logging.Logger): The logger to use.

    Raises:
    - SystemExit: If the cron job creation fails.
    """
    try:
        # Conditionally add semicolon
        command = (
            f"SHELL=$(command -v {user_shell})"
            + (";" if user_shell == "bash" else "")
            + f" {user_shell} {run_media_post_path} {media_post_path} {scheduled_post_file_path}"
        )
        job = cron.new(command=command)
        job.setall(post_date.strftime("%M %H %d %m *"))
    except Exception as e:
        log_and_exit(logger=logger, message=f"Failed to create cron job: {e}")
Enter fullscreen mode Exit fullscreen mode

该validate_post_date()函数检查日期时间字符串是否符合预期格式（不带秒数），并确保 Instagram 的指定发布日期不是过去的日期。

该create_cron_job()函数接受一个已配置的Crontab对象、shell 脚本的路径、 的路径media_post.py以及包含计划发布内容的文件的路径。然后，它会创建一个 cron 作业，并将SHELL 变量设置为用户的 shell（因为 cron 环境可能使用与当前用户不同的 shell），并安排该作业在指定时间执行。

如果在 cron 作业调度期间发生任何异常，该函数将记录错误并退出程序。

现在，是时候编写负责设置一切的主要函数了：

# 👇 insta-cron-post-automation/main.py

# Rest of the code...

def main() -> None:
    """
    Main function to schedule Instagram posts using cron jobs.

    This function performs the following tasks:
    1. Sets up logging to a file.
    2. Loads a list of posts from a JSON file.
    3. Creates a temporary JSON file for each post to be scheduled.
    4. Schedules a cron job to execute a script for each post at the specified date and time.
    5. Writes the cron jobs to the user's crontab.

    The cron job will execute the script `media_post.py` with the path to the temporary JSON file as an argument.
    """

    # Determine the current directory of the script
    current_dir = os.path.dirname(os.path.abspath(__file__))

    # Define paths for log file and posts JSON file
    log_path = os.path.join(current_dir, "logs", "post-activity.log")
    to_post_path = os.path.join(current_dir, "data", "to-post.json")
    media_post_path = os.path.join(current_dir, "src", "media_post.py")

    # Initialize logger
    logger = logger_config.get_logger(log_file=log_path)

    post_data_dir = os.path.join(current_dir, "data", "scheduled_posts")
    os.makedirs(post_data_dir, exist_ok=True)

    # Initialize PostList object and load posts from JSON file
    posts_list = post_list.PostList(log_path)

    posts_list.get_posts_from_json_file(posts_file_path=to_post_path)
    logger.info(f"Number of posts loaded: {len(posts_list.posts)}")

    user_shell = os.path.basename(environ.get("SHELL", "/bin/bash"))
    run_media_post_path = get_shell_script_to_run(
        user_shell=user_shell, current_dir=current_dir, logger=logger
    )

    # Access the current user's CronTab object.
    cron = CronTab(user=True)

    for post in posts_list.posts:
        # Create a unique identifier for each post file
        unique_id = "".join(
            secrets.choice(string.ascii_lowercase + string.digits) for _ in range(6)
        )

        post.post_date = validate_post_date(post_date=post.post_date, logger=logger)

        # Create a unique suffix for the temporary file based on the post date
        post_date_suffix = post.post_date.strftime("%Y-%m-%d-%H-%M")

        scheduled_post_file_path = os.path.join(
            post_data_dir, f"insta_post_{unique_id}_{post_date_suffix}.json"
        )

        # Write the post data to the temporary file
        try:
            with open(scheduled_post_file_path, "w") as f:
                json.dump(post.serialize(), f, default=str)
        except (IOError, json.JSONDecodeError) as e:
            log_and_exit(logger=logger, message=f"Failed to write post file: {e}")

        # Create a new cron job to run the Instagram post script with the temp file as an argument
        create_cron_job(
            cron=cron,
            user_shell=user_shell,
            run_media_post_path=run_media_post_path,
            media_post_path=media_post_path,
            scheduled_post_file_path=scheduled_post_file_path,
            post_date=post.post_date,
            logger=logger,
        )

    # Write the cron jobs to the user's crontab
    try:
        cron.write()
        logger.info(f"Cronjob added to the CronTab for the current user: {cron.user}")
    except Exception as e:
        log_and_exit(logger=logger, message=f"Failed to write to CronTab: {e}")


if __name__ == "__main__":
    main()
Enter fullscreen mode Exit fullscreen mode

该main()函数使用 cron 作业为 Instagram 帖子设置调度系统。它首先配置日志记录并从 JSON 文件 ( data/to-post.json) 加载帖子列表。对于每篇帖子，它会在data/scheduled-posts目录中创建一个包含帖子内容的 JSON 文件，并安排一个 cron 作业在指定的日期和时间运行处理帖子的脚本。

它还会确定用户的 shell 并设置要执行的适当脚本。在创建唯一的临时文件并安排好作业后，它会将所有 cron 作业写入用户的 crontab。如果在此过程中发生任何错误，则会记录这些错误并退出程序。

测试程序🧪

如果你对这个程序的工作原理感兴趣，我准备了一个名为 的示例脚本，populate_sample_posts.py它会将data/to-post.json一个示例帖子（包括描述、发布日期和图片）填充到文件中。你可以在这里找到它。

填充data/to-post.json文件并进入虚拟环境后，运行以下命令：

python3 main.py
Enter fullscreen mode Exit fullscreen mode

建议先用新的 Instagram 账号测试一下，然后再用主账号使用。满意后，就可以安排自己的 Instagram 帖子了！😉

免责声明⚠️
此脚本使用 Cron 作业，因此只有在系统运行时才能安排您的帖子发布。因此，最好在几乎全天候在线的云端虚拟机上运行它。

总结！⚡

哇，😮‍💨 这真是一段奇妙的旅程！如果你已经走到这一步，给自己一个当之无愧的鼓励吧！现在，你已经成功构建了一个 Python 应用程序，可以使用 Cron 作业自动在 Instagram 上发帖。🤯

这一定是你用 Python 构建的最酷、最独特的脚本之一。

而且我很确定这不是你能在互联网上轻易找到的东西。🥱

本文的完整源代码可以在这里找到：

https://github.com/shricodev/insta-cron-post-automation

非常感谢你的阅读！🎉 🫡

在下面的评论部分写下你的想法。👇

在社交媒体上关注我🐥

shricodev 图像

什里贾尔·阿查里亚

全栈 SDE • 开源贡献者 • 合作者 @Oppia • 协作邮件
文章来源：https://dev.to/shricodev/automate-your-instagram-posts-like-a-pro-with-cron-jobs-3idb
Devto 翻译
PREV
✨使用这些工具成为 10X Linux 用户😎💫
NEXT
30 个图标数据集和更多前端资源