code tested

Author: RaheesAhmed

.gitignore

@@ -3,4 +3,7 @@ __pycache__
 .env
 websites.json
 input_Data.csv
-Full Done - Copy.py
+Full Done - Copy.py
+/funs
+website_credentials.json
+full_working.py

LICENSE

@@ -0,0 +1,17 @@
+MIT License
+Copyright (c) 2024 Rahees Ahmed
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

app.py

@@ -0,0 +1,325 @@
+import streamlit as st
+import pandas as pd
+import asyncio
+import aiohttp
+import base64
+import logging
+import os
+import re
+import random
+import string
+from datetime import datetime, timedelta
+
+# Placeholder for OpenAI and Replicate API keys
+openai_api_key = "your_openai_api_key_here"
+replicate_api_token = "your_replicate_api_token_here"
+
+# Placeholder for websites dictionary
+websites = {
+    1: {
+        "url": "https://example.com/wp-json/wp/v2/",
+        "user": "username",
+        "password": "password",
+    },
+    # Add more websites as needed
+}
+
+# Setup logging
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
+)
+
+
+# Utility functions
+def format_url(url):
+    if not url.startswith(("http://", "https://")):
+        return "https://" + url
+    return url
+
+
+def clean_text(text):
+    return (
+        text.replace("##", "")
+        .replace("**", "")
+        .replace("###", "")
+        .replace("####", "")
+        .strip()
+    )
+
+
+def get_auth_header(user, password):
+    auth_string = f"{user}:{password}"
+    base64_auth = base64.b64encode(auth_string.encode("utf-8")).decode("utf-8")
+    return {"Authorization": f"Basic {base64_auth}", "User-Agent": "Mozilla/5.0"}
+
+
+def format_permalink(title):
+    permalink = re.sub(r"[^a-zA-Z0-9\s]", "", title)
+    words = permalink.lower().split()[:6]
+    formatted_permalink = "-".join(words)
+    return formatted_permalink
+
+
+def generate_random_string(length=4):
+    return "".join(random.choices(string.ascii_lowercase + string.digits, k=length))
+
+
+def get_custom_filename(keyword=None):
+    random_suffix = generate_random_string()
+    if keyword:
+        return f"{keyword}-{random_suffix}.webp"
+    else:
+        return f"{random_suffix}.webp"
+
+
+# Async functions
+async def fetch_serper_results_async(keyword, input_country):
+    url = "https://google.serper.dev/search"
+    headers = {
+        "X-API-KEY": "your_serper_api_key_here",
+        "Content-Type": "application/json",
+    }
+    payload = {"q": keyword, "gl": input_country}
+
+    async with aiohttp.ClientSession() as session:
+        try:
+            async with session.post(url, headers=headers, json=payload) as response:
+                if response.status == 200:
+                    data = await response.json()
+                    return data
+                else:
+                    st.error(f"Failed to fetch Serper results: {response.status}")
+        except Exception as e:
+            st.error(f"Error fetching Serper results: {e}")
+    return None
+
+
+async def generate_image_async(image_prompt, image_output_count, image_ai_model):
+    async with aiohttp.ClientSession() as session:
+        try:
+            async with session.post(
+                "https://api.replicate.com/v1/predictions",
+                json={"prompt": image_prompt, "num_outputs": image_output_count},
+                headers={"Authorization": f"Token {replicate_api_token}"},
+            ) as response:
+                if response.status == 200:
+                    data = await response.json()
+                    return data["outputs"][0]["url"]
+                else:
+                    st.error(f"Failed to generate image: {response.status}")
+        except Exception as e:
+            st.error(f"Error generating image: {e}")
+    return None
+
+
+async def create_post_async(
+    content,
+    website,
+    keyword,
+    description,
+    media_id,
+    category,
+    meta_keywords,
+    meta_description,
+    cached_title,
+    tag_names,
+    slug_function,
+    publish_status,
+    post_date_adjustment,
+):
+    headers = get_auth_header(website["user"], website["password"])
+    slug = slug_function(cached_title)
+    category_id = 1  # Placeholder, implement get_category_id function if needed
+    tag_ids = []  # Placeholder, implement get_tag_id function if needed
+    post_date = (datetime.now() + timedelta(days=post_date_adjustment)).strftime(
+        "%Y-%m-%dT%H:%M:%S"
+    )
+
+    data = {
+        "title": cached_title,
+        "slug": slug,
+        "content": content,
+        "status": publish_status,
+        "categories": [category_id],
+        "tags": tag_ids,
+        "featured_media": media_id,
+        "date": post_date,
+        "meta": {
+            "rank_math_focus_keyword": keyword,
+            "rank_math_description": meta_description,
+            "rank_math_keywords": meta_keywords,
+        },
+    }
+
+    async with aiohttp.ClientSession() as session:
+        try:
+            async with session.post(
+                f"{website['url']}posts", headers=headers, json=data
+            ) as response:
+                response.raise_for_status()
+                post_data = await response.json()
+                permalink = post_data.get("link", "")
+                post_id = post_data["id"]
+                st.success(f"Post published successfully. Permalink: {permalink}")
+                return permalink, post_id, cached_title
+        except Exception as e:
+            st.error(f"Failed to publish post: {e}")
+            return None, None, None
+
+
+# Streamlit UI
+st.title("WordPress Content Manager")
+
+# Sidebar for global settings
+st.sidebar.header("Global Settings")
+input_language = st.sidebar.selectbox(
+    "Input Language", ["English", "Spanish", "French", "German"]
+)
+input_openai_model = st.sidebar.selectbox("OpenAI Model", ["gpt-3.5-turbo", "gpt-4"])
+input_country = st.sidebar.selectbox("Search Country", ["us", "uk", "ca", "au"])
+image_output_count = st.sidebar.number_input(
+    "Number of Images to Generate", min_value=1, max_value=5, value=1
+)
+image_ai_model = st.sidebar.selectbox(
+    "Image AI Model", ["stability-ai/stable-diffusion", "midjourney/v4"]
+)
+post_date_adjustment = st.sidebar.number_input("Post Date Adjustment (days)", value=-5)
+publish_status = st.sidebar.selectbox(
+    "Publish Status", ["publish", "draft", "pending", "private"]
+)
+
+# Main content area
+tab1, tab2, tab3, tab4 = st.tabs(
+    ["Content Generation", "Post Editing", "Post Deletion", "Bulk Operations"]
+)
+
+with tab1:
+    st.header("Content Generation")
+
+    # Input for website URL and keywords
+    website_url = st.text_input("Website URL")
+    keywords = st.text_area("Keywords (one per line)")
+
+    if st.button("Generate Content"):
+        if website_url and keywords:
+            keywords_list = [k.strip() for k in keywords.split("\n") if k.strip()]
+
+            for keyword in keywords_list:
+                st.write(f"Processing keyword: {keyword}")
+
+                # Fetch SERP data
+                serper_data = asyncio.run(
+                    fetch_serper_results_async(keyword, input_country)
+                )
+
+                if serper_data:
+                    st.write("SERP data fetched successfully")
+
+                    # Generate SEO elements (placeholder function)
+                    seo_elements = {
+                        "meta_keywords": f"{keyword}, related, terms",
+                        "meta_description": f"Description for {keyword}",
+                        "img_alt": f"Image related to {keyword}",
+                        "website_title": f"Title for {keyword}",
+                        "ai_img_prompt": f"Generate an image for {keyword}",
+                        "website_category": "General",
+                    }
+
+                    # Generate content (placeholder function)
+                    content = f"<h1>Article about {keyword}</h1><p>This is a placeholder for the generated content.</p>"
+
+                    # Generate image
+                    image_url = asyncio.run(
+                        generate_image_async(
+                            seo_elements["ai_img_prompt"],
+                            image_output_count,
+                            image_ai_model,
+                        )
+                    )
+
+                    if image_url:
+                        st.image(image_url, caption=f"Generated image for {keyword}")
+
+                    # Publish post
+                    website = websites[1]  # Using the first website as an example
+                    media_id = 1  # Placeholder, implement upload_image_to_wp function if needed
+
+                    result = asyncio.run(
+                        create_post_async(
+                            content,
+                            website,
+                            keyword,
+                            seo_elements["meta_description"],
+                            media_id,
+                            seo_elements["website_category"],
+                            seo_elements["meta_keywords"],
+                            seo_elements["meta_description"],
+                            seo_elements["website_title"],
+                            [keyword],
+                            format_permalink,
+                            publish_status,
+                            post_date_adjustment,
+                        )
+                    )
+
+                    if result:
+                        permalink, post_id, article_title = result
+                        st.success(f"Post published: {article_title}")
+                        st.write(f"Permalink: {permalink}")
+                        st.write(f"Post ID: {post_id}")
+                else:
+                    st.error(f"Failed to fetch SERP data for {keyword}")
+        else:
+            st.warning("Please enter a website URL and at least one keyword.")
+
+with tab2:
+    st.header("Post Editing")
+
+    edit_post_id = st.number_input("Post ID to Edit", min_value=1)
+    edit_website = st.selectbox("Select Website", list(websites.keys()))
+
+    edit_title = st.text_input("New Title (optional)")
+    edit_content = st.text_area("New Content (optional)")
+
+    if st.button("Update Post"):
+        if edit_post_id and edit_website:
+            # Placeholder for post editing logic
+            st.success(
+                f"Post {edit_post_id} updated successfully on {websites[edit_website]['url']}"
+            )
+        else:
+            st.warning("Please enter a Post ID and select a website.")
+
+with tab3:
+    st.header("Post Deletion")
+
+    delete_post_id = st.number_input("Post ID to Delete", min_value=1)
+    delete_website = st.selectbox("Select Website for Deletion", list(websites.keys()))
+
+    if st.button("Delete Post"):
+        if delete_post_id and delete_website:
+            # Placeholder for post deletion logic
+            st.success(
+                f"Post {delete_post_id} deleted successfully from {websites[delete_website]['url']}"
+            )
+        else:
+            st.warning("Please enter a Post ID and select a website.")
+
+with tab4:
+    st.header("Bulk Operations")
+
+    uploaded_file = st.file_uploader("Upload CSV file", type="csv")
+
+    if uploaded_file is not None:
+        df = pd.read_csv(uploaded_file)
+        st.write(df)
+
+        if st.button("Process Bulk Operations"):
+            # Placeholder for bulk operations logic
+            st.info("Processing bulk operations...")
+            # Implement the logic to process the CSV file and perform bulk operations
+            st.success("Bulk operations completed successfully")
+
+# Footer
+st.markdown("---")
+st.write("WordPress Content Manager v1.0")

config.py

@@ -1,3 +1,4 @@
+import json
 import os
 from dotenv import load_dotenv
 from datetime import datetime, timedelta
@@ -28,7 +29,9 @@
 MAX_CONCURRENT_TASKS = 20
 
 # Website Credentials
-WEBSITE_CREDENTIALS_FILE = os.getenv("WEBSITE_CREDENTIALS_FILE", "websites.json")
+WEBSITE_CREDENTIALS_FILE = os.getenv(
+    "WEBSITE_CREDENTIALS_FILE", "website_credentials.json"
+)
 
 
 # Function to adjust the post date based on the number of days
@@ -66,3 +69,8 @@ async def get_default_settings():
         "image_ai_model": DEFAULT_IMAGE_AI_MODEL,
         "post_date_adjustment": DEFAULT_POST_DATE_ADJUSTMENT,
     }
+
+
+def load_website_credentials():
+    with open(WEBSITE_CREDENTIALS_FILE, "r") as file:
+        return json.load(file)