php淘宝商城网站源码,asp网站开发框架,平面设计速成培训机构,wordpress 无广告视频插件FLUX小红书极致真实V2图像生成工具Python爬虫#xff1a;内容创作者的自动化工作流实战 在小红书运营中#xff0c;高质量配图是内容传播的关键。但每天手动制作几十张风格统一、细节真实的图片#xff0c;对创作者来说几乎是不可能完成的任务。你是否也经历过这样的场景 Win64; x64) AppleWebKit/537.36 } # 模拟获取热门搜索页此处为示意代码 response requests.get(https://www.xiaohongshu.com/explore, headersheaders) soup BeautifulSoup(response.text, html.parser) # 提取热门话题卡片 trend_cards soup.find_all(div, class_trend-card) trends_data [] for card in trend_cards[:10]: # 取前10个热门话题 title card.find(h3).get_text(stripTrue) if card.find(h3) else desc card.find(p, class_desc).get_text(stripTrue) if card.find(p, class_desc) else # 关键词提取与清洗 keywords extract_keywords(title desc) trends_data.append({ title: title, description: desc, keywords: keywords, estimated_volume: estimate_search_volume(keywords) }) return trends_data def extract_keywords(text): 提取核心关键词过滤停用词 # 使用jieba分词并过滤常见停用词 words jieba.lcut(text) stop_words {的, 了, 在, 是, 我, 有, 和, 就, 不, 人, 都, 一, 一个} filtered_words [w for w in words if len(w) 1 and w not in stop_words] # 统计词频取前5个高频词 word_count Counter(filtered_words) return [word for word, count in word_count.most_common(5)] def estimate_search_volume(keywords): 基于关键词组合估算搜索热度简化版 # 实际项目中可对接百度指数、微信指数等API base_score len(keywords) * 100 if any(k in [推荐, 测评, 教程, 攻略] for k in keywords): base_score 200 if any(k in [平价, 学生党, 小众, 高级感] for k in keywords): base_score 150 return base_score # 使用示例 if __name__ __main__: trends fetch_xiaohongshu_trends() print(热门话题分析结果) for i, trend in enumerate(trends, 1): print(f{i}. {trend[title]} | 关键词: {, .join(trend[keywords])} | 预估热度: {trend[estimated_volume]})这段代码的核心价值在于它不只是抓取标题而是通过中文分词和语义分析将原始文本转化为可用于图像生成的结构化提示词。比如抓取到“春日野餐必备清单”这个标题系统会自动提取出“春日”、“野餐”、“布艺餐垫”、“藤编篮子”、“三明治”、“柠檬水”等具体视觉元素而不是直接用标题作为提示词。2.2 FLUX模型集成与批量生成模块FLUX小红书极致真实V2模型通常以LoRA形式存在需要与Stable Diffusion WebUI或ComfyUI配合使用。我们采用ComfyUI的API接口方式因为它更适合自动化任务且支持异步调用和批量处理。import json import time import requests from typing import List, Dict, Any class FluxImageGenerator: def __init__(self, comfyui_url: str http://localhost:8188): self.comfyui_url comfyui_url.rstrip(/) def queue_prompt(self, prompt_data: Dict[str, Any]) - str: 向ComfyUI提交生成任务 try: response requests.post( f{self.comfyui_url}/prompt, json{prompt: prompt_data}, timeout30 ) response.raise_for_status() return response.json().get(prompt_id, ) except Exception as e: print(f提交任务失败: {e}) return def get_history(self, prompt_id: str) - Dict[str, Any]: 获取生成历史 try: response requests.get( f{self.comfyui_url}/history/{prompt_id}, timeout10 ) return response.json() except Exception as e: print(f获取历史失败: {e}) return {} def wait_for_completion(self, prompt_id: str, timeout: int 600) - bool: 等待生成完成 start_time time.time() while time.time() - start_time timeout: history self.get_history(prompt_id) if prompt_id in history and status in history[prompt_id]: status history[prompt_id][status] if status.get(completed, False): return True elif status.get(failed, False): return False time.sleep(5) return False def generate_images_batch(self, prompts: List[Dict[str, Any]]) - List[str]: 批量生成图片 image_paths [] for i, prompt_config in enumerate(prompts): print(f正在生成第{i1}/{len(prompts)}张图片...) # 构建ComfyUI工作流JSON简化版 workflow self.build_flux_workflow(prompt_config) prompt_id self.queue_prompt(workflow) if not prompt_id: continue if self.wait_for_completion(prompt_id): history self.get_history(prompt_id) if prompt_id in history and outputs in history[prompt_id]: outputs history[prompt_id][outputs] for node_id, output_data in outputs.items(): if images in output_data: for img_info in output_data[images]: image_paths.append(img_info[filename]) else: print(f第{i1}张图片生成超时) # 添加间隔避免服务器压力过大 time.sleep(2) return image_paths def build_flux_workflow(self, config: Dict[str, Any]) - Dict[str, Any]: 构建FLUX小红书真实风格工作流 # 这里是ComfyUI节点工作流的JSON表示 # 实际使用时需根据你的ComfyUI配置调整 workflow { 3: { inputs: { seed: config.get(seed, 12345), steps: config.get(steps, 30), cfg: config.get(cfg, 3.5), sampler_name: dpmpp_2m_sde_gpu, scheduler: normal, denoise: 1, model: [4, 0], positive: [6, 0], negative: [7, 0], latent_image: [5, 0] }, class_type: KSampler, function: sample }, 4: { inputs: { ckpt_name: flux_realistic_v2.safetensors }, class_type: CheckpointLoaderSimple, function: load_checkpoint }, 5: { inputs: { width: config.get(width, 1024), height: config.get(height, 1365), batch_size: 1 }, class_type: EmptyLatentImage, function: generate }, 6: { inputs: { text: fxhs, {config[prompt]}, natural lighting, daily life style, realistic skin texture, high detail, 8k, clip: [4, 1] }, class_type: CLIPTextEncode, function: encode }, 7: { inputs: { text: text, watermark, signature, low quality, blurry, deformed, bad anatomy, clip: [4, 1] }, class_type: CLIPTextEncode, function: encode } } return workflow # 使用示例 if __name__ __main__: generator FluxImageGenerator() # 准备一批生成任务 generation_tasks [ { prompt: ins风北欧风客厅浅灰色布艺沙发原木茶几绿植点缀午后阳光斜射柔焦效果, width: 1024, height: 1365, steps: 30 }, { prompt: 手冲咖啡制作过程白色陶瓷手冲壶深棕色咖啡液流入玻璃分享壶木质背景蒸汽升腾特写镜头, width: 1024, height: 1365, steps: 30 } ] image_files generator.generate_images_batch(generation_tasks) print(f成功生成{len(image_files)}张图片: {image_files})这个模块的关键设计在于它将FLUX模型的特性融入了工作流默认添加了xhs触发词确保风格匹配正向提示词强化了“natural lighting”、“daily life style”等小红书真实风格关键词负向提示词则专门排除了AI生成常见的瑕疵。更重要的是它支持批量异步处理可以同时提交多个任务并在后台轮询状态大大提升了整体效率。2.3 图片后处理与多平台适配模块生成的图片需要经过一系列后处理才能直接用于发布。这个模块不仅做简单的裁剪和压缩还实现了智能适配和品牌化处理。from PIL import Image, ImageDraw, ImageFont import os import numpy as np class ImageProcessor: def __init__(self, font_path: str None): # 如果没有指定字体使用系统默认字体 self.font_path font_path or arial.ttf def resize_for_platform(self, image_path: str, platform: str) - Image.Image: 根据不同平台要求调整图片尺寸 with Image.open(image_path) as img: # 转换为RGB模式处理透明背景 if img.mode in (RGBA, LA, P): background Image.new(RGB, img.size, (255, 255, 255)) if img.mode P: img img.convert(RGBA) background.paste(img, maskimg.split()[-1] if img.mode RGBA else None) img background # 根据平台选择尺寸 sizes { xiaohongshu: (1024, 1365), # 3:4 竖版 weixin: (900, 500), # 16:9 横版 douyin: (1080, 1080), # 1:1 正方形 weibo: (1200, 675) # 16:9 横版 } target_size sizes.get(platform, (1024, 1365)) # 智能裁剪优先保留中心区域但检测主要对象位置 return self.smart_resize(img, target_size) def smart_resize(self, img: Image.Image, target_size: tuple) - Image.Image: 智能缩放保留主要内容区域 original_width, original_height img.size target_width, target_height target_size # 计算缩放比例 scale_w target_width / original_width scale_h target_height / original_height scale max(scale_w, scale_h) # 先按比例缩放 new_width int(original_width * scale) new_height int(original_height * scale) resized_img img.resize((new_width, new_height), Image.Resampling.LANCZOS) # 再居中裁剪 left (new_width - target_width) // 2 top (new_height - target_height) // 2 right left target_width bottom top target_height return resized_img.crop((left, top, right, bottom)) def add_watermark(self, image: Image.Image, text: str, position: str bottom-right) - Image.Image: 添加半透明水印 watermark image.copy() draw ImageDraw.Draw(watermark, RGBA) try: font ImageFont.truetype(self.font_path, 24) except: font ImageFont.load_default() # 计算文本尺寸 bbox draw.textbbox((0, 0), text, fontfont) text_width bbox[2] - bbox[0] text_height bbox[3] - bbox[1] # 设置位置 if position bottom-right: x image.width - text_width - 20 y image.height - text_height - 10 elif position top-left: x 20 y 10 else: # center x (image.width - text_width) // 2 y (image.height - text_height) // 2 # 绘制半透明背景 draw.rectangle([x-10, y-5, xtext_width10, ytext_height5], fill(0, 0, 0, 100)) # 绘制文字 draw.text((x, y), text, fill(255, 255, 255, 255), fontfont) # 混合原图和水印 return Image.alpha_composite( image.convert(RGBA), watermark ).convert(RGB) def optimize_for_web(self, image: Image.Image, quality: int 85) - bytes: 优化图片用于网页发布 from io import BytesIO buffer BytesIO() # 转换为RGB如果需要 if image.mode in (RGBA, LA): background Image.new(RGB, image.size, (255, 255, 255)) background.paste(image, maskimage.split()[-1]) image background # 保存为JPEG格式 image.save(buffer, formatJPEG, qualityquality, optimizeTrue) return buffer.getvalue() def batch_process(self, input_dir: str, output_dir: str, platforms: List[str] None): 批量处理目录中的图片 if platforms is None: platforms [xiaohongshu, weixin, douyin] # 创建输出目录 os.makedirs(output_dir, exist_okTrue) for filename in os.listdir(input_dir): if filename.lower().endswith((.png, .jpg, .jpeg)): image_path os.path.join(input_dir, filename) base_name os.path.splitext(filename)[0] for platform in platforms: try: # 调整尺寸 processed_img self.resize_for_platform(image_path, platform) # 添加水印 watermarked_img self.add_watermark( processed_img, f{platform}_content, bottom-right ) # 优化并保存 output_path os.path.join( output_dir, f{base_name}_{platform}.jpg ) # 保存优化后的图片 optimized_bytes self.optimize_for_web(watermarked_img) with open(output_path, wb) as f: f.write(optimized_bytes) print(f已处理: {filename} → {platform}) except Exception as e: print(f处理{filename}时出错: {e}) # 使用示例 if __name__ __main__: processor ImageProcessor() # 处理单张图片 processed_img processor.resize_for_platform( generated_image.png, xiaohongshu ) watermarked_img processor.add_watermark(processed_img, mybrand) # 批量处理 processor.batch_process( input_dir./generated/, output_dir./processed/, platforms[xiaohongshu, weixin] )这个后处理模块体现了真正的工程思维它不只是机械地裁剪图片而是通过智能缩放算法尽可能保留原图的主要内容区域水印添加考虑了不同平台的视觉习惯图片优化则平衡了文件大小和视觉质量。对于内容创作者来说这意味着生成的图片可以直接上传无需再打开Photoshop进行二次处理。3. 实战案例从零搭建个人内容工厂3.1 项目架构与环境准备要让这套自动化工作流稳定运行合理的架构设计至关重要。我们采用轻量级但可靠的方案避免过度工程化核心服务ComfyUI作为FLUX模型的运行环境配置GPU加速数据管道Python脚本作为主控程序负责调度爬虫、生成、后处理各环节存储管理本地文件系统 SQLite数据库记录每次生成的任务状态、参数、效果反馈监控告警简单的日志记录和邮件通知当生成失败率超过阈值时自动提醒环境准备步骤安装ComfyUI并配置FLUX小红书极致真实V2模型git clone https://github.com/comfyanonymous/ComfyUI.git cd ComfyUI python main.py --listen 0.0.0.0:8188下载FLUX模型文件到ComfyUI/models/loras/目录Flux_小红书真实风格丨日常照片丨极致逼真_V2.safetensors确保模型权重设置为0.7-0.8这是该模型的最佳表现区间安装Python依赖pip install requests beautifulsoup4 pillow jieba numpy创建项目目录结构xhs-automation/ ├── crawler/ │ ├── __init__.py │ └── xiaohongshu.py ├── generator/ │ ├── __init__.py │ └── flux_generator.py ├── processor/ │ ├── __init__.py │ └── image_processor.py ├── data/ │ ├── raw/ # 原始爬取数据 │ ├── generated/ # 生成的原始图片 │ └── processed/ # 处理后的成品 ├── logs/ └── config.py # 配置文件3.2 端到端工作流执行下面是一个完整的端到端执行脚本它将前面介绍的各个模块串联起来#!/usr/bin/env python3 # -*- coding: utf-8 -*- 小红书内容自动化工作流主程序 import os import time import logging from datetime import datetime from crawler.xiaohongshu import fetch_xiaohongshu_trends from generator.flux_generator import FluxImageGenerator from processor.image_processor import ImageProcessor # 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(logs/automation.log), logging.StreamHandler() ] ) logger logging.getLogger(__name__) class XiaohongshuContentFactory: def __init__(self): self.crawler None self.generator FluxImageGenerator() self.processor ImageProcessor() # 创建必要目录 for dir_path in [data/raw, data/generated, data/processed, logs]: os.makedirs(dir_path, exist_okTrue) def run_full_pipeline(self, max_trends: int 5, images_per_trend: int 3): 运行完整的内容生产流水线 logger.info( 启动小红书内容自动化工作流 ) # 步骤1采集热点话题 logger.info(步骤1采集小红书热点话题...) try: trends fetch_xiaohongshu_trends(max_trendsmax_trends) logger.info(f成功获取{len(trends)}个热门话题) except Exception as e: logger.error(f话题采集失败: {e}) return # 步骤2为每个话题生成图片 logger.info(步骤2生成高质量配图...) all_generated_files [] for i, trend in enumerate(trends): logger.info(f正在处理第{i1}个话题: {trend[title]}) # 构建提示词 base_prompt f{trend[title]} {trend[description]} # 添加小红书风格关键词 full_prompt fxhs, {base_prompt}, natural lighting, daily life style, realistic skin texture, high detail, 8k # 生成多张变体 generation_configs [] for j in range(images_per_trend): config { prompt: full_prompt, width: 1024, height: 1365, steps: 30, seed: int(time.time()) j } generation_configs.append(config) # 批量生成 try: generated_files self.generator.generate_images_batch(generation_configs) all_generated_files.extend(generated_files) logger.info(f为{trend[title]}生成{len(generated_files)}张图片) except Exception as e: logger.error(f生成图片时出错: {e}) # 步骤3后处理与多平台适配 logger.info(步骤3图片后处理与多平台适配...) try: self.processor.batch_process( input_dirdata/generated/, output_dirdata/processed/, platforms[xiaohongshu, weixin, douyin] ) logger.info(图片后处理完成) except Exception as e: logger.error(f图片处理失败: {e}) # 步骤4生成报告 self.generate_report(trends, all_generated_files) logger.info( 内容自动化工作流执行完成 ) def generate_report(self, trends: list, generated_files: list): 生成执行报告 report_path flogs/report_{datetime.now().strftime(%Y%m%d_%H%M%S)}.txt with open(report_path, w, encodingutf-8) as f: f.write(小红书内容自动化工作流执行报告\n) f.write( * 50 \n) f.write(f执行时间: {datetime.now().strftime(%Y-%m-%d %H:%M:%S)}\n) f.write(f处理话题数: {len(trends)}\n) f.write(f生成图片数: {len(generated_files)}\n) f.write(f处理平台: 小红书、微信公众号、抖音\n\n) f.write(热门话题列表:\n) for i, trend in enumerate(trends, 1): f.write(f{i}. {trend[title]}\n) f.write(f 关键词: {, .join(trend[keywords])}\n) f.write(f 预估热度: {trend[estimated_volume]}\n\n) f.write(生成图片统计:\n) for file in generated_files[:10]: # 只显示前10个 f.write(f - {os.path.basename(file)}\n) if len(generated_files) 10: f.write(f ... 还有{len(generated_files)-10}张图片\n) logger.info(f执行报告已保存至: {report_path}) # 主程序入口 if __name__ __main__: factory XiaohongshuContentFactory() # 运行完整流水线 factory.run_full_pipeline( max_trends3, # 每次处理3个热门话题 images_per_trend2 # 每个话题生成2张图片 )这个主程序展示了如何将各个技术模块有机整合。它不仅仅是功能的堆砌而是考虑了实际使用中的各种细节日志记录便于问题排查、错误处理保证流程稳定性、报告生成提供可视化结果、目录结构清晰便于维护。3.3 效果评估与持续优化任何自动化系统都需要持续的评估和优化。我们建立了一个简单的反馈机制让系统能够自我学习和改进效果评分体系为每张生成的图片设置基础评分维度风格匹配度是否符合小红书真实风格主题相关性是否准确反映了提示词内容视觉质量清晰度、细节丰富度、色彩协调性A/B测试框架对同一话题生成不同参数配置的图片发布后对比数据表现参数自动调优基于历史数据自动调整最佳的CFG值、采样步数等参数import sqlite3 import json from datetime import datetime class PerformanceTracker: def __init__(self, db_path: str data/performance.db): self.db_path db_path self.init_database() def init_database(self): 初始化性能跟踪数据库 conn sqlite3.connect(self.db_path) cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS generation_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT NOT NULL, prompt TEXT NOT NULL, parameters TEXT NOT NULL, image_path TEXT NOT NULL, platform TEXT NOT NULL, status TEXT DEFAULT pending, metrics TEXT, feedback TEXT ) ) conn.commit() conn.close() def log_generation(self, prompt: str, parameters: dict, image_path: str, platform: str): 记录生成日志 conn sqlite3.connect(self.db_path) cursor conn.cursor() cursor.execute( INSERT INTO generation_logs (timestamp, prompt, parameters, image_path, platform, status) VALUES (?, ?, ?, ?, ?, ?) , ( datetime.now().isoformat(), prompt, json.dumps(parameters), image_path, platform, success )) conn.commit() conn.close() def update_metrics(self, image_path: str, metrics: dict, feedback: str ): 更新图片效果指标 conn sqlite3.connect(self.db_path) cursor conn.cursor() cursor.execute( UPDATE generation_logs SET metrics ?, feedback ?, status completed WHERE image_path ? , ( json.dumps(metrics), feedback, image_path )) conn.commit() conn.close() # 使用示例在生成完成后记录性能数据 tracker PerformanceTracker() tracker.log_generation( promptins风北欧风客厅..., parameters{steps: 30, cfg: 3.5, width: 1024}, image_pathdata/generated/20241201_123456.png, platformxiaohongshu ) # 后续可以根据metrics数据进行分析和优化通过这样的跟踪机制系统不仅能告诉你“生成了多少张图片”还能告诉你“哪些参数组合效果最好”、“哪个话题的图片互动率最高”从而形成真正的数据驱动优化闭环。4. 实践中的经验与建议4.1 提示词工程的实用技巧FLUX小红书极致真实V2模型对提示词非常敏感但并不需要复杂的术语。经过多次实践我们总结出几个简单有效的技巧场景化描述优于抽象概念不要写“高质量人像”而要写“25岁女性穿着米色针织衫坐在窗边喝咖啡阳光从左侧45度角照射皮肤有自然光泽背景虚化”善用小红书专属触发词xhs是必须的它告诉模型要遵循小红书的视觉规范daily life style和natural lighting是提升真实感的关键短语控制细节层次先确定主体人物/物品再添加环境背景/光线最后补充质感材质/纹理。这样生成的图片结构更清晰负面提示词要具体除了通用的text, watermark, low quality还可以添加perfect symmetry, plastic skin, studio lighting等针对性的排除项4.2 硬件与性能优化建议这套工作流对硬件有一定要求但我们发现通过合理配置可以在消费级显卡上获得良好效果GPU选择RTX 3090及以上显卡可以流畅运行RTX 3060也能胜任只是生成速度稍慢内存管理ComfyUI默认会占用大量显存建议在启动时添加--gpu-only --lowvram参数批量策略不要一次性提交过多任务建议每次5-10个间隔3-5秒这样既保证效率又避免服务器崩溃缓存机制对常用的提示词组合建立缓存避免重复计算可以提升30%以上的整体效率4.3 内容合规与风险控制自动化内容生产必须重视合规性。我们的实践建议包括版权意识虽然FLUX模型生成的图片版权归使用者所有但提示词中应避免直接引用受版权保护的品牌名称、人物肖像等内容审核在发布前增加简单的AI审核环节检查是否包含敏感内容或不当元素人工复核设置关键节点的人工审核特别是首次使用新提示词或新主题时数据隔离爬虫获取的数据仅用于内部生成不存储用户个人信息符合基本的数据安全要求这套方案不是要完全取代人工创作而是解放创作者的生产力让他们