Local Image Generation

<IMAGE_GEN_DIR>\               ← 自动定位或创建（如 D:\image-gen-local）
├── image_gen\                 ← venv（Pre-flight 后 Step 2 自动创建）
├── generate_image.py          ← Step 4 自动写入
├── Z-Image-Turbo-int4-ov\     ← 首次自动从 ModelScope 下载（约 10 GB）
└── outputs\20260318_panda_bamboo.png  ← 文件名：YYYYMMDD_HHMMSS_topic_en.png

Pre-flight: 检查 Python ≥3.10 + git  → PYTHON_OK + GIT_OK
Step 0:     扩充提示词 + 提炼主题词   → 展示给用户
Step 1:     定位工作目录              → IMAGE_GEN_DIR
Step 2:     激活 venv，验证依赖       → DEP_CHECK=PASS
Step 3:     检查磁盘空间 + 模型       → MODEL_STATUS=READY 则跳过下载
Step 4:     写入脚本（版本检查）      → SCRIPT_UPDATE=DONE/SKIPPED
Step 5:     生图 + 预览               → 发送图像到对话

python --version

$f = "$env:TEMP\python-installer.exe"
Invoke-WebRequest "https://www.python.org/ftp/python/3.12.10/python-3.12.10-amd64.exe" -OutFile $f
Start-Process $f -ArgumentList "/quiet InstallAllUsers=0 PrependPath=1 Include_pip=1" -Wait
Remove-Item $f

git --version

$f = "$env:TEMP\git-installer.exe"
Invoke-WebRequest "https://github.com/git-for-windows/git/releases/download/v2.49.0.windows.1/Git-2.49.0-64-bit.exe" -OutFile $f
Start-Process $f -ArgumentList "/VERYSILENT /NORESTART /NOCANCEL /SP- /CLOSEAPPLICATIONS /RESTARTAPPLICATIONS /COMPONENTS=icons,ext\reg\shellhere,assoc,assoc_sh" -Wait
Remove-Item $f

📝 原始输入：{用户描述}
   扩充后：{完整提示词}
   主题词：{topic_en}（英文，用于文件命名）

python -c "
import string, shutil
from pathlib import Path
drives = [f'{d}:\\\\' for d in string.ascii_uppercase if Path(f'{d}:\\\\').exists()]
print(f'[INFO] 检测到磁盘: {drives}')
found = None
for drive in drives:
    candidate = Path(drive) / 'image-gen-local'
    if candidate.exists():
        found = candidate
        break
if not found:
    best = max(drives, key=lambda d: shutil.disk_usage(d).free)
    found = Path(best) / 'image-gen-local'
    found.mkdir(parents=True, exist_ok=True)
    print(f'[INFO] 已创建: {found}')
print(f'IMAGE_GEN_DIR={found}')
"

call "<IMAGE_GEN_DIR>\image_gen\Scripts\activate.bat"
python -c "import sys; print(sys.executable)"

python -m ensurepip --upgrade

python -m venv "<IMAGE_GEN_DIR>\image_gen"

call "<IMAGE_GEN_DIR>\image_gen\Scripts\activate.bat"

python -m pip install --upgrade pip

pip install -r "{baseDir}\requirements_imagegen.txt"

python -c "
import json, site
from pathlib import Path

# 期望的 git commit hash（前8位匹配）
EXPECTED_COMMITS = {
    'optimum_intel': '2f62e5ae',
    'diffusers':     'a1f36ee3',
}

def get_git_commit(pkg_name):
    dirs = site.getsitepackages()
    try: dirs += [site.getusersitepackages()]
    except Exception: pass
    for d in dirs:
        for dist in Path(d).glob(f'{pkg_name}*.dist-info'):
            url_file = dist / 'direct_url.json'
            if url_file.exists():
                data = json.loads(url_file.read_text(encoding='utf-8'))
                return data.get('vcs_info', {}).get('commit_id', 'no_vcs_info')
    return 'not_found'

results = {}

# 普通包
for pkg, imp in [('openvino','openvino'),('torch','torch'),('Pillow','PIL'),('modelscope','modelscope')]:
    try:
        ver = getattr(__import__(imp), '__version__', 'OK')
        results[pkg] = ('OK', ver)
    except ImportError as e:
        results[pkg] = ('MISSING', str(e))

# OVZImagePipeline 可用性
try:
    from optimum.intel import OVZImagePipeline
    results['OVZImagePipeline'] = ('OK', 'importable')
except ImportError as e:
    results['OVZImagePipeline'] = ('MISSING', str(e))

