(account_id)
| 105 | return code |
| 106 | |
| 107 | def login_with_code(account_id): |
| 108 | code = get_wx_code(account_id) |
| 109 | login_resp = session.get( |
| 110 | f"{LOGIN_BASE}/login/{code}", |
| 111 | headers=app_headers(), |
| 112 | timeout=30, |
| 113 | ) |
| 114 | login_resp.raise_for_status() |
| 115 | login_data = login_resp.json() |
| 116 | if str(login_data.get("code")) != "200" or not login_data.get("data"): |
| 117 | raise RuntimeError(f"code登录失败: {login_data}") |
| 118 | user_info = login_data["data"] |
| 119 | uid = user_info.get("uid") |
| 120 | open_id = user_info.get("openId") |
| 121 | union_id = user_info.get("unionId") |
| 122 | if not uid: |
| 123 | raise RuntimeError(f"登录响应缺少uid: {login_data}") |
| 124 | |
| 125 | token_resp = session.post( |
| 126 | f"{APP_BASE}/customer/loginByUid", |
| 127 | json={"uid": uid, "openId": open_id, "unionId": union_id}, |
| 128 | headers=app_headers(), |
| 129 | timeout=30, |
| 130 | ) |
| 131 | token_resp.raise_for_status() |
| 132 | token_data = token_resp.json() |
| 133 | if not is_success(token_data) or not (token_data.get("result") or {}).get("token"): |
| 134 | raise RuntimeError(f"业务token登录失败: {token_data}") |
| 135 | return { |
| 136 | "token": token_data["result"]["token"], |
| 137 | "uid": uid, |
| 138 | "openId": open_id, |
| 139 | "unionId": union_id, |
| 140 | "userInfo": user_info, |
| 141 | "customer": token_data.get("result", {}), |
| 142 | "updatedAt": int(time.time()), |
| 143 | } |
| 144 | |
| 145 | def get_cached_token(account_id): |
| 146 | return read_token_cache().get(account_id) |
no test coverage detected