How to Integrate Replit with SocialBee

1import os
2import requests
3from typing import Optional
4
5API_KEY = os.environ["SOCIALBEE_API_KEY"]
6WORKSPACE_ID = os.environ["SOCIALBEE_WORKSPACE_ID"]
7
8# SocialBee API base URL — verify current version at socialbee.com/api-docs
9BASE_URL = "https://app.socialbee.com/api/v1"
10
11HEADERS = {
12    "Authorization": f"Bearer {API_KEY}",
13    "Content-Type": "application/json",
14    "Accept": "application/json"
15}
16
17def get_workspaces() -> list:
18    """List all workspaces accessible with this API key."""
19    response = requests.get(f"{BASE_URL}/workspaces", headers=HEADERS)
20    response.raise_for_status()
21    data = response.json()
22    return data.get('workspaces', data.get('data', []))
23
24def get_categories(workspace_id: str = None) -> list:
25    """List all content categories in a workspace."""
26    ws_id = workspace_id or WORKSPACE_ID
27    url = f"{BASE_URL}/workspaces/{ws_id}/categories"
28    response = requests.get(url, headers=HEADERS)
29    response.raise_for_status()
30    data = response.json()
31    return data.get('categories', data.get('data', []))
32
33def get_profiles(workspace_id: str = None) -> list:
34    """List all connected social profiles in a workspace."""
35    ws_id = workspace_id or WORKSPACE_ID
36    url = f"{BASE_URL}/workspaces/{ws_id}/profiles"
37    response = requests.get(url, headers=HEADERS)
38    response.raise_for_status()
39    data = response.json()
40    return data.get('profiles', data.get('data', []))
41
42def create_post(
43    category_id: str,
44    content: str,
45    image_url: Optional[str] = None,
46    link_url: Optional[str] = None,
47    workspace_id: str = None
48) -> dict:
49    """
50    Create a new post in a SocialBee category.
51    The post will be scheduled according to the category's posting schedule.
52    """
53    ws_id = workspace_id or WORKSPACE_ID
54    url = f"{BASE_URL}/workspaces/{ws_id}/categories/{category_id}/posts"
55    
56    payload = {"content": content}
57    if image_url:
58        payload["image_url"] = image_url
59    if link_url:
60        payload["url"] = link_url
61
62    response = requests.post(url, json=payload, headers=HEADERS)
63    response.raise_for_status()
64    return response.json()
65
66def get_posts_in_category(category_id: str, workspace_id: str = None) -> list:
67    """List all posts in a specific category."""
68    ws_id = workspace_id or WORKSPACE_ID
69    url = f"{BASE_URL}/workspaces/{ws_id}/categories/{category_id}/posts"
70    response = requests.get(url, headers=HEADERS)
71    response.raise_for_status()
72    data = response.json()
73    return data.get('posts', data.get('data', []))
74
75def delete_post(post_id: str, workspace_id: str = None) -> bool:
76    """Delete a post from a category."""
77    ws_id = workspace_id or WORKSPACE_ID
78    url = f"{BASE_URL}/workspaces/{ws_id}/posts/{post_id}"
79    response = requests.delete(url, headers=HEADERS)
80    return response.status_code in [200, 204]
81
82# Example usage
83if __name__ == "__main__":
84    # List workspaces and categories to find correct IDs
85    workspaces = get_workspaces()
86    print("Workspaces:")
87    for ws in workspaces:
88        print(f"  [{ws.get('id')}] {ws.get('name')}")
89
90    categories = get_categories()
91    print("\nCategories:")
92    for cat in categories:
93        print(f"  [{cat.get('id')}] {cat.get('name')} — {cat.get('posts_count', 0)} posts")
94
95    # Create a test post (replace with actual category ID)
96    if categories:
97        cat_id = categories[0]['id']
98        post = create_post(
99            category_id=cat_id,
100            content="Check out our latest blog post! #content #marketing",
101            link_url="https://example.com/blog/latest"
102        )
103        print(f"\nPost created: {post.get('id', 'unknown')}")

