comi-logic/modalDeploy
7
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ADD 增加HeyGem纯享服务器配置

Browse Source 
ADD 增加AutoDL部署
FIX 修复json日志打印
ADD HeyGem允许多类型文件输入(http&upload)
This commit is contained in:
kyj@bowong.ai
2025-04-11 16:35:39 +08:00
parent 8678b7bb7d
commit 5f5d60092b
6 changed files with 324 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

245
AutoDL/AutoDL_pure_heygem.py Normal file
View File

@@ -0,0 +1,245 @@
import json
import os
import shutil
import socket
import subprocess
import time
import traceback
import uuid
from typing import Any, Optional, Union
import httpx
import loguru
import requests
import uvicorn
from fastapi import UploadFile, HTTPException, Depends, FastAPI
from fastapi.routing import APIRoute
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from starlette import status
class HeyGem:
def __init__(self):
def check_port_in_use(port, host='127.0.0.1'):
s = None
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(1)
s.connect((host, int(port)))
return True
except socket.error:
return False
finally:
if s:
s.close()
uid = str(uuid.uuid4())
cmd = "/bin/bash -c \"echo client_id: %s && source /root/.bashrc && nohup python /root/heygem.py >> /root/logs/%s.log 2>&1 &\"" % (uid, uid)
subprocess.run(cmd, shell=True, check=True)
timeout = 30
while timeout > 0:
check = check_port_in_use(8383)
if check:
loguru.logger.success("HeyGem Server started on port 8383")
break
else:
timeout -= 1
time.sleep(2)
if timeout == 0:
raise TimeoutError("HeyGem Server Start timed out")
self.server = FastAPI()
self.server.routes.append(APIRoute(path='/', endpoint=self.api, methods=['POST']))
def infer(self, code, video_path, audio_path) -> str:
def task_submit(uid, video, audio, heygem_url):
"""Submit a task to the API"""
task_submit_api = f'{heygem_url}/easy/submit'
result_json = {
'status': False, 'data': {}, 'msg': ''
}
try:
data = {
"code": uid,
"video_url": video,
"audio_url": audio,
"chaofen": 1,
"watermark_switch": 0,
"pn": 1
}
loguru.logger.info(f'data={json.dumps(data, ensure_ascii=False)}')
with httpx.Client() as client:
resp = client.post(task_submit_api, json=data)
resp_dict = resp.json()
loguru.logger.info(f'submit data: {json.dumps(resp_dict, ensure_ascii=False)}')
if resp_dict['code'] != 10000:
result_json['status'] = False
result_json['msg'] = result_json['msg']
else:
result_json['status'] = True
result_json['data'] = uid
result_json['msg'] = '任务提交成功'
return result_json
except Exception as e:
loguru.logger.info(f'submit task fail case by:{str(e)}')
raise RuntimeError(str(e))
def query_task_progress(heygem_url, heygem_temp_path, task_id: str, interval: int = 10, timeout: int = 60 * 35):
"""Query task progress and wait for completion"""
result_json = {'status': False, 'data': {}, 'msg': ''}
def query_result(t_id: str):
tmp_dict = {'status': True, 'data': dict(), 'msg': ''}
try:
query_task_url = f'{heygem_url}/easy/query'
params = {
'code': t_id
}
with httpx.Client() as client:
resp = client.get(query_task_url, params=params)
resp_dict = resp.json()
loguru.logger.info(f'query task data: {json.dumps(resp_dict, ensure_ascii=False)}')
status_code = resp_dict['code']
if status_code in (9999, 10002, 10003, 10001):
tmp_dict['status'] = False
tmp_dict['msg'] = resp_dict['msg']
elif status_code == 10000:
status_code = resp_dict['data'].get('status', 1)
if status_code == 3:
tmp_dict['status'] = False
tmp_dict['msg'] = resp_dict['data']['msg']
else:
process = resp_dict['data'].get('progress', 20)
if status_code == 2:
process = 100
else:
process = process
result = resp_dict['data'].get('result', '')
tmp_dict['data'] = {'progress': process,
'path': result,
}
else:
pass
except Exception as e:
loguru.logger.info(f'query task fail case by:{str(e)}')
raise RuntimeError(str(e))
return tmp_dict
end = time.time() + timeout
while time.time() < end:
tmp_result = query_result(task_id)
if not tmp_result['status'] or tmp_result['data'].__eq__({}):
result_json['status'] = False
result_json['msg'] = tmp_result['msg']
break
else:
process = tmp_result['data']['progress']
loguru.logger.info(f'query task progress :{process}')
if tmp_result['data']['progress'] < 100:
time.sleep(interval)
loguru.logger.info(f'wait next interval:{interval}')
else:
p = tmp_result['data']['path']
p = p.replace('/', '').replace('\\', '')
result_json['data'] = "%s/%s" % (heygem_temp_path, p)
result_json['status'] = True
break
return result_json
try:
heygem_url = "http://127.0.0.1:8383"
heygem_temp_path = "/code/data/temp"
submit_result = task_submit(code, video_path, audio_path, heygem_url)
if not submit_result['status']:
raise Exception(f"Task submission failed: {submit_result['msg']}")
task_id = submit_result['data']
loguru.logger.info(f'Submitted task: {task_id}')
# Query task progress
progress_result = query_task_progress(heygem_url, heygem_temp_path, task_id, interval=5)
if not progress_result['status']:
raise RuntimeError(f"Task processing failed: {progress_result['msg']}")
# Return the file for download
file_path = progress_result['data']
if not os.path.exists(file_path):
raise FileNotFoundError(f"Output file not found at {file_path}")
return file_path
except Exception as e:
loguru.logger.error(f"Error processing request || {str(e)}")
raise Exception(str(e))
finally:
try:
os.remove(video_path)
os.remove(audio_path)
except:
pass
def download_file(self, url, path):
with requests.get(url, stream=True) as r:
with open(path, 'wb') as f:
shutil.copyfileobj(r.raw, f)
async def api(self, video_file: Optional[Union[UploadFile, str]], audio_file: Optional[Union[UploadFile, str]], token: HTTPAuthorizationCredentials = Depends(HTTPBearer())):
if token.credentials != os.environ["AUTH_TOKEN"]:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect bearer token",
headers={"WWW-Authenticate": "Bearer"},
)
code = str(uuid.uuid4())
try:
if isinstance(video_file, UploadFile):
video_path = "/root/%s" % video_file.filename
with open(video_path, "wb") as f:
data = await video_file.read()
f.write(data)
elif isinstance(video_file, str) and video_file.startswith("http"):
video_path = "/root/%s" % video_file.split("?")[0].split("/")[-1]
self.download_file(video_file, video_path)
else:
return {"status": "fail", "msg": "video file save failed: Not a valid input"}
if isinstance(audio_file, UploadFile):
audio_path = "/root/%s" % audio_file.filename
with open(audio_path, "wb") as f:
data = await audio_file.read()
f.write(data)
elif isinstance(audio_file, str) and audio_file.startswith("http"):
audio_path = "/root/%s" % audio_file.split("?")[0].split("/")[-1]
self.download_file(audio_file, audio_path)
else:
return {"status": "fail", "msg": "audio file save failed: Not a valid input"}
except Exception as e:
return {"status": "fail", "msg": f"video or audio file save failed: {str(e)}"}
loguru.logger.info(f"Enter Inference Module")
try:
file_path = self.infer(code, video_path, audio_path)
try:
loguru.logger.info("Try move file to S3 manually")
# S3 Fallback
import boto3
import yaml
with open("/root/config.yaml", encoding="utf-8",
mode="r+") as config:
yaml_config = yaml.load(config, Loader=yaml.FullLoader)
awss3 = boto3.resource('s3', aws_access_key_id=yaml_config["aws_key_id"],
aws_secret_access_key=yaml_config["aws_access_key"])
awss3.meta.client.upload_file(file_path, "bw-heygem-output", file_path.split(os.path.sep)[-1])
except Exception as e:
return {"status": "fail", "msg": "Failed to move file to S3 manually: " + str(e)}
# resp = FileResponse(path=file_path, filename=Path(file_path).name, status_code=200, headers={
# "Content-Type": "application/octet-stream",
# "Content-Disposition": f"attachment; filename={Path(file_path).name}",
# })
return {"status": "success", "msg":f"{file_path.split(os.path.sep)[-1]}"}
except Exception as e:
traceback.print_exc()
return {"status": "fail", "msg": "Inference module failed: "+str(e)}
if __name__ == "__main__":
heygem = HeyGem()
uvicorn.run(app=heygem.server, host="0.0.0.0", port=6006)

