100 lines
4.2 KiB
Python
100 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
"""自动生成朋友圈素材，经 Frida/Hook 发布，并保存发布后截图。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import json
|
|
from pathlib import Path
|
|
import time
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.request import Request, urlopen
|
|
|
|
|
|
def request_json(method: str, url: str, payload: dict | None = None, timeout: int = 180) -> dict:
|
|
body = json.dumps(payload, ensure_ascii=False).encode() if payload is not None else None
|
|
req = Request(url, data=body, method=method, headers={"Content-Type": "application/json"})
|
|
try:
|
|
with urlopen(req, timeout=timeout) as response:
|
|
return json.loads(response.read().decode())
|
|
except HTTPError as exc:
|
|
raise RuntimeError(f"HTTP {exc.code}: {exc.read().decode(errors='replace')}") from exc
|
|
except URLError as exc:
|
|
raise RuntimeError(f"接口连接异常: {exc}") from exc
|
|
|
|
|
|
def device_ids(payload: dict) -> list[str]:
|
|
data = payload.get("data", payload)
|
|
rows = data.get("devices", data if isinstance(data, list) else []) if isinstance(data, (dict, list)) else []
|
|
return [str(row.get("device_id") or row.get("id") or row.get("serial")) for row in rows if isinstance(row, dict)]
|
|
|
|
|
|
def wait_device(api: str, selected: str | None, seconds: int) -> str:
|
|
deadline = time.time() + seconds
|
|
while True:
|
|
rows = device_ids(request_json("GET", f"{api}/api/v3/devices"))
|
|
if selected and selected in rows:
|
|
return selected
|
|
if not selected and rows:
|
|
return rows[0]
|
|
if time.time() >= deadline:
|
|
raise RuntimeError(f"等待设备上线超时，当前设备={rows}")
|
|
print(f"等待设备上线，剩余 {int(deadline - time.time())} 秒…", flush=True)
|
|
time.sleep(3)
|
|
|
|
|
|
def save_screenshot(payload: dict, output: Path) -> None:
|
|
data = payload.get("data", payload)
|
|
encoded = data.get("image") or data.get("base64") or data.get("screenshot")
|
|
if isinstance(encoded, str):
|
|
if encoded.startswith("data:"):
|
|
encoded = encoded.split(",", 1)[1]
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
output.write_bytes(base64.b64decode(encoded))
|
|
return
|
|
path = data.get("path") or data.get("file_path")
|
|
if path and Path(path).exists():
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
output.write_bytes(Path(path).read_bytes())
|
|
return
|
|
raise RuntimeError(f"截图响应缺少图像数据: {payload}")
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--api", default="http://127.0.0.1:8899")
|
|
parser.add_argument("--public-base-url", default="http://192.168.0.2:8899")
|
|
parser.add_argument("--device")
|
|
parser.add_argument("--topic", default="工作手机自动化验收")
|
|
parser.add_argument("--content")
|
|
parser.add_argument("--media-type", choices=["text", "image", "video"], default="image")
|
|
parser.add_argument("--duration", type=int, default=6)
|
|
parser.add_argument("--wait", type=int, default=600)
|
|
parser.add_argument("--screenshot-delay", type=int, default=8)
|
|
parser.add_argument("--output", default="开发文档/8、部署/05-测试验收/20260722_朋友圈自动生成发布/朋友圈发布结果.png")
|
|
args = parser.parse_args()
|
|
api = args.api.rstrip("/")
|
|
device = wait_device(api, args.device, args.wait)
|
|
payload = {
|
|
"device_id": device, "platform": "wechat", "topic": args.topic,
|
|
"content": args.content, "media_type": args.media_type, "duration": args.duration,
|
|
"public_base_url": args.public_base_url, "hook_only": True,
|
|
}
|
|
result = request_json("POST", f"{api}/api/v3/moments/auto", payload)
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
publish = result.get("publish", {})
|
|
channel = publish.get("channel_used", "")
|
|
if not result.get("success") or "hook" not in channel.lower():
|
|
raise RuntimeError(f"朋友圈Hook发布未通过: {result}")
|
|
time.sleep(max(1, args.screenshot_delay))
|
|
shot = request_json("POST", f"{api}/api/v3/devices/{device}/screenshot", {})
|
|
output = Path(args.output).resolve()
|
|
save_screenshot(shot, output)
|
|
print(f"截图已保存: {output}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|