模块化说明1
让我先区分前后端代码，然后为您详细注释并创建模块化说明：
前端代码
templates/index.html(注意：这个文件似乎缺失，需要创建)
后端代码
web_app.py(Flask服务器)websockets_api_example.py(ComfyUI API调用)workflow_api.json(工作流配置文件)
让我们先详细注释这些代码：
from flask import Flask, render_template, request, jsonify, send_file
import subprocess
import os
# 创建Flask应用实例
app = Flask(__name__)
# API路由：处理图像生成请求
@app.route('/generate_image', methods=['POST'])
def generate_image():
try:
# 从POST请求中获取提示词文本
# 如果没有提供提示词，使用"default"作为默认值
prompt_text = request.json.get("prompt_text", "default")
# 调用外部Python脚本处理图像生成
# subprocess.run 用于执行外部命令
# check=True 表示如果命令执行失败会抛出异常
# capture_output=True 捕获命令的输出
# text=True 将输出转换为文本格式
result = subprocess.run(
['python', 'C:/1/V2/websockets_api_example.py', prompt_text],
check=True,
capture_output=True,
text=True
)
# 返回成功响应
return jsonify({
"status": "success",
"message": "Image generation started",
"output": result.stdout
})
# 处理子进程执行错误
except subprocess.CalledProcessError as e:
error_message = e.stderr or str(e)
print("Error occurred:", error_message) # 打印错误信息到控制台
return jsonify({
"status": "error",
"message": "Failed to start image generation",
"error": error_message
}), 500
# 处理其他所有异常
except Exception as e:
print("Unexpected error:", str(e))
return jsonify({
"status": "error",
"message": str(e)
}), 500
# API路由：获取最新生成的图像
@app.route('/get_latest_image', methods=['GET'])
def get_latest_image():
try:
# 从文件中读取最新图像的路径
with open("latest_image.txt", "r") as f:
image_path = f.read().strip()
# 检查图像文件是否存在
if os.path.exists(image_path):
return send_file(image_path, mimetype='image/png')
else:
return jsonify({
"status": "error",
"message": "Image not found"
}), 404
except FileNotFoundError:
return jsonify({
"status": "error",
"message": "No image generated yet"
}), 404
# 网页路由：主页
@app.route('/')
def index():
return render_template('index.html')
# 程序入口点
if __name__ == "__main__":
app.run(port=5000) # 在5000端口启动Flask服务器
import websocket
import uuid
import json
import urllib.request
import urllib.parse
import os
import glob
import sys
# 服务器配置
server_address = "127.0.0.1:8188" # ComfyUI服务器地址
client_id = str(uuid.uuid4()) # 生成唯一客户端ID
def queue_prompt(prompt):
"""
将生成请求发送到ComfyUI服务器队列
参数:
prompt: dict, 包含生成参数的字典
返回:
dict: 服务器响应
"""
# 准备请求数据
request_data = {"prompt": prompt, "client_id": client_id}
data = json.dumps(request_data).encode('utf-8')
print("Sending prompt data:", request_data)
# 设置请求头
headers = {'Content-Type': 'application/json'}
req = urllib.request.Request(
f"http://{server_address}/prompt",
data=data,
headers=headers
)
try:
# 发送请求并获取响应
response = urllib.request.urlopen(req)
return json.loads(response.read())
except urllib.error.HTTPError as e:
print(f"HTTPError: {e.code} - {e.reason}")
print(e.read().decode())
raise
def get_image(filename, subfolder, folder_type):
"""
从服务器获取生成的图像
参数:
filename: str, 文件名
subfolder: str, 子文件夹名
folder_type: str, 文件夹类型
返回:
bytes: 图像数据
"""
data = {
"filename": filename,
"subfolder": subfolder,
"type": folder_type
}
url_values = urllib.parse.urlencode(data)
with urllib.request.urlopen(
"http://{}/view?{}".format(server_address, url_values)
) as response:
return response.read()
def get_history(prompt_id):
"""
获取特定提示词ID的历史记录
参数:
prompt_id: str, 提示词ID
返回:
dict: 历史记录数据
"""
with urllib.request.urlopen(
"http://{}/history/{}".format(server_address, prompt_id)
) as response:
return json.loads(response.read())
def get_images(ws, prompt):
"""
通过WebSocket获取图像生成状态
参数:
ws: WebSocket连接对象
prompt: dict, 生成参数
返回:
dict: 输出图像信息
"""
prompt_id = queue_prompt(prompt)['prompt_id']
output_images = {}
# 持续监听WebSocket消息
while True:
out = ws.recv()
if isinstance(out, str):
message = json.loads(out)
if message['type'] == 'executing':
data = message['data']
if data['node'] is None and data['prompt_id'] == prompt_id:
break # 生成完成
else:
continue # 跳过二进制预览数据
return output_images
def generate_image_with_comfyui(prompt_text):
"""
使用ComfyUI生成图像的主函数
参数:
prompt_text: str, 用户输入的提示词
"""
# 构建生成参数
prompt = {
# ... (省略详细配置，与原代码相同)
}
# 建立WebSocket连接
ws = websocket.WebSocket()
ws.connect(f"ws://{server_address}/ws?clientId={client_id}")
# 触发图像生成
images = get_images(ws, prompt)
ws.close()
# 查找生成的图像文件
output_directory = r"C:\AI\ComfyUI-mini\output"
list_of_files = glob.glob(os.path.join(output_directory, "GUO*.PNG"))
if not list_of_files:
print("未找到符合条件的图像文件，请检查生成过程。")
return
# 记录最新生成的图像路径
latest_file = max(list_of_files, key=os.path.getctime)
with open("latest_image.txt", "w") as f:
f.write(latest_file)
print(f"图像生成完毕，保存至: {latest_file}")
# 程序入口点
if __name__ == "__main__":
if len(sys.argv) > 1:
prompt_text = sys.argv[1] # 获取命令行参数作为提示词
generate_image_with_comfyui(prompt_text)
else:
print("提示词未提供。请传入一个提示词作为参数。")
接下来我会创建一个模块化说明文件...
# 项目模块化说明
## 1. Web服务模块 (web_app.py)
### 主要功能
- HTTP服务器启动
- 路由处理
- 请求响应处理
### 核心组件
1. 主页路由 (/)
- 功能：显示用户界面
- 入口：index()
2. 图像生成API (/generate_image)
- 功能：处理生成请求
- 入口：generate_image()
- 参数：prompt_text
3. 图像获取API (/get_latest_image)
- 功能：获取生成的图像
- 入口：get_latest_image()
## 2. ComfyUI接口模块 (websockets_api_example.py)
### 主要功能
- WebSocket通信
- 图像生成控制
- 文件管理
### 核心组件
1. 请求队列管理
- 功能：发送生成请求
- 入口：queue_prompt()
- 参数：prompt
2. WebSocket通信
- 功能：监听生成状态
- 入口：get_images()
- 参数：ws, prompt
3. 图像生成控制
- 功能：控制生成流程
- 入口：generate_image_with_comfyui()
- 参数：prompt_text
4. 文件管理
- 功能：处理生成的图像文件
- 组件：
- get_image()
- get_history()
## 3. 配置模块 (workflow_api.json)
### 主要功能
- 模型配置
- 参数设置
- 工作流定义
### 核心组件
1. 模型设置
- 模型选择
- 参数配置
2. 生成参数
- 图像尺寸
- 生成步骤
- 采样器设置
## 4. 数据存储模块
### 主要功能
- 图像路径记录
- 临时数据存储
### 核心组件
1. latest_image.txt
- 功能：记录最新图像路径
- 使用：图像获取API
## 5. 模块间通信
1. Web服务 -> ComfyUI接口
- 通过subprocess调用
- 参数传递
2. ComfyUI接口 -> 配置模块
- 读取配置信息
- 参数应用
3. ComfyUI接口 -> 数据存储
- 写入图像路径
- 状态记录
## 6. 错误处理
1. Web服务错误
- HTTP错误处理
- 请求异常处理
2. ComfyUI接口错误
- WebSocket连接错误
- 生成失败处理
3. 文件操作错误
- 文件不存在处理
- 权限错误处理
这个模块化说明文件详细描述了项目的各个组件及其功能。您需要我详细解释某个特定模块吗？
阅读建议
