scripting_tools/beidahuang/test.py

# coding=utf-8
import json
import random
import time
import hashlib
from webbrowser import get
# from plc_log import logger_pc
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util import Retry

root_url = "http://www.sukon-cloud.com"


def get_token():
    try:
        uid = "f49624126f654356a8e6b2669d3b8fbb"
        sid = "6b63484a788a4c2fbf245254dcfd7ec1"
        random_id = str(random.randint(100, 999)) + str("".join([chr(random.randint(97, 122)) for i in range(3)]))
        time_stamp = str(int(round(time.time()) * 1000))
        m = hashlib.md5()
        m.update(str(uid + sid + random_id + time_stamp).encode('utf-8'))
        signature = m.hexdigest().upper()
        url = root_url + "/api/v1/token/initToken"
        s = requests.Session()
        # 设置 post()方法进行重访问
        params = {"uid": uid, "sid": sid, "random": random_id, "timestamp": time_stamp, "signature": signature}
        # print("入参：", "\n", json.dumps(params, indent=4, ensure_ascii=False))
        s.mount('http://', HTTPAdapter(max_retries=Retry(total=5, allowed_methods=frozenset(['GET', 'POST']))))
        response = s.post(url=url, data=params, timeout=10)
        # print("出参：", "\n",  json.dumps(json.loads(response.text), indent=4, ensure_ascii=False))
        return json.loads(response.text).get("data")["token"]
    except Exception as e:
        print(e)
        return None


def get_projects():
    url = root_url + "/api/v1/base/projects"
    s = requests.Session()
    # 设置 post()方法进行重访问
    s.mount('http://', HTTPAdapter(max_retries=Retry(total=5, allowed_methods=frozenset(['GET', 'POST']))))
    params = {"token": get_token()}
    print("入参：", "\n", json.dumps(params, indent=4, ensure_ascii=False))
    response = s.post(url=url, data=params, timeout=10)
    try:
        print("出参：", "\n",  json.dumps(json.loads(response.text), indent=4, ensure_ascii=False))
        return json.loads(response.text).get("data")
    except Exception as e:
        return []


def get_box_id(project):
    try:
        url = root_url + "/api/v1/base/projectBoxes"
        s = requests.Session()
        # 设置 post()方法进行重访问
        s.mount('http://', HTTPAdapter(max_retries=Retry(total=5, allowed_methods=frozenset(['GET', 'POST']))))
        params = {"token": get_token(), "projectId": project}
        print("入参：", "\n", json.dumps(params, indent=4, ensure_ascii=False))
        response = s.post(url=url, data=params, timeout=10)
        print("出参：", "\n",  json.dumps(json.loads(response.text), indent=4, ensure_ascii=False))
        box_id = json.loads(response.text).get("data")
        return box_id
    except Exception as e:
        return [] 


def get_variants(box_id):
    try:
        url = root_url + "/api/v1/base/boxVariants"
        s = requests.Session()
        # 设置 post()方法进行重访问
        s.mount('http://', HTTPAdapter(max_retries=Retry(total=5, allowed_methods=frozenset(['GET', 'POST']))))
        params = {"token": get_token(), "boxId": box_id, "plcId": "0"}
        # print("入参：", "\n", json.dumps(params, indent=4, ensure_ascii=False))
        response = s.post(url=url, data=params, timeout=10)
        out_data = json.dumps(json.loads(response.text), indent=4, ensure_ascii=False)
        out_data = out_data[:400] + "......"
        # print("出参：", "\n",  out_data)
        variants = json.loads(response.text).get("data")
        return variants
    except Exception as e:
        return []


def get_variants_data(variants, box_id):
    try:
        url = root_url + "/api/v1/data/realtimeDatas"
        variants_ids = ":".join([str(i.get("variantId")) for i in variants])
        s = requests.Session()
        # 设置 post()方法进行重访问
        s.mount('http://', HTTPAdapter(max_retries=Retry(total=5, allowed_methods=frozenset(['GET', 'POST']))))
        params = {"token": get_token(), "variantIds": f"{box_id}({variants_ids})"}
        print("入参：", "\n", json.dumps(params, indent=4, ensure_ascii=False))
        response = s.post(url=url, data=params, timeout=10)
        out_data = json.dumps(json.loads(response.text), indent=4, ensure_ascii=False)
        out_data = out_data[:400] + "......"
        print("出参：", "\n",  out_data)
        variants_data = json.loads(response.text).get("data")
        new_variants_data = []
        # 合并数据
        for i in variants:
            for v in variants_data:
                v_id = v.get("id").split(":")[-1]
                if v_id == i.get("variantId"):
                    new_variants_data.append(dict(i, **v))
        return new_variants_data
    except Exception as e:
        return []


def test(id, box_id):
    url = root_url + "/api/v1/data/realtimeDatas"
    response = requests.post(url=url, data={"token": get_token(), "variantIds": f"{box_id}({id})"})
    variants_data = json.loads(response.text).get("data")
    print(variants_data)


def set_value():
    box_id = 'c1881d886ab54370aa81d3a1a8532ba4:150'
    url = "/api/v1/mndraw/setVariantValue"
    response = requests.post(url=url, data={"token": get_token(), "variantId": box_id, "variantVal": 907})
    variants_data = json.loads(response.text)
    print(variants_data)


if __name__ == '__main__':
    # box_id = get_box_id("mNCgKJSl75k")
    # for i in box_id:  39845c626cf645b9ad278a8b2817c593  f70a4e6dd29a40d0baa987b9bb067991
    #     print(i)
    # variants = get_variants('72f954c219324c839d14eb53911ec932')
    # for i in variant:
    #     print(i)
    # variants = [{'variantId': '26', 'name': 'MT101', 'addr': 'DB3.DBD680', 'type': '0', 'id': '72f954c219324c839d14eb53911ec932:26', 'status': '1', 'value': '1.16', 'time': 1725170642}]
    # variant_data = get_variants_data(variants, '72f954c219324c839d14eb53911ec932')
    # print(variant_data)
    # result = test(120, "86c16e6b5165464dab16a4f8320e6d5f")
    # box_id = "6499855e3d8b4836b158556646f3b9fd"
    # result = get_variants(box_id)
    # print(result)
    variants = get_variants("86c16e6b5165464dab16a4f8320e6d5f")
    get_variants_data(variants, "86c16e6b5165464dab16a4f8320e6d5f")