autotrain/scripts/calibrate.py
xinxin6623 7a01ed4db9 feat: add invoice batch wrap-up (name extraction + roster annotation) and web ui automation improvements
- add scripts/annotate_roster_invoices.py: match passenger names to invoice PDFs, write status + relative encoded hyperlinks into roster xlsx
- add scripts/add_invoice_passenger_name.py, scripts/calibrate.py
- add .opencode/skills/invoice-wrapup skill
- improve invoice_tool.py / web_ui.py / vision_fallback.py automation
- add device/screen/windows adaptation docs
- ignore out/ (delivered invoice PDFs contain sensitive data)
2026-07-10 00:11:45 +08:00

266 lines
9.9 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
"""设备校准工具:新手机屏幕适配。
流程:
1. 自动检测屏幕分辨率adb shell wm size
2. 逐个状态截屏用户手动导航到对应页面HTML 上点截图)
3. 对每张截图跑视觉分析(页面类型 + 按钮坐标)
4. 保存设备配置work/device_profile.json
5. invoice_tool.py 读取配置,按比例计算所有坐标
"""
from __future__ import annotations
import json
import subprocess
import time
from datetime import datetime
from pathlib import Path
PROJECT_ROOT = Path(__file__).resolve().parent.parent
CALIBRATION_DIR = PROJECT_ROOT / "work" / "calibration"
DEVICE_PROFILE_PATH = PROJECT_ROOT / "work" / "device_profile.json"
ADB = "/opt/homebrew/share/android-commandlinetools/platform-tools/adb"
VISION_SCRIPT = PROJECT_ROOT / "scripts" / "vision_fallback.py"
RATIOS = {
"swipe_x": 0.5,
"swipe_start_y": 0.857,
"swipe_end_y": 0.214,
"wake_tap_x": 0.5,
"wake_tap_y": 0.5,
}
SWIPE_SPEED = 2.25
CALIBRATION_STEPS = [
{
"id": "list_page",
"name": "H5 列表页",
"desc": "导航到扫码开票单列表页(显示乘客列表 + 「开具」按钮)",
"buttons": ["开具"],
"expect_page": "list",
},
{
"id": "verify_page",
"name": "身份核验页",
"desc": "点击某乘客的「开具」后进入身份核验页输入身份证后8位",
"buttons": ["核验"],
"expect_page": "verify",
},
{
"id": "invoice_info",
"name": "发票信息页",
"desc": "核验通过后,进入发票信息页(显示抬头/税号/提交按钮)",
"buttons": ["提交"],
"expect_page": "invoice_info",
},
{
"id": "invoice_confirm",
"name": "发票确认页",
"desc": "提交后进入发票确认页(显示确认按钮)",
"buttons": ["确认"],
"expect_page": "invoice_confirm",
},
{
"id": "success_page",
"name": "开票成功页",
"desc": "确认后进入成功页(显示发票预览/下载 + 继续开票)",
"buttons": ["发票预览/下载", "下载", "继续开票"],
"expect_page": "success",
},
{
"id": "no_ticket",
"name": "无票页",
"desc": "无可开发票客票的页面(发票管理页显示'您没有可开具电子发票的客票'",
"buttons": [],
"expect_page": "no_ticket",
},
{
"id": "loading",
"name": "开具中",
"desc": "发票正在开具中的 loading 页面",
"buttons": [],
"expect_page": "loading",
},
{
"id": "album_qr",
"name": "相册选码",
"desc": "扫码时进入相册选择界面显示QR码缩略图",
"buttons": [],
"expect_page": "other",
"special": "qr_thumbnail",
},
]
def get_screen_size() -> tuple[int, int]:
result = subprocess.run([ADB, "shell", "wm", "size"], capture_output=True, text=True, check=False)
for line in result.stdout.splitlines():
if "Physical size:" in line:
parts = line.split("Physical size:")[1].strip().split("x")
return int(parts[0]), int(parts[1])
return 1260, 2800
def capture_screenshot(step_id: str) -> dict:
CALIBRATION_DIR.mkdir(parents=True, exist_ok=True)
path = CALIBRATION_DIR / f"{step_id}.png"
result = subprocess.run([ADB, "exec-out", "screencap", "-p"], capture_output=True, check=False)
if result.returncode != 0:
return {"error": f"adb screencap 失败: {result.stderr.decode('utf-8', 'ignore')}"}
path.write_bytes(result.stdout)
return {"ok": True, "path": str(path.relative_to(PROJECT_ROOT)), "size": len(result.stdout)}
def _call_vision(image_path: Path, prompt: str) -> dict:
result = subprocess.run(
["python3", str(VISION_SCRIPT), "ping", "--image", str(image_path), "--prompt", prompt],
capture_output=True, text=True, check=False, cwd=str(PROJECT_ROOT),
)
if result.returncode != 0:
return {"error": result.stderr.strip()[:500]}
try:
return json.loads(result.stdout)
except json.JSONDecodeError:
return {"error": f"非JSON: {result.stdout[:300]}"}
def _screenshot_size(path: Path) -> tuple[int, int]:
data = path.read_bytes()
if len(data) >= 24 and data[:8] == b"\x89PNG\r\n\x1a\n":
return int.from_bytes(data[16:20], "big"), int.from_bytes(data[20:24], "big")
return 1260, 2800
def analyze_step(step_id: str) -> dict:
screenshot_path = CALIBRATION_DIR / f"{step_id}.png"
if not screenshot_path.exists():
return {"step_id": step_id, "error": "截图不存在"}
width, height = _screenshot_size(screenshot_path)
step = next((s for s in CALIBRATION_STEPS if s["id"] == step_id), None)
if not step:
return {"step_id": step_id, "error": "未知步骤"}
result = {
"step_id": step_id,
"name": step["name"],
"screenshot": str(screenshot_path.relative_to(PROJECT_ROOT)),
"screen_size": [width, height],
"buttons": {},
}
page_prompt = (
"判断这个手机截图属于哪种页面类型只返回JSON"
'{"page": "list|verify|invoice_info|invoice_confirm|success|no_ticket|loading|other", "summary": "一句话描述"}'
)
page_out = _call_vision(screenshot_path, page_prompt)
page_content = page_out.get("content", "")
try:
page_parsed = json.loads(page_content)
result["page_detected"] = page_parsed.get("page", "other")
result["page_summary"] = page_parsed.get("summary", "")
except Exception:
result["page_detected"] = "other"
result["page_summary"] = page_content[:200]
result["page_ok"] = result["page_detected"] == step["expect_page"]
for btn_text in step.get("buttons", []):
btn_prompt = (
f"找到屏幕上文字为「{btn_text}」的可点击按钮。"
f"用0-1000归一化坐标返回按钮中心点。"
f'只返回JSON: {{"found": true/false, "x": 0-1000, "y": 0-1000}}'
)
btn_out = _call_vision(screenshot_path, btn_prompt)
btn_content = btn_out.get("content", "")
usage = btn_out.get("usage", {})
elapsed = btn_out.get("elapsed_seconds", "?")
try:
btn_parsed = json.loads(btn_content)
if btn_parsed.get("found") and btn_parsed.get("x") is not None:
px = int(btn_parsed["x"] * width / 1000)
py = int(btn_parsed["y"] * height / 1000)
result["buttons"][btn_text] = {"found": True, "xy": [px, py], "usage": usage, "elapsed": elapsed}
else:
result["buttons"][btn_text] = {"found": False, "usage": usage, "elapsed": elapsed}
except Exception:
result["buttons"][btn_text] = {"found": False, "error": btn_content[:200]}
if step.get("special") == "qr_thumbnail":
qr_prompt = (
"这是手机相册选择界面。找到黑白QR二维码缩略图的中心。"
'只返回JSON: {"found": true/false, "x": 0-1000, "y": 0-1000}'
)
qr_out = _call_vision(screenshot_path, qr_prompt)
try:
qr_parsed = json.loads(qr_out.get("content", ""))
if qr_parsed.get("found") and qr_parsed.get("x") is not None:
px = int(qr_parsed["x"] * width / 1000)
py = int(qr_parsed["y"] * height / 1000)
result["qr_thumbnail"] = {"found": True, "xy": [px, py]}
else:
result["qr_thumbnail"] = {"found": False}
except Exception:
result["qr_thumbnail"] = {"found": False}
buttons_ok = all(b.get("found") for b in result["buttons"].values()) if result["buttons"] else True
result["verified"] = result["page_ok"] and buttons_ok
if step.get("special") == "qr_thumbnail":
result["verified"] = result["verified"] and result.get("qr_thumbnail", {}).get("found", False)
return result
def analyze_all() -> dict:
results = {}
for step in CALIBRATION_STEPS:
sid = step["id"]
path = CALIBRATION_DIR / f"{sid}.png"
if path.exists():
results[sid] = analyze_step(sid)
else:
results[sid] = {"step_id": sid, "name": step["name"], "skipped": True}
return results
def load_device_profile() -> dict:
if DEVICE_PROFILE_PATH.exists():
return json.loads(DEVICE_PROFILE_PATH.read_text(encoding="utf-8"))
return {}
def save_device_profile(calibration: dict | None = None) -> dict:
screen_w, screen_h = get_screen_size()
amplitude = int(screen_h * (RATIOS["swipe_start_y"] - RATIOS["swipe_end_y"]))
duration = max(300, int(amplitude / SWIPE_SPEED))
profile = {
"screen": {"width": screen_w, "height": screen_h},
"ratios": RATIOS,
"computed": {
"swipe_x": int(screen_w * RATIOS["swipe_x"]),
"swipe_start_y": int(screen_h * RATIOS["swipe_start_y"]),
"swipe_end_y": int(screen_h * RATIOS["swipe_end_y"]),
"swipe_amplitude": amplitude,
"swipe_duration_ms": duration,
"wake_tap": [int(screen_w * RATIOS["wake_tap_x"]), int(screen_h * RATIOS["wake_tap_y"])],
},
"calibration": calibration or {},
"created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
}
DEVICE_PROFILE_PATH.parent.mkdir(parents=True, exist_ok=True)
DEVICE_PROFILE_PATH.write_text(json.dumps(profile, ensure_ascii=False, indent=2), encoding="utf-8")
return profile
def get_steps() -> list[dict]:
return [{"id": s["id"], "name": s["name"], "desc": s["desc"],
"buttons": s.get("buttons", []), "expect_page": s.get("expect_page", "")}
for s in CALIBRATION_STEPS]
def get_capture_status() -> dict:
status = {}
for step in CALIBRATION_STEPS:
sid = step["id"]
path = CALIBRATION_DIR / f"{sid}.png"
status[sid] = {"captured": path.exists(), "name": step["name"]}
return status