File size: 1,981 Bytes
638cad0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 | import requests, time, struct, zlib, base64
def make_png():
w, h = 256, 192
rows = []
for y in range(h):
row = bytearray([0])
for x in range(w):
left = (60 <= x <= 110 and 60 <= y <= 130)
right = (150 <= x <= 200 and 60 <= y <= 130)
band = (110 <= x <= 150 and 30 <= y <= 45)
if left or right or band:
row += bytes([120, 80, 200])
else:
row += bytes([248, 248, 250])
rows.append(bytes(row))
raw = b"".join(rows)
def chunk(t, d):
c = t + d
return struct.pack(">I", len(d)) + c + struct.pack(">I", zlib.crc32(c) & 0xffffffff)
ihdr = struct.pack(">IIBBBBB", w, h, 8, 2, 0, 0, 0)
return b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", ihdr) + chunk(b"IDAT", zlib.compress(raw)) + chunk(b"IEND", b"")
png = make_png()
BASE = "http://127.0.0.1:8000"
H = {"Authorization": "Bearer sk-AQM1TfaVx2VSzEoTLiV1J2NH1B92g6AI"}
files = {"image": ("headphone.png", png, "image/png")}
data = {"client_task_id": "typewriter_demo_002", "prompt": "紫色头戴式蓝牙耳机，主动降噪，40mm动圈，适合通勤和游戏", "scenario": "detail_page", "platform": "jd", "output_count": "6"}
r = requests.post(BASE + "/api/product-visual/tasks", headers=H, data=data, files=files, timeout=30)
t = r.json()
print("CREATE:", r.status_code, "stage:", t.get("plan_stage"))
# 每 5 秒采样 analysis 是否出现且内容逐渐完整
prev_len = 0
for i in range(16):
time.sleep(5)
r2 = requests.get(BASE + "/api/product-visual/tasks?ids=typewriter_demo_002", headers=H, timeout=15)
tt = r2.json().get("items", [{}])[0]
stage = tt.get("plan_stage")
a = tt.get("plan_analysis") or {}
cat = a.get("category") or ""
print(f" [{i*5+5}s] stage={stage} analysis_len={len(cat)} status={tt.get('status')}")
if cat:
print(f" category: {cat[:60]}")
if stage == "done":
print(" DONE")
break
|