houhuan
129 lines
4.0 KiB
Python
129 lines
4.0 KiB
Python
"""
|
|
测试脚本：验证企业微信推送 + 小爱音箱播报
|
|
- 拉取最新订单，推送到企业微信和小爱音箱
|
|
- 显示 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()
|