实操审校说明
本文只用于学习合规自动化。请只采集自己拥有权限、允许转载或明确开放的数据，遵守 robots.txt、版权要求和各平台 API 协议；不要绕过登录、验证码、付费墙或反爬限制，也不要批量发布低质、重复或骚扰性内容。建议默认进入草稿箱，由人工审核后再发布。
Python 自动化实操教程：自动采集数据 + 自动发布内容
📅 更新时间：2026 年 6 月
👤 适合人群：有 Python 基础（会用 pip 安装包）的开发者、想自动化内容运营的创作者
⏱ 预计上手时间：跟着教程做，2 小时内跑通第一条自动化流水线
一、这套教程能帮你做什么？
手动操作（每天耗费 3 小时）：
打开网页 → 复制内容 → 粘贴整理 → 排版 → 手动发布 → 重复...
用 Python 自动化之后：
运行脚本 → 自动采集 → 自动清洗 → 自动发布 → 你去喝咖啡本教程覆盖的自动化场景：
- 自动采集网页文章内容
- AI 自动改写/清洗内容
- 自动发布到 WordPress 博客
- 自动推送到微信公众号草稿箱
- 自动发微博
- 一键全流程脚本
二、环境准备
# 安装必要的库
pip install requests # 发送 HTTP 请求
pip install beautifulsoup4 # 解析 HTML 页面
pip install selenium # 处理 JS 动态页面
pip install python-dotenv # 管理 API 密钥
pip install schedule # 定时任务
pip install openai # 调用 AI API（兼容 DeepSeek）
# 可选：Selenium 需要下载 ChromeDriver
# 访问 https://chromedriver.chromium.org 下载对应版本创建 .env 文件保存敏感信息（不要提交到 Git！）：
DEEPSEEK_API_KEY=sk-your-key-here
WP_URL=https://yourblog.com
WP_USERNAME=admin
WP_APP_PASSWORD=xxxx xxxx xxxx # WordPress 应用密码
WX_APPID=wx1234567890
WX_APPSECRET=your-appsecret
WEIBO_ACCESS_TOKEN=your-token三、模块一：自动采集网页数据
3.1 采集静态页面（最常用）
适用于：博客文章、新闻页面、一般内容网站
# scraper.py
import requests
from bs4 import BeautifulSoup
import time
import random
def scrape_article(url):
"""
采集单个网页的标题和正文
返回: {"title": str, "content": str, "images": list}
"""
headers = {
# 声明清晰的 User-Agent，遵守目标站点访问规则
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/120.0.0.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
}
try:
response = requests.get(url, headers=headers, timeout=15)
response.encoding = "utf-8"
soup = BeautifulSoup(response.text, "html.parser")
# 提取标题（优先级：h1 > title 标签）
title_tag = soup.find("h1") or soup.find("title")
title = title_tag.get_text(strip=True) if title_tag else "无标题"
# 提取正文（过滤掉导航、广告等无关内容）
# 策略：找包含最多文字的 div/article/section
content_candidates = soup.find_all(["article", "div", "section"])
best_content = ""
for tag in content_candidates:
text = tag.get_text(strip=True)
if len(text) > len(best_content):
best_content = text
# 也可以直接提取所有 p 标签
paragraphs = soup.find_all("p")
content = "\n\n".join([p.get_text(strip=True) for p in paragraphs
if len(p.get_text(strip=True)) > 20]) # 过滤太短的段落
# 提取图片
images = []
for img in soup.find_all("img"):
src = img.get("src", "")
if src and not src.startswith("data:"): # 排除 base64 图片
if not src.startswith("http"):
# 补全相对路径
from urllib.parse import urljoin
src = urljoin(url, src)
images.append(src)
return {
"title": title,
"content": content or best_content,
"images": images[:5], # 最多取 5 张图
"url": url
}
except requests.Timeout:
print(f"❌ 超时：{url}")
return None
except Exception as e:
print(f"❌ 采集失败 {url}: {e}")
return None
def batch_scrape(url_list):
"""
批量采集多个 URL
包含随机延迟，避免被封 IP
"""
results = []
total = len(url_list)
for i, url in enumerate(url_list, 1):
print(f"[{i}/{total}] 采集中: {url[:60]}...")
data = scrape_article(url)
if data:
results.append(data)
print(f" ✅ 成功：{data['title'][:30]}")
# 随机延迟 2-5 秒，模拟人类访问节奏
delay = random.uniform(2, 5)
time.sleep(delay)
print(f"\n✅ 采集完成：成功 {len(results)}/{total} 条")
return results
# ===== 测试运行 =====
if __name__ == "__main__":
test_urls = [
"https://www.36kr.com/p/2901234567890", # 替换成你想采集的真实 URL
]
articles = batch_scrape(test_urls)
for article in articles:
print(f"\n标题：{article['title']}")
print(f"内容前200字：{article['content'][:200]}")3.2 采集动态页面（JS 渲染的内容）
适用于：需要滚动加载的页面、自己有权限访问的后台页面或动态内容
# scraper_dynamic.py
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import time
def create_driver(headless=True):
"""创建 Chrome 浏览器驱动"""
options = Options()
if headless:
options.add_argument("--headless") # 无界面模式，不弹出浏览器窗口
options.add_argument("--no-sandbox")
options.add_argument("--disable-dev-shm-usage")
options.add_argument("--window-size=1920,1080")
options.add_argument("User-Agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0")
driver = webdriver.Chrome(options=options)
return driver
def scrape_dynamic_page(url, wait_selector="article"):
"""
采集动态渲染的页面
wait_selector: 等待某个 CSS 选择器出现后再开始采集
"""
driver = create_driver()
try:
driver.get(url)
# 等待关键元素加载（最多等 10 秒）
try:
WebDriverWait(driver, 10).until(
EC.presence_of_element_located((By.CSS_SELECTOR, wait_selector))
)
except:
time.sleep(3) # 如果等待超时，就固定等 3 秒
# 滚动到底部，触发懒加载
driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
time.sleep(2)
title = driver.find_element(By.TAG_NAME, "h1").text
# 提取所有段落
paragraphs = driver.find_elements(By.CSS_SELECTOR, "article p, .content p, .post-content p")
if not paragraphs:
# 如果上面的选择器没找到，尝试所有 p 标签
paragraphs = driver.find_elements(By.TAG_NAME, "p")
content = "\n\n".join([p.text for p in paragraphs if len(p.text) > 20])
return {"title": title, "content": content, "url": url}
finally:
driver.quit() # 确保浏览器被关闭四、模块二：AI 自动改写内容
避免直接搬运，用 AI 改写成原创内容：
# ai_rewriter.py
import os
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
# 使用 DeepSeek（便宜好用，中文最强）
client = OpenAI(
api_key=os.getenv("DEEPSEEK_API_KEY"),
base_url="https://api.deepseek.com/v1"
)
def rewrite_article(title, content, style="公众号"):
"""
用 AI 改写文章内容
style 可选: 公众号/知乎/简书/微博
"""
style_prompts = {
"公众号": "口语化、接地气、有情绪感染力，适合手机阅读，短段落多用",
"知乎": "深度分析、理性客观、引用数据，有专业感",
"微博": "简短有力，140字以内，加2-3个话题标签",
"简书": "文艺感、有故事性、适合长文",
}
prompt = f"""
你是一个资深内容编辑，请将以下文章改写为{style}风格的原创文章。
【改写要求】
1. 语言风格：{style_prompts.get(style, '通俗易懂')}
2. 改变句子结构，不能直接复制原文句子
3. 保留核心信息和观点，适当扩充细节
4. 去掉广告语、无关的链接引导
5. 生成3个备选标题
【原文标题】
{title}
【原文内容】
{content[:3000]}
请直接输出改写后的文章，格式：
【推荐标题1】：xxx
【推荐标题2】：xxx
【推荐标题3】：xxx
【正文】
（改写后的完整文章）
"""
response = client.chat.completions.create(
model="deepseek-v4-flash",
messages=[{"role": "user", "content": prompt}],
temperature=0.7, # 适当创意性
max_tokens=3000
)
result = response.choices[0].message.content
return result
def clean_content(raw_text):
"""清洗文本，去掉广告词、多余空白等"""
import re
# 去除多余空行
text = re.sub(r'\n{3,}', '\n\n', raw_text)
# 去除常见广告关键词
ad_words = ["点击阅读全文", "扫码关注", "免费领取", "限时优惠", "点击下载"]
for word in ad_words:
text = text.replace(word, "")
return text.strip()
# 测试
if __name__ == "__main__":
test_content = "这里放原文内容..."
result = rewrite_article("测试标题", test_content, style="公众号")
print(result)五、模块三：自动发布到各平台
5.1 发布到 WordPress 博客
# publisher_wordpress.py
import requests
import base64
import os
from dotenv import load_dotenv
load_dotenv()
def publish_to_wordpress(title, content, tags=[], category_ids=[1], status="publish"):
"""
发布文章到 WordPress
status: "draft"=保存草稿（建议默认），"publish"=确认后直接发布
"""
wp_url = os.getenv("WP_URL")
username = os.getenv("WP_USERNAME")
# 在 WordPress 后台 → 用户 → 个人资料 → 应用密码 生成
app_password = os.getenv("WP_APP_PASSWORD")
# Basic Auth 认证
credentials = f"{username}:{app_password}"
token = base64.b64encode(credentials.encode()).decode("utf-8")
headers = {
"Authorization": f"Basic {token}",
"Content-Type": "application/json"
}
payload = {
"title": title,
"content": content,
"status": status,
"categories": category_ids,
"tags": tags,
"format": "standard"
}
response = requests.post(
f"{wp_url}/wp-json/wp/v2/posts",
json=payload,
headers=headers,
timeout=30
)
if response.status_code == 201:
post_id = response.json()["id"]
post_url = response.json()["link"]
print(f"✅ WordPress 发布成功！")
print(f" 文章 ID: {post_id}")
print(f" 文章链接: {post_url}")
return post_id
else:
print(f"❌ WordPress 发布失败: {response.status_code}")
print(f" 错误信息: {response.text[:200]}")
return None5.2 发布到微信公众号草稿箱
# publisher_wechat.py
import requests
import os
from dotenv import load_dotenv
load_dotenv()
class WechatPublisher:
def __init__(self):
self.appid = os.getenv("WX_APPID")
self.appsecret = os.getenv("WX_APPSECRET")
self.access_token = self._get_access_token()
def _get_access_token(self):
"""获取微信 access_token（有效期 2 小时）"""
url = "https://api.weixin.qq.com/cgi-bin/token"
params = {
"grant_type": "client_credential",
"appid": self.appid,
"secret": self.appsecret
}
res = requests.get(url, params=params, timeout=10).json()
if "access_token" in res:
print(f"✅ 获取 access_token 成功")
return res["access_token"]
else:
print(f"❌ 获取 access_token 失败: {res}")
return None
def upload_thumb_image(self, image_path):
"""
上传封面图（必须先上传，获取 media_id）
image_path: 本地图片路径，要求 ≤ 64KB，格式 JPG
"""
url = f"https://api.weixin.qq.com/cgi-bin/material/add_material"
params = {"access_token": self.access_token, "type": "image"}
with open(image_path, "rb") as f:
files = {"media": f}
res = requests.post(url, params=params, files=files).json()
if "media_id" in res:
print(f"✅ 图片上传成功，media_id: {res['media_id']}")
return res["media_id"]
else:
print(f"❌ 图片上传失败: {res}")
return None
def add_draft(self, title, content, author="", digest="", thumb_media_id=""):
"""
添加草稿（建议在公众号后台手动审核后发布）
content: 支持 HTML 格式
digest: 摘要，显示在文章列表中（建议 50 字内）
thumb_media_id: 封面图 media_id（需提前上传）
"""
if not self.access_token:
print("❌ 无效的 access_token")
return None
url = f"https://api.weixin.qq.com/cgi-bin/draft/add"
params = {"access_token": self.access_token}
payload = {
"articles": [{
"title": title,
"author": author,
"digest": digest,
"content": content,
"content_source_url": "", # 原文链接（可选）
"thumb_media_id": thumb_media_id,
"need_open_comment": 1, # 开启评论
"only_fans_can_comment": 0 # 允许所有人评论
}]
}
res = requests.post(url, params=params, json=payload).json()
if "media_id" in res:
print(f"✅ 微信草稿添加成功！media_id: {res['media_id']}")
print(f" 请在公众号后台 → 草稿箱 中查看并发布")
return res["media_id"]
else:
print(f"❌ 添加草稿失败: {res}")
return None5.3 发微博（微博开放平台 API）
# publisher_weibo.py
import requests
import os
from dotenv import load_dotenv
load_dotenv()
def post_weibo_text(text):
"""
发布纯文字微博
注意：微博内容 ≤ 140 字
"""
access_token = os.getenv("WEIBO_ACCESS_TOKEN")
# 截断超长文字
if len(text) > 140:
text = text[:137] + "..."
url = "https://api.weibo.com/2/statuses/update.json"
data = {
"access_token": access_token,
"status": text
}
response = requests.post(url, data=data, timeout=10)
result = response.json()
if "id" in result:
print(f"✅ 微博发布成功！微博 ID: {result['id']}")
return result["id"]
else:
print(f"❌ 微博发布失败: {result}")
return None六、完整自动化流水线（一键运行）
把上面所有模块整合：
# pipeline.py - 完整自动化流水线
import json
import time
import os
from datetime import datetime
from dotenv import load_dotenv
# 导入上面写的各个模块
from scraper import batch_scrape
from ai_rewriter import rewrite_article, clean_content
from publisher_wordpress import publish_to_wordpress
from publisher_wechat import WechatPublisher
from publisher_weibo import post_weibo_text
load_dotenv()
def run_pipeline(
url_list,
publish_to=["wordpress", "wechat"],
save_backup=True
):
"""
完整采集-改写-发布流水线
参数：
- url_list: 要采集的 URL 列表
- publish_to: 发布到哪些平台 ["wordpress", "wechat", "weibo"]
- save_backup: 是否保存本地备份
"""
print("=" * 60)
print(f"🚀 流水线启动 | 时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}")
print(f" 待处理 URL: {len(url_list)} 个")
print(f" 发布平台: {', '.join(publish_to)}")
print("=" * 60)
# ===== 第一步：批量采集 =====
print("\n📡 第一步：采集内容...")
raw_articles = batch_scrape(url_list)
if not raw_articles:
print("❌ 没有采集到任何内容，流水线终止")
return
# ===== 第二步：AI 改写 =====
print(f"\n🤖 第二步：AI 改写 {len(raw_articles)} 篇文章...")
processed_articles = []
for i, article in enumerate(raw_articles, 1):
print(f"\n[{i}/{len(raw_articles)}] 改写: {article['title'][:30]}...")
try:
rewritten = rewrite_article(
title=article["title"],
content=clean_content(article["content"]),
style="公众号"
)
# 解析 AI 返回的标题和正文
lines = rewritten.split("\n")
new_title = article["title"] # 默认用原标题
content_start = 0
for j, line in enumerate(lines):
if "【推荐标题1】" in line:
new_title = line.replace("【推荐标题1】：", "").strip()
if "【正文】" in line:
content_start = j + 1
break
new_content = "\n".join(lines[content_start:])
processed_articles.append({
"original_title": article["title"],
"title": new_title,
"content": new_content,
"source_url": article["url"]
})
print(f" ✅ 改写完成: {new_title[:30]}")
time.sleep(1) # 避免 AI API 限速
except Exception as e:
print(f" ❌ 改写失败: {e}")
continue
# ===== 第三步：本地备份 =====
if save_backup:
backup_file = f"backup_{datetime.now().strftime('%Y%m%d_%H%M')}.json"
with open(backup_file, "w", encoding="utf-8") as f:
json.dump(processed_articles, f, ensure_ascii=False, indent=2)
print(f"\n💾 已备份到: {backup_file}")
# ===== 第四步：多平台发布 =====
print(f"\n📢 第四步：发布到 {len(publish_to)} 个平台...")
wx_publisher = WechatPublisher() if "wechat" in publish_to else None
for article in processed_articles:
title = article["title"]
content = article["content"]
print(f"\n 发布: {title[:30]}...")
if "wordpress" in publish_to:
publish_to_wordpress(title=title, content=content)
if "wechat" in publish_to and wx_publisher:
wx_publisher.add_draft(
title=title,
content=content,
digest=content[:50]
)
if "weibo" in publish_to:
weibo_text = f"{title} {content[:80]}... 详见主页"
post_weibo_text(weibo_text)
time.sleep(3) # 发布间隔，避免频率限制
print("\n" + "=" * 60)
print(f"✅ 流水线完成！共处理 {len(processed_articles)} 篇文章")
print("=" * 60)
# ===== 定时自动运行 =====
import schedule
def daily_task():
"""每天自动采集指定的 URL 列表"""
urls_to_scrape = [
# 把你想要采集的 URL 放这里
"https://www.36kr.com/...",
"https://juejin.cn/...",
]
run_pipeline(
url_list=urls_to_scrape,
publish_to=["wordpress", "wechat"],
)
# 每天早上 8 点自动运行
schedule.every().day.at("08:00").do(daily_task)
if __name__ == "__main__":
print("⏰ 定时任务已启动，每天 08:00 自动运行...")
print(" 按 Ctrl+C 停止")
while True:
schedule.run_pending()
time.sleep(60)七、实用技巧与注意事项
7.1 防止被封 IP 的策略
import time
import random
# 方法一：随机延迟
time.sleep(random.uniform(2, 8))
# 方法二：使用请求头轮换
USER_AGENTS = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) Safari/537.36",
"Mozilla/5.0 (X11; Linux x86_64) Firefox/121.0",
]
import random
headers = {"User-Agent": random.choice(USER_AGENTS)}7.2 API 调用频率控制
| 平台 | 限制 | 建议 |
| 微信公众号 | 每天群发 1 次 | 草稿可多次添加，手动选择发送时机 |
| 微博 API | 100 次/小时 | 加 sleep(36) 每次请求 |
| DeepSeek API | 60 次/分钟 | 加 sleep(1) |
| WordPress | 无严格限制 | 正常使用即可 |
7.3 合规使用原则
✅ 合法采集场景：
- 采集允许转载的内容（注明来源）
- 采集自己有权限的内容
- 用于学习和个人研究
⚠️ 注意事项：
- 遵守目标网站的 robots.txt
- 不采集需要登录才能看到的付费内容
- 转载必须注明原文来源链接
- 控制访问频率，不给服务器造成压力八、完整工具库速查
# 一键安装所有依赖
pip install requests beautifulsoup4 selenium openai schedule python-dotenv aiohttp jieba pandas| 库 | 用途 |
requests | 发送 HTTP 请求 |
beautifulsoup4 | 解析 HTML |
selenium | 控制浏览器（动态页面） |
openai | 调用 AI API（兼容 DeepSeek） |
schedule | 定时任务 |
python-dotenv | 读取 .env 配置 |
jieba | 中文分词（提取关键词） |
pandas | 数据处理、Excel 操作 |
总结：3 步开始你的自动化
第一步：先跑通单篇采集（scraper.py）
→ 选一个目标网页，能成功提取标题和内容
第二步：接入 AI 改写（ai_rewriter.py）
→ DeepSeek API 免费额度够用，申请就行
第三步：选一个平台发布（wordpress / wechat）
→ WordPress 最简单，先跑通再扩展到其他平台
三步全通了，再组装成完整流水线（pipeline.py）*参考：PingCode Python自动发布完整指南（2026.3）| 优采云内容自动化平台（2026.5）| Python官方文档*
