| 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")) |
|
|
| |
| 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 |
|
|
