聚水潭库存http请求

分享一下聚水潭库存http请求的大致思路,结尾会带上源代码

手动触发一下库存查询,然后在网络这里找到是这个请求获取的数据,那么就需要看一下需要哪些参数
载荷里面这两个是变量,所以需要先找到这两个参数

这两个查询的思路一样,我以__VIEWSTATE为例,从元素界面搜索,发现有多个__VIEWSTATE,

不过库存这个两个参数是一样的,倒是不要紧,但是有些的两个参数不一样,所以建议打开网页的时候直接打开iframe里面的页面

比如库存我打开的是这个https://www.erp321.com/app/item/SkuStock/WmsSkuStock.aspx?_c=jst-epaas&epaas=true
这个时候就剩一个__VIEWSTATE了,而这个页面如果和主页面的不一样,要用这个,有些页面会不一样,需要看清楚一下,最好是直接打开子页面

然后找到两个参数和cookie,准备工作就完成了

后面就可以直接请求了,

得到这个数据自己清洗一下就行了,我没做这部分,要清洗的话问ai写一段就行了
以下是rpa流程和python代码,python代码是ai写的,看不懂直接问ai,我也看不懂

# 使用提醒:
# 1. xbot包提供软件自动化、数据表格、Excel、日志、AI等功能
# 2. package包提供访问当前应用数据的功能，如获取元素、访问全局变量、获取资源文件等功能
# 3. 当此模块作为流程独立运行时执行main函数
# 4. 可视化流程中可以通过"调用模块"的指令使用此模块

import xbot
from xbot import print, sleep
from .import package
from .package import variables as glv

import requests
import time
import json

def query_stock(sku_id, viewstate, viewstategen, cookies_list, max_retries=3):
    """
    使用外部传入的 Cookie 列表查询库存

    :param sku_id: 要查询的 SKU ID
    :param viewstate: __VIEWSTATE 的值
    :param viewstategen: __VIEWSTATEGENERATOR 的值
    :param cookies_list: 从浏览器复制的 Cookie 列表，每个元素为字典，包含 'name' 和 'value'
    :param max_retries: 遇到空响应时的最大重试次数
    :return: 解析后的数据字典，如果失败返回 None
    """
    # 1. 创建会话并设置固定的请求头（模仿浏览器）
    session = requests.Session()
    session.headers.update({
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/145.0.0.0 Safari/537.36",
        "Accept": "*/*",
        "Accept-Language": "zh-CN,zh;q=0.9",
        "Origin": "https://www.erp321.com",
        "Referer": "https://www.erp321.com/app/item/SkuStock/WmsSkuStock.aspx?_c=jst-epaas&epaas=true",
        "X-Requested-With": "XMLHttpRequest",
        "sec-ch-ua": '"Not:A-Brand";v="99", "Google Chrome";v="145", "Chromium";v="145"',
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"Windows"',
        "sec-fetch-dest": "empty",
        "sec-fetch-mode": "cors",
        "sec-fetch-site": "same-origin",
    })

    # 2. 将传入的 Cookie 列表添加到会话中（只取 name 和 value）
    cookie_dict = {item['name']: item['value'] for item in cookies_list}
    session.cookies.update(cookie_dict)

    # 3. 构造过滤器数组（用于 __CALLBACKPARAM）
    filter_array = [{"k": "sku_id", "v": sku_id, "c": "like"}]
    filter_str = json.dumps(filter_array, ensure_ascii=False)  # '[{"k":"sku_id","v":"AS25K9086","c":"like"}]'

    # 4. 构造完整的 __CALLBACKPARAM 参数
    callback_param = {
        "Method": "LoadDataToJSON",
        "Args": ["1", filter_str, "{}"]
    }
    callback_param_str = json.dumps(callback_param, ensure_ascii=False)

    post_url = "https://www.erp321.com/app/item/SkuStock/WmsSkuStock.aspx?_c=jst-epaas&epaas=true&ts___={}&am___=LoadDataToJSON"

    for attempt in range(max_retries):
        ts = int(time.time() * 1000)  # 动态时间戳
        url = post_url.format(ts)

        data = {
            "__VIEWSTATE": viewstate,
            "__VIEWSTATEGENERATOR": viewstategen,
            "owner_co_id": "11717633",
            "authorize_co_id": "11717633",
            "i_id": "",
            "sku_id": sku_id,
            "name": "",
            "combine_sku_id": "",
            "properties_value": "",
            "qty": "",
            "qty": "",                      # 注意两个 qty 字段
            "wms_co_id_id": "",
            "wms_co_ids": "",
            "labels_search": "",
            "labels_exclude_search": "",
            "qtyminqty": "",
            "_jt_page_count_enabled": "",
            "_jt_page_size": "25",
            "_cbb_wms_co_ids": "",
            "__CALLBACKID": "JTable1",
            "__CALLBACKPARAM": callback_param_str
        }

        try:
            resp = session.post(url, data=data, timeout=10)
            if resp.status_code != 200:
                print(f"请求失败，状态码：{resp.status_code}")
                if attempt < max_retries - 1:
                    time.sleep(2 ** attempt)
                    continue
                else:
                    return None

            text = resp.text
            if text.startswith('0|'):
                # 成功返回数据，去掉前缀并解析 JSON
                json_str = text[2:]
                try:
                    result = json.loads(json_str)
                    return result
                except json.JSONDecodeError:
                    print("响应 JSON 解析失败")
                    return None
            elif text == '':
                print(f"第{attempt+1}次尝试返回空响应，可能是 Cookie 或参数失效")
                if attempt < max_retries - 1:
                    wait = 2 ** attempt
                    print(f"等待 {wait} 秒后重试...")
                    time.sleep(wait)
                else:
                    print("已达到最大重试次数，放弃。请检查 Cookie 是否有效，或重新复制。")
                    return None
            else:
                print(f"未知响应格式：{text[:200]}")
                return None

        except requests.RequestException as e:
            print(f"请求异常：{e}")
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)
            else:
                return None

    return None