簡介✨

在這個簡單易懂的教學中，您將學習如何使用 cron jobs從頭開始建立自己的 Instagram 自動化工具。 😎

您將學到什麼：👀

• 了解如何在 Python 專案中設定日誌記錄。

• 學習使用python-crontab模組在基於 Unix 的作業系統中新增 cron jobs。

• 了解如何使用instagrapi模組在 Instagram 上發佈。

那麼，您準備好建立最酷的 Instagram 自動化工具了嗎？ 😉

設定環境⚙️

在我們深入建立專案之前，先看一下專案架構以簡要了解佈局。

💁 我們將從頭開始建立這個專案，使其具備日誌記錄支援以及類別和函數中結構化的所有內容，使其可以投入生產。

初始化專案🛠️

建立一個資料夾來保存專案的所有原始程式碼：

mkdir insta-cron-post-automation
cd insta-cron-post-automation

建立一些新的子資料夾，我們將在其中儲存發布資料、日誌和 shell 腳本：

mkdir -p data logs src/scripts

現在初始資料夾結構已經設定完畢，是時候建立一個新的虛擬環境並安裝我們將在專案中使用的所有模組了。

執行以下命令在專案的根目錄中建立並啟動一個新的虛擬環境：

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

執行此命令來安裝我們將在專案中使用的所有必要模組：

pip3 install instagrapi python-crontab python-dotenv lorem numpy pillow

以下是每個模組的用途：

• instagrapi ：登入並發佈到 Instagram。

• python-crontab ：建立和編輯使用者的 crontable。

• python-dotenv ：從.env檔中讀取環境變數。

可選模組

• lorem ：產生用於建立範例貼文的虛擬描述。

• numpy ：產生隨機像素資料，用於為我們的範例貼文建立圖像。

• pillow ：使用 NumPy 中的像素資料建立樣本影像。

讓我們來編碼吧💻

設定日誌記錄 📋

💡 由於我們的工具在使用者透過 cron jobs提供的特定時間執行，因此我們不能依賴 print 語句來記錄輸出。一切都在背景發生，因此我們需要一個中心位置來查看程式的日誌，例如日誌檔案。

對於日誌記錄支持，我們將使用我們的舊 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

因此，這個get_logger()函數採用日誌檔案的路徑，它需要在其中儲存所有日誌，而不是將它們記錄到控制台。然後它建立一個記錄器實例並傳回它。

現在，設定此函數後，我們可以在專案中的任何位置呼叫它，並且它將維護相同的日誌記錄配置。 😎

實作 Instagram 登入 🔑

使用以下程式碼在src目錄中建立一個名為setup.py的新檔案：

# 👇 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

get_credentials()函數讀取使用者環境變數並傳回它們。如您可能已經猜到的，該函數要求您設定INSTA_USERNAME和INSTA_PASSWORD環境變數。

在專案的根目錄中建立一個新的.env文件，並定義這兩個變數。

INSTA_USERNAME=<your-insta-username>
INSTA_PASSWORD=<your-insta-password>

setup_instagrapi()函數會建立一個新的 Instagrapi 用戶端、登入 Instagram 並傳回客戶端。

定義類別🧩

我們將設定兩個不同的類別： Post和PostList 。 PostList類別將保存多個Post物件。

使用以下程式碼在src目錄中建立一個名為post.py的新檔案：

# 👇 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

該類別採用一些參數，例如帖子的描述、圖像路徑、帖子日期和可選的額外字段屬性，該屬性可用於傳遞帖子的附加元資料，如下所示：

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

這裡，二進位值 1 和 0 分別代表True和False 。

validate_extra_data()方法檢查提供的extra_data欄位是否僅包含有效鍵並刪除使用者提供的任何其他鍵。

serialize()方法檢查extra_data參數是否已傳遞給建構子。如果是，則將其加入到字典中並返回字典；否則，它會傳回不帶extra_data鍵的字典。

