#!/usr/bin/env python3 """Resolve a product release to its actual independently published components. This only reads committed release records; it never changes stable aliases or chooses a newer candidate automatically. stdout contains version settings only. """ import argparse import json import os from pathlib import Path import re import sys from urllib.parse import urljoin, urlsplit from urllib.request import Request, urlopen REGISTRY = 'registry.cn-shanghai.aliyuncs.com/' COMPONENTS = ( ('easyai-server', 'server-main', 'SERVER_VERSION', 'UPDATE_SERVER_VERSION', 'comfy-ai/comfy-server'), ('easyai-web', 'web', 'WEB_VERSION', 'UPDATE_WEB_VERSION', 'comfy-ai/one-ai'), ('ws-gateway', 'ws-gateway', 'WS_GATEWAY_VERSION', 'UPDATE_GATEWAY_VERSION', 'easyaigc/wsgateway'), ('model-conversion-worker', 'model-conversion-worker', 'MODEL_CONVERSION_WORKER_VERSION', 'UPDATE_WORKER_VERSION', 'easyaigc/model-conversion-worker'), ('easyai-updater', 'updater', 'UPDATER_VERSION', 'UPDATE_UPDATER_VERSION', 'easyaigc/easyai-updater'), ('easyai-asg', 'service-governance', 'ASG_VERSION', 'UPDATE_ASG_VERSION', 'easyaigc/agent-governance'), ('agent-memory', 'memory-service', 'AMS_VERSION', 'UPDATE_AMS_VERSION', 'easyaigc/agent-memory'), ) PRODUCT_VERSION = r'\d+\.\d+\.\d+\.\d{8}(?:\.\d+)?' SUFFIX = r'(?:-no-ssr|-hmg-ai|-easyai|-chinacuc)?' DIGEST = r'sha256:[a-f0-9]{64}' DEFAULT_RELEASE_URL = 'https://git.51easyai.com/wangbo/easyai/raw/branch/main/system-update/release-manifest.json' def env_values(path): result = {} for line in path.read_text(encoding='utf-8-sig').splitlines(): match = re.match(r'^([A-Za-z_][A-Za-z0-9_]*)=(.*)$', line) if match: result.setdefault(match[1], match[2].strip().strip('\"\'')) return result def load(path): try: return json.loads(path.read_text(encoding='utf-8-sig')) except (OSError, ValueError) as error: raise ValueError(f'无法读取发布记录 {path.name}；请完整更新部署仓库后重试') from error def remote_record(url): parts = urlsplit(url) if parts.scheme not in ('http', 'https') or not parts.netloc or parts.username or parts.password: raise ValueError('发布源必须是 HTTP(S) 地址，不能在地址中嵌入登录凭据') try: with urlopen(Request(url, headers={'Accept': 'application/json', 'Cache-Control': 'no-cache'}), timeout=20) as response: content = response.read(1024 * 1024 + 1) if len(content) > 1024 * 1024: raise ValueError('oversized record') return json.loads(content) except Exception as error: raise ValueError('无法读取正式发布源；保留当前服务，不使用历史清单宣称已更新') from error def resolve(root, values, selection, options): selected = set(selection.split()) requested = options.get('UPDATE_RELEASE_VERSION') or options.get('DEPLOY_RELEASE_VERSION') if requested and not re.fullmatch(PRODUCT_VERSION, requested): raise ValueError('产品版本必须为完整版本号，不能使用 latest/dev 或路径') mode = (options.get('UPDATE_MANIFEST_SOURCE', values.get('UPDATE_MANIFEST_SOURCE')) or 'remote').strip().lower() if mode not in ('local', 'remote'): raise ValueError('UPDATE_MANIFEST_SOURCE 只支持 remote 或 local') release_url = (options.get('UPDATE_MANIFEST_URL', values.get('UPDATE_MANIFEST_URL')) or DEFAULT_RELEASE_URL).strip() address = urlsplit(release_url) if (address.scheme not in ('http', 'https') or not address.netloc or address.username or address.password or any(ord(char) < 32 for char in release_url) or '\\' in release_url): raise ValueError('发布源必须是有效 HTTP(S) 地址，不能内嵌登录凭据或控制字符') path = root / 'system-update' / ('releases/' + requested + '.json' if requested else 'release-manifest.json') if mode == 'remote' and not requested: manifest = remote_record(release_url) elif path.exists(): manifest = load(path) elif mode == 'remote': manifest = remote_record(urljoin(release_url, 'releases/' + requested + '.json')) else: manifest = load(path) version = manifest.get('productVersion', '') if (manifest.get('schemaVersion') != 1 or not re.fullmatch(PRODUCT_VERSION, version) or manifest.get('platform') != 'linux/amd64' or manifest.get('channel') not in (('versioned', 'stable') if requested else ('stable',))): raise ValueError('发布记录的版本、通道或平台不正确，未修改配置或容器') if requested and version != requested: raise ValueError('发布记录与指定产品版本不一致') # Independently maintained services use the reviewed baseline shipped with # this deployment tool. Product-specific records override it when needed. support = load(root / 'system-update/deployment-components.json') records = {item['key']: item for item in support['components']} # Older updater manifests intentionally contain only its three supported # components. Use the matching product record for Worker / web variants. product_path = root / 'system-update/releases' / (version + '.json') if product_path.exists() or mode == 'remote': product = (load(product_path) if product_path.exists() else remote_record(urljoin(release_url, 'releases/' + version + '.json'))) if product.get('productVersion') != version: raise ValueError('配套产物记录与产品版本不一致') records.update({item['key']: item for item in product['components']}) records.update({item['key']: item for item in manifest['components']}) records.update({item['key']: item for item in manifest.get('deploymentComponents', [])}) plan = [] legacy = (values.get('SYSTEM_UPDATE_MODE') or options.get('EASYAI_UPDATE_MODE_HINT')) == 'legacy_watchtower' for service, component, key, option, repository in COMPONENTS: explicit = options.get(option, '') if selected and service not in selected: if explicit: raise ValueError(f'{option} 要求在 UPDATE_SERVICES 中指定 {service}') continue image = REGISTRY + repository stable_digest = '' if explicit: pattern = PRODUCT_VERSION + (SUFFIX if service == 'easyai-web' else '') if service == 'easyai-updater': pattern = r'\d+\.\d+\.\d+(?:\.\d{8}(?:\.\d+)?)?' if not re.fullmatch(pattern + '(?:@' + DIGEST + ')?', explicit): raise ValueError(f'{option} 必须是已发布的固定版本，不能使用浮动标签') target = explicit else: item = records.get(component) old_web = values.get('WEB_VERSION') or values.get('VERSION', '') variant = re.search(r'(-no-ssr|-hmg-ai|-easyai|-chinacuc)(?:@.*)?$', old_web) if service == 'easyai-web' and variant: variant_key = {'-no-ssr': 'web-no-ssr'}.get(variant[1], variant[1][1:]) item = records.get(variant_key) if not item: raise ValueError(f'所选发布没有当前网页变体 {variant[1]}；请显式指定 UPDATE_WEB_VERSION') if not item or item.get('image') != image or not re.fullmatch(DIGEST, item.get('digest', '')): raise ValueError(f'发布记录缺少有效配套产物 {component}，未修改配置或容器') tag = item.get('tag') or 'latest' if not re.fullmatch(r'[A-Za-z0-9_][A-Za-z0-9_.-]{0,127}', tag): raise ValueError(f'配套产物标签不正确：{component}') # A stable alias is allowed only WITH its immutable recorded digest. # The updater is infrastructure with an existing independent # Watchtower/cron maintenance contract. Keep its official stable # alias so old installations can still receive executor updates. target = tag if service == 'easyai-updater' and tag == 'latest' else tag + '@' + item['digest'] if legacy and not requested and component in ('server-main', 'web', 'ws-gateway'): # Legacy Watchtower must keep following the official stable # aliases. Pinning its containers to a digest strands updates. target = 'latest' + (variant[1] if service == 'easyai-web' and variant else '') stable_digest = item['digest'] plan.append({'Service': service, 'Key': key, 'Version': target, 'Image': image}) if component in ('server-main', 'web', 'ws-gateway'): plan.append({'Service': '', 'Key': key.replace('_VERSION', '_STABLE_DIGEST'), 'Version': stable_digest, 'Image': ''}) # Compose companion defaults are independent; changing VERSION cannot # silently repoint an unselected gateway / worker anymore. if not selected or {'easyai-server', 'easyai-web'} <= selected: plan.append({'Service': '', 'Key': 'VERSION', 'Version': version, 'Image': ''}) return plan def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--root', type=Path, default=Path(__file__).resolve().parent.parent) parser.add_argument('--env-file', type=Path, required=True) parser.add_argument('--format', choices=('json', 'env'), default='env') args = parser.parse_args() try: plan = resolve(args.root, env_values(args.env_file), os.environ.get('UPDATE_SERVICES', ''), os.environ) if args.format == 'json': print(json.dumps(plan, ensure_ascii=False)) else: for item in plan: print(item['Key'] + '=' + item['Version']) except (ValueError, KeyError, TypeError) as error: print('❌ ' + str(error), file=sys.stderr) return 1 return 0 if __name__ == '__main__': sys.exit(main())