1const express = require('express');
2const axios = require('axios');
3
4const app = express();
5app.use(express.json());
6
7const API_KEY = process.env.SOCIALBEE_API_KEY;
8const WORKSPACE_ID = process.env.SOCIALBEE_WORKSPACE_ID;
9const BLOG_CATEGORY_ID = process.env.SOCIALBEE_BLOG_CATEGORY_ID;
10const BASE_URL = 'https://app.socialbee.com/api/v1';
11
12const socialbee = axios.create({
13  baseURL: BASE_URL,
14  headers: {
15    'Authorization': `Bearer ${API_KEY}`,
16    'Content-Type': 'application/json',
17    'Accept': 'application/json'
18  }
19});
20
21// Generate caption variations for different social networks
22function generateCaptions(title, excerpt, url) {
23  const shortCaption = `${title}\n\n${url}`;
24  const mediumCaption = `${title}\n\n${excerpt.slice(0, 200)}...\n\n${url}`;
25  const linkedinCaption = `New article: ${title}\n\n${excerpt.slice(0, 400)}\n\nRead more: ${url}\n\n#content #marketing #blog`;
26  return [shortCaption, mediumCaption, linkedinCaption];
27}
28
29// List categories in the workspace
30app.get('/categories', async (req, res) => {
31  try {
32    const response = await socialbee.get(`/workspaces/${WORKSPACE_ID}/categories`);
33    const categories = response.data.categories || response.data.data || [];
34    res.json(categories);
35  } catch (err) {
36    console.error('Categories error:', err.response?.data);
37    res.status(err.response?.status || 500).json({ error: err.message });
38  }
39});
40
41// Create a single post in a category
42app.post('/posts', async (req, res) => {
43  const { categoryId, content, imageUrl, linkUrl } = req.body;
44  if (!categoryId || !content) {
45    return res.status(400).json({ error: 'categoryId and content are required' });
46  }
47  try {
48    const payload = { content };
49    if (imageUrl) payload.image_url = imageUrl;
50    if (linkUrl) payload.url = linkUrl;
51
52    const response = await socialbee.post(
53      `/workspaces/${WORKSPACE_ID}/categories/${categoryId}/posts`,
54      payload
55    );
56    res.status(201).json(response.data);
57  } catch (err) {
58    console.error('Create post error:', err.response?.data);
59    res.status(err.response?.status || 500).json({
60      error: err.response?.data?.message || err.message
61    });
62  }
63});
64
65// Publish blog post: create multiple caption variations in the blog category
66app.post('/publish-blog', async (req, res) => {
67  const { title, excerpt, url, imageUrl } = req.body;
68  if (!title || !url) {
69    return res.status(400).json({ error: 'title and url are required' });
70  }
71
72  const categoryId = BLOG_CATEGORY_ID;
73  if (!categoryId) {
74    return res.status(500).json({ error: 'SOCIALBEE_BLOG_CATEGORY_ID not configured in Secrets' });
75  }
76
77  const captions = generateCaptions(title, excerpt || '', url);
78  const results = [];
79
80  for (const caption of captions) {
81    try {
82      const payload = { content: caption };
83      if (imageUrl) payload.image_url = imageUrl;
84
85      const response = await socialbee.post(
86        `/workspaces/${WORKSPACE_ID}/categories/${categoryId}/posts`,
87        payload
88      );
89      results.push({ success: true, id: response.data.id || 'created' });
90
91      // Small delay between posts to avoid rate limits
92      await new Promise(r => setTimeout(r, 500));
93    } catch (err) {
94      results.push({ success: false, error: err.response?.data?.message || err.message });
95    }
96  }
97
98  const successCount = results.filter(r => r.success).length;
99  res.json({
100    message: `Created ${successCount}/${captions.length} posts in SocialBee`,
101    results
102  });
103});
104
105app.listen(3000, '0.0.0.0', () => {
106  console.log('SocialBee integration server running on port 3000');
107});

1from flask import Flask, request, jsonify
2import os
3import requests
4
5app = Flask(__name__)
6
7API_KEY = os.environ["SOCIALBEE_API_KEY"]
8WORKSPACE_ID = os.environ["SOCIALBEE_WORKSPACE_ID"]
9BLOG_CATEGORY_ID = os.environ.get("SOCIALBEE_BLOG_CATEGORY_ID", "")
10BASE_URL = "https://app.socialbee.com/api/v1"
11
12HEADERS = {
13    "Authorization": f"Bearer {API_KEY}",
14    "Content-Type": "application/json"
15}
16
17def create_socialbee_post(category_id: str, content: str, link_url: str = "") -> dict:
18    url = f"{BASE_URL}/workspaces/{WORKSPACE_ID}/categories/{category_id}/posts"
19    payload = {"content": content}
20    if link_url:
21        payload["url"] = link_url
22    response = requests.post(url, json=payload, headers=HEADERS)
23    response.raise_for_status()
24    return response.json()
25
26# Webhook from CMS (WordPress, Ghost, etc.) when new article published
27@app.route('/cms/webhook', methods=['POST'])
28def cms_webhook():
29    data = request.json
30    if not data:
31        return jsonify({'error': 'No payload'}), 400
32
33    title = data.get('title') or data.get('post', {}).get('title', '')
34    url = data.get('url') or data.get('post', {}).get('url', '')
35    excerpt = data.get('excerpt') or data.get('post', {}).get('excerpt', '')[:200]
36
37    if not title or not url:
38        return jsonify({'error': 'Missing title or url'}), 400
39
40    if not BLOG_CATEGORY_ID:
41        return jsonify({'error': 'SOCIALBEE_BLOG_CATEGORY_ID not configured'}), 500
42
43    captions = [
44        f"{title}\n\n{url}",
45        f"New post: {title}\n\n{excerpt}...\n\n{url}"
46    ]
47
48    created = 0
49    for caption in captions:
50        try:
51            create_socialbee_post(BLOG_CATEGORY_ID, caption, url)
52            created += 1
53        except Exception as e:
54            print(f"Failed to create post: {e}")
55
56    print(f"CMS webhook: created {created} SocialBee posts for '{title}'")
57    return jsonify({'created': created, 'title': title}), 200
58
59if __name__ == '__main__':
60    app.run(host='0.0.0.0', port=3000)