更新 tp/scripts/beauty.py

This commit is contained in:
lj091715 2026-05-19 13:53:11 +08:00
parent 7c5935b590
commit 5967e4d459

View File

@ -30,7 +30,7 @@ def get_key() -> str:
class NoRedirectHandler(urllib.request.HTTPRedirectHandler): class NoRedirectHandler(urllib.request.HTTPRedirectHandler):
"""不跟随重定向,让 open() 直接返回 302 响应""" """不跟随重定向,用于捕获 302 的 Location 头"""
def http_error_302(self, req, fp, code, msg, headers): def http_error_302(self, req, fp, code, msg, headers):
return fp return fp
http_error_301 = http_error_302 http_error_301 = http_error_302
@ -41,8 +41,6 @@ class NoRedirectHandler(urllib.request.HTTPRedirectHandler):
def fetch_single_image(api: dict) -> str | None: def fetch_single_image(api: dict) -> str | None:
"""从单个接口获取图片地址通过302重定向的Location头""" """从单个接口获取图片地址通过302重定向的Location头"""
key = get_key() key = get_key()
# 方式1通过 NoRedirectHandler 捕获 302 的 Location
try: try:
opener = urllib.request.build_opener(NoRedirectHandler()) opener = urllib.request.build_opener(NoRedirectHandler())
req = urllib.request.Request(api["url"], headers={"key": key}) req = urllib.request.Request(api["url"], headers={"key": key})
@ -50,23 +48,13 @@ def fetch_single_image(api: dict) -> str | None:
location = resp.headers.get("Location") location = resp.headers.get("Location")
if location: if location:
return location.strip() return location.strip()
# 方式2如果没Location头可能被跟随了重定向尝试读响应body中的URL
body = resp.read().decode("utf-8", errors="replace")
sys.stdout.write(f"[debug] {api['name']}: 无Location, body={body[:100]}\n")
except urllib.error.HTTPError as e: except urllib.error.HTTPError as e:
# 某些Python版本 NoRedirectHandler 仍会抛异常
if e.code in (301, 302, 303, 307): if e.code in (301, 302, 303, 307):
location = e.headers.get("Location") location = e.headers.get("Location")
if location: if location:
return location.strip() return location.strip()
sys.stdout.write(f"[debug] {api['name']}: HTTP {e.code}\n") except (urllib.error.URLError, TimeoutError, OSError):
except urllib.error.URLError as e: pass
sys.stdout.write(f"[debug] {api['name']}: 网络错误 {e.reason}\n")
except TimeoutError:
sys.stdout.write(f"[debug] {api['name']}: 请求超时\n")
except OSError as e:
sys.stdout.write(f"[debug] {api['name']}: 系统错误 {e}\n")
return None return None
@ -84,7 +72,6 @@ def send_images(image_urls: list[str]) -> bool:
robot_port = os.environ.get("ROBOT_WECHAT_CLIENT_PORT", "").strip() robot_port = os.environ.get("ROBOT_WECHAT_CLIENT_PORT", "").strip()
to_wxid = os.environ.get("ROBOT_FROM_WX_ID", "").strip() to_wxid = os.environ.get("ROBOT_FROM_WX_ID", "").strip()
if not robot_port or not to_wxid: if not robot_port or not to_wxid:
sys.stdout.write(f"[debug] 环境变量缺失: PORT={robot_port!r} WXID={to_wxid!r}\n")
return False return False
api_url = ( api_url = (
@ -109,29 +96,15 @@ def send_images(image_urls: list[str]) -> bool:
if 200 <= response.status < 300: if 200 <= response.status < 300:
return True return True
payload = json.load(response) payload = json.load(response)
except urllib.error.URLError as e: except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, OSError):
sys.stdout.write(f"[debug] 发图失败: 无法连接机器人 {e.reason}\n")
return False
except TimeoutError:
sys.stdout.write("[debug] 发图失败: 超时\n")
return False
except json.JSONDecodeError:
sys.stdout.write("[debug] 发图失败: 返回不是JSON\n")
return False
except OSError:
return False return False
code = payload.get("code") code = payload.get("code")
ok = code == 200 or code == 0 return code == 200 or code == 0
if not ok:
sys.stdout.write(f"[debug] 发图失败: code={code}\n")
return ok
def main() -> int: def main() -> int:
urls = fetch_all_images() urls = fetch_all_images()
if urls:
sys.stdout.write(f"[debug] 成功获取 {len(urls)} 张图片\n")
if len(urls) >= 2 and send_images(urls): if len(urls) >= 2 and send_images(urls):
return 0 return 0
sys.stdout.write(FALLBACK_TEXT) sys.stdout.write(FALLBACK_TEXT)