x11_template_vue2/templates/seeker/del_test_all_data.py

import argparse
import json
import sys
import uuid
import time
from pathlib import Path
from urllib import error
from urllib import request

PROJECT_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(PROJECT_ROOT))

from pre_data.common.client_tools import redis_client


SEEKER_TYPE_ID = "201c51890c31c308050100218008054251410000000000000000000000000040"


class HttpResponse:
    def __init__(self, status_code: int, text: str):
        self.status_code = status_code
        self.text = text


class HttpClient:
    def post(self, url: str, payload: dict):
        data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        req = request.Request(
            url,
            data=data,
            headers={"Content-Type": "application/json"},
            method="POST",
        )
        try:
            with request.urlopen(req) as resp:
                return HttpResponse(resp.status, resp.read().decode("utf-8"))
        except error.HTTPError as exc:
            return HttpResponse(exc.code, exc.read().decode("utf-8"))


def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--base-url",
        default="http://127.0.0.1:8000",
        help="reset 接口服务地址，例如 http://127.0.0.1:8000",
    )
    return parser.parse_args()


def get_test_device_ids(type_id: str) -> list[str]:
    seen: set[str] = set()
    device_ids: list[str] = []
    pattern = f"device-cycles:{type_id}:*"
    for key in redis_client.scan_iter(match=pattern, count=500):
        parts = key.split(":")
        if len(parts) < 4:
            continue
        device_id = parts[2]
        if not device_id.startswith(""):
            continue
        if device_id in seen:
            continue
        seen.add(device_id)
        device_ids.append(device_id)
    return device_ids


def reset_device(client: HttpClient, base_url: str, type_id: str, device_id: str):
    query_timestamp = str(int(time.time() * 1000))
    payload = {
        "device_info": [
            {
                "device_id": device_id,
                "type_id": type_id,
            }
        ],
        "query_timestamp": query_timestamp,
        "id": uuid.uuid4().hex,
    }
    return client.post(f"{base_url}/api/v1/reset/displayall", payload=payload)


#REDIS_PROFILE=redis_cluster python del_test_all_data.py  
if __name__ == "__main__":
    args = parse_args()
    client = HttpClient()
    device_ids = get_test_device_ids(SEEKER_TYPE_ID)
    print(f"found {len(device_ids)} test devices")
    # print(device_ids)
    # for device_id in device_ids:
    #     response = reset_device(
    #         client=client,
    #         base_url=args.base_url,
    #         type_id=SEEKER_TYPE_ID,
    #         device_id=device_id,
    #     )
    #     print(device_id, response.status_code, response.text)