webp fix v2
This commit is contained in:
tamakyi
@@ -0,0 +1,346 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""头像上传链路端到端验证（批次 A 分档 + B1 EXIF 归一）。
|
||||
|
||||
为什么走头像而不是提问图：
|
||||
- 提问图那条路要经过内容审核/提问入库才能拿到 key，链路长且易受其它因素干扰；
|
||||
- 头像强制 1:1 且 POST /user/profile/update 直接返回页面，便于立即取回 CDN URL。
|
||||
|
||||
验证两件事：
|
||||
A 分档生效：上传 1600x1600 的大图，头像档 maxSide=512 -> 存盘应为 512x512
|
||||
（改前是 1920，实测体积差 ~7 倍）
|
||||
B1 EXIF Orientation=6 归一：上传「像素 512x256 + EXIF 顺时针 90」的方形图，
|
||||
归一后仍应是 512x512（宽高互换但比例不变，故改用体积+尺寸双断言，
|
||||
并额外验证竖长条图 256x512 + Orientation=6 -> 应变成 512x256）
|
||||
|
||||
用法：
|
||||
python dev/devcheck-avatar.py [--base https://box-dev-api.337845818.best]
|
||||
依赖：Pillow（仅用于本地生成测试图片）
|
||||
清理：TamaBox ban --email <输出中的测试邮箱> --yes
|
||||
"""
|
||||
import argparse
|
||||
import http.cookiejar
|
||||
import io
|
||||
import json
|
||||
import re
|
||||
import struct
|
||||
import sys
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
import uuid
|
||||
|
||||
UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
||||
"(KHTML, like Gecko) Chrome/124.0 Safari/537.36")
|
||||
RESULTS = []
|
||||
|
||||
|
||||
class NoRedirect(urllib.request.HTTPRedirectHandler):
|
||||
def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: D102
|
||||
return None
|
||||
|
||||
|
||||
class Resp:
|
||||
def __init__(self, status, headers, body):
|
||||
self.status = status
|
||||
self.h = {k.lower(): v for k, v in headers.items()}
|
||||
self.body = body
|
||||
|
||||
@property
|
||||
def loc(self):
|
||||
return self.h.get("location", "")
|
||||
|
||||
@property
|
||||
def text(self):
|
||||
return self.body.decode("utf-8", "replace")
|
||||
|
||||
|
||||
class Client:
|
||||
def __init__(self, base):
|
||||
self.base = base
|
||||
self.cj = http.cookiejar.CookieJar()
|
||||
self.op = urllib.request.build_opener(
|
||||
urllib.request.HTTPCookieProcessor(self.cj), NoRedirect())
|
||||
|
||||
def _send(self, req, timeout=120):
|
||||
try:
|
||||
with self.op.open(req, timeout=timeout) as r:
|
||||
return Resp(r.status, dict(r.headers), r.read())
|
||||
except urllib.error.HTTPError as e:
|
||||
try:
|
||||
body = e.read()
|
||||
except Exception: # noqa: BLE001
|
||||
body = b""
|
||||
return Resp(e.code, dict(e.headers), body)
|
||||
except Exception as e: # noqa: BLE001
|
||||
return Resp(-1, {}, ("TRANSPORT_ERROR: %s" % e).encode())
|
||||
|
||||
def req(self, method, path, data=None, content_type=None, timeout=120):
|
||||
url = path if path.startswith("http") else self.base + path
|
||||
headers = {"User-Agent": UA, "Referer": self.base + path.split("?")[0]}
|
||||
body = None
|
||||
if data is not None:
|
||||
body = data if isinstance(data, bytes) else urllib.parse.urlencode(data).encode()
|
||||
headers["Content-Type"] = content_type or "application/x-www-form-urlencoded"
|
||||
return self._send(urllib.request.Request(url, data=body, headers=headers, method=method),
|
||||
timeout)
|
||||
|
||||
def get(self, p, **kw):
|
||||
r = self.req("GET", p, **kw)
|
||||
return r if r.status != -1 else self.req("GET", p, **kw)
|
||||
|
||||
def post(self, p, data=None, **kw):
|
||||
return self.req("POST", p, data=data, **kw)
|
||||
|
||||
def get_binary(self, url):
|
||||
return self._send(urllib.request.Request(
|
||||
url, headers={"User-Agent": UA}, method="GET"), timeout=60)
|
||||
|
||||
|
||||
def step(desc, ok, extra=""):
|
||||
RESULTS.append((desc, ok, extra))
|
||||
print(" [%s] %s%s" % ("PASS" if ok else "FAIL", desc, (" (%s)" % extra) if extra else ""))
|
||||
return ok
|
||||
|
||||
|
||||
def csrf_of(html):
|
||||
m = re.search(r'name="_csrf"\s+value="([^"]+)"', html)
|
||||
return m.group(1) if m else None
|
||||
|
||||
|
||||
def captcha(c):
|
||||
j = json.loads(c.get("/api/captcha/generate").text)
|
||||
pid, q = j["data"]["id"], j["data"]["question"]
|
||||
m = re.match(r"\s*(\d+)\s*([+\-×])\s*(\d+)\s*=", q)
|
||||
a, op, b = int(m.group(1)), m.group(2), int(m.group(3))
|
||||
return "%s:%d" % (pid, {"+": a + b, "-": a - b, "×": a * b}[op])
|
||||
|
||||
|
||||
def multipart(fields, files):
|
||||
boundary = "----tbav%s" % uuid.uuid4().hex
|
||||
buf = io.BytesIO()
|
||||
for k, v in fields.items():
|
||||
buf.write(("--%s\r\n" % boundary).encode())
|
||||
buf.write(('Content-Disposition: form-data; name="%s"\r\n\r\n' % k).encode())
|
||||
buf.write(("%s\r\n" % v).encode())
|
||||
for name, fn, ct, data in files:
|
||||
buf.write(("--%s\r\n" % boundary).encode())
|
||||
buf.write(('Content-Disposition: form-data; name="%s"; filename="%s"\r\n' % (name, fn)).encode())
|
||||
buf.write(("Content-Type: %s\r\n\r\n" % ct).encode())
|
||||
buf.write(data)
|
||||
buf.write(b"\r\n")
|
||||
buf.write(("--%s--\r\n" % boundary).encode())
|
||||
return buf.getvalue(), "multipart/form-data; boundary=%s" % boundary
|
||||
|
||||
|
||||
def sniff(d):
|
||||
if len(d) >= 12 and d[:4] == b"RIFF" and d[8:12] == b"WEBP":
|
||||
return "webp", webp_dims(d)
|
||||
if d[:3] == b"\xff\xd8\xff":
|
||||
return "jpeg", jpeg_dims(d)
|
||||
if d[:8] == b"\x89PNG\r\n\x1a\n":
|
||||
return "png", (struct.unpack(">II", d[16:24]) if len(d) >= 24 else None)
|
||||
return "unknown", None
|
||||
|
||||
|
||||
def webp_dims(d):
|
||||
c = d[12:16]
|
||||
if c == b"VP8 " and len(d) >= 30:
|
||||
return (struct.unpack("<H", d[26:28])[0] & 0x3FFF,
|
||||
struct.unpack("<H", d[28:30])[0] & 0x3FFF)
|
||||
if c == b"VP8L" and len(d) >= 25:
|
||||
b = struct.unpack("<I", d[21:25])[0]
|
||||
return (b & 0x3FFF) + 1, ((b >> 14) & 0x3FFF) + 1
|
||||
if c == b"VP8X" and len(d) >= 30:
|
||||
return ((d[24] | d[25] << 8 | d[26] << 16) + 1,
|
||||
(d[27] | d[28] << 8 | d[29] << 16) + 1)
|
||||
return None
|
||||
|
||||
|
||||
def jpeg_dims(d):
|
||||
i = 2
|
||||
while i + 9 < len(d):
|
||||
if d[i] != 0xFF:
|
||||
i += 1
|
||||
continue
|
||||
m = d[i + 1]
|
||||
if m in (0xC0, 0xC1, 0xC2, 0xC3):
|
||||
h, w = struct.unpack(">HH", d[i + 5:i + 9])
|
||||
return (w, h)
|
||||
if m in (0xD8, 0xD9) or 0xD0 <= m <= 0xD7:
|
||||
i += 2
|
||||
continue
|
||||
i += 2 + struct.unpack(">H", d[i + 2:i + 4])[0]
|
||||
return None
|
||||
|
||||
|
||||
def make_avatar(w, h, orientation=None):
|
||||
"""生成一张带明显方位特征的方形头像。"""
|
||||
from PIL import Image, ImageDraw
|
||||
im = Image.new("RGB", (w, h), (30, 120, 200))
|
||||
d = ImageDraw.Draw(im)
|
||||
# 顶部一条粗红条：旋转 90 度后它会跑到侧面，方向一眼可辨
|
||||
d.rectangle([0, 0, w, max(2, h // 12)], fill=(230, 30, 30))
|
||||
for i in range(0, w, max(8, w // 20)):
|
||||
d.line([(i, h // 4), (i, h)], fill=(255, 255, 255), width=2)
|
||||
buf = io.BytesIO()
|
||||
if orientation:
|
||||
exif = im.getexif()
|
||||
exif[0x0112] = orientation
|
||||
im.save(buf, "JPEG", quality=90, exif=exif)
|
||||
else:
|
||||
im.save(buf, "JPEG", quality=90)
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
def scrape_profile_fields(html):
|
||||
"""从 profile 页抓取所有 input 的 name/value，保证回填必填项。
|
||||
|
||||
UpdateProfile 的必填字段有十来个（昵称/介绍/5 个二维码颜色/点大小/透明度…），
|
||||
漏一个就会因 `valid:"required"` 被 Bind 挡下、头像根本不会走到上传分支。
|
||||
"""
|
||||
fields = {}
|
||||
for m in re.finditer(r'<input\b[^>]*>', html, re.I):
|
||||
tag = m.group(0)
|
||||
nm = re.search(r'name="([^"]+)"', tag, re.I)
|
||||
if not nm:
|
||||
continue
|
||||
name = nm.group(1)
|
||||
if name in ("avatar", "background", "backgroundimage", "_csrf", "submit"):
|
||||
continue
|
||||
typ = re.search(r'type="([^"]+)"', tag, re.I)
|
||||
if typ and typ.group(1).lower() in ("checkbox", "radio"):
|
||||
# 只保留勾选态的，避免把「本想取消」的项又勾上
|
||||
if not re.search(r'\bchecked\b', tag, re.I):
|
||||
continue
|
||||
val = re.search(r'value="([^"]*)"', tag, re.I)
|
||||
fields[name] = val.group(1) if val else ""
|
||||
# select 走第一个非空 option
|
||||
for m in re.finditer(r'<select\b[^>]*name="([^"]+)"[^>]*>(.*?)</select>', html, re.I | re.S):
|
||||
name, body = m.group(1), m.group(2)
|
||||
opts = re.findall(r'<option\b([^>]*)>', body, re.I)
|
||||
chosen = ""
|
||||
for o in opts:
|
||||
vm = re.search(r'value="([^"]*)"', o)
|
||||
v = vm.group(1) if vm else ""
|
||||
if re.search(r'\bselected\b', o, re.I):
|
||||
chosen = v
|
||||
break
|
||||
if not chosen:
|
||||
chosen = v
|
||||
fields[name] = chosen
|
||||
# textarea
|
||||
for m in re.finditer(r'<textarea\b[^>]*name="([^"]+)"[^>]*>(.*?)</textarea>', html, re.I | re.S):
|
||||
fields[m.group(1)] = re.sub(r"<[^>]+>", "", m.group(2)).strip()
|
||||
return fields
|
||||
|
||||
|
||||
def upload_avatar(c, data, fname="avatar.jpg"):
|
||||
page = c.get("/user/profile")
|
||||
if page.status != 200:
|
||||
return None, "profile HTTP %s" % page.status
|
||||
fields = scrape_profile_fields(page.text)
|
||||
fields["_csrf"] = csrf_of(page.text)
|
||||
body, ct = multipart(fields, [("avatar", fname, "image/jpeg", data)])
|
||||
res = c.post("/user/profile/update", data=body, content_type=ct)
|
||||
# 成功时是 302 跳回 /user/profile，头像 URL 在跳转后的页面里
|
||||
if res.status == 302 and res.loc:
|
||||
res = c.get(res.loc if res.loc.startswith("/") else "/user/profile")
|
||||
if res.status != 200:
|
||||
return None, "update 后页面 HTTP %s" % res.status
|
||||
# 从页面里取头像 URL：<img src="https://r2.../picture/..."> 或 og:image
|
||||
urls = re.findall(r'https://r2\.[^"\'\s<>]+/picture/[0-9]+/\d{2}/\d{2}/[0-9a-f]+', res.text)
|
||||
if not urls:
|
||||
m = re.search(r'(https://r2\.[^"\'\s<>]+/picture/[^"\'\s<>]+)', res.text)
|
||||
urls = [m.group(1)] if m else []
|
||||
if urls:
|
||||
return urls[-1], None
|
||||
# 没匹配到：落盘返回页并打印可疑片段，便于定位是「上传被拒」还是「正则没覆盖」
|
||||
dump = "C:/Users/Tamakyi Gekka/AppData/Local/Temp/tbav_profile.html"
|
||||
with open(dump, "wb") as f:
|
||||
f.write(res.body)
|
||||
alerts = re.findall(r'class="uk-alert[^"]*"[^>]*>(.*?)</div>', res.text, re.S)
|
||||
hint = " | ".join(re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", a)).strip()[:80]
|
||||
for a in alerts[:3])
|
||||
return None, "未匹配到 picture URL（已存 %s）%s%s" % (dump, (" 提示:" + hint) if hint else "",
|
||||
(" HTTP %s" % res.status))
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--base", default="https://box-dev-api.337845818.best")
|
||||
args = ap.parse_args()
|
||||
base = args.base.rstrip("/")
|
||||
pwd = "TbAv!2026"
|
||||
ts = time.strftime("%m%d%H%M%S")
|
||||
email, domain = "tbav%s@example.com" % ts, "tbav%s" % ts
|
||||
print("== 账号：%s / %s ==" % (email, domain))
|
||||
|
||||
# 方形 1600x1600：验证 A 分档（应存成 512x512）
|
||||
square = make_avatar(1600, 1600)
|
||||
# EXIF Orientation=6（顺时针 90）：像素存成横的 1536x512，归一后显示为竖的 512x1536，
|
||||
# 再经头像档限边 512 -> 512x512。
|
||||
# 注意必须用「像素横躺」的样本：头像有 1:1 比例校验，若像素本身就是竖的会被先拒掉，
|
||||
# 那样就测不到 EXIF 分支了。
|
||||
exif_shot = make_avatar(1536, 512, orientation=6)
|
||||
print(" 本地样本 square 1600x1600%7d bytes" % len(square))
|
||||
print(" 本地样本 exifshot 1536x512 (EXIF=6)%d bytes" % len(exif_shot))
|
||||
|
||||
c = Client(base)
|
||||
r = c.get("/register")
|
||||
reg = c.post("/register", {
|
||||
"email": email, "domain": domain, "name": "TBAv%s" % ts,
|
||||
"password": pwd, "repeat_password": pwd,
|
||||
"g-recaptcha-response": captcha(c), "language": "zh-CN",
|
||||
"_csrf": csrf_of(r.text),
|
||||
})
|
||||
if not step("注册", reg.status == 302, "HTTP %s -> %s" % (reg.status, reg.loc)):
|
||||
return 1
|
||||
|
||||
c = Client(base)
|
||||
r = c.get("/login")
|
||||
lg = c.post("/login", {"email": email, "password": pwd,
|
||||
"g-recaptcha-response": captcha(c), "_csrf": csrf_of(r.text)})
|
||||
if not step("登录", lg.status == 302, "HTTP %s -> %s" % (lg.status, lg.loc)):
|
||||
return 1
|
||||
|
||||
# ---- A 分档 ----
|
||||
url, err = upload_avatar(c, square)
|
||||
if not step("上传方形大头像（1600x1600）", url is not None, err or (url or "")):
|
||||
return 1
|
||||
img = c.get_binary(url)
|
||||
st, data = img.status, img.body
|
||||
if not step("拉回头像", st == 200 and bool(data), "HTTP %s" % st):
|
||||
return 1
|
||||
kind, dims = sniff(data)
|
||||
print(" 存储结果：%s %s %d bytes" % (kind, dims, len(data)))
|
||||
step("A 分档生效：大图被缩到头像档 512（改前是 1920）",
|
||||
dims == (512, 512), "实际 %s" % (dims,))
|
||||
|
||||
# ---- B1 EXIF ----
|
||||
# 关键：两次上传的 URL 必须不同，否则说明第二次没生效（回退到了旧头像）。
|
||||
url2, err2 = upload_avatar(c, exif_shot)
|
||||
if not step("上传 EXIF 横躺头像（1536x512, Orientation=6）", bool(url2), err2 or (url2 or "")):
|
||||
return 0
|
||||
if url2 == url:
|
||||
step("B1 EXIF 归一（第二次上传未生效，URL 未变）", False,
|
||||
"两次上传返回同一 URL，无法区分；请检查是否被 1:1 校验拒绝")
|
||||
return 0
|
||||
img2 = c.get_binary(url2)
|
||||
kind2, dims2 = sniff(img2.body)
|
||||
print(" 存储结果：%s %s %d bytes" % (kind2, dims2, len(img2.body)))
|
||||
# 归一后：像素 1536x512 -> 旋转成 512x1536 -> 再限边 512 -> 512x512
|
||||
step("B1 EXIF Orientation=6 归一（1536x512 -> 512x1536 -> 512x512）",
|
||||
dims2 == (512, 512), "实际 %s" % (dims2,))
|
||||
# 体积对照：未归一时 1536x512 的 WebP 明显更大
|
||||
step("B1 归一后体积已按 512 档压过（应与方形样本同量级）",
|
||||
len(img2.body) < 6000, "%d bytes" % len(img2.body))
|
||||
|
||||
print("\n== 结果：%d/%d 通过 ==" % (sum(1 for _, ok, _ in RESULTS if ok), len(RESULTS)))
|
||||
print("清理：TamaBox ban --email %s --yes" % email)
|
||||
return 0 if all(ok for _, ok, _ in RESULTS) else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,370 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""图片上传链路端到端验证（批次 A / B1 / B2）。
|
||||
|
||||
针对已部署环境验证三件事：
|
||||
B1 EXIF Orientation=6 的竖拍 JPEG 转换后必须变成竖图（核心验收点）
|
||||
B2 动画 GIF 必须被转成 .webp（cwebp 可用性 + 提问路径与头像路径一致性）
|
||||
A 转换后不比原图小时必须回退原图（用一张极小线框图验证）
|
||||
|
||||
做法：注册账号 -> 匿名提问（multipart 同时带三张图）-> 解析成功页里的 CDN URL
|
||||
-> 逐张拉回字节，解析真实容器格式与尺寸。
|
||||
|
||||
用法：
|
||||
python dev/devcheck-image.py [--base https://box-dev-api.337845818.best]
|
||||
|
||||
依赖：Pillow（仅用于在本地生成测试图片，不参与被测逻辑）。
|
||||
注意：会在目标环境真实创建账号并发送带图提问（测试邮箱前缀 tbimg<ts>@）。
|
||||
清理：TamaBox ban --email <该邮箱> --yes
|
||||
"""
|
||||
import argparse
|
||||
import http.cookiejar
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
import re
|
||||
import struct
|
||||
import sys
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
import uuid
|
||||
|
||||
UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
||||
"(KHTML, like Gecko) Chrome/124.0 Safari/537.36")
|
||||
RESULTS = []
|
||||
CREATED_EMAIL = ""
|
||||
|
||||
|
||||
class NoRedirect(urllib.request.HTTPRedirectHandler):
|
||||
def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: D102
|
||||
return None
|
||||
|
||||
|
||||
class Resp:
|
||||
def __init__(self, status, headers, body):
|
||||
self.status = status
|
||||
self.h = {k.lower(): v for k, v in headers.items()}
|
||||
self.body = body
|
||||
|
||||
@property
|
||||
def loc(self):
|
||||
return self.h.get("location", "")
|
||||
|
||||
@property
|
||||
def text(self):
|
||||
return self.body.decode("utf-8", "replace")
|
||||
|
||||
|
||||
class Client:
|
||||
def __init__(self, base):
|
||||
self.base = base
|
||||
self.cj = http.cookiejar.CookieJar()
|
||||
self.op = urllib.request.build_opener(
|
||||
urllib.request.HTTPCookieProcessor(self.cj), NoRedirect())
|
||||
|
||||
def _send(self, req, timeout=120):
|
||||
try:
|
||||
with self.op.open(req, timeout=timeout) as r:
|
||||
return Resp(r.status, dict(r.headers), r.read())
|
||||
except urllib.error.HTTPError as e:
|
||||
try:
|
||||
body = e.read()
|
||||
except Exception: # noqa: BLE001
|
||||
body = b""
|
||||
return Resp(e.code, dict(e.headers), body)
|
||||
except Exception as e: # noqa: BLE001
|
||||
return Resp(-1, {}, ("TRANSPORT_ERROR: %s" % e).encode())
|
||||
|
||||
def req(self, method, path, data=None, content_type=None, timeout=120):
|
||||
url = path if path.startswith("http") else self.base + path
|
||||
headers = {"User-Agent": UA, "Referer": self.base + path.split("?")[0]}
|
||||
body = None
|
||||
if data is not None:
|
||||
body = data if isinstance(data, bytes) else urllib.parse.urlencode(data).encode()
|
||||
headers["Content-Type"] = content_type or "application/x-www-form-urlencoded"
|
||||
return self._send(urllib.request.Request(url, data=body, headers=headers, method=method),
|
||||
timeout)
|
||||
|
||||
def get(self, p, **kw):
|
||||
r = self.req("GET", p, **kw)
|
||||
return r if r.status != -1 else self.req("GET", p, **kw)
|
||||
|
||||
def post(self, p, data=None, **kw):
|
||||
return self.req("POST", p, data=data, **kw)
|
||||
|
||||
def get_binary(self, url):
|
||||
return self._send(urllib.request.Request(
|
||||
url, headers={"User-Agent": UA}, method="GET"), timeout=60)
|
||||
|
||||
|
||||
def step(desc, ok, extra=""):
|
||||
RESULTS.append((desc, ok, extra))
|
||||
print(" [%s] %s%s" % ("PASS" if ok else "FAIL", desc, (" (%s)" % extra) if extra else ""))
|
||||
return ok
|
||||
|
||||
|
||||
def csrf_of(html):
|
||||
m = re.search(r'name="_csrf"\s+value="([^"]+)"', html)
|
||||
return m.group(1) if m else None
|
||||
|
||||
|
||||
def captcha(c):
|
||||
j = json.loads(c.get("/api/captcha/generate").text)
|
||||
pid, q = j["data"]["id"], j["data"]["question"]
|
||||
m = re.match(r"\s*(\d+)\s*([+\-×])\s*(\d+)\s*=", q)
|
||||
a, op, b = int(m.group(1)), m.group(2), int(m.group(3))
|
||||
return "%s:%d" % (pid, {"+": a + b, "-": a - b, "×": a * b}[op])
|
||||
|
||||
|
||||
def multipart(fields, files):
|
||||
boundary = "----tbimg%s" % uuid.uuid4().hex
|
||||
buf = io.BytesIO()
|
||||
for k, v in fields.items():
|
||||
buf.write(("--%s\r\n" % boundary).encode())
|
||||
buf.write(('Content-Disposition: form-data; name="%s"\r\n\r\n' % k).encode())
|
||||
buf.write(("%s\r\n" % v).encode())
|
||||
for name, fn, ct, data in files:
|
||||
buf.write(("--%s\r\n" % boundary).encode())
|
||||
buf.write(('Content-Disposition: form-data; name="%s"; filename="%s"\r\n' % (name, fn)).encode())
|
||||
buf.write(("Content-Type: %s\r\n\r\n" % ct).encode())
|
||||
buf.write(data)
|
||||
buf.write(b"\r\n")
|
||||
buf.write(("--%s--\r\n" % boundary).encode())
|
||||
return buf.getvalue(), "multipart/form-data; boundary=%s" % boundary
|
||||
|
||||
|
||||
# ---------- 容器格式解析（纯 stdlib） ----------
|
||||
|
||||
def sniff(data):
|
||||
"""返回 (kind, (w,h))。kind in {webp, jpeg, gif, png, unknown}"""
|
||||
if len(data) >= 12 and data[:4] == b"RIFF" and data[8:12] == b"WEBP":
|
||||
return "webp", webp_dims(data)
|
||||
if data[:3] == b"\xff\xd8\xff":
|
||||
return "jpeg", jpeg_dims(data)
|
||||
if data[:3] == b"GIF":
|
||||
return "gif", gif_dims(data)
|
||||
if data[:8] == b"\x89PNG\r\n\x1a\n":
|
||||
return "png", png_dims(data)
|
||||
return "unknown", None
|
||||
|
||||
|
||||
def webp_dims(d):
|
||||
c = d[12:16]
|
||||
if c == b"VP8 " and len(d) >= 30:
|
||||
return (struct.unpack("<H", d[26:28])[0] & 0x3FFF,
|
||||
struct.unpack("<H", d[28:30])[0] & 0x3FFF)
|
||||
if c == b"VP8L" and len(d) >= 25:
|
||||
b = struct.unpack("<I", d[21:25])[0]
|
||||
return (b & 0x3FFF) + 1, ((b >> 14) & 0x3FFF) + 1
|
||||
if c == b"VP8X" and len(d) >= 30:
|
||||
return ((d[24] | d[25] << 8 | d[26] << 16) + 1,
|
||||
(d[27] | d[28] << 8 | d[29] << 16) + 1)
|
||||
return None
|
||||
|
||||
|
||||
def jpeg_dims(d):
|
||||
i = 2
|
||||
while i + 9 < len(d):
|
||||
if d[i] != 0xFF:
|
||||
i += 1
|
||||
continue
|
||||
m = d[i + 1]
|
||||
if m in (0xC0, 0xC1, 0xC2, 0xC3):
|
||||
h, w = struct.unpack(">HH", d[i + 5:i + 9])
|
||||
return (w, h)
|
||||
if m in (0xD8, 0xD9) or 0xD0 <= m <= 0xD7:
|
||||
i += 2
|
||||
continue
|
||||
i += 2 + struct.unpack(">H", d[i + 2:i + 4])[0]
|
||||
return None
|
||||
|
||||
|
||||
def gif_dims(d):
|
||||
return struct.unpack("<HH", d[6:10]) if len(d) >= 10 else None
|
||||
|
||||
|
||||
def png_dims(d):
|
||||
return struct.unpack(">II", d[16:24]) if len(d) >= 24 else None
|
||||
|
||||
|
||||
# ---------- 测试图片生成（Pillow 仅此一处使用） ----------
|
||||
|
||||
def make_images():
|
||||
from PIL import Image, ImageDraw
|
||||
import io as _io
|
||||
|
||||
# B1：EXIF Orientation=6。像素存 240x120，正确归一后应显示 120x240。
|
||||
base = Image.new("RGB", (240, 120))
|
||||
dr = ImageDraw.Draw(base)
|
||||
for i in range(0, 240, 20):
|
||||
dr.line([(i, 0), (i, 120)], fill=(200, 30, 30), width=3)
|
||||
dr.rectangle([0, 0, 239, 119], outline=(0, 0, 255), width=4)
|
||||
exif = base.getexif()
|
||||
exif[0x0112] = 6
|
||||
buf = _io.BytesIO()
|
||||
base.save(buf, "JPEG", quality=88, exif=exif)
|
||||
portrait = buf.getvalue()
|
||||
|
||||
# B2：6 帧动画 GIF。加大到 640x480，确保转换后仍小于原图（否则会走回退）。
|
||||
frames = []
|
||||
pal_frames = []
|
||||
for f in range(6):
|
||||
im = Image.new("RGB", (640, 480), (240, 240, 240))
|
||||
d = ImageDraw.Draw(im)
|
||||
d.ellipse([40 + f * 60, 120, 200 + f * 60, 280], fill=(220, 40, 40))
|
||||
d.rectangle([300, 60 + f * 20, 600, 200], fill=(30, 90, 220))
|
||||
pal_frames.append(im.convert("P", palette=Image.ADAPTIVE, colors=64))
|
||||
gbuf = _io.BytesIO()
|
||||
pal_frames[0].save(gbuf, "GIF", save_all=True, append_images=pal_frames[1:],
|
||||
duration=120, loop=0)
|
||||
animated = gbuf.getvalue()
|
||||
|
||||
# A：极小线框图（200x200），WebP 有损编码后反而比 PNG 大 -> 应回退原图。
|
||||
wire = Image.new("RGB", (200, 200), (255, 255, 255))
|
||||
d = ImageDraw.Draw(wire)
|
||||
for i in range(0, 200, 20):
|
||||
d.line([(i, 0), (i, 199)], fill=(0, 0, 0))
|
||||
d.line([(0, i), (199, i)], fill=(0, 0, 0))
|
||||
wbuf = _io.BytesIO()
|
||||
wire.save(wbuf, "PNG")
|
||||
wireframe = wbuf.getvalue()
|
||||
|
||||
return {"portrait": portrait, "anim": animated, "wire": wireframe}
|
||||
|
||||
|
||||
def main():
|
||||
global CREATED_EMAIL
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--base", default="https://box-dev-api.337845818.best")
|
||||
args = ap.parse_args()
|
||||
base = args.base.rstrip("/")
|
||||
pwd = "TbImg!2026"
|
||||
ts = time.strftime("%m%d%H%M%S")
|
||||
email, domain = "tbimg%s@example.com" % ts, "tbimg%s" % ts
|
||||
CREATED_EMAIL = email
|
||||
print("== 账号：%s / %s ==" % (email, domain))
|
||||
|
||||
imgs = make_images()
|
||||
for k, v in imgs.items():
|
||||
print(" 本地样本 %-9s %7d bytes" % (k, len(v)))
|
||||
|
||||
c = Client(base)
|
||||
r = c.get("/register")
|
||||
reg = c.post("/register", {
|
||||
"email": email, "domain": domain, "name": "TBImg%s" % ts,
|
||||
"password": pwd, "repeat_password": pwd,
|
||||
"g-recaptcha-response": captcha(c), "language": "zh-CN",
|
||||
"_csrf": csrf_of(r.text),
|
||||
})
|
||||
if not step("注册", reg.status == 302, "HTTP %s -> %s" % (reg.status, reg.loc)):
|
||||
return 1
|
||||
|
||||
c = Client(base)
|
||||
r = c.get("/login")
|
||||
lg = c.post("/login", {"email": email, "password": pwd,
|
||||
"g-recaptcha-response": captcha(c), "_csrf": csrf_of(r.text)})
|
||||
if not step("登录", lg.status == 302, "HTTP %s -> %s" % (lg.status, lg.loc)):
|
||||
return 1
|
||||
|
||||
page = c.get("/_/%s" % domain)
|
||||
if not step("打开提问箱", page.status == 200, "HTTP %s" % page.status):
|
||||
return 1
|
||||
|
||||
# route/question/page.go 只取 f.Images[0]（单张上限），
|
||||
# 故每张样本各发一条提问，逐张验证。
|
||||
fetched = []
|
||||
for key, (fname, ctype) in (("portrait", ("portrait.jpg", "image/jpeg")),
|
||||
("anim", ("anim.gif", "image/gif")),
|
||||
("wire", ("wire.png", "image/png"))):
|
||||
box = c.get("/_/%s" % domain)
|
||||
body, ct = multipart({
|
||||
"content": "图片上传链路验证 %s（可忽略）" % key,
|
||||
"g-recaptcha-response": captcha(c),
|
||||
"receive_reply_email": "", "private": "", "ask_password": "",
|
||||
"start_chat_val": "", "_csrf": csrf_of(box.text),
|
||||
}, [("images", fname, ctype, imgs[key])])
|
||||
|
||||
res = c.post("/_/%s" % domain, data=body, content_type=ct)
|
||||
if not step("提交带图提问 [%s]" % key, res.status in (200, 302),
|
||||
"HTTP %s -> %s" % (res.status, res.loc)):
|
||||
continue
|
||||
|
||||
# 成功页不渲染图片（templates/question/success.html 无图片区），
|
||||
# 故从成功页取出 question ID，再访问详情页取图片 key。
|
||||
qid = re.search(r'/_/%s/(\d+)' % re.escape(domain), res.text)
|
||||
if not qid:
|
||||
step(" └ 取提问 ID [%s]" % key, False, "成功页未给出 /_/<domain>/<id>")
|
||||
continue
|
||||
detail = c.get("/_/%s/%s" % (domain, qid.group(1)))
|
||||
# 提问图的 key 不带 picture/ 前缀（是 "<年>/<月>/<纳秒>.<ext>"），
|
||||
# 模板拼成 https://<CDN>/<key>，故按 CDN host + 数字开头路径来匹配。
|
||||
urls = sorted(set(re.findall(
|
||||
r'https://r2\.[^"\'\s<>]+/\d{4}/\d{1,2}/\d+\.\w+', detail.text)))
|
||||
if not urls:
|
||||
step(" └ 详情页回显图片 [%s]" % key, False,
|
||||
"HTTP %s，未找到 picture URL" % detail.status)
|
||||
continue
|
||||
step(" └ 详情页回显图片 [%s]" % key, True, "%d 个" % len(urls))
|
||||
|
||||
for u in urls:
|
||||
st, data = c.get_binary(u)
|
||||
if st != 200 or not data:
|
||||
step(" └ 拉回图片 [%s]" % key, False, "HTTP %s" % st)
|
||||
continue
|
||||
kind, dims = sniff(data)
|
||||
fetched.append((key, kind, dims, len(data)))
|
||||
print(" %-14s %-7s %-12s %7d bytes" % (u.rsplit("/", 1)[-1], kind, dims, len(data)))
|
||||
|
||||
|
||||
print("\n== 断言 ==")
|
||||
|
||||
# B1
|
||||
portrait = [f for f in fetched if f[0] == "portrait" and f[1] in ("webp", "jpeg")
|
||||
and f[2] == (120, 240)]
|
||||
land = [f for f in fetched if f[0] == "portrait" and f[1] in ("webp", "jpeg")
|
||||
and f[2] == (240, 120)]
|
||||
if portrait:
|
||||
step("B1 EXIF Orientation=6 已归一（240x120 -> 120x240）", True,
|
||||
"%s, %d bytes" % (portrait[0][1], portrait[0][3]))
|
||||
elif land:
|
||||
step("B1 EXIF Orientation=6 已归一", False, "仍是 240x120，说明未应用 EXIF 旋转")
|
||||
else:
|
||||
step("B1 EXIF 方向归一", False, "未匹配到 240x120 / 120x240 的图片，实际: %s" %
|
||||
[(f[1], f[2]) for f in fetched])
|
||||
|
||||
# B2：GIF 样本必须变成 webp；关键是按 key 归属判定，而不是看整体有没有 webp
|
||||
gif_res = [f for f in fetched if f[0] == "anim"]
|
||||
if not gif_res:
|
||||
step("B2 动画 GIF 转 WebP", False, "未取到 anim 样本结果")
|
||||
elif gif_res[0][1] == "webp":
|
||||
step("B2 动画 GIF 已转为 WebP", True,
|
||||
"%dx%d, %d bytes（原 GIF %d bytes，-%.0f%%）" % (
|
||||
gif_res[0][2][0], gif_res[0][2][1], gif_res[0][3], len(imgs["anim"]),
|
||||
(1 - gif_res[0][3] / len(imgs["anim"])) * 100))
|
||||
elif gif_res[0][1] == "gif":
|
||||
step("B2 动画 GIF 已转为 WebP", False,
|
||||
"仍是原始 GIF %d bytes -> cwebp 不可用或转换失败" % gif_res[0][3])
|
||||
else:
|
||||
step("B2 动画 GIF 转 WebP", False, "得到 %s（预期 webp）" % gif_res[0][1])
|
||||
|
||||
# A：线框图应回退原图（PNG），而非转成更大的 WebP
|
||||
wire = [f for f in fetched if f[0] == "wire"]
|
||||
if not wire:
|
||||
step("A 线框图兜底回退", False, "未取到 wire 样本结果")
|
||||
elif wire[0][1] == "png":
|
||||
step("A 转换无收益时回退原图（线框图保持 PNG）", True,
|
||||
"%d bytes（原 PNG %d bytes）" % (wire[0][3], len(imgs["wire"])))
|
||||
else:
|
||||
step("A 线框图兜底回退", False,
|
||||
"被转成了 %s（%d bytes）——本应回退原图" % (wire[0][1], wire[0][3]))
|
||||
|
||||
print("\n== 结果：%d/%d 通过 ==" % (sum(1 for _, ok, _ in RESULTS if ok), len(RESULTS)))
|
||||
print("清理：TamaBox ban --email %s --yes" % email)
|
||||
return 0 if all(ok for _, ok, _ in RESULTS) else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -7,7 +7,6 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"io"
|
||||
"mime/multipart"
|
||||
)
|
||||
|
||||
@@ -35,7 +34,7 @@ func UploadBackgroundToS3(file multipart.File, fileHeader *multipart.FileHeader)
|
||||
return "", nil
|
||||
}
|
||||
|
||||
func ConvertStaticImageToWebP(r io.Reader, spec EncodeSpec) ([]byte, error) { return nil, nil }
|
||||
func ConvertStaticImageToWebP(data []byte, spec EncodeSpec) ([]byte, error) { return nil, nil }
|
||||
|
||||
func ConvertGifToAnimatedWebP(gifData []byte) ([]byte, error) { return nil, nil }
|
||||
|
||||
|
||||
Reference in New Issue
Block a user