miniapp_qr_portable_bundle/miniapp_qr_poc.py

import argparse
import base64
import json
import os
import re
import subprocess
import urllib.request
from datetime import datetime
from pathlib import Path
from zoneinfo import ZoneInfo

API_URL = "https://jsh.szsentry.com/api//mine/one_see_secret_word"
BASE_DIR = Path(__file__).resolve().parent
HELPER_JS = BASE_DIR / "js" / "21D38FE5F73FD4DF47B5E7E2FB120D83.js"
QR_JS = BASE_DIR / "js" / "0C723952F73FD4DF6A145155C4220D83.js"
FOUND_IDS_FILE = BASE_DIR / "found_ids.txt"
FOUND_ID_PATTERN = re.compile(r"ID:\s*(\d+)\s*\|\s*Village:\s*(.+)")
DEFAULT_TIMEZONE = ZoneInfo(os.getenv("MINIAPP_QR_TIMEZONE", "Asia/Shanghai"))


def fetch(member_id: int):
    body = json.dumps({"member_id": member_id}).encode()
    req = urllib.request.Request(
        API_URL,
        data=body,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=20) as resp:
        return json.loads(resp.read().decode("utf-8"))


def get_command(qr_hex: str, dt: datetime | None = None) -> bytes:
    if dt is None:
        dt = datetime.now(DEFAULT_TIMEZONE)
    elif dt.tzinfo is None:
        dt = dt.replace(tzinfo=DEFAULT_TIMEZONE)
    else:
        dt = dt.astimezone(DEFAULT_TIMEZONE)
    c = [
        f"{dt.year % 100:02d}",
        f"{dt.month:02d}",
        f"{dt.day:02d}",
        f"{dt.hour:02d}",
        f"{dt.minute:02d}",
        f"{dt.second:02d}",
    ]
    m = [int(x, 16) for x in c]
    s = m[0] ^ m[2]
    s = (((s + m[4]) ^ m[1]) + m[3]) ^ m[5]
    s &= 0xFF

    pairs = re.findall(r"[0-9A-Fa-f]{2}", qr_hex)
    g = [f"{(int(x, 16) ^ s):02X}" for x in pairs]
    f = ("".join(c) + "".join(g)).upper()
    d = [ord(ch) for ch in f]
    h = [0x03] + d + [0x04]
    chk = 0
    for x in h:
        chk ^= x
    chk_hex = f"{chk:02X}"
    h.extend(ord(ch) for ch in chk_hex)
    h.append(0x0D)
    return bytes(h)


def render_with_bundled_js(qr_hex: str) -> str:
    js = f'''
const helper = require({json.dumps(str(HELPER_JS))});
const qr = require({json.dumps(str(QR_JS))});
const out = qr.QR(helper.getCommand({json.dumps(qr_hex)}).toString());
console.log(out);
'''
    cp = subprocess.run(["node", "-e", js], capture_output=True, text=True)
    if cp.returncode != 0:
        raise RuntimeError(f"node failed\nstdout:\n{cp.stdout}\nstderr:\n{cp.stderr}")
    return cp.stdout.strip()


def save_data_url(data_url: str, out_path: str):
    m = re.match(r"data:.*?;base64,(.*)", data_url)
    if not m:
        raise ValueError("unexpected data URL")
    raw = base64.b64decode(m.group(1))
    with open(out_path, "wb") as f:
        f.write(raw)


def load_found_ids(path: str | Path = FOUND_IDS_FILE) -> list[dict]:
    path = Path(path)
    items = []
    seen = set()
    if not path.exists():
        return items

    with path.open("r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            m = FOUND_ID_PATTERN.match(line)
            if not m:
                continue
            member_id = int(m.group(1))
            if member_id in seen:
                continue
            seen.add(member_id)
            village_name = m.group(2).strip()
            items.append(
                {
                    "member_id": member_id,
                    "village_name": village_name,
                    "label": f"{member_id} | {village_name}",
                }
            )
    return items


def summarize_response(resp: dict) -> dict:
    data = resp.get("data") or {}
    return {
        "status": resp.get("status"),
        "msg": resp.get("msg"),
        "member_id": data.get("member_id"),
        "village_name": data.get("village_name"),
        "uuid": data.get("uuid"),
        "qr_prefix": str(data.get("qr", ""))[:40],
    }


def generate_member_qr(member_id: int, include_data_url: bool = False) -> dict:
    resp = fetch(member_id)
    data = resp.get("data") or {}
    qr_hex = data.get("qr")
    if not qr_hex:
        raise ValueError(f"member_id={member_id} did not return qr data")

    payload = get_command(qr_hex)
    result = summarize_response(resp)
    result.update(
        {
            "qr_hex": qr_hex,
            "payload_len": len(payload),
            "payload_hex_prefix": payload.hex()[:160],
        }
    )
    if include_data_url:
        result["data_url"] = render_with_bundled_js(qr_hex)
    return result


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--member-id", type=int)
    ap.add_argument("--qr")
    ap.add_argument("--json-out")
    ap.add_argument("--gif-out")
    args = ap.parse_args()

    resp = None
    qr_hex = args.qr
    if args.member_id is not None:
        resp = fetch(args.member_id)
        if args.json_out:
            with open(args.json_out, "w", encoding="utf-8") as f:
                json.dump(resp, f, ensure_ascii=False, indent=2)
        qr_hex = resp["data"]["qr"]

    if not qr_hex:
        raise SystemExit("provide --member-id or --qr")

    payload = get_command(qr_hex)
    print("payload_len=", len(payload))
    print("payload_hex_prefix=", payload.hex()[:160])

    if args.gif_out:
        data_url = render_with_bundled_js(qr_hex)
        save_data_url(data_url, args.gif_out)
        print("saved", args.gif_out)

    if resp:
        print(json.dumps(summarize_response(resp), ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()