# git commit 精确验证
for pkg_name, exp in EXPECTED_COMMITS.items():
    actual = get_git_commit(pkg_name)
    if actual == 'not_found':
        results[f'{pkg_name}@commit'] = ('MISSING', 'not installed via git+https')
    elif actual.startswith(exp):
        results[f'{pkg_name}@commit'] = ('OK', actual[:16])
    else:
        results[f'{pkg_name}@commit'] = ('WRONG', f'got {actual[:16]} want {exp}...')

all_ok = all(v[0] == 'OK' for v in results.values())
for k, (status, detail) in results.items():
    icon = '✅' if status == 'OK' else ('⚠️' if status == 'WRONG' else '❌')
    print(f'  {icon} {k}: {detail}')
print('DEP_CHECK=PASS' if all_ok else 'DEP_CHECK=FAIL')
"

pip uninstall optimum-intel diffusers -y
pip install -r "{baseDir}\requirements_imagegen.txt" --no-cache-dir

python -c "
import shutil
from pathlib import Path
target = Path(r'<IMAGE_GEN_DIR>')
free_gb = shutil.disk_usage(target).free / (1024**3)
print(f'DISK_FREE={free_gb:.1f}GB')
if free_gb < 15:
    print('DISK_STATUS=LOW')
    print(f'[WARN] 可用空间不足 15 GB，模型下载可能中途失败')
else:
    print('DISK_STATUS=OK')
"

python -c "
from pathlib import Path
model_dir = Path(r'<IMAGE_GEN_DIR>') / 'Z-Image-Turbo-int4-ov'
required = ['transformer', 'vae_decoder', 'text_encoder']
missing = [r for r in required if not (model_dir / r).exists()]
print('MODEL_STATUS=READY') if not missing else print('MODEL_STATUS=MISSING')
print(f'MODEL_DIR={model_dir}')
if missing: print(f'MISSING_DIRS={missing}')
"

📥 模型首次下载约 10 GB，预计耗时：
   • 100 Mbps 宽带：约 15 分钟
   •  50 Mbps 宽带：约 30 分钟
   •  10 Mbps 宽带：约 2 小时
   下载期间请保持终端开启，支持断点续传——中断后重跑此步会自动从断点继续，无需重头下载。
   ✅ 准备好了，开始自动下载
   📂 我想手动下载，跳过自动下载

python -c "
import sys, time, threading
from pathlib import Path
from modelscope import snapshot_download

model_dir = Path(r'<IMAGE_GEN_DIR>') / 'Z-Image-Turbo-int4-ov'
model_dir.mkdir(parents=True, exist_ok=True)

# ── 后台监控线程：每 30 秒打印目录已下载大小和速度 ──
_stop = threading.Event()

def watchdog():
    prev = 0
    while not _stop.wait(30):
        try:
            total = sum(f.stat().st_size for f in model_dir.rglob('*') if f.is_file())
            speed = (total - prev) / 30
            print(f'[进度] 已下载 {total/1024**3:.2f} GB  速度 {speed/1024**2:.1f} MB/s', flush=True)
            prev = total
        except Exception:
            pass

t = threading.Thread(target=watchdog, daemon=True)
t.start()

try:
    # snapshot_download 内置 tqdm 进度条，自动断点续传
    snapshot_download(
        'snake7gun/Z-Image-Turbo-int4-ov',
        local_dir=str(model_dir),
        ignore_file_pattern=[r'\.git.*']
    )
    print('MODEL_DOWNLOAD=DONE')
except KeyboardInterrupt:
    print('[WARN] 下载被中断，已保留进度。重新运行此步将从断点续传。')
    print('MODEL_DOWNLOAD=INTERRUPTED')
except Exception as e:
    err = str(e).lower()
    if 'disk' in err or 'space' in err or 'no space' in err:
        print(f'[ERROR] 磁盘空间不足: {e}')
        print('MODEL_DOWNLOAD=FAIL_DISK')
    elif 'timeout' in err or 'connection' in err or 'network' in err:
        print(f'[ERROR] 网络错误: {e}')
        print('MODEL_DOWNLOAD=FAIL_NETWORK')
    else:
        print(f'[ERROR] 未知错误: {e}')
        print('MODEL_DOWNLOAD=FAIL_UNKNOWN')
finally:
    _stop.set()
"

pip install modelscope
python -c "from modelscope import snapshot_download; snapshot_download('snake7gun/Z-Image-Turbo-int4-ov', local_dir=r'<IMAGE_GEN_DIR>\Z-Image-Turbo-int4-ov')"

<IMAGE_GEN_DIR>\Z-Image-Turbo-int4-ov\
├── transformer\
├── vae_decoder\
└── text_encoder\

