Initial commit: Fengxiang order monitor with WeChat & Xiaomi speaker push

2026-04-30 14:25:34 +08:00
commit 1bb7bba970
10 changed files with 970 additions and 0 deletions
@@ -0,0 +1,128 @@
+"""
+测试脚本：验证企业微信推送 + 小爱音箱播报
+- 拉取最新订单，推送到企业微信和小爱音箱
+- 显示 API 返回结果，方便排查问题
+"""
+
+import json
+import time
+import requests
+import asyncio
+from pathlib import Path
+from aiohttp import ClientSession
+from miservice import MiAccount, MiNAService
+
+WEBHOOK_URL = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=644ab6d9-3b66-4166-88e9-5a8a89e3731d"
+DATA_API = "https://fs.szfx.com/saasmerchant/pcweb/order/quickpayorder/list"
+SHOP_ID = "20434543575189"
+
+XIAOMI_USER_ID = "1136458602"
+XIAOMI_TOKEN_PATH = str(Path.home() / ".mi.token")
+XIAOMI_SPEAKER_DID = "3ba2c1e8-d8cb-45c5-b88a-15624e7a02f3"
+
+
+def fmt_money(cents):
+    return f"{cents / 100:.2f}元" if cents else "0.00元"
+
+
+def fmt_ts(ts):
+    import datetime
+    return datetime.datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S") if ts else "-"
+
+
+def load_cookies():
+    with open("cookies.json", "r", encoding="utf-8") as f:
+        return json.load(f)
+
+
+def fetch_latest_order():
+    """使用 cookies.json 拉取最新订单"""
+    cookies = load_cookies()
+    headers = {
+        "Content-Type": "application/json",
+        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
+        "Referer": "https://fs.szfx.com/MMS",
+        "Origin": "https://fs.szfx.com",
+    }
+    payload = {"shopId": SHOP_ID, "page": 1, "pageSize": 1}
+    resp = requests.post(DATA_API, json=payload, headers=headers, cookies=cookies, timeout=30)
+    data = resp.json()
+    if data.get("errno") != 0:
+        print(f"[FAIL] API error: errno={data.get('errno')}, errmsg={data.get('errmsg')}")
+        return None
+    return data["data"]["list"][0]
+
+
+def test_wecom(msg):
+    """发送到企业微信"""
+    print("\n--- Test: WeChat Work ---")
+    payload = {"msgtype": "text", "text": {"content": msg}}
+    try:
+        resp = requests.post(WEBHOOK_URL, json=payload, timeout=10)
+        result = resp.json()
+        print(f"  Response: {json.dumps(result, ensure_ascii=False)}")
+        if result.get("errcode") == 0:
+            print("  [OK] WeChat push success")
+        else:
+            print("  [FAIL] WeChat push failed")
+    except Exception as e:
+        print(f"  [FAIL] Exception: {e}")
+
+
+async def _tts(text):
+    async with ClientSession() as session:
+        account = MiAccount(session, XIAOMI_USER_ID, None, XIAOMI_TOKEN_PATH)
+        mina = MiNAService(account)
+        return await mina.text_to_speech(XIAOMI_SPEAKER_DID, text)
+
+
+def test_speaker(msg):
+    """发送到小爱音箱"""
+    print("\n--- Test: Xiaomi Speaker ---")
+    print(f"  Text: {msg}")
+    try:
+        result = asyncio.run(_tts(msg))
+        print(f"  Response: {json.dumps(result, ensure_ascii=False)}")
+        if result and result.get("code") == 0:
+            print("  [OK] Speaker TTS success")
+        else:
+            print("  [FAIL] Speaker TTS failed - token may be expired, run setup_xiaomi.py")
+    except Exception as e:
+        print(f"  [FAIL] Exception: {e}")
+
+
+def main():
+    print("=" * 50)
+    print("  Fengxiang Order Monitor - Test Script")
+    print("=" * 50)
+
+    # 1. Fetch latest order
+    print("\n1. Fetch latest order...")
+    order = fetch_latest_order()
+    if not order:
+        print("[FAIL] Cannot fetch order, check cookies.json")
+        return
+
+    print(f"   orderId: {order['orderId']}")
+    print(f"   shopName: {order['shopName']}")
+    print(f"   allPrice: {fmt_money(order['allPrice'])}")
+    print(f"   amountPayable: {fmt_money(order['amountPayable'])}")
+    print(f"   payTime: {fmt_ts(order['payTime'])}")
+
+    # 2. Format message
+    msg = f"【丰享丰食】订单收款成功，收款{fmt_money(order['amountPayable'])}"
+    print(f"\n2. Message to send:\n   {msg}")
+
+    # 3. Test WeChat and Speaker
+    test_wecom(msg)
+    test_speaker(msg)
+
+    print("\n" + "=" * 50)
+    print("  Test complete. Check:")
+    print("  - WeChat: open the enterprise WeChat group")
+    print("  - Speaker: the L15A speaker in the shop")
+    print("=" * 50)
+
+
+if __name__ == "__main__":
+    main()