12
AutoDL/heygem.py Normal file
View File

@@ -0,0 +1,12 @@
import site
import subprocess
import sys
pkg_path = site.getsitepackages()[0]
sys.path.append(f"{pkg_path}/heygem/code")
p = subprocess.Popen(f"cd {pkg_path}/heygem/code && "
"python3.8 app_local.py", shell=True)
p.communicate()

70
server_pure_heygem.py
View File

@@ -7,16 +7,14 @@ import subprocess
import time
import traceback
import uuid
from pathlib import Path
from typing import Any
from typing import Any, Optional, Union
import httpx
import loguru
import modal
import requests
from fastapi import Depends, HTTPException, status, UploadFile
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi.responses import Response, FileResponse
image = (
modal.Image.debian_slim( # start from basic Linux with Python
python_version="3.10"
@@ -39,7 +37,7 @@ image = (
.run_commands("python3.8 -m pip install https://github.com/pydata/numexpr/releases/download/v2.8.6/numexpr-2.8.6-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl")
.add_local_file("heygem.py", "/root/heygem.py", copy=True)
.add_local_file("config.yaml", "/root/config.yaml", copy=True)
.workdir("/root").pip_install("boto3").pip_install("PyYAML")
.workdir("/root").pip_install("boto3").pip_install("PyYAML").pip_install("requests")
)
app = modal.App(name="heygem", image=image)
@@ -50,7 +48,7 @@ bucket_output = "bw-heygem-output"
@app.cls(
allow_concurrent_inputs=1, # required for UI startup process which runs several API calls concurrently
max_containers=25, # limit interactive session to 1 container
gpu="T4", # good starter GPU for inference
gpu="L40S", # good starter GPU for inference
cpu=(2,32),
memory=(6144, 32768),
timeout=2160,
@@ -104,7 +102,7 @@ class HeyGem:
@modal.method()
def infer(self, code, video_path, audio_path) -> str:
def task_submit(uid, video, audio, heygem_url) -> dict[str, Any]:
def task_submit(uid, video, audio, heygem_url):
"""Submit a task to the API"""
task_submit_api = f'{heygem_url}/easy/submit'
result_json = {
@@ -119,11 +117,11 @@ class HeyGem:
"watermark_switch": 0,
"pn": 1
}
loguru.logger.info(f'data={data}')
loguru.logger.info(f'data={json.dumps(data, ensure_ascii=False)}')
with httpx.Client() as client:
resp = client.post(task_submit_api, json=data)
resp_dict = resp.json()
loguru.logger.info(f'submit data: {resp_dict}')
loguru.logger.info(f'submit data: {json.dumps(resp_dict, ensure_ascii=False)}')
if resp_dict['code'] != 10000:
result_json['status'] = False
result_json['msg'] = result_json['msg']
@@ -136,7 +134,7 @@ class HeyGem:
loguru.logger.info(f'submit task fail case by:{str(e)}')
raise RuntimeError(str(e))
def query_task_progress(heygem_url, heygem_temp_path, task_id: str, interval: int = 10, timeout: int = 60 * 35) -> dict[str, Any]:
def query_task_progress(heygem_url, heygem_temp_path, task_id: str, interval: int = 10, timeout: int = 60 * 35):
"""Query task progress and wait for completion"""
result_json = {'status': False, 'data': {}, 'msg': ''}
@@ -150,7 +148,7 @@ class HeyGem:
with httpx.Client() as client:
resp = client.get(query_task_url, params=params)
resp_dict = resp.json()
loguru.logger.info(f'query task data: {json.dumps(resp_dict)}')
loguru.logger.info(f'query task data: {json.dumps(resp_dict, ensure_ascii=False)}')
status_code = resp_dict['code']
if status_code in (9999, 10002, 10003, 10001):
tmp_dict['status'] = False
@@ -229,9 +227,13 @@ class HeyGem:
except:
pass
def download_file(self, url, path):
with requests.get(url, stream=True) as r:
with open(path, 'wb') as f:
shutil.copyfileobj(r.raw, f)
@modal.fastapi_endpoint(method="POST")
async def api(self, video_file: UploadFile, audio_file: UploadFile, token: HTTPAuthorizationCredentials = Depends(auth_scheme)) -> Response:
async def api(self, video_file: Optional[Union[UploadFile, str]], audio_file: Optional[Union[UploadFile, str]], token: HTTPAuthorizationCredentials = Depends(auth_scheme)):
if token.credentials != os.environ["AUTH_TOKEN"]:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
@@ -240,17 +242,29 @@ class HeyGem:
)
code = str(uuid.uuid4())
try:
video_path = "/root/%s" % video_file.filename
with open(video_path, "wb") as f:
data = await video_file.read()
f.write(data)
if isinstance(video_file, UploadFile):
video_path = "/root/%s" % video_file.filename
with open(video_path, "wb") as f:
data = await video_file.read()
f.write(data)
elif isinstance(video_file, str) and video_file.startswith("http"):
video_path = "/root/%s" % video_file.split("?")[0].split("/")[-1]
self.download_file(video_file, video_path)
else:
return {"status": "fail", "msg": "video file save failed: Not a valid input"}
audio_path = "/root/%s" % audio_file.filename
with open(audio_path, "wb") as f:
data = await audio_file.read()
f.write(data)
if isinstance(audio_file, UploadFile):
audio_path = "/root/%s" % audio_file.filename
with open(audio_path, "wb") as f:
data = await audio_file.read()
f.write(data)
elif isinstance(audio_file, str) and audio_file.startswith("http"):
audio_path = "/root/%s" % audio_file.split("?")[0].split("/")[-1]
self.download_file(audio_file, audio_path)
else:
return {"status": "fail", "msg": "audio file save failed: Not a valid input"}
except Exception as e:
return Response({"status": "fail", "msg": f"video or audio file save failed: {str(e)}"})
return {"status": "fail", "msg": f"video or audio file save failed: {str(e)}"}
loguru.logger.info(f"Enter Inference Module")
try:
@@ -271,14 +285,14 @@ class HeyGem:
aws_secret_access_key=yaml_config["aws_access_key"])
awss3.meta.client.upload_file(file_path, bucket_output, file_path.split(os.path.sep)[-1])
except:
raise Exception("Failed to move file to S3 manually")
resp = FileResponse(path=file_path, filename=Path(file_path).name, status_code=200, headers={
"Content-Type": "application/octet-stream",
"Content-Disposition": f"attachment; filename={Path(file_path).name}",
})
return resp
return {"status": "fail", "msg": "Failed to move file to S3 manually: " + str(e)}
# resp = FileResponse(path=file_path, filename=Path(file_path).name, status_code=200, headers={
# "Content-Type": "application/octet-stream",
# "Content-Disposition": f"attachment; filename={Path(file_path).name}",
# })
return {"status": "success", "msg":f"{file_path.split(os.path.sep)[-1]}"}
except Exception as e:
traceback.print_exc()
return Response({"status": "fail", "msg": "Inference module failed: "+str(e)})
return {"status": "fail", "msg": "Inference module failed: "+str(e)}

13
server_with_s3.py
View File

@@ -6,6 +6,7 @@ import subprocess
import uuid
from typing import Dict
import loguru
import modal
image = ( # build up a Modal Image to run ComfyUI, step by step
@@ -53,7 +54,7 @@ image = (
.run_commands("comfy node install https://github.com/melMass/comfy_mtb")
.run_commands("echo 3 && comfy node install https://e.coding.net/g-ldyi2063/dev/ComfyUI_SparkTTS.git")
.run_commands("comfy node install https://e.coding.net/g-ldyi2063/dev/ComfyUI-CustomNode.git")
.run_commands("comfy node install https://e.coding.net/g-ldyi2063/dev/cosyvoice_comfyui.git")
.run_commands("echo 04111 && comfy node install https://e.coding.net/g-ldyi2063/dev/cosyvoice_comfyui.git")
.run_commands("echo 4 && comfy node install https://e.coding.net/g-ldyi2063/dev/ComfyUI-LatentSync-Node.git")
.run_commands(
"mkdir -p /root/comfy/ComfyUI/models/ComfyUI-CustomNode/model && rm -rf /root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/model && ln -s /root/comfy/ComfyUI/models/ComfyUI-CustomNode/model /root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/model"
@@ -98,8 +99,7 @@ output_dir = "/root/comfy/ComfyUI/output"
cpu=(2,16),
# memory=(32768, 32768), # (内存预留量, 内存使用上限)
memory=(20480,40960),
retries=1,
enable_memory_snapshot=True,
enable_memory_snapshot=False,
secrets=[secret],
volumes={
"/root/comfy/ComfyUI/models": vol,
@@ -198,7 +198,6 @@ class ComfyUI:
@modal.fastapi_endpoint(method="POST")
def api(self, item: Dict):
from fastapi import Response
try:
# save this updated workflow to a new file
new_workflow_file = item["prompt"]
@@ -207,8 +206,8 @@ class ComfyUI:
if fname is None:
raise RuntimeError("Output File not found")
j = {"file_name": fname}
print("json", j)
return Response(content=json.dumps(j), media_type="application/json")
loguru.logger.success(j)
return j
except Exception as e:
raise RuntimeError(str(e))
finally:
@@ -222,7 +221,7 @@ class ComfyUI:
try:
subprocess.run(cmd, shell=True, check=True)
except:
raise Exception("Failed to launch ComfyUI")
pass
def poll_server_health(self):

23
server_with_s3_auth.py
View File

@@ -6,6 +6,7 @@ import subprocess
import uuid
from typing import Dict
import loguru
import modal
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
@@ -55,7 +56,7 @@ image = (
.run_commands("comfy node install https://github.com/melMass/comfy_mtb")
.run_commands("echo 3 && comfy node install https://e.coding.net/g-ldyi2063/dev/ComfyUI_SparkTTS.git")
.run_commands("comfy node install https://e.coding.net/g-ldyi2063/dev/ComfyUI-CustomNode.git")
.run_commands("comfy node install https://e.coding.net/g-ldyi2063/dev/cosyvoice_comfyui.git")
.run_commands("echo 04111 && comfy node install https://e.coding.net/g-ldyi2063/dev/cosyvoice_comfyui.git")
.run_commands("echo 4 && comfy node install https://e.coding.net/g-ldyi2063/dev/ComfyUI-LatentSync-Node.git")
.run_commands(
"mkdir -p /root/comfy/ComfyUI/models/ComfyUI-CustomNode/model && rm -rf /root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/model && ln -s /root/comfy/ComfyUI/models/ComfyUI-CustomNode/model /root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/model"
@@ -101,8 +102,7 @@ auth_scheme = HTTPBearer()
cpu=(2,16),
# memory=(32768, 32768), # (内存预留量, 内存使用上限)
memory=(20480,40960),
retries=1,
enable_memory_snapshot=True,
enable_memory_snapshot=False,
secrets=[secret, modal.Secret.from_name("web_auth_token")],
volumes={
"/root/comfy/ComfyUI/models": vol,
@@ -204,7 +204,6 @@ class ComfyUI:
detail="Incorrect bearer token",
headers={"WWW-Authenticate": "Bearer"},
)
from fastapi import Response
try:
# save this updated workflow to a new file
new_workflow_file = item["prompt"]
@@ -212,11 +211,13 @@ class ComfyUI:
fname = self.infer.local(new_workflow_file)
if fname is None:
raise RuntimeError("Output File not found")
j = {"file_name": fname}
print("json", j)
return Response(content=json.dumps(j), media_type="application/json")
j = {"status":"success", "file_name": fname}
loguru.logger.success(j)
return j
except Exception as e:
raise RuntimeError(str(e))
j = {"status":"fail", "msg": str(e)}
loguru.logger.error(j)
return j
finally:
print("Purge Garbage By Restart Comfy")
cmd = "comfy stop"
@@ -228,7 +229,7 @@ class ComfyUI:
try:
subprocess.run(cmd, shell=True, check=True)
except:
raise Exception("Failed to launch ComfyUI")
pass
def poll_server_health(self):
import socket
@@ -257,8 +258,4 @@ class ComfyUI:
modal.experimental.stop_fetching_inputs()
raise Exception("ComfyUI server is not healthy, restart failed, stopping container")
# @modal.web_endpoint(method="POST", label="tk")
# def tk_api(self, item: Dict):
# pass
#

16
server_with_s3_auth_heygem.py
View File

@@ -6,6 +6,7 @@ import subprocess
import uuid
from typing import Dict
import loguru
import modal
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
@@ -55,7 +56,7 @@ image = (
.run_commands("comfy node install https://github.com/melMass/comfy_mtb")
.run_commands("echo 3 && comfy node install https://e.coding.net/g-ldyi2063/dev/ComfyUI_SparkTTS.git")
.run_commands("echo 4 && comfy node install https://e.coding.net/g-ldyi2063/dev/ComfyUI-CustomNode.git")
.run_commands("comfy node install https://e.coding.net/g-ldyi2063/dev/cosyvoice_comfyui.git")
.run_commands("echo 04111 && comfy node install https://e.coding.net/g-ldyi2063/dev/cosyvoice_comfyui.git")
.run_commands("echo 4 && comfy node install https://e.coding.net/g-ldyi2063/dev/ComfyUI-LatentSync-Node.git")
.run_commands(
"mkdir -p /root/comfy/ComfyUI/models/ComfyUI-CustomNode/model && rm -rf /root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/model && ln -s /root/comfy/ComfyUI/models/ComfyUI-CustomNode/model /root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/model"
@@ -258,7 +259,6 @@ class ComfyUI:
detail="Incorrect bearer token",
headers={"WWW-Authenticate": "Bearer"},
)
from fastapi import Response
try:
# save this updated workflow to a new file
new_workflow_file = item["prompt"]
@@ -266,11 +266,13 @@ class ComfyUI:
fname = self.infer.local(new_workflow_file)
if fname is None:
raise RuntimeError("Output File not found")
j = {"file_name": fname}
print("json", j)
return Response(content=json.dumps(j), media_type="application/json")
j = {"status": "success", "file_name": fname}
loguru.logger.success(j)
return j
except Exception as e:
raise RuntimeError(str(e))
j = {"status":"fail", "msg": str(e)}
loguru.logger.error(j)
return j
finally:
print("Purge Garbage By Restart Comfy")
cmd = "comfy stop"
@@ -282,7 +284,7 @@ class ComfyUI:
try:
subprocess.run(cmd, shell=True, check=True)
except:
raise Exception("Failed to launch ComfyUI")
pass
def poll_server_health(self):
import socket