python -c "
from pathlib import Path
model_dir = Path(r'<IMAGE_GEN_DIR>') / 'Z-Image-Turbo-int4-ov'
required = ['transformer', 'vae_decoder', 'text_encoder']
missing = [r for r in required if not (model_dir / r).exists()]
print('MODEL_STATUS=READY') if not missing else print(f'MODEL_STATUS=MISSING  缺少: {missing}')
"

python -c "
from pathlib import Path
import re

CURRENT_VERSION = 'v2.0'
script = Path(r'<IMAGE_GEN_DIR>') / 'generate_image.py'

existing_version = None
if script.exists():
    m = re.search(r\"SKILL_VERSION\s*=\s*[\\\"'](.*?)[\\\"']\", script.read_text(encoding='utf-8', errors='ignore'))
    if m: existing_version = m.group(1)

if existing_version == CURRENT_VERSION:
    print('SCRIPT_UPDATE=SKIPPED (already up to date)')
    print(f'EXISTS={script.exists()}')
else:
    print(f'SCRIPT_VERSION_OLD={existing_version} -> NEW={CURRENT_VERSION}')
    print('SCRIPT_UPDATE=WRITING...')
    code = r'''
SKILL_VERSION = \"v2.0\"

import sys, io, os, subprocess, argparse, string, re
from datetime import datetime
from pathlib import Path
import openvino as ov
import torch
from optimum.intel import OVZImagePipeline
from PIL import Image

def get_image_gen_dir():
    for d in string.ascii_uppercase:
        c = Path(f\"{d}:\\\\\") / \"image-gen-local\"
        if c.exists(): return c
    return Path(__file__).resolve().parent

def get_device():
    core = ov.Core()
    devs = core.available_devices
    print(f\"[INFO] 可用设备: {devs}\")
    for d in devs:
        if \"GPU\" in d:
            print(f\"[INFO] 使用 Intel GPU: {d}\")
            return d
    print(\"[INFO] 使用 CPU\")
    return \"CPU\"

def make_filename(prompt, topic):
    date_str = datetime.now().strftime(\"%Y%m%d_%H%M%S\")
    src = topic if topic else prompt[:30]
    safe = re.sub(r'[^\\w]', '_', src.strip())[:30].strip('_')
    return f\"{date_str}_{safe}.png\"

def generate(prompt, topic='', steps=9, width=512, height=512, seed=42, output_path=None):
    root = get_image_gen_dir()
    model_dir = root / \"Z-Image-Turbo-int4-ov\"
    out_dir = root / \"outputs\"
    out_dir.mkdir(parents=True, exist_ok=True)

    missing = [r for r in [\"transformer\",\"vae_decoder\",\"text_encoder\"] if not (model_dir/r).exists()]
    if missing:
        print(f\"[ERROR] 模型不完整: {missing}，请执行 Step 3\")
        sys.exit(1)

    device = get_device()
    print(f\"[INFO] 加载模型: {model_dir}\")
    pipe = OVZImagePipeline.from_pretrained(str(model_dir), device=device)
    print(\"[INFO] 模型加载完成\")

    gen = torch.Generator(\"cpu\").manual_seed(seed) if seed >= 0 else None
    print(f\"[INFO] 推理: steps={steps}, {width}x{height}, seed={seed}\")
    image = pipe(prompt=prompt, height=height, width=width,
                 num_inference_steps=steps, guidance_scale=0.0, generator=gen).images[0]

    if output_path is None:
        output_path = str(out_dir / make_filename(prompt, topic))
    image.save(output_path)
    print(f\"[SUCCESS] 图像已保存: {output_path}\")
    try:
        # 用 explorer.exe 打开：独立进程、原生 Unicode 路径、不依赖 Shell 消息循环
        subprocess.Popen(['explorer', output_path])
        print(\"[INFO] 已用默认程序打开图像\")
    except Exception as e:
        print(f\"[WARN] 打开图像失败: {e}\")
    return output_path

if __name__ == \"__main__\":
    # 强制 stdout UTF-8，避免中文路径在 Windows GBK 终端乱码
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
    p = argparse.ArgumentParser()
    p.add_argument(\"--prompt\", required=True)
    p.add_argument(\"--topic\",  default='')
    p.add_argument(\"--steps\",  type=int, default=9)
    p.add_argument(\"--width\",  type=int, default=512)
    p.add_argument(\"--height\", type=int, default=512)
    p.add_argument(\"--seed\",   type=int, default=42)
    p.add_argument(\"--output\", default=None)
    args = p.parse_args()
    print(generate(args.prompt, args.topic, args.steps, args.width, args.height, args.seed, args.output))
'''
    script.write_text(code.strip(), encoding='utf-8')
    print('SCRIPT_UPDATE=DONE')

print(f'EXISTS={script.exists()}')
"