現在Post類別已準備就緒，讓我們建立另一個類別PostList來保存Post物件。

在src目錄中建立一個名為post_list.py的新檔案並新增以下程式碼行：

# 👇 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

_log_and_exit()方法，顧名思義，是一個私有方法，它將訊息記錄到檔案並退出程式。

to_json()方法，顧名思義，以 JSON 字串的形式傳回貼文清單。

parse_post_date()方法採用post_date變數並傳回字串格式的日期，不帶秒部分，因為我們在 cron jobs中不需要秒。

get_posts_from_json_file()方法讀取 JSON 文件，將每個貼文作為Post物件填入 posts 陣列，並處理讀取文件內容時可能發生的各種異常。

編寫媒體貼文腳本📝

現在我們已經設定了所有類，是時候編寫負責在 Instagram 上發布的主要 Python 腳本了。

在src目錄中建立一個名為media_post.py的新檔案。這個檔案會很長，所以我們將把程式碼分成每個函數，我會一路解釋它。

# 👇 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)

這些都是相當簡單的函數。 log_and_exit()函數將訊息記錄到檔案中並退出程式。

is_valid_image_extension()函數檢查圖片是否具有允許在 Instagram 上發布的有效副檔名。

💁 我不完全確定是否允許其他擴展，但這些似乎是標準擴展。如果有其他擴展，請隨時進行相應更新。

一旦我們嘗試將帖子上傳到 Instagram，我們需要將其從data目錄中的to-post.json檔案中刪除，我們在其中加入了我們想要安排的所有帖子。無論上傳是否成功，我們都會將貼文新增至data目錄內的error.json或success.json檔案。

建立一個處理此過程的新函數。

# 👇 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)

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)

parse_post_file_to_json()函數取得 JSON 檔案的路徑並嘗試將其內容解析為 JSON。如果解析失敗，則使用handle_invalid_post_file()函數來處理失敗。它將 success 布林值設為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,
        )

prepare_upload_params()函數取得貼文內容並準備上傳參數。它包括對extra_data字段的明確驗證，以確保所有鍵都屬於預期類型，並最終返回整個上傳參數集。

upload_to_instagram()函數使用提供的客戶端和upload_params將媒體上傳到 Instagram。如果上傳成功，它會記錄帖子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()

我們首先設定日誌記錄並驗證帖子檔案路徑是否存在。然後，我們初始化 Instagrapi 用戶端並讀取貼文件案的內容，檢查檔案路徑和圖片副檔名的有效性。

如果偵測到任何問題，例如無效的檔案路徑或不受支援的影像類型，我們會將其記錄到日誌檔案中。

驗證完成後，函數會準備上傳參數並將其上傳到 Instagram。 ✨

建置 Shell 腳本 🧰

🤔為什麼需要寫shell腳本？

我們將在 Cron 作業中使用 shell 腳本來執行media_post.py ，因為我們需要在執行 Python 腳本之前取得虛擬環境，因為所有模組都安裝在那裡。如果我們不需要取得虛擬環境，我們可以直接將 Python 腳本作為 Cron 作業命令執行，而無需編寫此 shell 腳本。

使用下列程式碼行在src/scripts目錄中建立一個名為run_media_post.sh的新檔案：

💁 如果您使用的是fish shell，您可以在這裡找到與fish語法相同的程式碼。在src/scripts目錄中建立一個名為run_media_post.fish的新文件，並加入連結中的程式碼。

#!/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 -

該腳本旨在自動執行 Python 腳本media_post.py ，該腳本負責使用指定參數將內容上傳到 Instagram，同時確保事先正確設定環境。

它首先檢查是否提供了正確數量的參數（兩個檔案路徑），然後驗證這些檔案是否存在並具有正確的副檔名（Python 腳本為.py ，post 資料檔為.json ）。

