7.5. 配图生成
目的：由于小红书发布文章要求图文格式，因此在生成文案的基础上，我们还需要生成图文搭配文案进行发布。
原实现思路：
起初我打算使用deepseek的文生图模型Janus进行本地部署生成，参考博客：Deepseek发布的Janus-Pro-1B初体验但后来尝试使用后发现Janus现阶段对于这类特定任务的生成图还不太能够胜任。以下是我尝试使用文案让Janus生成的图片：
现实现思路：
- 当下普遍的方案是使用文案生成一段相关的html代码，再使用python中的自动化库来进行相应部分的截图，最后将截图与文案进行组合，形成图文格式。
代码实现：
html生成：
''' Author: yeffky Date: 2025-02-14 08:43:28 LastEditTime: 2025-02-15 19:28:28 ''' import requests import json import os from datetime import datetime def build_prompt(drafts): prompt = "根据下面的小红书文案，帮我生成一个html页面，包含小红书的封面（需要一个卡片状的封面，上面只需文案内容即可，需要吸引眼球），以及下方几个要点内容，要点内容和封面我希望制作成卡片形式，并且每一部分的div请为我附上属性id，id为'card1', 'card2', ...。要求符合小红书平台的图文要求规则以及平替风格，还要符合小红书平台的用户审美。回复只要给出代码即可，请不要添加多余表达" return f"""{prompt} \n\n小红书文案：\n\n{drafts}""" def get_deepseek_response(prompt, api_key): url = "https://api.deepseek.com/chat/completions" headers = { "Authorization": f"Bearer {api_key}", 'Content-Type': 'application/json', 'Accept': 'application/json', } payload = json.dumps({ "messages": [ { "content": prompt, "role": "user" } ], "model": "deepseek-reasoner", "frequency_penalty": 0, "max_tokens": 2048, "presence_penalty": 0, "response_format": { "type": "text" }, "stop": None, "stream": False, "stream_options": None, "temperature": 1, "top_p": 1, "tools": None, "tool_choice": "none", "logprobs": False, "top_logprobs": None }) response = None while not response: try: print("发送请求") response = requests.post(url, data=payload, headers=headers, timeout=200) response.raise_for_status() if not response.json(): response = None except requests.exceptions.RequestException as e: print(f"请求失败：{str(e)}，开始重试...") response = None return response.json()['choices'][0]['message']['content'] def generate_html(): api_key = os.getenv("DEEPSEEK_API_KEY") today = datetime.now().strftime("%Y-%m-%d") file_path = "./xiaohongshu_drafts/小红书_推广文案_千战系列" + today +".txt" drafts = open(file_path, "r", encoding="utf-8").read() prompt = build_prompt(drafts=drafts) response = get_deepseek_response(prompt, api_key) print(response) with open('./pic_generate/pic.html', 'w', encoding='utf-8') as f: f.write(response)截图：
''' Author: yeffky Date: 2025-02-14 09:41:09 LastEditTime: 2025-02-15 10:44:51 ''' from playwright.sync_api import sync_playwright import time import re def generate_pic(url): # 启动浏览器 player = sync_playwright().start() # 初始化Playwright并启动 chrome_driver = player.chromium # 获取Chromium浏览器实例 browser = chrome_driver.launch(headless=False) # 启动浏览器，headless=False表示以非无头模式启动，即显示浏览器窗口 context = browser.new_context() # 创建一个新的浏览器上下文（类似于一个新的浏览器窗口） page = context.new_page() # 在该上下文中创建一个新的页面（标签页） # 访问页面 card_cnt = 0 with(open('./pic_generate/pic.html', 'r', encoding='utf-8')) as f: page_content = f.read() card_cnt = len(re.findall(r'<div class="card" id="card\d+">', page_content)) print(card_cnt) page.goto(url) # 导航到指定的URL # 截取相关卡片的截图 for i in range(1, card_cnt + 1): card_pic = page.query_selector(f"id=card{i}") # 使用CSS选择器查找页面中的搜索按钮元素 card_pic.screenshot(path=f"./pictures/card{i}.png") # 对搜索按钮元素进行截图并保存为b.png # 停止访问 context.close() # 关闭浏览器上下文 browser.close() # 关闭浏览器 player.stop() # 停止Playwright if __name__ == '__main__': url = 'D:/Project/UUCrawl/Code/pic_generate/pic.html' generate_pic(url)
7.6. 自动化发布
- 目的：将生成的图片和文案自动发布到小红书
- 实现思路：
- 1.使用python中的selenium库，模拟页面操作，登陆后需要将cookie保存下来，下次使用时直接读取cookie，避免重复登陆。同时保存一份token，每次调用登录时检查token是否过期，如未过期则无需登录操作。
- 2.登录后，模拟页面操作前往发布页面，使用send_keys()方法输入标题和正文，使用click()方法点击发布按钮。
- 参考开源项目：xhs_ai_publisher
- 代码实现：
'''
Author: yeffky
Date: 2025-02-15 20:28:32
LastEditTime: 2025-02-17 14:08:45
'''
import sys
sys.path.append("./")
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
from selenium.webdriver.common.keys import Keys
from utils import line_process
import time
import json
import os
class XiaohongshuClient:
def __init__(self):
self.driver = webdriver.Chrome()
self.wait = WebDriverWait(self.driver, 10)
# 获取当前执行文件所在目录
current_dir = os.path.dirname(os.path.abspath(__file__))
self.token_file = os.path.join(current_dir, "xiaohongshu_token.json")
self.cookies_file = os.path.join(current_dir, "xiaohongshu_cookies.json")
self.token = self._load_token()
self._load_cookies()
def _load_token(self):
"""从文件加载token"""
if os.path.exists(self.token_file):
try:
with open(self.token_file, 'r') as f:
token_data = json.load(f)
# 检查token是否过期
if token_data.get('expire_time', 0) > time.time():
return token_data.get('token')
except:
pass
return None
def _save_token(self, token):
"""保存token到文件"""
token_data = {
'token': token,
# token有效期设为30天
'expire_time': time.time() + 30 * 24 * 3600
}
with open(self.token_file, 'w') as f:
json.dump(token_data, f)
def _load_cookies(self):
"""从文件加载cookies"""
if os.path.exists(self.cookies_file):
try:
with open(self.cookies_file, 'r') as f:
cookies = json.load(f)
self.driver.get("https://creator.xiaohongshu.com")
for cookie in cookies:
self.driver.add_cookie(cookie)
except:
pass
def _save_cookies(self):
"""保存cookies到文件"""
cookies = self.driver.get_cookies()
with open(self.cookies_file, 'w') as f:
json.dump(cookies, f)
def login(self, phone, country_code="+86"):
"""登录小红书"""
# 如果token有效则直接返回
if self.token:
return
# 尝试加载cookies进行登录
self.driver.get("https://creator.xiaohongshu.com/login")
self._load_cookies()
self.driver.refresh()
time.sleep(3)
# 检查是否已经登录
if self.driver.current_url != "https://creator.xiaohongshu.com/login":
print("使用cookies登录成功")
self.token = self._load_token()
self._save_cookies()
time.sleep(2)
return
else:
# 清理无效的cookies
self.driver.delete_all_cookies()
print("无效的cookies，已清理")
# 如果cookies登录失败，则进行手动登录
self.driver.get("https://creator.xiaohongshu.com/login")
# 等待登录页面加载完成
time.sleep(5)
# 点击国家区号输入框
country_input = self.wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, "input[placeholder='请选择选项']")))
country_input.click()
time.sleep(5)
# 等待区号列表出现并点击+886
# 等待区号列表出现并点击+86
try:
self.driver.find_element(By.XPATH, "/html/body/div[1]/div/div/div/div[2]/div[1]/div[2]/div/div/div/div/div/div[2]/div[1]/div[1]/div/div/div[1]/input").click()
time.sleep(3)
self.driver.find_element(By.XPATH, "/html/body/div[1]/div/div/div/div[2]/div[1]/div[2]/div/div/div/div/div/div[2]/div[1]/div[1]/div/div/div[1]/input").send_keys(country_code)
time.sleep(3)
# self.driver.find_element(By.XPATH, "/html/body/div[6]/div/div").click()
# china_option = self.wait.until(EC.element_to_be_clickable((By.XPATH, "//div[contains(@class, 'css-cqcgee')]//div[contains(text(), '+86')]")))
time.sleep(5)
except Exception as e:
print("无法找到国家区号选项")
print(e)
# 定位手机号输入框
phone_input = self.wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, "input[placeholder='手机号']")))
phone_input.clear()
phone_input.send_keys(phone)
# 点击发送验证码按钮
try:
send_code_btn = self.wait.until(EC.element_to_be_clickable((By.CSS_SELECTOR, ".css-uyobdj")))
send_code_btn.click()
except:
# 尝试其他可能的选择器
try:
send_code_btn = self.wait.until(EC.element_to_be_clickable((By.CSS_SELECTOR, ".css-1vfl29")))
send_code_btn.click()
except:
try:
send_code_btn = self.wait.until(EC.element_to_be_clickable((By.XPATH, "//button[contains(text(),'发送验证码')]")))
send_code_btn.click()
except:
print("无法找到发送验证码按钮")
# 输入验证码
verification_code = input("请输入验证码: ")
code_input = self.wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, "input[placeholder='验证码']")))
code_input.clear()
code_input.send_keys(verification_code)
# 点击登录按钮
login_button = self.wait.until(EC.element_to_be_clickable((By.CSS_SELECTOR, ".beer-login-btn")))
login_button.click()
# 等待登录成功,获取token
time.sleep(3)
# 保存cookies
self._save_cookies()
# 关闭浏览器
# self.driver.quit()
# print(f"获取到的token: {token}")
# if token:
# self._save_token(token)
# self.token = token
# else:
# print("未能获取到token")
def post_article(self, title, content, images=None):
"""发布文章
Args:
title: 文章标题
content: 文章内容
images: 图片路径列表
"""
# 如果token失效则重新登录
# 设置token
# self.driver.execute_script(f'localStorage.setItem("token", "{self.token}")')
time.sleep(3)
print("点击发布按钮")
# 点击发布按钮
publish_btn = self.wait.until(EC.element_to_be_clickable((By.CSS_SELECTOR, ".btn.el-tooltip__trigger.el-tooltip__trigger")))
publish_btn.click()
# 如果是发布视频，则不操作这一步
# 切换到上传图文
time.sleep(3)
tabs = self.driver.find_elements(By.CSS_SELECTOR, ".creator-tab")
if len(tabs) > 1:
tabs[1].click()
time.sleep(3)
# # 输入标题和内容
# title_input = self.wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, ".title-input")))
# content_input = self.driver.find_element(By.CSS_SELECTOR, ".content-input")
# title_input.send_keys(title)
# content_input.send_keys(content)
# 上传图片
if images:
upload_input = self.driver.find_element(By.CSS_SELECTOR,'input[type="file"]')
# 将所有图片路径用\n连接成一个字符串一次性上传
upload_input.send_keys('\n'.join(images))
time.sleep(1)
time.sleep(3)
JS_ADD_TEXT_TO_INPUT = """
var elm = arguments[0], txt = arguments[1];
elm.value += txt;
elm.dispatchEvent(new Event('change'));
"""
title_input = self.wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, ".d-text")))
self.driver.execute_script(JS_ADD_TEXT_TO_INPUT, title_input, title)
# title_input.send_keys(title)
# Start of Selection
# Start of Selection
print(content)
JS_ADD_TEXT_TO_P = """
var elm = arguments[0], txt = arguments[1];
elm.textContent = txt;
"""
content_input = self.wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, ".ql-editor")))
p_element = content_input.find_element(By.CSS_SELECTOR, "p")
print(p_element)
self.driver.execute_script(JS_ADD_TEXT_TO_P, p_element, content)
# content_input.send_keys(123)
# 发布
time.sleep(600)
submit_btn = self.driver.find_element(By.CSS_SELECTOR, ".el-button.publishBtn")
submit_btn.click()
def close(self):
"""关闭浏览器"""
self.driver.quit()
def post_article():
poster = XiaohongshuClient()
phone = open('./docs/phone.txt').read()
poster.login(phone)
print("登录成功")
print("开始发布文章")
print(os.getcwd())
title = open('./xiaohongshu_drafts/小红书_推广文案_千战系列2025-02-15.txt', 'r', encoding='utf-8').readline()
article = line_process.get_article('./xiaohongshu_drafts/小红书_推广文案_千战系列2025-02-15.txt')
print(article)
images = os.listdir('./pictures')
images = map(lambda x: os.path.join(r"D:\Project\UUCrawl\Code\pictures", x), images)
poster.post_article(title, article, images)
poster.close()
7.7. 主程序
from crawler import ip_crawler, data_crawler
from analysis import data_analysis
from pic_generate import pic_generate, html_generate
from post import xiaohongshu_post
import file_handler
if __name__ == '__main__':
url = 'D:/Project/UUCrawl/Code/pic_generate/pic.html'
# 获取IP
ip = ip_crawler.crawl_ip()
# 获取数据
data = data_crawler.crawl_data()
# 数据分析
data_analysis.analysis_data()
file_handler.start_observer()
# 生成html
html_generate.generate_html()
# 生成图片
pic_generate.generate_pic(url)
# 发布小红书
xiaohongshu_post.post_article()