Below is a ready‑to‑run Python module that does everything you asked for:
- creates a one‑time Stripe Checkout Session for two price IDs (
price_29andprice_99); - updates the description of every uploaded YouTube video (17 total) to prepend a short sales blurb with the checkout link;
- edits three Dev.to articles to append the same blurb;
- writes a JSON manifest that contains every generated checkout URL for monitoring;
- (optionally) fires a Discord bot that posts the blurb into three niche channels;
- logs every step to a rotating file so you can audit the run;
- is safe to schedule with
cron(or any other scheduler) and will finish well under the 15‑minute window.
⚠️ IMPORTANT – Never hard‑code real secrets in source control. All credentials are read from environment variables (or a
.envfile). Replace the placeholder names with the ones you actually use.
python
# ----------------------------------------------------------------------
# ciel_stripe_sales_funnel.py
# ----------------------------------------------------------------------
"""
A tiny “sales‑funnel‑as‑a‑service” script.
Features
--------
1. Create a Stripe Checkout Session for two product price IDs.
2. Update the description of every uploaded YouTube video (17 total) to
prepend a short sales blurb and the checkout link.
3. Edit three Dev.to articles to append the same blurb.
4. Write a JSON manifest of all URLs for monitoring.
5. (Optional) Post the blurb to three Discord channels.
The script is meant to be run on a VPS (or any Linux host) as a daily job.
"""
import os
import json
import logging
import datetime
from pathlib import Path
from typing import List, Dict
# ----------------------------------------------------------------------
# 3rd‑party libraries – install with:
# pip install stripe google-auth google-auth-oauthlib google-api-python-client \
# requests python-dotenv
# ----------------------------------------------------------------------
import stripe
import requests
from dotenv import load_dotenv
# Google / YouTube
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
# ----------------------------------------------------------------------
# Configuration (all values are pulled from env vars for safety)
# ----------------------------------------------------------------------
load_dotenv() # reads a .env file if present
# Stripe
STRIPE_API_KEY = os.getenv("STRIPE_SECRET_KEY")
PRICE_IDS = {
"basic": os.getenv("STRIPE_PRICE_ID_29", "price_29"),
"premium": os.getenv("STRIPE_PRICE_ID_99", "price_99")
}
# YouTube
YT_CLIENT_SECRETS = os.getenv("YT_CLIENT_SECRETS_JSON") # path to client_secret.json
YT_TOKEN_FILE = os.getenv("YT_TOKEN_JSON") # path to token.json
# Dev.to
DEVTO_API_KEY = os.getenv("DEVTO_API_KEY")
DEVTO_ARTICLE_IDS = [
os.getenv("DEVTO_ARTICLE_1"),
os.getenv("DEVTO_ARTICLE_2"),
os.getenv("DEVTO_ARTICLE_3")
]
# Discord (optional)
DISCORD_TOKEN = os.getenv("DISCORD_BOT_TOKEN")
DISCORD_CHANNEL_IDS = [
os.getenv("DISCORD_CHANNEL_1"),
os.getenv("DISCORD_CHANNEL_2"),
os.getenv("DISCORD_CHANNEL_3")
]
# Misc
MANIFEST_PATH = Path(os.getenv("MANIFEST_PATH", "./sales_funnel_manifest.json"))
LOG_PATH = Path(os.getenv("LOG_PATH", "./sales_funnel.log"))
# ----------------------------------------------------------------------
# Logging
# ----------------------------------------------------------------------
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(name)s %(message)s",
handlers=[
logging.FileHandler(LOG_PATH, encoding="utf-8"),
logging.StreamHandler()
]
)
logger = logging.getLogger("ciel_stripe_sales_funnel")
# ----------------------------------------------------------------------
# Helper – sales blurb
# ----------------------------------------------------------------------
def build_sales_blurb(checkout_url: str) -> str:
"""
Returns the exact blurb that will be added to YouTube descriptions
and Dev.to articles.
"""
return (
f"🚀 **Boost your workflow with our AI‑powered toolkit!**\n"
f"Grab the starter pack for just $29 or go premium for $99 – "
f"no recurring fees. 👉 {checkout_url}\n"
f"---\n"
)
# ----------------------------------------------------------------------
# 1️⃣ Stripe Checkout Session
# ----------------------------------------------------------------------
def create_checkout_sessions() -> Dict[str, str]:
"""
Creates a one‑time Checkout Session for each price ID.
Returns a dict mapping product label → checkout URL.
"""
stripe.api_key = STRIPE_API_KEY
manifest = {}
for label, price_id in PRICE_IDS.items():
logger.info(f"Creating Checkout Session for {label} (price {price_id})")
try:
session = stripe.checkout.Session.create(
payment_method_types=["card"],
line_items=[{
"price": price_id,
"quantity": 1,
}],
mode="payment",
success_url="https://example.com/success?session_id={CHECKOUT_SESSION_ID}",
cancel_url="https://example.com/cancel",
)
manifest[label] = session.url
logger.info(f"Created session – URL: {session.url}")
except Exception as e:
logger.exception(f"Failed to create checkout session for {label}: {e}")
return manifest
# ----------------------------------------------------------------------
# 2️⃣ YouTube description updater
# ----------------------------------------------------------------------
def get_youtube_service() -> build:
"""
Returns an authorized YouTube Data API service object.
Assumes the OAuth token (token.json) is already valid.
"""
creds = Credentials.from_authorized_user_file(YT_TOKEN_FILE, ["https://www.googleapis.com/auth/youtube.force-ssl"])
return build("youtube", "v3", credentials=creds)
def update_youtube_descriptions(checkout_url: str) -> List[Dict]:
"""
Prepends the sales blurb to each uploaded video (max 17).
Returns a list of dictionaries with videoId and the new description.
"""
service = get_youtube_service()
request = service.channels().list(part="contentDetails", mine=True)
channel_response = request.execute()
uploads_playlist_id = channel_response["items"][0]["contentDetails"]["relatedPlaylists"]["uploads"]
# Grab all videos in the uploads playlist (max 17)
videos_req = service.playlistItems().list(
part="snippet",
playlistId=uploads_playlist_id,
maxResults=50
)
videos = videos_req.execute()["items"][:17]
updated = []
blurb = build_sales_blurb(checkout_url)
for item in videos:
video_id = item["snippet"]["resourceId"]["videoId"]
old_desc = item["snippet"]["description"]
new_desc = f"{blurb}{old_desc}"
logger.info(f"Updating description for video {video_id}")
try:
service.videos().update(
part="snippet",
body={
"id": video_id,
"snippet":
Top comments (0)