icp-api/mainapi.py

import base64
import json
import requests
import hashlib
import time
import os
from urllib import parse
import uuid
from flask import Flask, request, jsonify ,Response
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
from crack import Crack  # 假设 Crack 是一个你自己实现的类

app = Flask(__name__)

CACHE_DIR = './cache'  # 缓存文件目录

# 创建缓存目录（如果不存在）
if not os.path.exists(CACHE_DIR):
    os.makedirs(CACHE_DIR)


def auth():
    t = str(round(time.time()))
    data = {
        "authKey": hashlib.md5(("testtest" + t).encode()).hexdigest(),
        "timeStamp": t
    }
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
        "Referer": "https://beian.miit.gov.cn/",
        "Content-Type": "application/x-www-form-urlencoded",
        "Connection": "keep-alive",
        "Accept": "application/json, text/plain, */*",
        "Accept-Encoding": "gzip, deflate, br",
        "Accept-Language": "zh-CN,zh;q=0.9",
        "Origin": "https://beian.miit.gov.cn"
    }
    try:
        resp = requests.post("https://hlwicpfwc.miit.gov.cn/icpproject_query/api/auth", headers=headers,
                             data=parse.urlencode(data)).text
        return json.loads(resp)["params"]["bussiness"]
    except Exception:
        time.sleep(5)
        resp = requests.post("https://hlwicpfwc.miit.gov.cn/icpproject_query/api/auth", headers=headers,
                             data=parse.urlencode(data)).text
        return json.loads(resp)["params"]["bussiness"]


def getImage(token):
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
        "Referer": "https://beian.miit.gov.cn/",
        "Token": token,
        "Connection": "keep-alive",
        "Accept": "application/json, text/plain, */*",
        "Accept-Encoding": "gzip, deflate, br",
        "Accept-Language": "zh-CN,zh;q=0.9",
        "Origin": "https://beian.miit.gov.cn"
    }
    payload = {
        "clientUid": "point-" + str(uuid.uuid4())
    }
    try:
        resp = requests.post("https://hlwicpfwc.miit.gov.cn/icpproject_query/api/image/getCheckImagePoint",
                             headers=headers, json=payload).json()
        return resp["params"], payload["clientUid"]
    except Exception:
        time.sleep(5)
        resp = requests.post("https://hlwicpfwc.miit.gov.cn/icpproject_query/api/image/getCheckImagePoint",
                             headers=headers, json=payload).json()
        return resp["params"], payload["clientUid"]


def aes_ecb_encrypt(plaintext: bytes, key: bytes, block_size=16):
    backend = default_backend()
    cipher = Cipher(algorithms.AES(key), modes.ECB(), backend=backend)

    padding_length = block_size - (len(plaintext) % block_size)
    plaintext_padded = plaintext + bytes([padding_length]) * padding_length

    encryptor = cipher.encryptor()
    ciphertext = encryptor.update(plaintext_padded) + encryptor.finalize()

    return base64.b64encode(ciphertext).decode('utf-8')


def generate_pointjson(big_img, small_img, secretKey):
    crack = Crack()  # 假设 Crack 是一个你实现的类
    boxes = crack.detect(big_img)
    if boxes:
        print("文字检测成功")
    else:
        print("文字检测失败,请重试")
        raise Exception("文字检测失败,请重试")
    points = crack.siamese(small_img, boxes)
    print("文字匹配成功")
    new_points = [[p[0] + 20, p[1] + 20] for p in points]
    pointJson = [{"x": p[0], "y": p[1]} for p in new_points]
    enc_pointJson = aes_ecb_encrypt(json.dumps(pointJson).replace(" ", "").encode(), secretKey.encode())
    return enc_pointJson


def checkImage(uuid_token, secretKey, clientUid, pointJson, token):
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
        "Referer": "https://beian.miit.gov.cn/",
        "Token": token,
        "Connection": "keep-alive",
        "Accept": "application/json, text/plain, */*",
        "Accept-Encoding": "gzip, deflate, br",
        "Accept-Language": "zh-CN,zh;q=0.9",
        "Origin": "https://beian.miit.gov.cn"
    }
    data = {
        "token": uuid_token,
        "secretKey": secretKey,
        "clientUid": clientUid,
        "pointJson": pointJson
    }
    resp = requests.post("https://hlwicpfwc.miit.gov.cn/icpproject_query/api/image/checkImage", headers=headers,
                         json=data).json()
    if resp["code"] == 200:
        return resp["params"]["sign"]
    return False


def query(sign, uuid_token, domain, token):
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
        "Referer": "https://beian.miit.gov.cn/",
        "Token": token,
        "Sign": sign,
        "Uuid": uuid_token,
        "Connection": "keep-alive",
        "Accept": "application/json, text/plain, */*",
        "Accept-Encoding": "gzip, deflate, br",
        "Accept-Language": "zh-CN,zh;q=0.9",
        "Origin": "https://beian.miit.gov.cn",
        "Content-Type": "application/json",
        "Cookie": "__jsluid_s=" + str(uuid.uuid4().hex[:32])
    }
    data = {"pageNum": "", "pageSize": "", "unitName": domain, "serviceType": 1}
    resp = requests.post("https://hlwicpfwc.miit.gov.cn/icpproject_query/api/icpAbbreviateInfo/queryByCondition",
                         headers=headers, data=json.dumps(data).replace(" ", "")).text
    return resp


def save_to_cache(domain, data):
    """保存数据到缓存文件"""
    cache_path = os.path.join(CACHE_DIR, f"{domain}.json")
    with open(cache_path, 'w', encoding='utf-8') as f:
        json.dump(data, f)


def load_from_cache(domain):
    """从缓存文件中加载数据"""
    cache_path = os.path.join(CACHE_DIR, f"{domain}.json")
    if os.path.exists(cache_path):
        with open(cache_path, 'r', encoding='utf-8') as f:
            return json.load(f)
    return None


@app.route('/query', methods=['GET'])
def query_api():
    try:
        # 从 URL 查询参数中获取 domain 和 type
        domain = request.args.get('domain')
        query_type = request.args.get('type', '')  # 默认为空字符串
        
        if not domain:
            return jsonify({"status": "failed", "message": "Missing 'domain' parameter"}), 400
        
        # 如果 type=cache，则检查缓存
        if query_type == 'cache':
            cached_data = load_from_cache(domain)
            if cached_data:
                json_data = json.dumps({"status": "successful", "data": cached_data})
                return Response(json_data, content_type='application/json')

        # 如果没有缓存，或未指定 type=cache，则继续进行查询
        crack = Crack()
        token = auth()
        time.sleep(0.1)
        params, clientUid = getImage(token)
        pointjson = generate_pointjson(params["bigImage"], params["smallImage"], params["secretKey"])
        time.sleep(0.5)
        sign = checkImage(params["uuid"], params["secretKey"], clientUid, pointjson, token)
        time.sleep(0.5)

        if sign:
            result = query(sign, params["uuid"], domain, token)
            response = json.loads(result)
            json_data = json.dumps({"status": "successful", "data": response['params']['list']})

            # 如果 type=cache，则将查询结果缓存
            if query_type == 'cache':
                save_to_cache(domain, response['params']['list'])

            return Response(json_data, content_type='application/json')

        else:
            return jsonify({"status": "failed", "message": "Captcha verification failed"}), 400
    
    except Exception as e:
        return jsonify({"status": "error", "message": str(e)}), 500


if __name__ == '__main__':
    app.run(debug=True, port=5000)