API 集成
平台 API 现状
首先,一个重要的事实:大部分国内平台没有公开的发布 API。
| 平台 | 官方 API | 替代方案 | 难度 | 稳定性 |
|---|---|---|---|---|
| 微信公众号 | 无(仅编辑) | 定制化开发 | 高 | 中 |
| 知乎 | 无 | 爬虫 | 中 | 低 |
| 小红书 | 无 | 爬虫+逆向 | 高 | 低 |
| 掘金 | 有 | 官方 API | 低 | 高 |
| CSDN | 无 | 爬虫 | 中 | 低 |
| Medium | 有 | 官方 API | 低 | 高 |
| WordPress | 有 | 官方 API | 低 | 高 |
可行方案
方案 A:只支持有官方 API 的平台
平台:掘金、Medium、WordPress
优点:
- 稳定可靠
- 不违反服务条款
- 维护成本低
缺点:
- 覆盖面有限
适合:MVP 验证
方案 B:模拟浏览器操作(Selenium/Playwright)
原理:用代码模拟人操作浏览器
优点:
- 可以支持没有 API 的平台
缺点:
- 速度慢
- 容易被检测和封禁
- 维护成本高(页面改版就得改代码)
适合:学习和个人使用
方案 C:使用第三方服务
服务:
优点:
- 简单
- 稳定
缺点:
- 可能需要付费
- 功能受限
技术选型
我们用 方案 B(Selenium) 作为学习示例。
安装 Selenium
bash
pip install selenium
pip install webdriver-manager安装浏览器驱动
python
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager
# 自动下载并配置 ChromeDriver
driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()))掘金 API 集成(示例)
掘金有相对开放的平台,我们先从它开始。
获取授权
- 登录 掘金
- 进入设置,查看 Cookie
- 复制必要的认证信息
发布文章
python
import requests
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
class JuejinPublisher:
def __init__(self):
self.driver = None
self.is_logged_in = False
def login(self, cookie_string):
"""使用 Cookie 登录"""
self.driver = webdriver.Chrome()
# 先访问主页
self.driver.get("https://juejin.cn/")
# 添加 Cookie
for item in cookie_string.split("; "):
name, value = item.split("=", 1)
self.driver.add_cookie({
"name": name,
"value": value,
"domain": ".juejin.cn"
})
# 刷新页面
self.driver.refresh()
self.is_logged_in = True
def publish(self, title, content, tags=[]):
"""发布文章"""
if not self.is_logged_in:
raise Exception("请先登录")
# 进入发布页面
self.driver.get("https://juejin.cn/editor")
# 等待编辑器加载
wait = WebDriverWait(self.driver, 10)
editor = wait.until(
EC.presence_of_element_located((By.CSS_SELECTOR, ".markdown-body"))
)
# 输入标题
title_input = self.driver.find_element(
By.CSS_SELECTOR,
"input[placeholder*='请输入标题']"
)
title_input.clear()
title_input.send_keys(title)
# 输入内容(这里需要根据实际页面结构调整)
content_area = self.driver.find_element(
By.CSS_SELECTOR,
".markdown-body"
)
content_area.send_keys(content)
# 添加标签
for tag in tags:
tag_input = self.driver.find_element(
By.CSS_SELECTOR,
"input[placeholder*='搜索标签']"
)
tag_input.send_keys(tag)
tag_input.send_keys("\n") # 模拟回车
# 点击发布按钮
publish_btn = self.driver.find_element(
By.CSS_SELECTOR,
"button[type='submit']"
)
publish_btn.click()
print("发布成功!")
def close(self):
"""关闭浏览器"""
if self.driver:
self.driver.quit()
# 使用示例
publisher = JuejinPublisher()
# 从浏览器开发者工具复制的 Cookie
cookie = "your-cookie-string-here"
publisher.login(cookie)
publisher.publish(
title="测试文章",
content="这是一篇测试文章...",
tags=["Python", "API"]
)
publisher.close()通用发布接口设计
为了支持多个平台,设计一个统一的接口:
python
from abc import ABC, abstractmethod
class PlatformPublisher(ABC):
"""平台发布器基类"""
@abstractmethod
def login(self, credentials):
"""登录"""
pass
@abstractmethod
def publish(self, title, content, **kwargs):
"""发布内容"""
pass
@abstractmethod
def format_content(self, content):
"""格式化内容以适应平台"""
pass掘金实现
python
class JuejinPublisher(PlatformPublisher):
def __init__(self):
self.driver = None
self.is_logged_in = False
def login(self, credentials):
"""使用 Cookie 登录"""
self.driver = webdriver.Chrome()
self.driver.get("https://juejin.cn/")
# 添加 Cookie
cookie = credentials.get("cookie")
for item in cookie.split("; "):
name, value = item.split("=", 1)
self.driver.add_cookie({
"name": name,
"value": value,
"domain": ".juejin.cn"
})
self.driver.refresh()
self.is_logged_in = True
def format_content(self, content):
"""格式化 Markdown 为掘金格式"""
# 掘金原生支持 Markdown,基本不需要转换
return content
def publish(self, title, content, tags=[], cover=""):
"""发布文章"""
if not self.is_logged_in:
raise Exception("请先登录")
self.driver.get("https://juejin.cn/editor")
# 填写表单...(省略具体实现)
return {"success": True, "url": "发布后的文章链接"}WordPress 实现(使用 REST API)
python
from requests.auth import HTTPBasicAuth
class WordPressPublisher(PlatformPublisher):
def __init__(self, base_url):
self.base_url = base_url
self.auth = None
def login(self, credentials):
"""设置认证信息"""
self.auth = HTTPBasicAuth(
credentials["username"],
credentials["app_password"]
)
def format_content(self, content):
"""将 Markdown 转为 HTML"""
import markdown
return markdown.markdown(content)
def publish(self, title, content, status="draft"):
"""发布文章"""
url = f"{self.base_url}/wp/v2/posts"
data = {
"title": title,
"content": self.format_content(content),
"status": status # draft, publish
}
response = requests.post(
url,
json=data,
auth=self.auth
)
if response.status_code == 201:
return {
"success": True,
"id": response.json()["id"],
"url": response.json()["link"]
}
else:
return {
"success": False,
"error": response.text
}内容格式化
不同平台对内容格式要求不同:
Markdown 转换
python
import markdown
def markdown_to_html(md_content):
"""将 Markdown 转为 HTML"""
html = markdown.markdown(
md_content,
extensions=['extra', 'codehilite', 'toc']
)
return html提取摘要
python
def extract_summary(content, max_length=200):
"""提取文章摘要"""
# 移除 Markdown 标记
import re
clean_content = re.sub(r'[#*`_\[\]]', '', content)
# 截取前 N 个字符
summary = clean_content[:max_length]
if len(clean_content) > max_length:
summary += "..."
return summary提取封面图
python
def extract_first_image(content):
"""提取文章中的第一张图片"""
import re
# Markdown 格式的图片
img_pattern = r'!\[.*?\]\((.*?)\)'
match = re.search(img_pattern, content)
if match:
return match.group(1)
return None批量发布管理器
python
class ContentDistributor:
"""内容分发管理器"""
def __init__(self):
self.publishers = []
def add_publisher(self, publisher):
"""添加发布平台"""
self.publishers.append(publisher)
def distribute(self, title, content, platforms=None):
"""分发到多个平台"""
if platforms is None:
platforms = [p.__class__.__name__ for p in self.publishers]
results = {}
for publisher in self.publishers:
publisher_name = publisher.__class__.__name__
if publisher_name in platforms:
try:
result = publisher.publish(title, content)
results[publisher_name] = result
except Exception as e:
results[publisher_name] = {
"success": False,
"error": str(e)
}
return results
# 使用示例
distributor = ContentDistributor()
# 添加平台
juejin = JuejinPublisher()
juejin.login({"cookie": "your-cookie"})
distributor.add_publisher(juejin)
wordpress = WordPressPublisher("https://yourblog.com/wp-json")
wordpress.login({"username": "user", "app_password": "pass"})
distributor.add_publisher(wordpress)
# 分发
results = distributor.distribute(
title="我的第一篇文章",
content="文章内容...",
platforms=["JuejinPublisher", "WordPressPublisher"]
)
print(results)错误处理和重试
python
import time
from functools import wraps
def retry_on_failure(max_retries=3, delay=2):
"""失败重试装饰器"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_retries - 1:
raise
print(f"第 {attempt + 1} 次失败,{delay} 秒后重试...")
time.sleep(delay)
return None
return wrapper
return decorator
# 使用
@retry_on_failure(max_retries=3)
def publish_with_retry(title, content):
return publisher.publish(title, content)安全考虑
保护敏感信息
python
# 不要硬编码敏感信息
# 使用环境变量
import os
cookie = os.getenv("JUEJIN_COOKIE")
username = os.getenv("WORDPRESS_USERNAME")
password = os.getenv("WORDPRESS_PASSWORD")速率限制
python
import time
class RateLimiter:
def __init__(self, calls_per_minute=10):
self.calls_per_minute = calls_per_minute
self.calls = []
def wait_if_needed(self):
"""如果需要,等待以遵守速率限制"""
now = time.time()
# 移除超过 1 分钟的记录
self.calls = [call for call in self.calls if now - call < 60]
if len(self.calls) >= self.calls_per_minute:
# 等待到最早的调用超过 1 分钟
sleep_time = 60 - (now - self.calls[0])
if sleep_time > 0:
time.sleep(sleep_time)
self.calls.append(now)下一步
API 集成完成,接下来处理内容格式化。