該腳本還會檢查系統上是否安裝了Python和Bash並設定虛擬環境。它僅支援 Bash shell，並且會在執行 Python 腳本之前啟動虛擬環境。

執行後，該腳本透過與grep命令進行反向匹配來刪除觸發其執行的 cron jobs。

寫main.py檔 🧑‍💻

這是填充data/to-post.json檔案後我們需要手動執行的唯一 Python 腳本。

我們將分塊編寫此文件並一路解釋它。在專案的根目錄中建立一個名為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

👀 請注意，我們使用sys.path.insert()方法插入src目錄的路徑，以確保 Python 可以從src目錄找到並匯入模組。

log_and_exit()函數與以前相同 - 如果出現問題，它會記錄錯誤並退出程式。 get_shell_script_to_run()函數根據使用者的 shell 是Bash還是Fish ，傳回應在 cron jobs中執行的 shell 腳本的路徑。如果使用者的 shell 不是其中之一，程式將會退出。

現在，讓我們新增一個輔助函數來驗證發布日期，並使用提供的參數新增一個 cron jobs。

# 👇 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}")

validate_post_date()函數檢查日期時間字串是否採用預期格式（不含秒），並確保指定的 Instagram 發布日期不是過去的日期。

create_cron_job()函數採用配置的Crontab物件、shell 腳本的路徑、 media_post.py的路徑以及包含計畫發佈內容的檔案的路徑。然後，它會建立一個 cron jobs，並將SHELL 變數設為使用者的 shell（因為 cron 環境可能使用與目前使用者的 shell 不同的 shell），並安排該作業在指定時間執行。

如果在 cron jobs的排程過程中發生任何異常，則函數會記錄錯誤並退出程式。

現在，是時候編寫負責設定所有內容的主函數了：

# 👇 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()

main()函數使用 cron jobs為 Instagram 貼文設定一個排程系統。首先配置日誌記錄並從 JSON 檔案 ( data/to-post.json ) 載入貼文清單。對於每個帖子，它都會在data/scheduled-posts目錄中建立一個包含帖子內容的 JSON 文件，並安排一個 cron jobs來執行處理在指定日期和時間發布的腳本。

它還確定使用者的 shell 並設定要執行的適當腳本。建立唯一的暫存檔案並排程作業後，它將所有 cron jobs寫入使用者的 crontab。如果在這些過程中發生任何錯誤，它們都會被記錄下來，並且程式會退出。

測試程序🧪

如果您對這個程式的工作原理感到好奇，我準備了一個名為populate_sample_posts.py的範例腳本，它將使用範例貼文填充data/to-post.json文件，包括描述、發布日期和圖像。您可以在這裡找到它。

填充data/to-post.json檔案並進入虛擬環境後，執行以下命令：

python3 main.py

建議先使用新的 Instagram 帳戶進行測試，然後再與主帳戶一起使用。一旦您感到滿意，就可以安排您自己的 Instagram 貼文了！ 😉

免責聲明⚠️

該腳本使用 Cron 作業，因此只有在系統執行時它才能安排您的帖子。因此，最好在幾乎 24/7 在線的基於雲端的 VM 上執行它。

包起來！ ⚡

哇哦，😮‍💨這是一趟多麼美好的旅程啊！如果您已經做到了這一步，請給自己一個當之無愧的鼓勵。到目前為止，您已經成功建立了一個 Python 應用程式來使用 Cron 作業自動執行 Instagram 發布。 🤯

這一定是您使用 Python 建置的最酷、最獨特的腳本之一。

我很確定這不是您可以在網路上輕鬆找到的東西。 🥱

本文的完整記錄原始碼可在此處取得：

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

非常感謝您的閱讀！ 🎉🫡

在下面的評論部分寫下你的想法。 👇

https://linktr.ee/shricodev

在社群上關注我 🐥

https://dev.to/shricodev

原文出處：https://dev.to/shricodev/automate-your-instagram-posts-like-a-pro-with-cron-jobs-3idb