diff --git a/README.md b/README.md index 991acd728..922ad45da 100644 --- a/README.md +++ b/README.md @@ -24,7 +24,7 @@ MPV2 (MoneyPrinter Version 2) is, as the name suggests, the second version of th ## Features - [x] Twitter Bot (with CRON Jobs => `scheduler`) -- [x] YouTube Shorts Automator (with CRON Jobs => `scheduler`) +- [x] PostBridge-first Video Publishing (with CRON Jobs => `scheduler`) - [x] Affiliate Marketing (Amazon + Twitter) - [x] Find local businesses & cold outreach @@ -75,7 +75,7 @@ All relevant documents can be found [here](docs/). For easier usage, there are some scripts in the `scripts` directory that can be used to directly access the core functionality of MPV2 without the need for user interaction. -All scripts need to be run from the root directory of the project, e.g. `bash scripts/upload_video.sh`. +All scripts need to be run from the root directory of the project, e.g. `bash scripts/publish_video.sh`. ## Contributing diff --git a/config.example.json b/config.example.json index ac6d9b9c3..fd25c479a 100644 --- a/config.example.json +++ b/config.example.json @@ -2,6 +2,11 @@ "verbose": true, "firefox_profile": "", "headless": false, + "video_publishing": { + "profile_name": "Default Publisher", + "niche": "", + "language": "English" + }, "ollama_base_url": "http://127.0.0.1:11434", "ollama_model": "", "twitter_language": "English", @@ -35,8 +40,8 @@ "post_bridge": { "enabled": false, "api_key": "", - "platforms": ["tiktok", "instagram"], + "platforms": ["youtube", "tiktok", "instagram"], "account_ids": [], - "auto_crosspost": false + "auto_publish": false } } diff --git a/docs/Configuration.md b/docs/Configuration.md index 6901e922b..4c452c603 100644 --- a/docs/Configuration.md +++ b/docs/Configuration.md @@ -5,8 +5,12 @@ All your configurations will be in a file in the root directory, called `config. ## Values - `verbose`: `boolean` - If `true`, the application will print out more information. -- `firefox_profile`: `string` - The path to your Firefox profile. This is used to use your Social Media Accounts without having to log in every time you run the application. +- `firefox_profile`: `string` - The path to your Firefox profile. This is only needed for browser-based Twitter automation. - `headless`: `boolean` - If `true`, the application will run in headless mode. This means that the browser will not be visible. +- `video_publishing`: `object`: + - `profile_name`: `string` - Friendly label for the configured publisher profile. + - `niche`: `string` - The topic or niche used for generated videos. + - `language`: `string` - The language used for generated videos. - `ollama_base_url`: `string` - Base URL of your local Ollama server (default: `http://127.0.0.1:11434`). - `ollama_model`: `string` - Ollama model to use for text generation (e.g. `llama3.2:3b`). If empty, the app queries Ollama at startup and lets you pick from the available models interactively. - `twitter_language`: `string` - The language that will be used to generate & post tweets. @@ -39,11 +43,11 @@ All your configurations will be in a file in the root directory, called `config. - `imagemagick_path`: `string` - The path to the ImageMagick binary. This is used by MoviePy to manipulate images. Install ImageMagick from [here](https://imagemagick.org/script/download.php) and set the path to the `magick.exe` on Windows, or on Linux/MacOS the path to `convert` (usually /usr/bin/convert). - `script_sentence_length`: `number` - The number of sentences in the generated video script (default: `4`). - `post_bridge`: `object`: - - `enabled`: `boolean` - Enables Post Bridge cross-posting after successful YouTube uploads. + - `enabled`: `boolean` - Enables Post Bridge as the primary video publishing backend. - `api_key`: `string` - Your Post Bridge API key. If empty, MPV2 falls back to `POST_BRIDGE_API_KEY`. - - `platforms`: `string[]` - Platforms to target. Supported values in v1 are `tiktok` and `instagram`. - - `account_ids`: `number[]` - Optional fixed Post Bridge account IDs to avoid account-selection prompts. - - `auto_crosspost`: `boolean` - If `true`, cross-post automatically after a successful YouTube upload. If `false`, interactive runs ask and cron runs skip. + - `platforms`: `string[]` - Platforms to target. Supported values include `youtube`, `tiktok`, `instagram`, `facebook`, `twitter`, `threads`, `linkedin`, `bluesky`, and `pinterest`. + - `account_ids`: `number[]` - Fixed Post Bridge account IDs. The setup wizard stores one account ID per selected platform. + - `auto_publish`: `boolean` - If `true`, generated videos are published automatically. If `false`, interactive runs ask and cron runs skip. ## Example @@ -52,6 +56,11 @@ All your configurations will be in a file in the root directory, called `config. "verbose": true, "firefox_profile": "", "headless": false, + "video_publishing": { + "profile_name": "Default Publisher", + "niche": "", + "language": "English" + }, "ollama_base_url": "http://127.0.0.1:11434", "ollama_model": "", "twitter_language": "English", @@ -85,9 +94,9 @@ All your configurations will be in a file in the root directory, called `config. "post_bridge": { "enabled": false, "api_key": "", - "platforms": ["tiktok", "instagram"], + "platforms": ["youtube", "tiktok", "instagram"], "account_ids": [], - "auto_crosspost": false + "auto_publish": false } } ``` diff --git a/docs/PostBridge.md b/docs/PostBridge.md index b1b412182..6604baaf2 100644 --- a/docs/PostBridge.md +++ b/docs/PostBridge.md @@ -1,33 +1,39 @@ -# Post Bridge Integration +# Post Bridge Video Publishing -MoneyPrinterV2 can optionally hand off a successfully uploaded YouTube Short to [Post Bridge](https://api.post-bridge.com/reference), which then publishes the same asset to connected TikTok and Instagram accounts. +MoneyPrinterV2 now uses [Post Bridge](https://api.post-bridge.com/reference) as the primary backend for publishing generated videos. ## What Post Bridge Does -Post Bridge is a publishing API for social platforms. In this integration, MoneyPrinterV2 uses it to: +In the current flow, MoneyPrinterV2 still generates the video locally, but Post Bridge handles publishing: -1. Look up your connected social accounts. -2. Request a signed upload URL for the generated video. -3. Upload the video asset to Post Bridge storage. -4. Create a post for the selected TikTok and Instagram accounts. +1. Fetch the connected social accounts that should receive the video. +2. Request a signed upload URL for the generated media. +3. Upload the local video asset to Post Bridge storage. +4. Create a post for the selected platform accounts. +5. Expose publish history and per-platform URLs through the Post Bridge API. -MoneyPrinterV2 still owns video generation and the initial YouTube upload. Post Bridge only starts after YouTube upload succeeds. +This replaces the old browser-driven YouTube upload path for normal usage. ## Setup 1. Create a Post Bridge account. -2. Connect the TikTok and Instagram accounts you want to publish to. +2. Connect the social accounts you want to publish to. 3. Generate an API key from Post Bridge. -4. Add the `post_bridge` block to `config.json`, or set `POST_BRIDGE_API_KEY` in your environment. +4. Run the in-app publisher setup wizard, or configure `video_publishing` and `post_bridge` manually in `config.json`. ```json { + "video_publishing": { + "profile_name": "Default Publisher", + "niche": "finance", + "language": "English" + }, "post_bridge": { "enabled": true, "api_key": "pb_your_api_key_here", - "platforms": ["tiktok", "instagram"], - "account_ids": [], - "auto_crosspost": false + "platforms": ["youtube", "tiktok", "instagram"], + "account_ids": [101, 202, 303], + "auto_publish": false } } ``` @@ -36,45 +42,45 @@ MoneyPrinterV2 still owns video generation and the initial YouTube upload. Post | Key | Type | Default | Description | | --- | --- | --- | --- | -| `enabled` | `boolean` | `false` | Enables the Post Bridge integration. | -| `api_key` | `string` | `""` | Post Bridge API key. Falls back to `POST_BRIDGE_API_KEY` when blank. | -| `platforms` | `string[]` | `["tiktok", "instagram"]` when omitted | Platform filters used when looking up connected accounts. Unsupported values inside the list are ignored. | -| `account_ids` | `number[]` | `[]` | Exact Post Bridge account IDs to post to. When provided, MoneyPrinterV2 uses these directly and skips account lookup. | -| `auto_crosspost` | `boolean` | `false` | Automatically cross-post after a successful YouTube upload. | +| `video_publishing.profile_name` | `string` | `"Default Publisher"` | Friendly label shown in the CLI wizard. | +| `video_publishing.niche` | `string` | `""` | Topic or niche used for generated videos. | +| `video_publishing.language` | `string` | `"English"` | Language used for generated videos. | +| `post_bridge.enabled` | `boolean` | `false` | Enables Post Bridge publishing. | +| `post_bridge.api_key` | `string` | `""` | Post Bridge API key. Falls back to `POST_BRIDGE_API_KEY` when blank. | +| `post_bridge.platforms` | `string[]` | `["youtube", "tiktok", "instagram"]` when omitted | Platforms targeted for publishing. | +| `post_bridge.account_ids` | `number[]` | `[]` | Exact Post Bridge account IDs to publish to. The setup wizard stores one account per selected platform. | +| `post_bridge.auto_publish` | `boolean` | `false` | Automatically publish after generation. Interactive runs prompt when disabled; cron runs skip. | -## How The Integration Works +## Publish Flow -### Interactive YouTube flow +### Interactive publishing -- If `enabled` is `false`, nothing happens. -- If `enabled` is `true` and `auto_crosspost` is `false`, MoneyPrinterV2 asks whether to cross-post after a successful YouTube upload. -- If `account_ids` is configured, those IDs are used directly. -- If `account_ids` is empty, MoneyPrinterV2 fetches connected Post Bridge accounts for the configured platforms. -- If there is exactly one connected account for a platform, it is selected automatically. -- If there are multiple connected accounts for a platform, MoneyPrinterV2 prompts you to choose one. -- After interactive selection, the chosen IDs are printed so you can copy them into `config.json`, but the app does not edit your config file for you. +- Use the `Video Publishing` menu in `src/main.py`. +- `Setup Publisher` runs the config-writing wizard. +- `Publish Video` generates a video locally and publishes it through Post Bridge. +- `Show Recent Publishes` fetches recent posts and post results live from the API. -### Cron / scheduled uploads +### Scheduled publishing -- Cron uses the same integration after a successful YouTube upload. -- If `auto_crosspost` is `false`, cron skips Post Bridge and logs why. -- If `auto_crosspost` is `true`, cron cross-posts automatically. -- If `account_ids` is empty and multiple connected accounts exist for a platform, cron skips cross-posting instead of hanging on an interactive prompt. +- Cron now uses `publish` mode instead of `youtube`. +- `scripts/publish_video.sh` runs the publish cron entrypoint directly. +- If `post_bridge.auto_publish` is `false`, cron skips publishing and logs why. -## Current v1 Behavior +## Content Mapping -- The generated YouTube title is used as the default caption. -- TikTok receives the YouTube title as its platform-specific `title` override. -- Post Bridge account lookup follows the API’s pagination. -- Instagram cover-image customization is intentionally not included in this v1 integration. -- Cross-posting only runs after `upload_video()` returns success. +- Global caption: generated description +- `youtube.title`: generated title +- `youtube.caption`: generated description +- `tiktok.title`: generated title + +Only configured platform overrides are sent. ## Troubleshooting | Issue | What to check | | --- | --- | -| Cross-post prompt never appears | Verify `post_bridge.enabled` is `true`. | -| Cross-post is skipped in cron | Set `auto_crosspost` to `true`. | -| No accounts are found | Make sure the accounts are connected in Post Bridge and that `platforms` matches the accounts you connected. | -| Cron skips because multiple accounts exist | Add the desired `account_ids` to `config.json` so cron does not need to prompt. | -| API key seems ignored | Set `post_bridge.api_key`, or leave it blank and export `POST_BRIDGE_API_KEY`. | +| The publisher wizard cannot continue | Verify `video_publishing.niche` and a Post Bridge API key are set. | +| No accounts are found | Make sure the accounts are connected in Post Bridge and that `post_bridge.platforms` matches the connected accounts. | +| Cron skips publishing | Set `post_bridge.auto_publish` to `true`. | +| Publish history is empty | Confirm that posts exist for the configured platforms and that the API key has access. | +| Old cron commands stopped working | Use `publish` mode instead of `youtube`. | diff --git a/docs/YouTube.md b/docs/YouTube.md index fdd4ff89d..498c870fa 100644 --- a/docs/YouTube.md +++ b/docs/YouTube.md @@ -1,12 +1,14 @@ -# YouTube Shorts Automater +# Legacy YouTube Automation -MPV2 uses a similar implementation of V1 (see [MPV1](https://github.com/FujiwaraChoki/MoneyPrinter)), to generate Video-Files and upload them to YouTube Shorts. +This document describes the older browser-driven YouTube workflow. -In contrast to V1, V2 uses AI generated images as the visuals for the video, instead of using stock footage. This makes the videos more unique and less likely to be flagged by YouTube. V2 also supports music right from the get-go. +MoneyPrinterV2 now uses Post Bridge as the primary publishing backend for generated videos. See [PostBridge.md](./PostBridge.md) for the current flow. + +The legacy implementation used a Firefox profile plus Selenium to upload videos directly to YouTube Shorts after generation. ## Relevant Configuration -In your `config.json`, you need the following attributes filled out, so that the bot can function correctly. +If you are still experimenting with the legacy code path, you need the following attributes filled out: ```json { diff --git a/scripts/preflight_local.py b/scripts/preflight_local.py index dcc46b26c..0de93828b 100755 --- a/scripts/preflight_local.py +++ b/scripts/preflight_local.py @@ -61,7 +61,40 @@ def main() -> int: else: warn(f"firefox_profile does not exist: {firefox_profile}") else: - warn("firefox_profile is empty. Twitter/YouTube automation requires this.") + warn("firefox_profile is empty. Twitter automation requires this.") + + video_cfg = cfg.get("video_publishing", {}) + if isinstance(video_cfg, dict): + niche = str(video_cfg.get("niche", "")).strip() + language = str(video_cfg.get("language", "")).strip() + if niche: + ok(f"video_publishing.niche is set: {niche}") + else: + warn("video_publishing.niche is empty. Video publishing setup is incomplete.") + if language: + ok(f"video_publishing.language is set: {language}") + else: + warn("video_publishing.language is empty. Defaulting to English at runtime.") + else: + warn("video_publishing config block is missing or invalid.") + + post_bridge = cfg.get("post_bridge", {}) + if isinstance(post_bridge, dict) and post_bridge.get("enabled"): + api_key = str(post_bridge.get("api_key", "")).strip() or os.environ.get( + "POST_BRIDGE_API_KEY", + "", + ).strip() + if api_key: + ok("Post Bridge API key is set") + else: + fail("Post Bridge is enabled but no API key is configured") + failures += 1 + + platforms = post_bridge.get("platforms", []) + if platforms: + ok(f"Post Bridge platforms configured: {', '.join(platforms)}") + else: + warn("Post Bridge is enabled but no platforms are configured.") # Ollama (LLM) base = str(cfg.get("ollama_base_url", "http://127.0.0.1:11434")).rstrip("/") diff --git a/scripts/publish_video.sh b/scripts/publish_video.sh new file mode 100755 index 000000000..a4a24deb0 --- /dev/null +++ b/scripts/publish_video.sh @@ -0,0 +1,11 @@ +#!/bin/bash + +# Script to generate and publish a video via Post Bridge. + +if [ -x "$(command -v python3)" ]; then + PYTHON=python3 +else + PYTHON=python +fi + +"$PYTHON" src/cron.py publish diff --git a/scripts/setup_local.sh b/scripts/setup_local.sh index cacaf758e..d3bfff4ed 100755 --- a/scripts/setup_local.sh +++ b/scripts/setup_local.sh @@ -59,6 +59,19 @@ cfg.setdefault("whisper_model", "base") cfg.setdefault("whisper_device", "auto") cfg.setdefault("whisper_compute_type", "int8") +video_publishing = cfg.setdefault("video_publishing", {}) +video_publishing.setdefault("profile_name", "Default Publisher") +video_publishing.setdefault("niche", "") +video_publishing.setdefault("language", "English") + +post_bridge = cfg.setdefault("post_bridge", {}) +post_bridge.setdefault("enabled", False) +post_bridge.setdefault("api_key", "") +post_bridge.setdefault("platforms", ["youtube", "tiktok", "instagram"]) +post_bridge.setdefault("account_ids", []) +if "auto_publish" not in post_bridge: + post_bridge["auto_publish"] = bool(post_bridge.pop("auto_crosspost", False)) + magick_path = os.environ.get("MAGICK_PATH", "") if magick_path: cfg["imagemagick_path"] = magick_path @@ -109,6 +122,7 @@ print(f"[setup] Updated {cfg_path}") print(f"[setup] llm_provider={cfg.get('llm_provider')} model={cfg.get('ollama_model')}") print(f"[setup] image_provider={cfg.get('image_provider')}") print(f"[setup] stt_provider={cfg.get('stt_provider')}") +print(f"[setup] video_publishing platforms={post_bridge.get('platforms')}") PY echo "[setup] Running local preflight..." diff --git a/scripts/upload_video.sh b/scripts/upload_video.sh old mode 100644 new mode 100755 index a57b855f9..80bffe9d9 --- a/scripts/upload_video.sh +++ b/scripts/upload_video.sh @@ -1,34 +1,5 @@ #!/bin/bash -# Script to generate & Upload a video to YT Shorts - -# Check which interpreter to use (python) -if [ -x "$(command -v python3)" ]; then - PYTHON=python3 -else - PYTHON=python -fi - -# Read .mp/youtube.json file, loop through accounts array, get each id and print every id -youtube_ids=$($PYTHON -c "import json; print('\n'.join([account['id'] for account in json.load(open('.mp/youtube.json'))['accounts']]))") - -echo "What account do you want to upload the video to?" - -# Print the ids -for id in $youtube_ids; do - echo $id -done - -# Ask for the id -read -p "Enter the id: " id - -# Check if the id is in the list -if [[ " ${youtube_ids[@]} " =~ " ${id} " ]]; then - echo "ID found" -else - echo "ID not found" - exit 1 -fi - -# Run python script -$PYTHON src/cron.py youtube $id +echo "The legacy YouTube upload helper has been removed." +echo "Use 'bash scripts/publish_video.sh' instead." +exit 1 diff --git a/src/classes/PostBridge.py b/src/classes/PostBridge.py index db5cd8f39..1e0c9f678 100644 --- a/src/classes/PostBridge.py +++ b/src/classes/PostBridge.py @@ -60,30 +60,80 @@ def list_social_accounts( for platform in platforms: params.append(("platform", platform)) - url = f"{self.API_BASE}/social-accounts" - accounts = [] - is_first_request = True + return self._list_paginated_request( + f"{self.API_BASE}/social-accounts", + params=params, + invalid_payload_message="Post Bridge returned an invalid social accounts payload.", + max_items=None, + ) - while url: - response_json = self._request_json( - "GET", - url, - params=params if is_first_request else None, - ) - is_first_request = False + def list_posts( + self, + platforms: Optional[Sequence[str]] = None, + statuses: Optional[Sequence[str]] = None, + limit: int = 20, + offset: int = 0, + ) -> list[dict]: + """ + Fetch posts with optional platform and status filters. - page_accounts = response_json.get("data", response_json) - if not isinstance(page_accounts, list): - raise PostBridgeClientError( - "Post Bridge returned an invalid social accounts payload." - ) + Args: + platforms (Sequence[str] | None): Optional platform filters. + statuses (Sequence[str] | None): Optional post status filters. + limit (int): Max number of posts to return. + offset (int): Initial pagination offset. - accounts.extend(page_accounts) + Returns: + posts (list[dict]): Post data. + """ + params = [("limit", limit), ("offset", offset)] + if platforms: + for platform in platforms: + params.append(("platform", platform)) + if statuses: + for status in statuses: + params.append(("status", status)) - meta = response_json.get("meta", {}) - url = meta.get("next") if isinstance(meta, dict) else None + return self._list_paginated_request( + f"{self.API_BASE}/posts", + params=params, + invalid_payload_message="Post Bridge returned an invalid posts payload.", + max_items=limit, + ) - return accounts + def list_post_results( + self, + post_ids: Optional[Sequence[str]] = None, + platforms: Optional[Sequence[str]] = None, + limit: int = 100, + offset: int = 0, + ) -> list[dict]: + """ + Fetch post results with optional post ID and platform filters. + + Args: + post_ids (Sequence[str] | None): Optional post IDs. + platforms (Sequence[str] | None): Optional platform filters. + limit (int): Max number of results to return. + offset (int): Initial pagination offset. + + Returns: + post_results (list[dict]): Post result data. + """ + params = [("limit", limit), ("offset", offset)] + if post_ids: + for post_id in post_ids: + params.append(("post_id", post_id)) + if platforms: + for platform in platforms: + params.append(("platform", platform)) + + return self._list_paginated_request( + f"{self.API_BASE}/post-results", + params=params, + invalid_payload_message="Post Bridge returned an invalid post results payload.", + max_items=limit, + ) def upload_media(self, file_path: str) -> str: """ @@ -175,6 +225,53 @@ def create_post( json=payload, ) + def _list_paginated_request( + self, + url: str, + *, + params: Optional[list[tuple[str, object]]] = None, + invalid_payload_message: str, + max_items: Optional[int], + ) -> list[dict]: + """ + Fetch list endpoints that use the standard paginated response shape. + + Args: + url (str): Endpoint URL. + params (list[tuple[str, object]] | None): Query params for the first page. + invalid_payload_message (str): Error message for invalid payloads. + max_items (int | None): Optional max number of items to return. + + Returns: + items (list[dict]): Collected items. + """ + items = [] + next_url = url + is_first_request = True + + while next_url and (max_items is None or len(items) < max_items): + response_json = self._request_json( + "GET", + next_url, + params=params if is_first_request else None, + ) + is_first_request = False + + page_items = response_json.get("data", response_json) + if not isinstance(page_items, list): + raise PostBridgeClientError(invalid_payload_message) + + if max_items is None: + items.extend(page_items) + else: + remaining = max_items - len(items) + items.extend(page_items[:remaining]) + + meta = response_json.get("meta", {}) + next_url = meta.get("next") if isinstance(meta, dict) else None + + return items + def _guess_mime_type(self, file_path: str) -> str: guessed_type = mimetypes.guess_type(file_path)[0] if guessed_type in {"image/png", "image/jpeg", "video/mp4", "video/quicktime"}: diff --git a/src/classes/YouTube.py b/src/classes/YouTube.py index 79640f661..dcbe3f364 100644 --- a/src/classes/YouTube.py +++ b/src/classes/YouTube.py @@ -75,29 +75,9 @@ def __init__( self._language: str = language self.images = [] - - # Initialize the Firefox profile - self.options: Options = Options() - - # Set headless state of browser - if get_headless(): - self.options.add_argument("--headless") - - if not os.path.isdir(self._fp_profile_path): - raise ValueError( - f"Firefox profile path does not exist or is not a directory: {self._fp_profile_path}" - ) - - self.options.add_argument("-profile") - self.options.add_argument(self._fp_profile_path) - - # Set the service - self.service: Service = Service(GeckoDriverManager().install()) - - # Initialize the browser - self.browser: webdriver.Firefox = webdriver.Firefox( - service=self.service, options=self.options - ) + self.options: Options | None = None + self.service: Service | None = None + self.browser: webdriver.Firefox | None = None @property def niche(self) -> str: @@ -692,7 +672,7 @@ def get_channel_id(self) -> str: Returns: channel_id (str): The Channel ID. """ - driver = self.browser + driver = self._ensure_browser() driver.get("https://studio.youtube.com") time.sleep(2) channel_id = driver.current_url.split("/")[-1] @@ -710,7 +690,7 @@ def upload_video(self) -> bool: try: self.get_channel_id() - driver = self.browser + driver = self._ensure_browser() verbose = get_verbose() # Go to youtube.com/upload @@ -849,7 +829,8 @@ def upload_video(self) -> bool: return True except: - self.browser.quit() + if self.browser is not None: + self.browser.quit() return False def get_videos(self) -> List[dict]: @@ -876,3 +857,32 @@ def get_videos(self) -> List[dict]: videos = account["videos"] return videos + + def _ensure_browser(self) -> webdriver.Firefox: + """ + Lazily initialize the Firefox browser only when legacy upload behavior is used. + + Returns: + browser (webdriver.Firefox): Initialized Firefox webdriver. + """ + if self.browser is not None: + return self.browser + + if not self._fp_profile_path or not os.path.isdir(self._fp_profile_path): + raise ValueError( + "Legacy browser-based YouTube upload requires a valid firefox_profile." + ) + + self.options = Options() + if get_headless(): + self.options.add_argument("--headless") + + self.options.add_argument("-profile") + self.options.add_argument(self._fp_profile_path) + + self.service = Service(GeckoDriverManager().install()) + self.browser = webdriver.Firefox( + service=self.service, + options=self.options, + ) + return self.browser diff --git a/src/config.py b/src/config.py index fb26d8d4e..ea91f87c1 100644 --- a/src/config.py +++ b/src/config.py @@ -7,6 +7,18 @@ ROOT_DIR = os.path.dirname(sys.path[0]) +POST_BRIDGE_SUPPORTED_PLATFORMS = [ + "youtube", + "tiktok", + "instagram", + "facebook", + "twitter", + "threads", + "linkedin", + "bluesky", + "pinterest", +] + def assert_folder_structure() -> None: """ Make sure that the nessecary folder structure is present. @@ -36,8 +48,7 @@ def get_email_credentials() -> dict: Returns: credentials (dict): The email credentials """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file)["email"] + return load_config()["email"] def get_verbose() -> bool: """ @@ -46,8 +57,7 @@ def get_verbose() -> bool: Returns: verbose (bool): The verbose flag """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file)["verbose"] + return load_config()["verbose"] def get_firefox_profile_path() -> str: """ @@ -56,8 +66,7 @@ def get_firefox_profile_path() -> str: Returns: path (str): The path to the Firefox profile """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file)["firefox_profile"] + return load_config()["firefox_profile"] def get_headless() -> bool: """ @@ -66,8 +75,7 @@ def get_headless() -> bool: Returns: headless (bool): The headless flag """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file)["headless"] + return load_config()["headless"] def get_ollama_base_url() -> str: """ @@ -76,8 +84,7 @@ def get_ollama_base_url() -> str: Returns: url (str): The Ollama base URL """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file).get("ollama_base_url", "http://127.0.0.1:11434") + return load_config().get("ollama_base_url", "http://127.0.0.1:11434") def get_ollama_model() -> str: """ @@ -86,8 +93,7 @@ def get_ollama_model() -> str: Returns: model (str): The Ollama model name, or empty string if not set. """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file).get("ollama_model", "") + return load_config().get("ollama_model", "") def get_twitter_language() -> str: """ @@ -96,8 +102,7 @@ def get_twitter_language() -> str: Returns: language (str): The Twitter language """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file)["twitter_language"] + return load_config()["twitter_language"] def get_nanobanana2_api_base_url() -> str: """ @@ -106,11 +111,10 @@ def get_nanobanana2_api_base_url() -> str: Returns: url (str): API base URL """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file).get( - "nanobanana2_api_base_url", - "https://generativelanguage.googleapis.com/v1beta", - ) + return load_config().get( + "nanobanana2_api_base_url", + "https://generativelanguage.googleapis.com/v1beta", + ) def get_nanobanana2_api_key() -> str: """ @@ -119,9 +123,8 @@ def get_nanobanana2_api_key() -> str: Returns: key (str): API key """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - configured = json.load(file).get("nanobanana2_api_key", "") - return configured or os.environ.get("GEMINI_API_KEY", "") + configured = load_config().get("nanobanana2_api_key", "") + return configured or os.environ.get("GEMINI_API_KEY", "") def get_nanobanana2_model() -> str: """ @@ -130,8 +133,7 @@ def get_nanobanana2_model() -> str: Returns: model (str): Model name """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file).get("nanobanana2_model", "gemini-3.1-flash-image-preview") + return load_config().get("nanobanana2_model", "gemini-3.1-flash-image-preview") def get_nanobanana2_aspect_ratio() -> str: """ @@ -140,8 +142,7 @@ def get_nanobanana2_aspect_ratio() -> str: Returns: ratio (str): Aspect ratio """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file).get("nanobanana2_aspect_ratio", "9:16") + return load_config().get("nanobanana2_aspect_ratio", "9:16") def get_threads() -> int: """ @@ -150,8 +151,7 @@ def get_threads() -> int: Returns: threads (int): Amount of threads """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file)["threads"] + return load_config()["threads"] def get_zip_url() -> str: """ @@ -160,8 +160,7 @@ def get_zip_url() -> str: Returns: url (str): The URL to the zip file """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file)["zip_url"] + return load_config()["zip_url"] def get_is_for_kids() -> bool: """ @@ -170,8 +169,7 @@ def get_is_for_kids() -> bool: Returns: is_for_kids (bool): The is for kids flag """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file)["is_for_kids"] + return load_config()["is_for_kids"] def get_google_maps_scraper_zip_url() -> str: """ @@ -180,8 +178,7 @@ def get_google_maps_scraper_zip_url() -> str: Returns: url (str): The URL to the zip file """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file)["google_maps_scraper"] + return load_config()["google_maps_scraper"] def get_google_maps_scraper_niche() -> str: """ @@ -190,8 +187,7 @@ def get_google_maps_scraper_niche() -> str: Returns: niche (str): The niche """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file)["google_maps_scraper_niche"] + return load_config()["google_maps_scraper_niche"] def get_scraper_timeout() -> int: """ @@ -200,8 +196,7 @@ def get_scraper_timeout() -> int: Returns: timeout (int): The timeout """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file)["scraper_timeout"] or 300 + return load_config()["scraper_timeout"] or 300 def get_outreach_message_subject() -> str: """ @@ -210,8 +205,7 @@ def get_outreach_message_subject() -> str: Returns: subject (str): The outreach message subject """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file)["outreach_message_subject"] + return load_config()["outreach_message_subject"] def get_outreach_message_body_file() -> str: """ @@ -220,8 +214,7 @@ def get_outreach_message_body_file() -> str: Returns: file (str): The outreach message body file """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file)["outreach_message_body_file"] + return load_config()["outreach_message_body_file"] def get_tts_voice() -> str: """ @@ -230,8 +223,7 @@ def get_tts_voice() -> str: Returns: voice (str): The TTS voice """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file).get("tts_voice", "Jasper") + return load_config().get("tts_voice", "Jasper") def get_assemblyai_api_key() -> str: """ @@ -240,8 +232,7 @@ def get_assemblyai_api_key() -> str: Returns: key (str): The AssemblyAI API key """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file)["assembly_ai_api_key"] + return load_config()["assembly_ai_api_key"] def get_stt_provider() -> str: """ @@ -250,8 +241,7 @@ def get_stt_provider() -> str: Returns: provider (str): The STT provider """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file).get("stt_provider", "local_whisper") + return load_config().get("stt_provider", "local_whisper") def get_whisper_model() -> str: """ @@ -260,8 +250,7 @@ def get_whisper_model() -> str: Returns: model (str): Whisper model name """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file).get("whisper_model", "base") + return load_config().get("whisper_model", "base") def get_whisper_device() -> str: """ @@ -270,8 +259,7 @@ def get_whisper_device() -> str: Returns: device (str): Whisper device """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file).get("whisper_device", "auto") + return load_config().get("whisper_device", "auto") def get_whisper_compute_type() -> str: """ @@ -280,8 +268,7 @@ def get_whisper_compute_type() -> str: Returns: compute_type (str): Whisper compute type """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file).get("whisper_compute_type", "int8") + return load_config().get("whisper_compute_type", "int8") def equalize_subtitles(srt_path: str, max_chars: int = 10) -> None: """ @@ -303,8 +290,7 @@ def get_font() -> str: Returns: font (str): The font """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file)["font"] + return load_config()["font"] def get_fonts_dir() -> str: """ @@ -322,8 +308,7 @@ def get_imagemagick_path() -> str: Returns: path (str): The path to ImageMagick """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - return json.load(file)["imagemagick_path"] + return load_config()["imagemagick_path"] def get_script_sentence_length() -> int: """ @@ -333,12 +318,83 @@ def get_script_sentence_length() -> int: Returns: length (int): Length of script's sentence """ - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - config_json = json.load(file) - if (config_json.get("script_sentence_length") is not None): - return config_json["script_sentence_length"] - else: - return 4 + config_json = load_config() + if config_json.get("script_sentence_length") is not None: + return config_json["script_sentence_length"] + return 4 + +def load_config() -> dict: + """ + Loads the full application configuration. + + Returns: + config (dict): Parsed config.json contents. + """ + with open(os.path.join(ROOT_DIR, "config.json"), "r", encoding="utf-8") as file: + return json.load(file) + +def save_config(config: dict) -> None: + """ + Persists the full application configuration. + + Args: + config (dict): Config payload to write. + + Returns: + None + """ + with open(os.path.join(ROOT_DIR, "config.json"), "w", encoding="utf-8") as file: + json.dump(config, file, indent=2) + file.write("\n") + +def update_config_section(section: str, values: dict) -> dict: + """ + Merges values into a top-level config section and persists the change. + + Args: + section (str): Top-level config key. + values (dict): Values to merge into the section. + + Returns: + updated (dict): Updated config payload. + """ + config = load_config() + current_values = config.get(section, {}) + if not isinstance(current_values, dict): + current_values = {} + + merged_values = dict(current_values) + merged_values.update(values) + config[section] = merged_values + save_config(config) + return config + +def get_video_publishing_config() -> dict: + """ + Gets the config-backed video generation profile. + + Returns: + config (dict): Sanitized video publishing configuration. + """ + defaults = { + "profile_name": "Default Publisher", + "niche": "", + "language": "English", + } + + raw_config = load_config().get("video_publishing", {}) + if not isinstance(raw_config, dict): + raw_config = {} + + return { + "profile_name": str( + raw_config.get("profile_name", defaults["profile_name"]) + ).strip() or defaults["profile_name"], + "niche": str(raw_config.get("niche", defaults["niche"])).strip(), + "language": str( + raw_config.get("language", defaults["language"]) + ).strip() or defaults["language"], + } def get_post_bridge_config() -> dict: """ @@ -350,16 +406,13 @@ def get_post_bridge_config() -> dict: defaults = { "enabled": False, "api_key": "", - "platforms": ["tiktok", "instagram"], + "platforms": ["youtube", "tiktok", "instagram"], "account_ids": [], - "auto_crosspost": False, + "auto_publish": False, } - supported_platforms = {"tiktok", "instagram"} - - with open(os.path.join(ROOT_DIR, "config.json"), "r") as file: - config_json = json.load(file) + supported_platforms = set(POST_BRIDGE_SUPPORTED_PLATFORMS) - raw_config = config_json.get("post_bridge", {}) + raw_config = load_config().get("post_bridge", {}) if not isinstance(raw_config, dict): raw_config = {} @@ -399,7 +452,10 @@ def get_post_bridge_config() -> dict: "api_key": api_key, "platforms": normalized_platforms, "account_ids": normalized_account_ids, - "auto_crosspost": bool( - raw_config.get("auto_crosspost", defaults["auto_crosspost"]) + "auto_publish": bool( + raw_config.get( + "auto_publish", + raw_config.get("auto_crosspost", defaults["auto_publish"]), + ) ), } diff --git a/src/constants.py b/src/constants.py index 72bf9d73f..6d5a3b989 100644 --- a/src/constants.py +++ b/src/constants.py @@ -6,7 +6,7 @@ TWITTER_POST_BUTTON_XPATH = "/html/body/div[1]/div/div/div[2]/main/div/div/div/div[1]/div/div[3]/div/div[2]/div[1]/div/div/div/div[2]/div[2]/div[2]/div/div/div/div[3]" OPTIONS = [ - "YouTube Shorts Automation", + "Video Publishing", "Twitter Bot", "Affiliate Marketing", "Outreach", @@ -27,14 +27,15 @@ "Quit" ] -YOUTUBE_OPTIONS = [ - "Upload Short", - "Show all Shorts", +VIDEO_OPTIONS = [ + "Setup Publisher", + "Publish Video", + "Show Recent Publishes", "Setup CRON Job", "Quit" ] -YOUTUBE_CRON_OPTIONS = [ +VIDEO_CRON_OPTIONS = [ "Once a day", "Twice a day", "Thrice a day", diff --git a/src/cron.py b/src/cron.py index 9d002dc73..89720d9cc 100644 --- a/src/cron.py +++ b/src/cron.py @@ -3,23 +3,26 @@ from status import * from cache import get_accounts +from config import get_ollama_model from config import get_verbose +from config import get_video_publishing_config from classes.Tts import TTS from classes.Twitter import Twitter from classes.YouTube import YouTube from llm_provider import select_model -from post_bridge_integration import maybe_crosspost_youtube_short +from post_bridge_integration import ensure_post_bridge_publishing_ready +from post_bridge_integration import publish_video def main(): - """Main function to post content to Twitter or upload videos to YouTube. + """Main function to post content to Twitter or publish generated videos. This function determines its operation based on command-line arguments: - If the purpose is "twitter", it initializes a Twitter account and posts a message. - - If the purpose is "youtube", it initializes a YouTube account, generates a video with TTS, and uploads it. + - If the purpose is "publish", it generates a video and publishes it via Post Bridge. Command-line arguments: - sys.argv[1]: A string indicating the purpose, either "twitter" or "youtube". - sys.argv[2]: A string representing the account UUID. + sys.argv[1]: A string indicating the purpose, either "twitter" or "publish". + sys.argv[2]: Twitter account UUID for twitter mode, or optional model name for publish mode. The function also handles verbose output based on user settings and reports success or errors as appropriate. @@ -28,14 +31,23 @@ def main(): Returns: None. The function performs operations based on the purpose and account UUID and does not return any value.""" + if len(sys.argv) < 2: + error("No cron purpose provided.") + sys.exit(1) + purpose = str(sys.argv[1]) - account_id = str(sys.argv[2]) - model = str(sys.argv[3]) if len(sys.argv) > 3 else None + account_id = str(sys.argv[2]) if len(sys.argv) > 2 else "" + configured_model = get_ollama_model() + + if purpose == "twitter": + model = str(sys.argv[3]) if len(sys.argv) > 3 else configured_model + else: + model = str(sys.argv[2]) if len(sys.argv) > 2 else configured_model if model: select_model(model) else: - error("No Ollama model specified. Pass model name as third argument.") + error("No Ollama model specified. Set ollama_model or pass it as a cron argument.") sys.exit(1) verbose = get_verbose() @@ -60,38 +72,41 @@ def main(): if verbose: success("Done posting.") break - elif purpose == "youtube": - tts = TTS() - - accounts = get_accounts("youtube") - - if not account_id: - error("Account UUID cannot be empty.") + elif purpose == "publish": + if not ensure_post_bridge_publishing_ready(interactive=False): + error( + "Video publishing is not configured. Run the interactive Post Bridge " + "setup wizard from src/main.py first." + ) + sys.exit(1) - for acc in accounts: - if acc["id"] == account_id: - if verbose: - info("Initializing YouTube...") - youtube = YouTube( - acc["id"], - acc["nickname"], - acc["firefox_profile"], - acc["niche"], - acc["language"] - ) - youtube.generate_video(tts) - upload_success = youtube.upload_video() - if upload_success: - if verbose: - success("Uploaded Short.") - maybe_crosspost_youtube_short( - video_path=youtube.video_path, - title=youtube.metadata.get("title", ""), - interactive=False, - ) - else: - warning("YouTube upload failed. Skipping Post Bridge cross-post.") - break + tts = TTS() + video_config = get_video_publishing_config() + + if verbose: + info("Initializing Video Publishing...") + + video_generator = YouTube( + "post-bridge-publisher", + video_config["profile_name"], + "", + video_config["niche"], + video_config["language"], + ) + video_generator.generate_video(tts) + publish_success = publish_video( + video_path=video_generator.video_path, + title=video_generator.metadata.get("title", ""), + description=video_generator.metadata.get("description", ""), + interactive=False, + ) + if publish_success and verbose: + success("Published video.") + elif purpose == "youtube": + error( + "The 'youtube' cron mode has been removed. Use 'publish' instead." + ) + sys.exit(1) else: error("Invalid Purpose, exiting...") sys.exit(1) diff --git a/src/main.py b/src/main.py index 8bb1542b3..f30445bce 100644 --- a/src/main.py +++ b/src/main.py @@ -16,15 +16,144 @@ from classes.Outreach import Outreach from classes.AFM import AffiliateMarketing from llm_provider import list_models, select_model, get_active_model -from post_bridge_integration import maybe_crosspost_youtube_short +from post_bridge_integration import ensure_post_bridge_publishing_ready +from post_bridge_integration import get_publish_history +from post_bridge_integration import publish_video +from post_bridge_integration import run_post_bridge_setup_wizard + + +def build_video_generator() -> YouTube: + """ + Build a generator instance from config-backed video publishing settings. + + Returns: + generator (YouTube): Video generator instance. + """ + video_config = get_video_publishing_config() + return YouTube( + "post-bridge-publisher", + video_config["profile_name"], + "", + video_config["niche"], + video_config["language"], + ) + + +def show_publish_history() -> None: + """ + Display recent Post Bridge publish history. + + Returns: + None + """ + try: + history = get_publish_history() + except Exception as exc: + warning(f"Could not load publish history: {exc}") + return + + if len(history) == 0: + warning("No recent publishes found.") + return + + history_table = PrettyTable() + history_table.field_names = ["ID", "Created", "Status", "Platforms", "URLs"] + + for item in history: + urls = "\n".join(item["urls"][:2]) if item["urls"] else "-" + history_table.add_row( + [ + colored(item["id"], "cyan"), + colored(item["created_at"] or "-", "blue"), + colored(item["status"], "green"), + colored(", ".join(item["platforms"]), "magenta"), + colored(urls, "yellow"), + ] + ) + + print(history_table) + + +def run_video_publishing_menu() -> None: + """ + Run the PostBridge-first video publishing menu. + + Returns: + None + """ + info("Starting Video Publishing...") + + while True: + rem_temp_files() + info("\n============ OPTIONS ============", False) + + for idx, video_option in enumerate(VIDEO_OPTIONS): + print(colored(f" {idx + 1}. {video_option}", "cyan")) + + info("=================================\n", False) + + user_input = int(question("Select an option: ")) + + if user_input == 1: + run_post_bridge_setup_wizard() + elif user_input == 2: + if not ensure_post_bridge_publishing_ready(interactive=True): + continue + + video_generator = build_video_generator() + tts = TTS() + video_generator.generate_video(tts) + publish_video( + video_path=video_generator.video_path, + title=video_generator.metadata.get("title", ""), + description=video_generator.metadata.get("description", ""), + interactive=True, + ) + elif user_input == 3: + show_publish_history() + elif user_input == 4: + info("How often do you want to publish?") + + info("\n============ OPTIONS ============", False) + for idx, cron_option in enumerate(VIDEO_CRON_OPTIONS): + print(colored(f" {idx + 1}. {cron_option}", "cyan")) + + info("=================================\n", False) + + user_input = int(question("Select an Option: ")) + + cron_script_path = os.path.join(ROOT_DIR, "src", "cron.py") + command = ["python", cron_script_path, "publish", get_active_model()] + + def job(): + subprocess.run(command) + + if user_input == 1: + schedule.every(1).day.do(job) + success("Set up CRON Job.") + elif user_input == 2: + schedule.every().day.at("10:00").do(job) + schedule.every().day.at("16:00").do(job) + success("Set up CRON Job.") + elif user_input == 3: + schedule.every().day.at("08:00").do(job) + schedule.every().day.at("12:00").do(job) + schedule.every().day.at("18:00").do(job) + success("Set up CRON Job.") + else: + break + elif user_input == 5: + if get_verbose(): + info(" => Climbing Options Ladder...", False) + break def main(): """Main entry point for the application, providing a menu-driven interface - to manage YouTube, Twitter bots, Affiliate Marketing, and Outreach tasks. + to manage video publishing, Twitter bots, Affiliate Marketing, and Outreach tasks. This function allows users to: - 1. Start the YouTube Shorts Automater to manage YouTube accounts, - generate and upload videos, and set up CRON jobs. + 1. Start the Video Publishing flow to configure Post Bridge, + generate videos, publish them, and set up CRON jobs. 2. Start a Twitter Bot to manage Twitter accounts, post tweets, and schedule posts using CRON jobs. 3. Manage Affiliate Marketing by creating pitches and sharing them via @@ -66,162 +195,7 @@ def main(): # Start the selected option if user_input == 1: - info("Starting YT Shorts Automater...") - - cached_accounts = get_accounts("youtube") - - if len(cached_accounts) == 0: - warning("No accounts found in cache. Create one now?") - user_input = question("Yes/No: ") - - if user_input.lower() == "yes": - generated_uuid = str(uuid4()) - - success(f" => Generated ID: {generated_uuid}") - nickname = question(" => Enter a nickname for this account: ") - fp_profile = question(" => Enter the path to the Firefox profile: ") - niche = question(" => Enter the account niche: ") - language = question(" => Enter the account language: ") - - account_data = { - "id": generated_uuid, - "nickname": nickname, - "firefox_profile": fp_profile, - "niche": niche, - "language": language, - "videos": [], - } - - add_account("youtube", account_data) - - success("Account configured successfully!") - else: - table = PrettyTable() - table.field_names = ["ID", "UUID", "Nickname", "Niche"] - - for account in cached_accounts: - table.add_row([cached_accounts.index(account) + 1, colored(account["id"], "cyan"), colored(account["nickname"], "blue"), colored(account["niche"], "green")]) - - print(table) - info("Type 'd' to delete an account.", False) - - user_input = question("Select an account to start (or 'd' to delete): ").strip() - - if user_input.lower() == "d": - delete_input = question("Enter account number to delete: ").strip() - account_to_delete = None - - for account in cached_accounts: - if str(cached_accounts.index(account) + 1) == delete_input: - account_to_delete = account - break - - if account_to_delete is None: - error("Invalid account selected. Please try again.", "red") - else: - confirm = question(f"Are you sure you want to delete '{account_to_delete['nickname']}'? (Yes/No): ").strip().lower() - - if confirm == "yes": - remove_account("youtube", account_to_delete["id"]) - success("Account removed successfully!") - else: - warning("Account deletion canceled.", False) - - return - - selected_account = None - - for account in cached_accounts: - if str(cached_accounts.index(account) + 1) == user_input: - selected_account = account - - if selected_account is None: - error("Invalid account selected. Please try again.", "red") - main() - else: - youtube = YouTube( - selected_account["id"], - selected_account["nickname"], - selected_account["firefox_profile"], - selected_account["niche"], - selected_account["language"] - ) - - while True: - rem_temp_files() - info("\n============ OPTIONS ============", False) - - for idx, youtube_option in enumerate(YOUTUBE_OPTIONS): - print(colored(f" {idx + 1}. {youtube_option}", "cyan")) - - info("=================================\n", False) - - # Get user input - user_input = int(question("Select an option: ")) - tts = TTS() - - if user_input == 1: - youtube.generate_video(tts) - upload_to_yt = question("Do you want to upload this video to YouTube? (Yes/No): ") - if upload_to_yt.lower() == "yes": - upload_success = youtube.upload_video() - if upload_success: - maybe_crosspost_youtube_short( - video_path=youtube.video_path, - title=youtube.metadata.get("title", ""), - interactive=True, - ) - else: - warning("YouTube upload failed. Skipping Post Bridge cross-post.") - elif user_input == 2: - videos = youtube.get_videos() - - if len(videos) > 0: - videos_table = PrettyTable() - videos_table.field_names = ["ID", "Date", "Title"] - - for video in videos: - videos_table.add_row([ - videos.index(video) + 1, - colored(video["date"], "blue"), - colored(video["title"][:60] + "...", "green") - ]) - - print(videos_table) - else: - warning(" No videos found.") - elif user_input == 3: - info("How often do you want to upload?") - - info("\n============ OPTIONS ============", False) - for idx, cron_option in enumerate(YOUTUBE_CRON_OPTIONS): - print(colored(f" {idx + 1}. {cron_option}", "cyan")) - - info("=================================\n", False) - - user_input = int(question("Select an Option: ")) - - cron_script_path = os.path.join(ROOT_DIR, "src", "cron.py") - command = ["python", cron_script_path, "youtube", selected_account['id'], get_active_model()] - - def job(): - subprocess.run(command) - - if user_input == 1: - # Upload Once - schedule.every(1).day.do(job) - success("Set up CRON Job.") - elif user_input == 2: - # Upload Twice a day - schedule.every().day.at("10:00").do(job) - schedule.every().day.at("16:00").do(job) - success("Set up CRON Job.") - else: - break - elif user_input == 4: - if get_verbose(): - info(" => Climbing Options Ladder...", False) - break + run_video_publishing_menu() elif user_input == 2: info("Starting Twitter Bot...") diff --git a/src/post_bridge_integration.py b/src/post_bridge_integration.py index 274ad12d5..60264419a 100644 --- a/src/post_bridge_integration.py +++ b/src/post_bridge_integration.py @@ -4,22 +4,112 @@ from classes.PostBridge import PostBridge from classes.PostBridge import PostBridgeClientError +from config import POST_BRIDGE_SUPPORTED_PLATFORMS from config import get_post_bridge_config +from config import get_video_publishing_config +from config import load_config +from config import update_config_section from status import info from status import question from status import success from status import warning +PromptFn = Callable[[str], str] + + +def _is_yes(value: str) -> bool: + return value.strip().lower() in {"y", "yes"} + + +def _normalize_csv_numbers( + raw_value: str, + max_value: int, +) -> list[int]: + values = [] + seen_values = set() + + for chunk in raw_value.split(","): + chunk = chunk.strip() + if not chunk: + continue + + try: + selected_index = int(chunk) + except ValueError: + return [] + + if not 1 <= selected_index <= max_value: + return [] + + if selected_index not in seen_values: + values.append(selected_index) + seen_values.add(selected_index) + + return values + + +def _build_publish_caption(video_path: str, description: str, title: str) -> str: + cleaned_description = description.strip() + if cleaned_description: + return cleaned_description + + cleaned_title = title.strip() + if cleaned_title: + return cleaned_title + + return os.path.splitext(os.path.basename(video_path))[0] + + +def _prompt_for_platforms( + prompt_fn: PromptFn, + default_platforms: list[str], +) -> list[str]: + info("Select the platforms this video publisher should target:") + for index, platform in enumerate(POST_BRIDGE_SUPPORTED_PLATFORMS, start=1): + info(f" {index}. {platform}", False) + + default_selection = [ + str(POST_BRIDGE_SUPPORTED_PLATFORMS.index(platform) + 1) + for platform in default_platforms + if platform in POST_BRIDGE_SUPPORTED_PLATFORMS + ] + default_label = ",".join(default_selection) + + while True: + response = prompt_fn( + f"Enter comma-separated platform numbers [{default_label}]: " + ).strip() + + if not response: + selected_indexes = [ + int(value) for value in default_selection + ] + else: + selected_indexes = _normalize_csv_numbers( + response, + max_value=len(POST_BRIDGE_SUPPORTED_PLATFORMS), + ) + + if not selected_indexes: + warning("Please select at least one valid platform.") + continue + + return [ + POST_BRIDGE_SUPPORTED_PLATFORMS[selected_index - 1] + for selected_index in selected_indexes + ] + + def resolve_social_account_ids( client: PostBridge, configured_account_ids: list[int], platforms: list[str], interactive: bool, - prompt_fn: Optional[Callable[[str], str]] = None, + prompt_fn: Optional[PromptFn] = None, ) -> list[int]: """ - Resolve the social account IDs to use for a cross-post. + Resolve the social account IDs to use for a publish. Args: client (PostBridge): Post Bridge client instance. @@ -94,129 +184,430 @@ def resolve_social_account_ids( warning("Invalid selection. Please try again.") if resolved_account_ids: - info( - "Tip: Add these Post Bridge account IDs to config.json to skip prompts:" - ) + info("Tip: These Post Bridge account IDs can be persisted in config.json:") info(f' "account_ids": {resolved_account_ids}', False) return resolved_account_ids -def build_platform_configurations(title: str) -> dict: +def build_platform_configurations( + title: str, + description: str, + platforms: list[str], +) -> dict: """ Build platform-specific post overrides for Post Bridge. Args: - title (str): Video title generated by YouTube flow. + title (str): Generated video title. + description (str): Generated video description. + platforms (list[str]): Configured target platforms. Returns: platform_configurations (dict): Platform override payload. """ cleaned_title = title.strip() - if not cleaned_title: - return {} + cleaned_description = description.strip() + platform_configurations = {} - return { - "tiktok": { - "title": cleaned_title, - } - } + if "youtube" in platforms: + youtube_config = {} + if cleaned_title: + youtube_config["title"] = cleaned_title + if cleaned_description: + youtube_config["caption"] = cleaned_description + if youtube_config: + platform_configurations["youtube"] = youtube_config + if "tiktok" in platforms and cleaned_title: + platform_configurations["tiktok"] = {"title": cleaned_title} -def maybe_crosspost_youtube_short( + return platform_configurations + + +def run_post_bridge_setup_wizard( + prompt_fn: Optional[PromptFn] = None, +) -> Optional[dict]: + """ + Guide the user through a config-backed Post Bridge publisher setup. + + Args: + prompt_fn (Callable | None): Optional prompt function override for tests. + + Returns: + config (dict | None): Persisted Post Bridge config on success. + """ + if prompt_fn is None: + prompt_fn = question + + raw_config = load_config() + raw_post_bridge_config = raw_config.get("post_bridge", {}) + if not isinstance(raw_post_bridge_config, dict): + raw_post_bridge_config = {} + + current_post_bridge_config = get_post_bridge_config() + current_video_config = get_video_publishing_config() + + info("Starting the Post Bridge video publishing setup wizard...") + profile_name = ( + prompt_fn( + f"Publisher name [{current_video_config['profile_name']}]: " + ).strip() + or current_video_config["profile_name"] + ) + niche = ( + prompt_fn( + "Video niche/topic " + f"[{current_video_config['niche'] or 'required'}]: " + ).strip() + or current_video_config["niche"] + ) + language = ( + prompt_fn( + f"Video language [{current_video_config['language']}]: " + ).strip() + or current_video_config["language"] + ) + + stored_api_key = str(raw_post_bridge_config.get("api_key", "")).strip() + entered_api_key = prompt_fn( + "Post Bridge API key " + "[leave blank to keep current value or use POST_BRIDGE_API_KEY]: " + ).strip() + if entered_api_key: + api_key = entered_api_key + else: + api_key = current_post_bridge_config["api_key"] + + if not niche: + warning("A video niche is required to generate content.") + return None + + if not api_key: + warning( + "No Post Bridge API key is configured. Set one in config.json " + "or export POST_BRIDGE_API_KEY." + ) + return None + + default_platforms = ( + current_post_bridge_config["platforms"] + or ["youtube", "tiktok", "instagram"] + ) + platforms = _prompt_for_platforms(prompt_fn, default_platforms) + + client = PostBridge(api_key) + try: + account_ids = resolve_social_account_ids( + client=client, + configured_account_ids=[], + platforms=platforms, + interactive=True, + prompt_fn=prompt_fn, + ) + except PostBridgeClientError as exc: + warning(f"Unable to fetch Post Bridge accounts: {exc}") + return None + + if len(account_ids) != len(platforms): + warning( + "The setup wizard could not resolve one account for every selected " + "platform. Connect the missing accounts in Post Bridge and try again." + ) + return None + + auto_publish_default = "yes" if current_post_bridge_config["auto_publish"] else "no" + auto_publish = _is_yes( + prompt_fn( + f"Auto-publish generated videos without confirmation? " + f"(Yes/No) [{auto_publish_default}]: " + ).strip() + or auto_publish_default + ) + + update_config_section( + "video_publishing", + { + "profile_name": profile_name, + "niche": niche, + "language": language, + }, + ) + updated_config = update_config_section( + "post_bridge", + { + "enabled": True, + "api_key": entered_api_key or stored_api_key, + "platforms": platforms, + "account_ids": account_ids, + "auto_publish": auto_publish, + }, + ) + + success("Saved Post Bridge video publishing settings to config.json.") + return updated_config + + +def ensure_post_bridge_publishing_ready( + interactive: bool, + prompt_fn: Optional[PromptFn] = None, +) -> bool: + """ + Ensure the app has enough config to generate and publish videos. + + Args: + interactive (bool): Whether the user can be prompted. + prompt_fn (Callable | None): Optional prompt function override for tests. + + Returns: + ready (bool): Whether publishing can proceed. + """ + if prompt_fn is None: + prompt_fn = question + + post_bridge_config = get_post_bridge_config() + video_config = get_video_publishing_config() + + missing_settings = [] + if not post_bridge_config["enabled"]: + missing_settings.append("post_bridge.enabled") + if not post_bridge_config["api_key"]: + missing_settings.append("post_bridge.api_key") + if ( + not post_bridge_config["platforms"] + and not post_bridge_config["account_ids"] + ): + missing_settings.append("post_bridge.platforms") + if not video_config["niche"]: + missing_settings.append("video_publishing.niche") + + if not missing_settings: + return True + + warning( + "Video publishing is not configured yet. Missing: " + + ", ".join(missing_settings) + ) + if not interactive: + return False + + if _is_yes(prompt_fn("Run the Post Bridge setup wizard now? (Yes/No): ")): + return run_post_bridge_setup_wizard(prompt_fn=prompt_fn) is not None + + return False + + +def publish_video( video_path: str, title: str, + description: str, interactive: bool, - prompt_fn: Optional[Callable[[str], str]] = None, + prompt_fn: Optional[PromptFn] = None, ) -> Optional[bool]: """ - Cross-post a successfully uploaded YouTube Short via Post Bridge. + Publish a generated video via Post Bridge. Args: video_path (str): Path to generated video file. - title (str): Generated YouTube title. + title (str): Generated video title. + description (str): Generated video description. interactive (bool): Whether prompting is allowed. prompt_fn (Callable | None): Optional prompt function override for tests. Returns: - result (bool | None): True when posted, False when attempted and failed, + result (bool | None): True when published, False when attempted and failed, None when skipped. """ - config = get_post_bridge_config() + if prompt_fn is None: + prompt_fn = question - if not config["enabled"]: + if not ensure_post_bridge_publishing_ready( + interactive=interactive, + prompt_fn=prompt_fn, + ): return None - if not config["api_key"]: - warning( - "Post Bridge is enabled but no API key is configured. " - "Set post_bridge.api_key or POST_BRIDGE_API_KEY." - ) - return None + post_bridge_config = get_post_bridge_config() if not os.path.exists(video_path): - warning(f"Cannot cross-post because the video file was not found: {video_path}") + warning(f"Cannot publish because the video file was not found: {video_path}") return False - platforms = config["platforms"] - configured_account_ids = config["account_ids"] - if not platforms and not configured_account_ids: - warning("Post Bridge is enabled but no supported platforms are configured.") - return None - - if prompt_fn is None: - prompt_fn = question - if interactive: - should_crosspost = config["auto_crosspost"] - if not should_crosspost: - platform_label = ", ".join(platforms) if platforms else "configured Post Bridge accounts" + should_publish = post_bridge_config["auto_publish"] + if not should_publish: + platform_label = ", ".join(post_bridge_config["platforms"]) response = prompt_fn( - f"Cross-post this video to {platform_label} via Post Bridge? (Yes/No): " - ).strip().lower() - should_crosspost = response in {"y", "yes"} + f"Publish this video to {platform_label} via Post Bridge? (Yes/No): " + ).strip() + should_publish = _is_yes(response) else: - if not config["auto_crosspost"]: + if not post_bridge_config["auto_publish"]: info( - "Post Bridge is enabled, but auto_crosspost is disabled. " - "Skipping cross-post in cron mode." + "Post Bridge is enabled, but auto_publish is disabled. " + "Skipping publish in cron mode." ) return None - should_crosspost = True + should_publish = True - if not should_crosspost: + if not should_publish: return None - post_caption = title.strip() - if not post_caption: - post_caption = os.path.splitext(os.path.basename(video_path))[0] - - client = PostBridge(config["api_key"]) + caption = _build_publish_caption(video_path, description, title) + client = PostBridge(post_bridge_config["api_key"]) try: account_ids = resolve_social_account_ids( client=client, - configured_account_ids=configured_account_ids, - platforms=platforms, + configured_account_ids=post_bridge_config["account_ids"], + platforms=post_bridge_config["platforms"], interactive=interactive, prompt_fn=prompt_fn, ) if not account_ids: - warning("No Post Bridge accounts were resolved. Skipping cross-post.") + warning("No Post Bridge accounts were resolved. Skipping publish.") return None media_id = client.upload_media(video_path) result = client.create_post( - caption=post_caption, + caption=caption, social_account_ids=account_ids, media_ids=[media_id], - platform_configurations=build_platform_configurations(title), + platform_configurations=build_platform_configurations( + title=title, + description=description, + platforms=post_bridge_config["platforms"], + ), ) - success(f"Cross-posted via Post Bridge (post ID: {result.get('id', 'unknown')}).") + success( + f"Published via Post Bridge (post ID: {result.get('id', 'unknown')})." + ) for warning_message in result.get("warnings", []): warning(f"Post Bridge warning: {warning_message}") return True except PostBridgeClientError as exc: - warning(f"Post Bridge cross-post failed: {exc}") + warning(f"Post Bridge publishing failed: {exc}") return False + + +def get_publish_history(limit: int = 10) -> list[dict]: + """ + Fetch recent publish history from Post Bridge. + + Args: + limit (int): Number of posts to load. + + Returns: + history (list[dict]): Recent posts enriched with post results. + """ + post_bridge_config = get_post_bridge_config() + if not post_bridge_config["enabled"] or not post_bridge_config["api_key"]: + warning( + "Post Bridge publishing is not configured. Run the setup wizard first." + ) + return [] + + client = PostBridge(post_bridge_config["api_key"]) + posts = client.list_posts( + platforms=post_bridge_config["platforms"], + statuses=["posted", "scheduled", "processing"], + limit=limit, + ) + if not posts: + return [] + + social_accounts = client.list_social_accounts(platforms=post_bridge_config["platforms"]) + platform_by_social_account_id = {} + for social_account in social_accounts: + social_account_id = social_account.get("id") + platform = social_account.get("platform") + if social_account_id is None or not platform: + continue + try: + platform_by_social_account_id[int(social_account_id)] = platform + except (TypeError, ValueError): + continue + + post_ids = [post.get("id") for post in posts if post.get("id")] + result_limit = max(limit * max(len(post_bridge_config["platforms"]), 1), limit) + post_results = client.list_post_results( + post_ids=post_ids, + platforms=post_bridge_config["platforms"], + limit=result_limit, + ) + + results_by_post_id = {} + for post_result in post_results: + post_id = post_result.get("post_id") + if not post_id: + continue + results_by_post_id.setdefault(post_id, []).append(post_result) + + history = [] + for post in posts: + target_platforms = [] + seen_platforms = set() + for social_account_id in post.get("social_accounts") or []: + try: + normalized_account_id = int(social_account_id) + except (TypeError, ValueError): + continue + + platform = platform_by_social_account_id.get(normalized_account_id) + if platform and platform not in seen_platforms: + target_platforms.append(platform) + seen_platforms.add(platform) + + if not target_platforms: + platform_configurations = post.get("platform_configurations") or {} + for platform in platform_configurations.keys(): + if platform not in seen_platforms: + target_platforms.append(platform) + seen_platforms.add(platform) + + if not target_platforms: + target_platforms = post_bridge_config["platforms"] + + results = results_by_post_id.get(post.get("id"), []) + urls = [] + failures = [] + for result in results: + platform_data = result.get("platform_data") or {} + if platform_data.get("url"): + urls.append(platform_data["url"]) + if not result.get("success"): + error_payload = result.get("error") or {} + failures.append(str(error_payload or "Unknown publishing error")) + + history.append( + { + "id": post.get("id", ""), + "created_at": post.get("created_at", ""), + "status": post.get("status", "unknown"), + "caption": post.get("caption", ""), + "platforms": target_platforms, + "urls": urls, + "failures": failures, + } + ) + + return history + + +def maybe_crosspost_youtube_short( + video_path: str, + title: str, + interactive: bool, + prompt_fn: Optional[PromptFn] = None, +) -> Optional[bool]: + """ + Backward-compatible alias for legacy tests and callers. + """ + return publish_video( + video_path=video_path, + title=title, + description=title, + interactive=interactive, + prompt_fn=prompt_fn, + ) diff --git a/tests/test_config.py b/tests/test_config.py index 0a0976bfc..78b96b582 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -20,40 +20,42 @@ def write_config(self, directory: str, payload: dict) -> None: with open(os.path.join(directory, "config.json"), "w", encoding="utf-8") as handle: json.dump(payload, handle) - def test_missing_platforms_uses_defaults(self) -> None: + def read_config(self, directory: str) -> dict: + with open(os.path.join(directory, "config.json"), "r", encoding="utf-8") as handle: + return json.load(handle) + + def test_missing_platforms_uses_publish_defaults(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: self.write_config(temp_dir, {"post_bridge": {"enabled": True}}) with patch.object(config, "ROOT_DIR", temp_dir): post_bridge_config = config.get_post_bridge_config() - self.assertEqual(post_bridge_config["platforms"], ["tiktok", "instagram"]) + self.assertEqual( + post_bridge_config["platforms"], + ["youtube", "tiktok", "instagram"], + ) - def test_invalid_or_empty_platforms_do_not_expand_to_defaults(self) -> None: + def test_legacy_auto_crosspost_maps_to_auto_publish(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: self.write_config( temp_dir, - { - "post_bridge": { - "enabled": True, - "platforms": ["youtube", "tik-tok"], - } - }, + {"post_bridge": {"enabled": True, "auto_crosspost": True}}, ) with patch.object(config, "ROOT_DIR", temp_dir): post_bridge_config = config.get_post_bridge_config() - self.assertEqual(post_bridge_config["platforms"], []) + self.assertTrue(post_bridge_config["auto_publish"]) - def test_non_list_platforms_fail_closed(self) -> None: + def test_invalid_platforms_are_filtered_but_youtube_is_supported(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: self.write_config( temp_dir, { "post_bridge": { "enabled": True, - "platforms": "tiktok", + "platforms": ["youtube", "tik-tok", "instagram"], } }, ) @@ -61,23 +63,44 @@ def test_non_list_platforms_fail_closed(self) -> None: with patch.object(config, "ROOT_DIR", temp_dir): post_bridge_config = config.get_post_bridge_config() - self.assertEqual(post_bridge_config["platforms"], []) + self.assertEqual(post_bridge_config["platforms"], ["youtube", "instagram"]) - def test_non_object_post_bridge_config_falls_back_to_defaults(self) -> None: + def test_update_config_section_preserves_unrelated_keys(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: self.write_config( temp_dir, { - "post_bridge": None, + "verbose": True, + "post_bridge": {"enabled": False}, + "video_publishing": {"niche": "finance"}, }, ) with patch.object(config, "ROOT_DIR", temp_dir): - post_bridge_config = config.get_post_bridge_config() + config.update_config_section( + "post_bridge", + { + "enabled": True, + "platforms": ["youtube"], + }, + ) + + saved_config = self.read_config(temp_dir) + + self.assertTrue(saved_config["verbose"]) + self.assertEqual(saved_config["video_publishing"]["niche"], "finance") + self.assertEqual(saved_config["post_bridge"]["platforms"], ["youtube"]) + + def test_video_publishing_defaults_are_safe(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + self.write_config(temp_dir, {}) + + with patch.object(config, "ROOT_DIR", temp_dir): + video_config = config.get_video_publishing_config() - self.assertEqual(post_bridge_config["platforms"], ["tiktok", "instagram"]) - self.assertEqual(post_bridge_config["account_ids"], []) - self.assertFalse(post_bridge_config["enabled"]) + self.assertEqual(video_config["profile_name"], "Default Publisher") + self.assertEqual(video_config["language"], "English") + self.assertEqual(video_config["niche"], "") if __name__ == "__main__": diff --git a/tests/test_cron_post_bridge.py b/tests/test_cron_post_bridge.py index ee16c3a65..22fb26d44 100644 --- a/tests/test_cron_post_bridge.py +++ b/tests/test_cron_post_bridge.py @@ -2,7 +2,6 @@ import sys import types import unittest -from unittest.mock import Mock from unittest.mock import patch @@ -40,48 +39,59 @@ class CronPostBridgeTests(unittest.TestCase): - @patch("cron.maybe_crosspost_youtube_short") + @patch("cron.publish_video") @patch("cron.YouTube") @patch("cron.TTS") - @patch("cron.get_accounts") + @patch("cron.get_video_publishing_config") + @patch("cron.ensure_post_bridge_publishing_ready", return_value=True) @patch("cron.select_model") @patch("cron.get_verbose") - def test_crosspost_does_not_run_when_youtube_upload_fails( + def test_publish_mode_generates_and_publishes_video( self, get_verbose_mock, select_model_mock, - get_accounts_mock, + _ensure_ready_mock, + get_video_config_mock, tts_cls_mock, youtube_cls_mock, - crosspost_mock, + publish_video_mock, ) -> None: get_verbose_mock.return_value = False - get_accounts_mock.return_value = [ - { - "id": "yt-1", - "nickname": "Channel", - "firefox_profile": "/tmp/profile", - "niche": "finance", - "language": "English", - } - ] + get_video_config_mock.return_value = { + "profile_name": "Default Publisher", + "niche": "finance", + "language": "English", + } youtube_instance = youtube_cls_mock.return_value - youtube_instance.upload_video.return_value = False youtube_instance.video_path = "/tmp/video.mp4" - youtube_instance.metadata = {"title": "Title"} + youtube_instance.metadata = { + "title": "Title", + "description": "Description", + } + publish_video_mock.return_value = True - with patch.object( - sys, - "argv", - ["cron.py", "youtube", "yt-1", "llama3.2:3b"], - ): + with patch.object(sys, "argv", ["cron.py", "publish", "llama3.2:3b"]): cron.main() select_model_mock.assert_called_once_with("llama3.2:3b") tts_cls_mock.assert_called_once() youtube_instance.generate_video.assert_called_once() - youtube_instance.upload_video.assert_called_once() - crosspost_mock.assert_not_called() + publish_video_mock.assert_called_once_with( + video_path="/tmp/video.mp4", + title="Title", + description="Description", + interactive=False, + ) + + def test_legacy_youtube_mode_exits_with_migration_message(self) -> None: + with patch.object(sys, "argv", ["cron.py", "youtube", "llama3.2:3b"]), patch( + "cron.get_ollama_model", + return_value="llama3.2:3b", + ), patch("cron.select_model"): + with self.assertRaises(SystemExit) as raised: + cron.main() + + self.assertEqual(raised.exception.code, 1) if __name__ == "__main__": diff --git a/tests/test_post_bridge_client.py b/tests/test_post_bridge_client.py index 60638d68b..49ad47d41 100644 --- a/tests/test_post_bridge_client.py +++ b/tests/test_post_bridge_client.py @@ -79,6 +79,38 @@ def test_list_social_accounts_follows_pagination(self, _sleep_mock) -> None: self.assertEqual(second_call.args[1], "https://api.post-bridge.com/v1/social-accounts?offset=1&limit=1") self.assertIsNone(second_call.kwargs["params"]) + @patch("classes.PostBridge.time.sleep") + def test_list_posts_uses_platform_and_status_filters(self, _sleep_mock) -> None: + session = Mock() + session.request.return_value = MockResponse( + 200, + { + "data": [{"id": "post-1", "status": "posted"}], + "meta": {"next": None}, + }, + ) + client = PostBridge("token", session=session) + + posts = client.list_posts( + platforms=["youtube", "tiktok"], + statuses=["posted"], + limit=5, + ) + + self.assertEqual(posts, [{"id": "post-1", "status": "posted"}]) + first_call = session.request.call_args_list[0] + self.assertEqual(first_call.args[1], "https://api.post-bridge.com/v1/posts") + self.assertEqual( + first_call.kwargs["params"], + [ + ("limit", 5), + ("offset", 0), + ("platform", "youtube"), + ("platform", "tiktok"), + ("status", "posted"), + ], + ) + @patch("classes.PostBridge.time.sleep") def test_create_post_retries_after_rate_limit(self, sleep_mock) -> None: session = Mock() diff --git a/tests/test_post_bridge_integration.py b/tests/test_post_bridge_integration.py index c47743ff7..c6ed27aeb 100644 --- a/tests/test_post_bridge_integration.py +++ b/tests/test_post_bridge_integration.py @@ -1,3 +1,4 @@ +import json import os import sys import tempfile @@ -12,11 +13,23 @@ if SRC_DIR not in sys.path: sys.path.insert(0, SRC_DIR) -from post_bridge_integration import maybe_crosspost_youtube_short +import config +import post_bridge_integration +from post_bridge_integration import get_publish_history +from post_bridge_integration import publish_video from post_bridge_integration import resolve_social_account_ids +from post_bridge_integration import run_post_bridge_setup_wizard class PostBridgeIntegrationTests(unittest.TestCase): + def write_config(self, directory: str, payload: dict) -> None: + with open(os.path.join(directory, "config.json"), "w", encoding="utf-8") as handle: + json.dump(payload, handle) + + def read_config(self, directory: str) -> dict: + with open(os.path.join(directory, "config.json"), "r", encoding="utf-8") as handle: + return json.load(handle) + def test_resolve_social_account_ids_interactive_prompts_for_ambiguous_accounts(self) -> None: client = Mock() client.list_social_accounts.return_value = [ @@ -53,53 +66,85 @@ def test_resolve_social_account_ids_skips_non_interactive_when_multiple_accounts self.assertEqual(account_ids, []) + @patch("post_bridge_integration.ensure_post_bridge_publishing_ready", return_value=True) @patch("post_bridge_integration.PostBridge") @patch("post_bridge_integration.get_post_bridge_config") - def test_cron_mode_skips_when_auto_crosspost_is_disabled( + def test_cron_mode_skips_when_auto_publish_is_disabled( self, get_config_mock, post_bridge_cls_mock, + _ensure_ready_mock, ) -> None: get_config_mock.return_value = { "enabled": True, "api_key": "token", - "platforms": ["tiktok", "instagram"], + "platforms": ["youtube", "tiktok"], "account_ids": [12, 34], - "auto_crosspost": False, + "auto_publish": False, } with tempfile.NamedTemporaryFile(suffix=".mp4") as media_file: - result = maybe_crosspost_youtube_short( + result = publish_video( video_path=media_file.name, title="My title", + description="My description", interactive=False, ) self.assertIsNone(result) post_bridge_cls_mock.assert_not_called() + @patch("post_bridge_integration.get_video_publishing_config") + @patch("post_bridge_integration.get_post_bridge_config") + def test_readiness_allows_fixed_account_ids_without_platform_filters( + self, + get_config_mock, + get_video_config_mock, + ) -> None: + get_config_mock.return_value = { + "enabled": True, + "api_key": "token", + "platforms": [], + "account_ids": [12, 34], + "auto_publish": True, + } + get_video_config_mock.return_value = { + "profile_name": "Default Publisher", + "niche": "finance", + "language": "English", + } + + self.assertTrue( + post_bridge_integration.ensure_post_bridge_publishing_ready( + interactive=False, + ) + ) + + @patch("post_bridge_integration.ensure_post_bridge_publishing_ready", return_value=True) @patch("post_bridge_integration.PostBridge") @patch("post_bridge_integration.get_post_bridge_config") - def test_interactive_crosspost_uploads_and_posts( + def test_interactive_publish_uploads_and_posts( self, get_config_mock, post_bridge_cls_mock, + _ensure_ready_mock, ) -> None: get_config_mock.return_value = { "enabled": True, "api_key": "token", - "platforms": ["tiktok", "instagram"], + "platforms": ["youtube", "tiktok"], "account_ids": [12, 34], - "auto_crosspost": False, + "auto_publish": False, } client = post_bridge_cls_mock.return_value client.upload_media.return_value = "media-123" client.create_post.return_value = {"id": "post-123", "warnings": []} with tempfile.NamedTemporaryFile(suffix=".mp4") as media_file: - result = maybe_crosspost_youtube_short( + result = publish_video( video_path=media_file.name, title="My title", + description="My description", interactive=True, prompt_fn=lambda _: "yes", ) @@ -107,15 +152,97 @@ def test_interactive_crosspost_uploads_and_posts( self.assertTrue(result) client.upload_media.assert_called_once() client.create_post.assert_called_once_with( - caption="My title", + caption="My description", social_account_ids=[12, 34], media_ids=["media-123"], - platform_configurations={"tiktok": {"title": "My title"}}, + platform_configurations={ + "youtube": {"title": "My title", "caption": "My description"}, + "tiktok": {"title": "My title"}, + }, + ) + + @patch("post_bridge_integration.PostBridge") + def test_setup_wizard_persists_selected_accounts(self, post_bridge_cls_mock) -> None: + client = post_bridge_cls_mock.return_value + client.list_social_accounts.return_value = [ + {"id": 11, "platform": "youtube", "username": "yt_brand"}, + {"id": 21, "platform": "tiktok", "username": "tt_brand"}, + {"id": 31, "platform": "instagram", "username": "ig_brand"}, + ] + + responses = iter( + [ + "Launch Profile", + "finance", + "English", + "pb_live_token", + "", + "yes", + ] + ) + + with tempfile.TemporaryDirectory() as temp_dir: + self.write_config(temp_dir, {"verbose": True}) + + with patch.object(config, "ROOT_DIR", temp_dir): + result = run_post_bridge_setup_wizard(prompt_fn=lambda _: next(responses)) + + saved_config = self.read_config(temp_dir) + + self.assertIsNotNone(result) + self.assertEqual(saved_config["video_publishing"]["profile_name"], "Launch Profile") + self.assertEqual(saved_config["video_publishing"]["niche"], "finance") + self.assertEqual( + saved_config["post_bridge"]["platforms"], + ["youtube", "tiktok", "instagram"], ) + self.assertEqual(saved_config["post_bridge"]["account_ids"], [11, 21, 31]) + self.assertTrue(saved_config["post_bridge"]["auto_publish"]) + + @patch.dict(os.environ, {"POST_BRIDGE_API_KEY": "pb_env_token"}, clear=False) + @patch("post_bridge_integration.PostBridge") + def test_setup_wizard_does_not_persist_env_only_api_key(self, post_bridge_cls_mock) -> None: + client = post_bridge_cls_mock.return_value + client.list_social_accounts.return_value = [ + {"id": 11, "platform": "youtube", "username": "yt_brand"}, + {"id": 21, "platform": "tiktok", "username": "tt_brand"}, + {"id": 31, "platform": "instagram", "username": "ig_brand"}, + ] + + responses = iter( + [ + "", + "finance", + "", + "", + "", + "no", + ] + ) + + with tempfile.TemporaryDirectory() as temp_dir: + self.write_config( + temp_dir, + { + "post_bridge": { + "enabled": False, + "api_key": "", + } + }, + ) + + with patch.object(config, "ROOT_DIR", temp_dir): + result = run_post_bridge_setup_wizard(prompt_fn=lambda _: next(responses)) + + saved_config = self.read_config(temp_dir) + + self.assertIsNotNone(result) + self.assertEqual(saved_config["post_bridge"]["api_key"], "") + self.assertFalse(saved_config["post_bridge"]["auto_publish"]) @patch("post_bridge_integration.PostBridge") @patch("post_bridge_integration.get_post_bridge_config") - def test_account_ids_work_without_platform_filters( + def test_get_publish_history_merges_posts_and_results( self, get_config_mock, post_bridge_cls_mock, @@ -123,29 +250,52 @@ def test_account_ids_work_without_platform_filters( get_config_mock.return_value = { "enabled": True, "api_key": "token", - "platforms": [], + "platforms": ["youtube", "tiktok"], "account_ids": [12, 34], - "auto_crosspost": True, + "auto_publish": True, } client = post_bridge_cls_mock.return_value - client.upload_media.return_value = "media-123" - client.create_post.return_value = {"id": "post-123", "warnings": []} + client.list_social_accounts.return_value = [ + {"id": 12, "platform": "youtube", "username": "yt_brand"}, + {"id": 34, "platform": "instagram", "username": "ig_brand"}, + ] + client.list_posts.return_value = [ + { + "id": "post-1", + "created_at": "2026-03-26T10:00:00Z", + "status": "posted", + "caption": "hello", + "platform_configurations": { + "youtube": {"title": "Title"}, + }, + "social_accounts": [12, 34], + } + ] + client.list_post_results.return_value = [ + { + "post_id": "post-1", + "success": True, + "platform_data": {"url": "https://youtube.com/watch?v=abc"}, + }, + { + "post_id": "post-1", + "success": True, + "platform_data": {"url": "https://instagram.com/p/123"}, + }, + ] - with tempfile.NamedTemporaryFile(suffix=".mp4") as media_file: - result = maybe_crosspost_youtube_short( - video_path=media_file.name, - title="My title", - interactive=False, - ) + history = get_publish_history(limit=5) - self.assertTrue(result) - client.upload_media.assert_called_once() - client.create_post.assert_called_once_with( - caption="My title", - social_account_ids=[12, 34], - media_ids=["media-123"], - platform_configurations={"tiktok": {"title": "My title"}}, + self.assertEqual(len(history), 1) + self.assertEqual(history[0]["id"], "post-1") + self.assertEqual( + history[0]["urls"], + [ + "https://youtube.com/watch?v=abc", + "https://instagram.com/p/123", + ], ) + self.assertEqual(history[0]["platforms"], ["youtube", "instagram"]) if __name__ == "__main__":