| """Generic ComfyUI/Irodori app backend supplied by the HF repository. |
| |
| The fixed v53 image executes this code as uid10001. HTML is hosted separately. |
| Only a job-scoped Cloudflare capability is received; no provider keys are needed. |
| """ |
| from __future__ import annotations |
| import base64,hashlib,json,math,mimetypes,os,re,shutil,subprocess,sys,time,urllib.error,urllib.parse,urllib.request,uuid |
| from pathlib import Path |
|
|
| TRUSTED_ORIGIN=os.environ['APP_SERVICE_ORIGIN'] |
| USER_AGENT='app-HF/1.0' |
| INPUT_ROOT=Path('/comfyui/input') |
| OUTPUT_ROOT=Path('/comfyui/output') |
| TEMP_ROOT=Path('/comfyui/temp') |
| CHUNK_BYTES=8*1024*1024 |
| MAX_ARTIFACT=512*1024*1024 |
| MAX_TOTAL=2*1024*1024*1024 |
| MAX_FILES=16 |
| MEDIA_MARKERS=('loadimage','loadvideo','loadaudio') |
| MEDIA_KEYS={'image','images','imagefile','imagepath','video','videos','videofile','videopath','audio','audiofile','audiopath','file','filename','path','filepath','directory','folder','url','uri'} |
| MODEL_KEYS={'ckpt_name','unet_name','lora_name','vae_name','clip_name','clip_name1','clip_name2','clip_name3','clip_name4','control_net_name','model_name','text_encoder','text_encoder_name','tiny_vae','mmaudio_model','clip_model','vae_model','synchformer_model'} |
| DENIED_MARKERS={'python','shell','terminal','subprocess','executecommand','systemcommand','commandline','comfyinstall','comfymanager'} |
| MIMES={'.mp4':'video/mp4','.webm':'video/webm','.mov':'video/quicktime','.mkv':'video/x-matroska','.avi':'video/x-msvideo','.m4v':'video/mp4','.wav':'audio/wav','.flac':'audio/flac','.mp3':'audio/mpeg','.ogg':'audio/ogg','.opus':'audio/ogg','.m4a':'audio/mp4','.png':'image/png','.jpg':'image/jpeg','.jpeg':'image/jpeg','.webp':'image/webp','.gif':'image/gif','.avif':'image/avif'} |
|
|
| class Deadline: |
| def __init__(self,seconds=280):self.end=time.monotonic()+seconds |
| def remaining(self,limit=None): |
| remaining=self.end-time.monotonic() |
| if remaining<=0:raise TimeoutError('Generation and artifact transfer exceeded the 280-second app limit') |
| return min(remaining,limit) if limit is not None else remaining |
|
|
| class NoRedirect(urllib.request.HTTPRedirectHandler): |
| def redirect_request(self,*args,**kwargs):raise ValueError('Media service redirects are not accepted') |
|
|
| def trusted_url(value): |
| if not isinstance(value,str) or len(value)>4096:raise ValueError('Invalid media service URL') |
| parsed=urllib.parse.urlsplit(value) |
| if parsed.scheme!='https' or parsed.netloc!=urllib.parse.urlsplit(TRUSTED_ORIGIN).netloc or parsed.username or parsed.password or parsed.fragment: |
| raise ValueError('Media URL must use the trusted Cloudflare service origin') |
| return value |
|
|
| def set_read_timeout(response,deadline): |
| remaining=deadline.remaining(25) |
| |
| |
| sock=getattr(getattr(getattr(response,'fp',None),'raw',None),'_sock',None) |
| if sock is not None:sock.settimeout(remaining) |
|
|
| def service_error(label,error,deadline): |
| details=['HTTP '+str(error.code)] |
| |
| try: |
| set_read_timeout(error,deadline) |
| snippet=error.read(8192).decode('utf-8',errors='replace') |
| match=re.search(r'(?:error[ _-]*code|cloudflare[ _-]*error)\s*[:=]\s*["\']?(1[0-9]{3})\b',snippet,re.I) |
| if match:details.append('Cloudflare '+match.group(1)) |
| if error.headers.get('cf-mitigated')=='challenge':details.append('Cloudflare challenge') |
| ray=error.headers.get('cf-ray','') |
| if re.fullmatch(r'[0-9a-fA-F]{8,32}-[A-Z]{3}',ray):details.append('Ray '+ray) |
| except Exception:pass |
| return RuntimeError(label+' ('+', '.join(details)+')') |
|
|
| def trusted_network_test(url,deadline): |
| trusted_url(url) |
| request=urllib.request.Request(url,headers={'User-Agent':USER_AGENT,'Range':'bytes=0-0','Accept':'application/octet-stream'}) |
| try: |
| with urllib.request.build_opener(NoRedirect()).open(request,timeout=deadline.remaining(25)) as response: |
| set_read_timeout(response,deadline);received=len(response.read(1)) |
| return {'status':response.status,'sample_bytes':received,'content_type':response.headers.get('Content-Type'),'content_length':response.headers.get('Content-Length'),'cache_status':response.headers.get('CF-Cache-Status')} |
| except urllib.error.HTTPError as error:raise service_error('Network diagnostic failed',error,deadline) from None |
| except urllib.error.URLError:raise RuntimeError('Network diagnostic connection failed') from None |
|
|
| def request_json(url,method,body,deadline,headers=None): |
| trusted_url(url) |
| raw=body if isinstance(body,bytes) else json.dumps(body,separators=(',',':'),allow_nan=False).encode() |
| actual={'User-Agent':USER_AGENT,'Content-Type':'application/octet-stream' if isinstance(body,bytes) else 'application/json',**(headers or {})} |
| req=urllib.request.Request(url,data=raw,method=method,headers=actual) |
| try: |
| with urllib.request.build_opener(NoRedirect()).open(req,timeout=deadline.remaining(25)) as response: |
| set_read_timeout(response,deadline) |
| encoded=response.read(65537) |
| if len(encoded)>65536:raise ValueError('Media service JSON response is too large') |
| result=json.loads(encoded) |
| if not isinstance(result,dict):raise ValueError('Invalid media service response') |
| return result |
| except urllib.error.HTTPError as error: |
| |
| raise service_error('Media service request failed',error,deadline) from None |
| except urllib.error.URLError as error: |
| raise RuntimeError('Media service connection failed: '+str(type(error.reason).__name__)) from None |
|
|
| def relative_name(value): |
| return isinstance(value,str) and bool(value) and len(value)<=512 and not value.startswith('/') and '\\' not in value and ':' not in value and not any(part in ('','..','.') for part in value.split('/')) and '\x00' not in value |
|
|
| def reference_name(value): |
| if not isinstance(value,str):return None |
| if value.startswith('input/'):value=value[6:] |
| if value.endswith(' [input]'):value=value[:-8] |
| return value if relative_name(value) else None |
|
|
| def safe_cache_path(root,value,directory=False): |
| if not relative_name(value):raise ValueError('Invalid cached model path') |
| source=Path(root)/value |
| |
| resolved=source.resolve();repo=Path(root).parent.parent.resolve() |
| if not (resolved.is_relative_to(Path(root).resolve()) or resolved.is_relative_to(repo)): |
| raise ValueError('Cached model escaped its repository') |
| if not (source.is_dir() if directory else source.is_file()):raise ValueError('Required file is missing from the native HF cache: '+value) |
| return source |
|
|
| def prepare_files(files,job,deadline): |
| if files is None:files=[] |
| if not isinstance(files,list) or len(files)>20:raise ValueError('files must contain at most20 items') |
| directory=INPUT_ROOT/'hf-jobs'/job |
| directory.mkdir(parents=True,exist_ok=False) |
| references={};total=0 |
| try: |
| for item in files: |
| deadline.remaining() |
| if not isinstance(item,dict):raise ValueError('Invalid input file') |
| name=item.get('name') |
| if not isinstance(name,str) or not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9._-]{0,159}',name) or name in references: |
| raise ValueError('Input file names must be unique safe basenames') |
| if Path(name).suffix.lower() not in MIMES:raise ValueError('Unsupported input media extension: '+name) |
| expected=item.get('size');expected_hash=item.get('sha256') |
| if expected is not None and (isinstance(expected,bool) or not isinstance(expected,int) or expected<1 or expected>MAX_ARTIFACT): |
| raise ValueError('Input file size is invalid or exceeds512MiB') |
| if expected_hash is not None and not re.fullmatch(r'[a-f0-9]{64}',str(expected_hash)): |
| raise ValueError('Input SHA256 must be lowercase hexadecimal') |
| target=directory/(uuid.uuid4().hex+Path(name).suffix.lower()) |
| digest=hashlib.sha256();size=0 |
| with target.open('xb') as out: |
| if item.get('url'): |
| trusted_url(item['url']) |
| if expected is None or expected_hash is None:raise ValueError('Staged inputs require size and sha256') |
| try: |
| request=urllib.request.Request(item['url'],headers={'User-Agent':USER_AGENT,'Accept':'application/octet-stream'}) |
| response=urllib.request.build_opener(NoRedirect()).open(request,timeout=deadline.remaining(25)) |
| except urllib.error.HTTPError as error:raise service_error('Input media download failed',error,deadline) from None |
| except urllib.error.URLError as error:raise RuntimeError('Input media download connection failed') from None |
| with response: |
| while True: |
| deadline.remaining() |
| set_read_timeout(response,deadline) |
| chunk=response.read(256*1024) |
| if not chunk:break |
| size+=len(chunk);total+=len(chunk) |
| if size>expected or size>MAX_ARTIFACT or total>MAX_TOTAL:raise ValueError('Input media size exceeds declared or job limits') |
| digest.update(chunk);out.write(chunk) |
| elif isinstance(item.get('data'),str): |
| encoded=item['data'].split(',',1)[-1] |
| if len(encoded)>768*1024:raise ValueError('Inline media is too large; stage it through Cloudflare first') |
| try:raw=base64.b64decode(encoded,validate=True) |
| except ValueError:raise ValueError('Invalid input base64') from None |
| size=len(raw);total+=size;digest.update(raw);out.write(raw) |
| else:raise ValueError('Input media requires a staged URL') |
| if not size or (expected is not None and size!=expected):raise ValueError('Input media size does not match its manifest') |
| if expected_hash is not None and digest.hexdigest()!=expected_hash:raise ValueError('Input media SHA256 does not match its manifest') |
| references[name]=target.relative_to(INPUT_ROOT).as_posix() |
| return references,directory |
| except BaseException: |
| shutil.rmtree(directory,ignore_errors=True) |
| raise |
|
|
| def normalize_and_validate(workflow,object_info,references,job): |
| if not isinstance(workflow,dict) or not workflow or len(workflow)>512:raise ValueError('workflow must be a non-empty API graph with at most512 nodes') |
| workflow=json.loads(json.dumps(workflow,allow_nan=False)) |
| for node_id,node in workflow.items(): |
| if not isinstance(node_id,str) or not re.fullmatch(r'[A-Za-z0-9_-]{1,80}',node_id):raise ValueError('Invalid workflow node ID') |
| if not isinstance(node,dict) or not isinstance(node.get('inputs'),dict):raise ValueError('Invalid workflow node: '+node_id) |
| node_type=node.get('class_type') |
| if not isinstance(node_type,str) or node_type not in object_info:raise ValueError('Node type is not installed: '+str(node_type)) |
| normalized=re.sub('[^a-z0-9]','',node_type.lower()) |
| if any(marker in normalized for marker in DENIED_MARKERS):raise ValueError('This node cannot run in the shared runtime: '+node_type) |
| definitions=object_info[node_type].get('input',{}) |
| definitions={**definitions.get('required',{}),**definitions.get('optional',{})} |
| if object_info[node_type].get('output_node') and 'filename_prefix' in definitions and 'filename_prefix' not in node['inputs']: |
| node['inputs']['filename_prefix']='output' |
| for key,value in list(node['inputs'].items()): |
| lower=str(key).lower() |
| if isinstance(value,str) and len(value)>20000:raise ValueError('Workflow text is too long') |
| if isinstance(value,list) and len(value)==2 and isinstance(value[0],str) and isinstance(value[1],int): |
| if value[0] not in workflow or not 0<=value[1]<=128:raise ValueError('Broken workflow link in node '+node_id) |
| if isinstance(value,(int,float)) and not isinstance(value,bool): |
| if not math.isfinite(value):raise ValueError('Workflow numbers must be finite') |
| limits={'width':8192,'height':8192,'target_width':8192,'target_height':8192,'output_width':8192,'output_height':8192, |
| 'length':2000,'num_frames':2000,'frames':2000,'video_frames':2000,'frame_count':2000, |
| 'fps':240,'frame_rate':240,'steps':120,'batch':8,'batch_size':8,'batch_count':8,'duration':300,'duration_seconds':300} |
| if lower in limits and not 0<=value<=limits[lower]:raise ValueError('Workflow exceeds '+lower+' limit') |
| if (lower in MODEL_KEYS or re.fullmatch(r'lora_\d+',lower)) and isinstance(value,str) and value.lower() not in ('none','disabled'): |
| if not relative_name(value):raise ValueError('Unsafe model path in '+node_id) |
| definition=definitions.get(key) |
| options=definition[0] if isinstance(definition,list) and definition and isinstance(definition[0],list) else None |
| if options and value not in options: |
| basename=Path(value).name |
| matches=[option for option in options if isinstance(option,str) and Path(option).name==basename] |
| if len(matches)==1:node['inputs'][key]=matches[0] |
| else:raise ValueError('Model is missing from the native HF cache mapping: '+value) |
| if lower in ('filename_prefix','subfolder') and isinstance(value,str): |
| if value and not relative_name(value):raise ValueError('Unsafe output path') |
| node['inputs'][key]='hf-jobs/'+job+'/'+(value or 'output') |
| media_key=re.sub('[^a-z0-9]','',lower) |
| if any(marker in normalized for marker in MEDIA_MARKERS) and media_key in MEDIA_KEYS and isinstance(value,str): |
| source=reference_name(value) |
| if source not in references:raise ValueError('Workflow input media was not supplied by this job: '+str(source)) |
| mapped=references[source] |
| if node_type=='VHS_LoadVideoPath' and media_key=='video': |
| |
| target=(INPUT_ROOT/mapped).resolve() |
| job_root=(INPUT_ROOT/'hf-jobs'/job).resolve() |
| if not target.is_relative_to(job_root) or not target.is_file(): |
| raise ValueError('Video path must reference an existing input from this job') |
| mapped=str(target) |
| node['inputs'][key]=mapped |
| return workflow |
|
|
| def snapshot_outputs(): |
| state={} |
| for root in (OUTPUT_ROOT,TEMP_ROOT): |
| if not root.exists():continue |
| for path in root.rglob('*'): |
| try: |
| resolved=path.resolve() |
| if path.is_file() and resolved.is_relative_to(root.resolve()): |
| stat=path.stat();state[resolved]=(stat.st_size,stat.st_mtime_ns) |
| except OSError:pass |
| return state |
|
|
| def output_files(history,before): |
| paths=[];owned=[];seen=set() |
| def visit(value,depth=0): |
| if depth>8:return |
| if isinstance(value,list): |
| for child in value:visit(child,depth+1) |
| elif isinstance(value,dict): |
| if isinstance(value.get('filename'),str): |
| name=value['filename'];subfolder=value.get('subfolder','');kind=value.get('type','output') |
| roots={'output':OUTPUT_ROOT,'temp':TEMP_ROOT} |
| if kind not in roots or not relative_name(name) or (subfolder and not relative_name(subfolder)):return |
| root=roots[kind].resolve();path=(root/subfolder/name).resolve() |
| if not path.is_relative_to(root) or not path.is_file() or path in seen:return |
| seen.add(path);paths.append(path) |
| stat=path.stat() |
| if before.get(path)!=(stat.st_size,stat.st_mtime_ns):owned.append(path) |
| else: |
| for child in value.values():visit(child,depth+1) |
| visit(history.get('outputs',{})) |
| if not paths:raise RuntimeError('ComfyUI completed without an image, video or audio artifact') |
| if len(paths)>MAX_FILES:raise ValueError('Workflow produced more than16 output artifacts') |
| return paths,owned |
|
|
| def comfy_execution_ms(history): |
| """Use ComfyUI execution lifecycle timestamps; exclude input/output transfer.""" |
| events={} |
| for item in (history.get('status') or {}).get('messages',[]): |
| if not isinstance(item,(list,tuple)) or len(item)!=2:continue |
| name,data=item |
| if name not in ('execution_start','execution_success') or not isinstance(data,dict):continue |
| stamp=data.get('timestamp') |
| if isinstance(stamp,(int,float)) and not isinstance(stamp,bool) and math.isfinite(stamp):events[name]=(data.get('prompt_id'),stamp) |
| start=events.get('execution_start');end=events.get('execution_success') |
| if not start or not end or not start[0] or start[0]!=end[0] or end[1]<start[1]:return None |
| return math.ceil(end[1]-start[1]) |
|
|
|
|
| def run_comfy(workflow,context,deadline): |
| identifier=None |
| with context.lock: |
| execution_started=time.monotonic() |
| try: |
| result=context.comfy('/prompt',{'prompt':workflow,'client_id':'hf-'+uuid.uuid4().hex}) |
| identifier=result.get('prompt_id') |
| if not isinstance(identifier,str):raise RuntimeError('ComfyUI rejected the workflow: '+json.dumps(result)[:1800]) |
| while True: |
| deadline.remaining(10) |
| history=context.comfy('/history/'+identifier).get(identifier) |
| if isinstance(history,dict): |
| status=history.get('status') or {} |
| if status.get('status_str')=='error':raise RuntimeError('ComfyUI execution failed: '+json.dumps(status.get('messages',[]))[:1800]) |
| if status.get('completed') or status.get('status_str')=='success': |
| |
| return {**history,'_execution_ms':elapsed_ms(execution_started)} |
| time.sleep(min(.25,deadline.remaining())) |
| except (TimeoutError,KeyboardInterrupt): |
| try:context.comfy('/interrupt',{}) |
| except Exception:pass |
| raise |
| except urllib.error.HTTPError as error: |
| try:message=error.read(16384).decode(errors='replace') |
| except Exception:message='' |
| raise RuntimeError('ComfyUI rejected the workflow (HTTP '+str(error.code)+'): '+message[:1800]) from None |
| finally: |
| if identifier: |
| try:context.comfy('/history',{'delete':[identifier]}) |
| except Exception:pass |
|
|
| def file_metadata(path,deadline): |
| size=path.stat().st_size |
| if not 0<size<=MAX_ARTIFACT:raise ValueError('Output artifact must be1byte..512MiB') |
| digest=hashlib.sha256() |
| with path.open('rb') as source: |
| while True: |
| deadline.remaining();chunk=source.read(CHUNK_BYTES) |
| if not chunk:break |
| digest.update(chunk) |
| return {'filename':path.name,'mime_type':MIMES.get(path.suffix.lower()) or mimetypes.guess_type(path.name)[0] or 'application/octet-stream', |
| 'size':size,'sha256':digest.hexdigest()} |
|
|
| def upload_outputs(paths,capability,deadline): |
| if len(paths)>MAX_FILES or sum(path.stat().st_size for path in paths)>MAX_TOTAL:raise ValueError('Output artifacts exceed the per-job limit') |
| if capability: |
| trusted_url(capability) |
| parsed=urllib.parse.urlsplit(capability) |
| if parsed.query or not re.fullmatch(r'/a/[A-Za-z0-9_-]{43}',parsed.path):raise ValueError('Invalid job-scoped artifact capability') |
| elif sum(path.stat().st_size for path in paths)>7*1024*1024: |
| raise ValueError('A Cloudflare artifact upload capability is required for output larger than7MiB') |
| outputs=[] |
| for path in paths: |
| metadata=file_metadata(path,deadline) |
| if not capability: |
| outputs.append({**metadata,'data':base64.b64encode(path.read_bytes()).decode()});continue |
| response=request_json(capability+'/begin','POST',metadata,deadline) |
| identifier=response.get('id') |
| if not isinstance(identifier,str) or not re.fullmatch(r'[A-Za-z0-9_-]{8,100}',identifier):raise RuntimeError('Invalid artifact upload ID') |
| parts=[] |
| with path.open('rb') as source: |
| index=0 |
| while True: |
| deadline.remaining();chunk=source.read(CHUNK_BYTES) |
| if not chunk:break |
| digest=hashlib.sha256(chunk).hexdigest() |
| response=request_json(capability+'/'+identifier+'/parts/'+str(index),'PUT',chunk,deadline,{'X-Part-SHA256':digest}) |
| if response.get('sha256') and response['sha256']!=digest:raise RuntimeError('Artifact part checksum mismatch') |
| parts.append({'index':index,'sha256':digest,'size':len(chunk)});index+=1 |
| completed=request_json(capability+'/'+identifier+'/complete','POST',{'parts':parts,'sha256':metadata['sha256'],'size':metadata['size']},deadline) |
| url=completed.get('url');trusted_url(url) |
| if completed.get('sha256') and completed['sha256']!=metadata['sha256']:raise RuntimeError('Uploaded artifact checksum mismatch') |
| if completed.get('size') is not None and completed['size']!=metadata['size']:raise RuntimeError('Uploaded artifact size mismatch') |
| outputs.append({**metadata,'url':url,'expires_at':completed.get('expires_at')}) |
| return outputs |
|
|
| def finite_number(body,key,default,minimum,maximum,integer=False): |
| raw=body.get(key,default) |
| if isinstance(raw,bool):raise ValueError(key+' must be numeric') |
| try:value=float(raw) |
| except (TypeError,ValueError):raise ValueError(key+' must be numeric') from None |
| if not math.isfinite(value) or not minimum<=value<=maximum or (integer and value!=int(value)): |
| raise ValueError(key+' must be between'+str(minimum)+' and'+str(maximum)) |
| return int(value) if integer else value |
|
|
| def run_irodori(payload,context,references,job,deadline): |
| config=json.loads((context.root/'app-runtime.json').read_text(encoding='utf-8-sig')).get('irodori') |
| if not isinstance(config,dict):raise ValueError('HF app is missing its irodori model configuration') |
| text=payload.get('text');caption=payload.get('caption','') |
| if not isinstance(text,str) or not text.strip() or len(text)>1000:raise ValueError('Irodori text must contain1..1000 characters') |
| if not isinstance(caption,str) or len(caption)>2000:raise ValueError('Irodori caption exceeds2000 characters') |
| checkpoint=safe_cache_path(context.root,config.get('checkpoint')) |
| codec=safe_cache_path(context.root,config.get('codec')) |
| tokenizer=safe_cache_path(context.root,config.get('tokenizer','tokenizer'),directory=True) |
| if not (tokenizer/'tokenizer_config.json').is_file():raise ValueError('Irodori tokenizer_config.json is missing') |
| directory=OUTPUT_ROOT/'hf-jobs'/job;directory.mkdir(parents=True,exist_ok=True) |
| output=directory/'irodori.wav' |
| reference=payload.get('reference_audio') |
| many=payload.get('reference_audios') |
| if reference is not None and many is not None:raise ValueError('Use one reference representation') |
| selected=many if many is not None else ([reference] if reference else []) |
| if not isinstance(selected,list) or len(selected)>20 or any(name not in references for name in selected):raise ValueError('Irodori reference audio was not supplied by this job') |
| if sum((INPUT_ROOT/references[name]).stat().st_size for name in selected)>20*1024*1024:raise ValueError('Irodori reference audio exceeds20MiB combined') |
| request={'checkpoint':str(checkpoint),'codec':str(codec),'tokenizer':str(tokenizer),'output':str(output), |
| 'text':text.strip(),'caption':caption.strip(),'reference_wav':str(INPUT_ROOT/references[reference]) if reference else None, |
| 'reference_wavs':[str(INPUT_ROOT/references[name]) for name in many] if many is not None else None, |
| 'num_steps':finite_number(payload,'num_steps',60,4,80,True),'duration_scale':finite_number(payload,'duration_scale',1,.5,2), |
| 'cfg_scale_text':finite_number(payload,'cfg_scale_text',3,0,20),'cfg_scale_caption':finite_number(payload,'cfg_scale_caption',3,0,20), |
| 'cfg_scale_speaker':finite_number(payload,'cfg_scale_speaker',10,0,20), |
| 'seed':None if payload.get('seed') is None else finite_number(payload,'seed',0,0,2**53-1,True), |
| 'seconds':None if payload.get('seconds') is None else finite_number(payload,'seconds',5,.5,45)} |
| runner=context.root/'irodori_runner.py' |
| if not runner.is_file():raise ValueError('irodori_runner.py is missing from the HF app') |
| log=directory/'irodori.stderr' |
| with log.open('wb') as error_log: |
| try: |
| result=subprocess.run(['/opt/irodori-venv/bin/python',str(runner)],input=json.dumps(request),text=True, |
| stdout=subprocess.PIPE,stderr=error_log,timeout=deadline.remaining(),check=False, |
| env={**os.environ,'HF_HUB_OFFLINE':'1','TRANSFORMERS_OFFLINE':'1'}) |
| except subprocess.TimeoutExpired:raise TimeoutError('Irodori exceeded the remaining280-second app budget') from None |
| if result.returncode: |
| with log.open('rb') as error_log:error_log.seek(max(0,log.stat().st_size-1600));detail=error_log.read().decode(errors='replace') |
| raise RuntimeError('Irodori generation failed: '+detail) |
| try:summary=json.loads(result.stdout) |
| except ValueError:raise RuntimeError('Irodori returned invalid metadata') from None |
| if summary.get('error'):raise RuntimeError(str(summary['error'])[:1600]) |
| if not output.is_file():raise RuntimeError('Irodori returned no WAV file') |
| return [output],summary |
|
|
| def cleanup_job(directory,roots): |
| if directory is None:return |
| resolved=directory.resolve() |
| if any(resolved.is_relative_to((root/'hf-jobs').resolve()) and resolved!=(root/'hf-jobs').resolve() for root in roots): |
| shutil.rmtree(directory,ignore_errors=True) |
|
|
| def elapsed_ms(started): |
| return max(0,math.ceil((time.monotonic()-started)*1000)) |
|
|
| def generate(payload,context): |
| if not isinstance(payload,dict):raise ValueError('Generation payload must be an object') |
| deadline=Deadline(280);started=time.monotonic();job=uuid.uuid4().hex |
| metadata=payload.get('_app') or {} |
| if not isinstance(metadata,dict):raise ValueError('Invalid artifact metadata') |
| capability=metadata.get('artifact_upload_url') |
| if capability:trusted_url(capability) |
| references={};input_directory=None;owned=[];additional=[] |
| try: |
| references,input_directory=prepare_files(payload.get('files'),job,deadline) |
| if payload.get('task')=='diagnostics': |
| execution_started=time.monotonic() |
| result=diagnostics(payload,context) |
| result['summary']={'backend':'diagnostics','executed_ms':elapsed_ms(execution_started)} |
| return result |
| if payload.get('task')=='chat': |
| execution_started=time.monotonic() |
| result=run_chat(payload,context,deadline) |
| result['summary']['executed_ms']=elapsed_ms(execution_started) |
| result['summary']['elapsed_seconds']=round(time.monotonic()-started,3) |
| return result |
| if payload.get('task')=='irodori': |
| execution_started=time.monotonic() |
| paths,summary=run_irodori(payload,context,references,job,deadline);owned=paths |
| summary['executed_ms']=elapsed_ms(execution_started) |
| else: |
| additional_ms=0 |
| if payload.get('task')=='dialogue': |
| settings=payload.get('irodori') |
| if not isinstance(settings,dict):raise ValueError('Dialogue requires Irodori synthesis settings') |
| execution_started=time.monotonic() |
| audio,tts_summary=run_irodori(settings,context,references,job,deadline) |
| additional_ms=elapsed_ms(execution_started) |
| tts_summary['executed_ms']=additional_ms |
| reference=input_directory/'dialogue-generated.wav' |
| shutil.copyfile(audio[0],reference) |
| references['dialogue-generated.wav']=reference.relative_to(INPUT_ROOT).as_posix() |
| additional.extend(audio);owned.extend(audio) |
| continuation=prepare_continuation(payload.get('continuation'),references,input_directory,deadline) |
| workflow=payload.get('workflow') |
| if workflow is None: |
| default=context.root/'workflow.json' |
| if not default.is_file():raise ValueError('Provide an API workflow or include workflow.json in the HF app') |
| workflow=json.loads(default.read_text(encoding='utf-8-sig')) |
| if payload.get('task')=='dialogue':workflow=dialogue_timing(workflow,tts_summary) |
| info=context.comfy('/object_info') |
| workflow=normalize_and_validate(workflow,info,references,job) |
| before=snapshot_outputs();history=run_comfy(workflow,context,deadline) |
| paths,new_outputs=output_files(history,before);owned.extend(new_outputs) |
| paths,merged=finish_continuation(paths,continuation,job,deadline);owned.extend(merged) |
| executed_ms=comfy_execution_ms(history) |
| if executed_ms is None:executed_ms=history['_execution_ms'] |
| paths.extend(additional);summary={'backend':'comfyui','nodes':len(workflow),'executed_ms':executed_ms+additional_ms} |
| if additional:summary['irodori']=tts_summary |
| if continuation:summary['continuation']=True |
| outputs=upload_outputs(paths,capability,deadline) |
| return {'files':outputs,'summary':{**summary,'elapsed_seconds':round(time.monotonic()-started,3)}} |
| finally: |
| cleanup_job(input_directory,(INPUT_ROOT,)) |
| for path in owned: |
| try: |
| resolved=path.resolve() |
| if any(resolved.is_relative_to(root.resolve()) for root in (OUTPUT_ROOT,TEMP_ROOT)):path.unlink(missing_ok=True) |
| except OSError:pass |
| for root in (OUTPUT_ROOT,TEMP_ROOT):cleanup_job(root/'hf-jobs'/job,(root,)) |
|
|
| def run_chat(payload,context,deadline): |
| config=json.loads((context.root/'app-runtime.json').read_text(encoding='utf-8-sig')).get('chat') or {} |
| model_path=context.root if config.get('model_path','.')=='.' else safe_cache_path(context.root,config['model_path'],directory=True) |
| if not (model_path/'config.json').is_file():raise ValueError('Chat model config.json is missing from the native HF cache') |
| messages=payload.get('messages') |
| if messages is None: |
| messages=[] |
| if payload.get('system_prompt'):messages.append({'role':'system','content':payload['system_prompt']}) |
| if isinstance(payload.get('history'),list):messages.extend(payload['history']) |
| messages.append({'role':'user','content':payload.get('message','')}) |
| if not isinstance(messages,list) or not 1<=len(messages)<=64:raise ValueError('Chat requires1..64 messages') |
| for message in messages: |
| if not isinstance(message,dict) or message.get('role') not in ('system','user','assistant') or not isinstance(message.get('content'),str) or not message['content'].strip() or len(message['content'])>20000: |
| raise ValueError('Chat messages require role and nonempty text content') |
| request={'model_path':str(model_path),'messages':messages,'max_tokens':finite_number(payload,'max_tokens',512,1,2048,True), |
| 'temperature':finite_number(payload,'temperature',.78,0,2),'top_p':finite_number(payload,'top_p',.9,.01,1), |
| 'repetition_penalty':finite_number(payload,'repetition_penalty',1.08,1,2), |
| 'seed':None if payload.get('seed') is None else finite_number(payload,'seed',0,0,2**53-1,True), |
| 'generation_seconds':deadline.remaining()} |
| directory=Path('/workspace/hf-app');directory.mkdir(parents=True,exist_ok=True) |
| log=directory/('chat-'+uuid.uuid4().hex+'.stderr') |
| try: |
| with log.open('wb') as error_log: |
| try:result=subprocess.run(['/opt/venv/bin/python',str(context.root/'chat_runner.py')],input=json.dumps(request),text=True, |
| stdout=subprocess.PIPE,stderr=error_log,timeout=deadline.remaining(),check=False, |
| env={**os.environ,'HF_HUB_OFFLINE':'1','TRANSFORMERS_OFFLINE':'1'}) |
| except subprocess.TimeoutExpired:raise TimeoutError('Chat exceeded the remaining280-second app budget') from None |
| try:output=json.loads(result.stdout) |
| except ValueError:raise RuntimeError('Chat returned invalid JSON') from None |
| if result.returncode or output.get('error'):raise RuntimeError('Chat failed: '+str(output.get('error','runtime exited'))[:1600]) |
| return output |
| finally:log.unlink(missing_ok=True) |
|
|
| def prepare_continuation(value,references,directory,deadline): |
| import bernini_continuation as module |
| config=module.validate_continuation(value) |
| if config is None:return None |
| name=config['source_video_name'] |
| if name not in references:raise ValueError('Continuation source video was not supplied') |
| source=INPUT_ROOT/references[name] |
| frame=directory/(uuid.uuid4().hex+'.png') |
| resize=f"scale={config['width']}:{config['height']}:force_original_aspect_ratio=decrease,pad={config['width']}:{config['height']}:(ow-iw)/2:(oh-ih)/2:color=black,setsar=1" |
| module._DEADLINE=deadline |
| errors=[] |
| for offset in ('-0.1','-0.5','-1.0'): |
| frame.unlink(missing_ok=True) |
| try:module._run(['ffmpeg','-hide_banner','-loglevel','error','-y','-sseof',offset,'-i',str(source),'-map','0:v:0','-an','-frames:v','1','-vf',resize,str(frame)],'continuation frame extraction',30) |
| except (RuntimeError,subprocess.TimeoutExpired) as error:errors.append(str(error)) |
| if frame.is_file() and frame.stat().st_size:break |
| else: |
| frame.unlink(missing_ok=True) |
| module._run(['ffmpeg','-hide_banner','-loglevel','error','-y','-i',str(source),'-map','0:v:0','-an','-vf',resize,'-fps_mode','passthrough','-update','1',str(frame)],'continuation full-decode frame extraction',deadline.remaining()) |
| if not frame.is_file() or not frame.stat().st_size:raise RuntimeError('Continuation produced no reference frame') |
| references[config['last_frame_name']]=frame.relative_to(INPUT_ROOT).as_posix() |
| return {**config,'source_path':source} |
|
|
| def finish_continuation(paths,config,job,deadline): |
| if config is None:return paths,[] |
| import bernini_continuation as module |
| module._DEADLINE=deadline |
| outputs=[];created=[] |
| for path in paths: |
| if path.suffix.lower() not in module.VIDEO_SUFFIXES:outputs.append(path);continue |
| target=OUTPUT_ROOT/'hf-jobs'/job/('continued-'+str(len(created))+'.mp4') |
| module.merge_continuation_output(config['source_path'],path,target,config) |
| outputs.append(target);created.append(target) |
| if not created:raise RuntimeError('Continuation workflow produced no video') |
| return outputs,created |
|
|
|
|
| def diagnostics(payload,context): |
| """Bounded, read-only inspection for version/mapping and workflow smoke checks.""" |
| network=trusted_network_test(payload['network_url'],Deadline(25)) if payload.get('network_url') else None |
| if payload.get('network_only'): |
| if network is None:raise ValueError('network_only requires a trusted network_url') |
| return {'network':network} |
| requested=payload.get('requested_class_types',[]) |
| if not isinstance(requested,list) or len(requested)>128 or any(not isinstance(name,str) or len(name)>160 for name in requested): |
| raise ValueError('requested_class_types must contain at most128 node names') |
| info=context.comfy('/object_info') |
| result={'network':network,'available_node_count':len(info),'missing':[name for name in requested if name not in info], 'nodes':{}} |
| for name in requested: |
| if name not in info:continue |
| node=info[name] |
| result['nodes'][name]={'input':node.get('input',{}),'output':node.get('output',[]),'output_node':bool(node.get('output_node'))} |
| stats=context.comfy('/system_stats') |
| system=stats.get('system',{}) |
| result['versions']={key:system.get(key) for key in ('comfyui_version','python_version','pytorch_version','embedded_python') if key in system} |
| workflow=payload.get('workflow') |
| if workflow is not None: |
| references={} |
| for item in payload.get('files',[]): |
| if isinstance(item,dict) and isinstance(item.get('name'),str):references[item['name']]='hf-jobs/diagnostics/'+item['name'] |
| normalized=normalize_and_validate(workflow,info,references,'diagnostics') |
| result['workflow']={'valid':True,'nodes':len(normalized),'model_inputs':{node_id:{key:value for key,value in node['inputs'].items() if key.lower() in MODEL_KEYS or re.fullmatch(r'lora_\d+',key.lower())} for node_id,node in normalized.items()}} |
| if len(json.dumps(result,separators=(',',':')).encode())>512*1024:raise ValueError('Requested diagnostic schema exceeds512KiB; request fewer nodes') |
| return result |
|
|
|
|
| def dialogue_timing(workflow,summary): |
| """Preserve the host dialogue graph's speech-duration dependent frame count.""" |
| if not isinstance(workflow,dict):raise ValueError('Dialogue workflow must be an object') |
| workflow=json.loads(json.dumps(workflow,allow_nan=False)) |
| expected={'796':'mxSlider','535':'LTXVEmptyLatentAudio','542':'PrimitiveFloat','549':'VHS_VideoCombine'} |
| if any(workflow.get(key,{}).get('class_type')!=kind for key,kind in expected.items()): |
| raise ValueError('Dialogue workflow is missing its duration/frame-rate controls') |
| fps=finite_number(workflow['542']['inputs'],'value',24,1,120) |
| duration=finite_number(summary,'duration_seconds',0,.001,45) |
| seconds=max(3,math.ceil(duration+.25));frames=int(seconds*fps) |
| workflow['796']['inputs'].update(Xi=frames,Xf=frames) |
| workflow['535']['inputs'].update(frames_number=frames+1,frame_rate=fps) |
| workflow['542']['inputs']['value']=fps |
| workflow['549']['inputs']['frame_rate']=fps |
| return workflow |
|
|
