huanghai/dsProject
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

'commit'

Browse Source
This commit is contained in:
HuangHai
2025-08-21 09:16:43 +08:00
parent c577687727
commit e543809bc2
3 changed files with 106 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
dsLightRag/JiMeng/T1_JmTxt2Img.py
View File

@@ -1,22 +1,39 @@
import base64
import json
import logging
import os
import uuid
from Config.Config import OBS_SERVER, OBS_PREFIX
from Config.Config import OBS_SERVER, OBS_PREFIX, OBS_BUCKET
from JiMeng.Kit.JmCommon import JmCommon
from JiMeng.Kit.JmErrorCode import JmErrorCode
from Util.ObsUtil import ObsUploader
# 设置日志
logger = logging.getLogger(__name__)
logger.level = logging.INFO
# 添加控制台处理器
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
# 设置日志格式
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
console_handler.setFormatter(formatter)
# 将处理器添加到logger
logger.addHandler(console_handler)
class JmTxt2Img:
action = "CVProcess"
# V2.1
req_key = "jimeng_high_aes_general_v21_L"
# V3.0
# req_key="jimeng_t2i_v30"
@staticmethod
def generate_image(prompt, save_img_path):
def generate_image(prompt):
"""生成图片"""
# 创建请求体
req = {
@@ -37,47 +54,39 @@ class JmTxt2Img:
# 获取图片Base64数据
img_base64 = jo.get("data", {}).get("binary_data_base64", [""])[0]
# 确保目录存在
os.makedirs(os.path.dirname(save_img_path), exist_ok=True)
# 对 Base64 字符串进行解码并保存为文件
try:
# 注意：有些Base64字符串可能有前缀，需要去除
if img_base64.startswith('data:image'):
img_base64 = img_base64.split(',')[1]
bytes_data = base64.b64decode(img_base64)
with open(save_img_path, 'wb') as f:
f.write(bytes_data)
print(f"文件保存成功！文件位置: {save_img_path}")
# 创建上传器实例
uploader = ObsUploader()
# 生成一个随机的图片文件名
jpg_file_name = f"{str(uuid.uuid4())}.jpg"
# 上传参数
object_key = OBS_PREFIX + "/" + file_path
object_key = OBS_PREFIX + "/JiMeng/" + jpg_file_name
# 执行上传
success, result = uploader.upload_file(object_key, file_path)
success, result = uploader.upload_base64_image(object_key, bytes_data)
# 处理结果
if success:
logger.info(f'{file_path}上传成功!')
logger.info(f'{jpg_file_name}上传成功!')
# 获取上传文件的 URL
file_url = f"https://{OBS_BUCKET}.{OBS_SERVER}/{object_key}"
img_url_list.append(file_url)
return file_url
except Exception as e:
print(f"保存图片失败: {str(e)}")
raise Exception(f"保存图片失败: {str(e)}")
return None
@staticmethod
def main():
# 示例提示词
prompt = "雨天街头，一位头戴黄色安全帽、身穿荧光黄安全背心与棕色外套、搭配蓝色牛仔裤和棕色工装靴的中国工人，背着一个中国小女孩。工人脖子上侧挂一个小熊玩偶包，拖住小女孩臀部的手同时挂着一袋小熊饼干。小女孩双手搂着工人的脖子。小女孩穿红色雨衣，戴红色雨衣帽。两个人露出微笑。背景是居民楼和树木，地面潮湿，氛围温馨。"
# 保存图片路径
save_image_path = r"D:\Text2Img.jpg"
JmTxt2Img.generate_image(prompt, save_image_path)
print(f"保存图片路径：{save_image_path}")
url = JmTxt2Img.generate_image(prompt)
logger.info(f"图片URL: {url}")
if __name__ == "__main__":

77
dsLightRag/Util/ObsUtil.py
View File

@@ -1,6 +1,7 @@
# pip install esdk-obs-python --trusted-host pypi.org
import traceback
import logging
import base64 # 添加base64模块导入
from obs import ObsClient
from obs import PutObjectHeader
@@ -31,7 +32,7 @@ class ObsUploader:
:param bucket_name: 桶名称
:param object_key: 对象键（上传后的文件名）
:param file_path: 本地文件路径
:param meta元数据
:param metadata: 元数据
:param headers: 请求头
:return: 成功返回True和响应信息，失败返回False和错误信息
"""
@@ -76,3 +77,77 @@ class ObsUploader:
logger.error(error_msg)
logger.error(traceback.format_exc())
return False, {'error': str(e), 'traceback': traceback.format_exc()}
def upload_base64_image(self, object_key, base64_data, bucket_name=OBS_BUCKET, metadata=None, headers=None):
"""
上传Base64编码的图片或原始二进制图片数据到OBS
:param bucket_name: 桶名称
:param object_key: 对象键（上传后的文件名）
:param base64_data: Base64编码的图片字符串、字节形式或原始二进制图片数据
:param metadata: 元数据
:param headers: 请求头
:return: 成功返回True和响应信息，失败返回False和错误信息
"""
try:
# 初始化默认参数
if metadata is None:
metadata = {}
if headers is None:
headers = PutObjectHeader()
# 处理数据
if isinstance(base64_data, bytes):
try:
# 尝试解码为UTF-8字符串（假设是Base64编码的字符串的字节形式）
base64_string = base64_data.decode('utf-8')
# 移除Base64前缀（如果有）
if 'base64,' in base64_string:
base64_string = base64_string.split('base64,')[1]
# 解码Base64数据
image_data = base64.b64decode(base64_string)
except UnicodeDecodeError:
# 解码失败，假设是原始二进制图片数据
image_data = base64_data
else:
# 假设是字符串
base64_string = str(base64_data)
# 移除Base64前缀（如果有）
if 'base64,' in base64_string:
base64_string = base64_string.split('base64,')[1]
# 解码Base64数据
image_data = base64.b64decode(base64_string)
# 执行对象上传
resp = self.obs_client.putObject(
bucket_name,
object_key,
content=image_data,
metadata=metadata,
headers=headers
)
# 检查响应状态
if resp.status < 300:
logger.info(f'图片上传成功: -> {bucket_name}/{object_key}')
logger.info(f'requestId: {resp.requestId}')
logger.info(f'etag: {resp.body.etag}')
return True, {
'requestId': resp.requestId,
'etag': resp.body.etag,
'versionId': resp.body.versionId,
'storageClass': resp.body.storageClass
}
else:
error_msg = f'图片上传失败: {resp.errorCode} - {resp.errorMessage}'
logger.error(error_msg)
logger.error(f'requestId: {resp.requestId}')
return False, {
'requestId': resp.requestId,
'errorCode': resp.errorCode,
'errorMessage': resp.errorMessage
}
except Exception as e:
error_msg = f'图片上传异常: {str(e)}'
logger.error(error_msg)
logger.error(traceback.format_exc())
return False, {'error': str(e), 'traceback': traceback.format_exc()}

BIN
dsLightRag/Util/__pycache__/ObsUtil.cpython-310.pyc
View File

Binary file not shown.