腾讯混元调用代码实践
目录
查看资源是否用尽：
ai3d的资源包，可以免费领取
api调用实例，亲测ok：
查看资源是否用尽：
https://console.cloud.tencent.com/hunyuan/packagesai3d的资源包，可以免费领取
https://console.cloud.tencent.com/ai3d/packagesapi调用实例，亲测ok：
# -*- coding: utf-8 -*- import time import json import requests from tencentcloud.common import credential from tencentcloud.common.profile.client_profile import ClientProfile from tencentcloud.common.profile.http_profile import HttpProfile from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException from tencentcloud.hunyuan.v20230901 import hunyuan_client, models SECRET_ID = "SECRET_ID" SECRET_KEY = "SECRET_KEY " REGION = "ap-guangzhou" def init_client(): """初始化混元API客户端""" try: cred = credential.Credential(SECRET_ID, SECRET_KEY) http_profile = HttpProfile() http_profile.endpoint = "hunyuan.tencentcloudapi.com" http_profile.reqTimeout = 60 # 超时时间60秒 client_profile = ClientProfile() client_profile.httpProfile = http_profile client = hunyuan_client.HunyuanClient(cred, REGION, client_profile) print("✅ API客户端初始化成功") return client except Exception as e: print(f"❌ API客户端初始化失败: {e}") return None def submit_hunyuan_image_job(prompt, resolution="1024:1024", num=1, style=None): client = init_client() if not client: return None print("提交混元图片生成任务") if style: print(f"风格: {style}") print("-" * 60) try: # 创建请求对象 req = models.SubmitHunyuanImageJobRequest() req.Prompt = prompt req.Resolution = resolution req.Num = num # 可选参数：风格 if style: req.Style = style # 提交任务 resp = client.SubmitHunyuanImageJob(req) print(f"✅ 任务提交成功 Job ID: {resp.JobId}") return resp.JobId except TencentCloudSDKException as e: print(f"\n❌ 提交失败: {e}") print(f" 错误代码: {e.code}") print(f" 错误信息: {e.message}") return None except Exception as e: print(f"\n❌ 未知错误: {e}") return None def query_hunyuan_image_job(job_id, max_retries=60, interval=5): client = init_client() if not client: return None print(f"\n开始查询任务: {job_id}") print("-" * 60) req = models.QueryHunyuanImageJobRequest() req.JobId = job_id for retry_count in range(max_retries): try: print(f"[{retry_count + 1}/{max_retries}] 查询任务状态...") resp = client.QueryHunyuanImageJob(req) status_code = resp.JobStatusCode status_msg = resp.JobStatusMsg print(f" 状态码: {status_code} 状态信息: {status_msg}") # 状态码判断（关键修改） if status_code == '5': # 任务完成 print("✅ 任务处理完成！") print("=" * 60) result = {"job_id": job_id, "status": "completed", "image_urls": list(resp.ResultImage) if hasattr(resp, 'ResultImage') and resp.ResultImage else [], "revised_prompts": list(resp.RevisedPrompt) if hasattr(resp, 'RevisedPrompt') and resp.RevisedPrompt else [], "request_id": resp.RequestId} # 打印图片URL if result['image_urls']: print(f"📷 生成的图片 ({len(result['image_urls'])}张):") for idx, url in enumerate(result['image_urls'], 1): print(f" {idx}. {url}") else: print("⚠️ 任务完成但未返回图片URL") # 打印优化后的提示词 if result['revised_prompts']: print(f"📝 优化后的提示词:") for idx, prompt in enumerate(result['revised_prompts'], 1): print(f" {idx}. {prompt}") return result elif status_code in ['0', '1', '2']: # 等待、处理中、运行中 - 继续等待 status_desc = {'0': '⏳ 任务排队中', '1': '⏳ 任务初始化中', '2': '🎨 任务运行中（生成图片）'}.get(status_code, '⏳ 任务处理中') print(f"{status_desc}，等待 {interval} 秒后重试...") elif status_code in ['3', '4']: # 真正的失败状态 print(f"\n❌ 任务失败: {status_msg}") if hasattr(resp, 'JobErrorMsg') and resp.JobErrorMsg: print(f" 错误详情: {resp.JobErrorMsg}") return None else: # 未知状态码 print(f"\n⚠️ 未知状态码: {status_code}") print(f" 状态信息: {status_msg}") # 如果不是完成状态且不是已知的处理状态，可能是失败 if status_code != '5': print(" 任务可能失败，停止查询") return None # 等待下次查询 if retry_count < max_retries - 1: time.sleep(interval) except TencentCloudSDKException as e: print(f"❌ 查询异常: {e}") if "JobNotFound" in str(e): print("任务不存在，请检查Job ID") return None time.sleep(interval) except Exception as e: print(f"❌ 未知错误: {e}") time.sleep(interval) print("\n❌ 查询超时，任务可能仍在处理中") return None def get_fresh_image_url(job_id): """获取新的有效图片URL""" cred = credential.Credential(SECRET_ID, SECRET_KEY) http_profile = HttpProfile() http_profile.endpoint = "hunyuan.tencentcloudapi.com" client_profile = ClientProfile() client_profile.httpProfile = http_profile client = hunyuan_client.HunyuanClient(cred, REGION, client_profile) req = models.QueryHunyuanImageJobRequest() req.JobId = job_id try: resp = client.QueryHunyuanImageJob(req) if resp.JobStatusCode == '5' and resp.ResultImage: return resp.ResultImage[0] except Exception as e: print(f"查询失败: {e}") return None def submit_and_wait(prompt, resolution="1024:1024", num=1, style=None, max_retries=30, interval=3): job_id = submit_hunyuan_image_job(prompt, resolution, num, style) if not job_id: return None result = query_hunyuan_image_job(job_id, max_retries, interval) return result def download_image(url, save_path=None): """下载生成的图片""" if not url: print("❌ 图片URL为空") return None if save_path is None: import hashlib url_hash = hashlib.md5(url.encode()).hexdigest()[:8] save_path = f"generated_image_{url_hash}.png" # print(f"\n开始下载图片: {save_path}") headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'} try: response = requests.get(url, headers=headers, timeout=30) if response.status_code == 200: with open(save_path, 'wb') as f: f.write(response.content) print(f"✅ 图片已保存: {save_path}") return save_path else: print(f"❌ 下载失败: HTTP {response.status_code}") return None except Exception as e: print(f"❌ 下载异常: {e}") return None if __name__ == "__main__": prompt1 = "生成一个金黄色戒指的3D设计稿，然后拆分3d打印零件" result1 = submit_and_wait(prompt1, resolution="1024:1024", num=1,interval=2) if result1 and result1['image_urls']: # 下载图片 for idx, url in enumerate(result1['image_urls'], 1): download_image(url, f"example1_image_{idx}.png") # prompt2 = "山水风景，有山有水有云雾，中国传统水墨画风格" # # 风格代码: 101-水墨画, 102-概念艺术, 103-油画, 104-素描, 105-水彩画 # result2 = submit_and_wait(prompt2, resolution="1280:720", num=1, style="101")