使用DeepSeek+本地知识库，尝试从0到1搭建高度定制化工作流（自动化篇）

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 = " api.deepseek /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 mon.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver mon.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(" creator.xiaohongshu ") 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(" creator.xiaohongshu /login") self._load_cookies() self.driver.refresh() time.sleep(3) # 检查是否已经登录 if self.driver.current_url != " creator.xiaohongshu /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(" creator.xiaohongshu /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()