9eb6778a6f
- web_ui.py: major refactor with improved runner, mode selection, progress tracking - scripts: add device_runtime, device_calibrate, calibration_trace, record_android_flow - docs: add phone-control-manual, update device-adaptation and windows-migration - invoice_tool: enhance H5 automation flow and visual fallback - calibrate.py: expand calibration capabilities - add skills/ and tests/ directories - remove obsolete .opencode/skills/invoice-wrapup Co-Authored-By: Claude <noreply@anthropic.com>
366 lines
14 KiB
Python
366 lines
14 KiB
Python
#!/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 shutil
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
import device_runtime
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||
CALIBRATION_DIR = PROJECT_ROOT / "work" / "calibration"
|
||
SAMPLE_META_PATH = CALIBRATION_DIR / "trusted_samples.json"
|
||
# 旧单设备配置只用于首次迁移；新配置统一按 ADB serial 写入 device_profiles.json。
|
||
DEVICE_PROFILE_PATH = PROJECT_ROOT / "work" / "device_profile.json"
|
||
VISION_SCRIPT = PROJECT_ROOT / "scripts" / "vision_fallback.py"
|
||
|
||
|
||
def _strip_codeblocks(text: str) -> str:
|
||
"""剥掉可能的 markdown 代码块包裹，只保留 JSON 内容"""
|
||
text = text.strip()
|
||
if text.startswith("```json"):
|
||
text = text[7:]
|
||
elif text.startswith("```"):
|
||
text = text[3:]
|
||
if text.endswith("```"):
|
||
text = text[:-3]
|
||
return text.strip()
|
||
|
||
|
||
def _parse_json_lenient(text: str) -> dict:
|
||
"""lenient JSON parsing, handles markdown codeblocks and trailing garbage"""
|
||
text = _strip_codeblocks(text)
|
||
# Find first { and last }
|
||
start = text.find("{")
|
||
end = text.rfind("}")
|
||
if start != -1 and end != -1 and end > start:
|
||
text = text[start:end+1]
|
||
return json.loads(text)
|
||
|
||
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": "album_qr",
|
||
"name": "相册选码",
|
||
"desc": "扫码时进入相册选择界面（显示QR码缩略图）",
|
||
"buttons": [],
|
||
"expect_page": "other",
|
||
"special": "qr_thumbnail",
|
||
},
|
||
{
|
||
"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": "loading",
|
||
"name": "开具中",
|
||
"desc": "确认后，发票正在开具中的 loading 页面",
|
||
"buttons": [],
|
||
"expect_page": "loading",
|
||
},
|
||
{
|
||
"id": "success_page",
|
||
"name": "开票成功页",
|
||
"desc": "开具完成后进入成功页（显示发票预览/下载 + 继续开票）",
|
||
"buttons": ["发票预览/下载", "下载", "继续开票"],
|
||
"expect_page": "success",
|
||
},
|
||
{
|
||
"id": "no_ticket",
|
||
"name": "无票页",
|
||
"desc": "无可开发票客票的页面（发票管理页显示'您没有可开具电子发票的客票'）",
|
||
"buttons": [],
|
||
"expect_page": "no_ticket",
|
||
},
|
||
]
|
||
|
||
|
||
def get_screen_size() -> tuple[int, int]:
|
||
try:
|
||
size = str(device_runtime.device_fingerprint().get("physical_size") or "")
|
||
if "x" in size:
|
||
width, height = size.split("x", 1)
|
||
return int(width), int(height)
|
||
except (ValueError, device_runtime.DeviceError):
|
||
pass
|
||
return 0, 0
|
||
|
||
|
||
def capture_screenshot(step_id: str) -> dict:
|
||
CALIBRATION_DIR.mkdir(parents=True, exist_ok=True)
|
||
path = CALIBRATION_DIR / f"{step_id}.png"
|
||
result = device_runtime.run_adb_bytes(["exec-out", "screencap", "-p"], 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(
|
||
[sys.executable, 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 = (
|
||
"分析这个手机截图，先写一句话描述，然后判断属于哪种页面类型。"
|
||
"注意："
|
||
"- list：显示多个乘客的列表页，有「开具」按钮，"
|
||
"- verify：身份核验页，输入身份证后8位，有「核验」按钮，"
|
||
"- invoice_info：发票抬头税号页，有「提交」按钮，"
|
||
"- invoice_confirm：发票信息确认页，有「确认」按钮，"
|
||
"- success：开票成功页，有发票预览/下载，有「继续开票」或「下载」按钮，"
|
||
"- no_ticket：无可开票客票页，显示「您没有可开具电子发票的客票」，"
|
||
"- loading：开具中/加载中页面，"
|
||
"- other：以上都不是。"
|
||
'只返回JSON，不要markdown包裹：{"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 = _parse_json_lenient(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归一化坐标返回按钮中心点（x: 0=左, 1000=右; y:0=上,1000=下）。"
|
||
f'只返回JSON，不要markdown包裹：{{"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 = _parse_json_lenient(btn_content)
|
||
if btn_parsed.get("found") and btn_parsed.get("x") is not None:
|
||
px = int(float(btn_parsed["x"]) * width / 1000)
|
||
py = int(float(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二维码缩略图的中心。"
|
||
'用0-1000归一化坐标返回，不要markdown包裹：{"found": true/false, "x": 0-1000, "y": 0-1000}'
|
||
)
|
||
qr_out = _call_vision(screenshot_path, qr_prompt)
|
||
try:
|
||
qr_parsed = _parse_json_lenient(qr_out.get("content", ""))
|
||
if qr_parsed.get("found") and qr_parsed.get("x") is not None:
|
||
px = int(float(qr_parsed["x"]) * width / 1000)
|
||
py = int(float(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:
|
||
try:
|
||
device_runtime.migrate_legacy_profile()
|
||
return device_runtime.profile_status().get("profile") or {}
|
||
except device_runtime.DeviceError:
|
||
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))
|
||
calibration = calibration or {}
|
||
verified = bool(calibration) and all(
|
||
bool((calibration.get(step["id"]) or {}).get("verified"))
|
||
for step in CALIBRATION_STEPS
|
||
)
|
||
profile = {
|
||
"status": "verified" if verified else "unverified",
|
||
"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,
|
||
"created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
}
|
||
saved = device_runtime.upsert_profile(profile)
|
||
return {**profile, "device": saved.get("fingerprint", {})}
|
||
|
||
|
||
def get_device_status() -> dict:
|
||
"""给 Web UI 的设备与档案状态；异常转成可展示的 JSON。"""
|
||
try:
|
||
device_runtime.migrate_legacy_profile()
|
||
return device_runtime.profile_status()
|
||
except device_runtime.DeviceError as exc:
|
||
return {"status": "unavailable", "error": str(exc), "device": {}, "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
|
||
|
||
|
||
def _known_step_ids() -> set[str]:
|
||
return {str(step["id"]) for step in CALIBRATION_STEPS}
|
||
|
||
|
||
def _load_sample_meta() -> dict:
|
||
if not SAMPLE_META_PATH.exists():
|
||
return {}
|
||
try:
|
||
data = json.loads(SAMPLE_META_PATH.read_text(encoding="utf-8"))
|
||
return data if isinstance(data, dict) else {}
|
||
except (OSError, json.JSONDecodeError):
|
||
return {}
|
||
|
||
|
||
def trusted_sample_path(step_id: str) -> Path | None:
|
||
"""只返回人工确认过、且与步骤绑定的样图，旧占位图不会被展示。"""
|
||
if step_id not in _known_step_ids():
|
||
return None
|
||
record = _load_sample_meta().get(step_id) or {}
|
||
filename = str(record.get("filename") or "")
|
||
path = CALIBRATION_DIR / filename
|
||
if filename == f"{step_id}_sample.png" and path.exists():
|
||
return path
|
||
return None
|
||
|
||
|
||
def approve_current_as_sample(step_id: str) -> dict:
|
||
"""把当前已截图的页面设为该步骤唯一可信的参考样图。"""
|
||
if step_id not in _known_step_ids():
|
||
return {"ok": False, "error": "未知校准步骤"}
|
||
source = CALIBRATION_DIR / f"{step_id}.png"
|
||
if not source.exists():
|
||
return {"ok": False, "error": "请先截取当前步骤的正确页面"}
|
||
CALIBRATION_DIR.mkdir(parents=True, exist_ok=True)
|
||
target = CALIBRATION_DIR / f"{step_id}_sample.png"
|
||
shutil.copy2(source, target)
|
||
metadata = _load_sample_meta()
|
||
metadata[step_id] = {
|
||
"filename": target.name,
|
||
"approved_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
}
|
||
SAMPLE_META_PATH.write_text(json.dumps(metadata, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
return {"ok": True, "path": str(target.relative_to(PROJECT_ROOT))}
|