使用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 = "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 = Nonewhile not response:try:print("發送請求")response = requests.post(url, data=payload, headers=headers, timeout=200)response.raise_for_status()if not response.json():response = Noneexcept requests.exceptions.RequestException as e:print(f"請求失敗：{str(e)}，開始重試...")response = Nonereturn 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 redef 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 = 0with(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()  # 停止Playwrightif __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 osclass 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:passreturn Nonedef _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:passdef _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)returnelse:# 清理無效的cookiesself.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# 等待區號列表出現并點擊+86try: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()# 等待登錄成功,獲取tokentime.sleep(3)# 保存cookiesself._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 Selectionprint(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_handlerif __name__ == '__main__':url = 'D:/Project/UUCrawl/Code/pic_generate/pic.html'# 獲取IPip = ip_crawler.crawl_ip()# 獲取數據data = data_crawler.crawl_data()# 數據分析data_analysis.analysis_data()file_handler.start_observer()# 生成htmlhtml_generate.generate_html()# 生成圖片pic_generate.generate_pic(url)# 發布小紅書xiaohongshu_post.post_article()