ai_qback/mock/mock_request.py

# -*- coding:utf-8 -*-
#
import requests
import time
import json 
from functools import wraps

product_adress = "http://111.231.167.191" 
test_address = "http://111.231.167.191:8004" 
test_address2 = "http://111.231.167.191:8003" 

local_adress = "http://127.0.0.1:9000" 

use_address = test_address 


def time_use(fn):
    @wraps(fn)
    def cc(*args, **kwargs): 
        f_time = time.time()
        res = fn(*args, **kwargs)

        cha = round(time.time() - f_time, 3)
        if cha > 0.1:
            print(f'函数:{fn.__name__} 一共用时', cha, '秒')
        return res 

    return cc 


def test_connect():
    """连接测试"""
    try:
        r = requests.post(use_address)
        assert r.text == 'hello world'
        r = requests.get(use_address)
        assert r.text == 'hello world'
        print("测试连接成功")
    except AssertionError:
        print("测试连接失败")


def get_article():
    """同时检验article接口和query_oss_file接口"""
    json_data = {"meaning_ids": [300751, 300756, 300757, 300758, 300454, 302096, 302102], "callback_url": "http://localhost/callback",
                 "student_stage": 1, "vocabulary": 700, "class_id": 123456}

    r = requests.post(f"{use_address}/article", json=json_data)
   
    key = r.json()['key']
    time.sleep(120)
    query_file_content(key)


def query_file_content(key):
    json_data = {"key": key}
    try:
        r = requests.post(f"{use_address}/query_oss_file", json=json_data)
        r.raise_for_status() 
        response_data = r.json()
        assert response_data['wordCount'] > 0, "词数为0"
    except requests.RequestException as e:
        print(f"请求失败: {e}")
    except json.JSONDecodeError as e:
        print(f"JSON解析错误: {e}")
    except AssertionError as e:
        print(f"断言错误: {e}")
    except Exception as e:
        print(f"未知错误: {e}")


def get_audio():
    word = "cat"
    r1 = requests.post(f"{use_address}/tts", json={"text": word, "type": 0}) 
    r2 = requests.post(f"{use_address}/tts", json={"text": word, "type": 2}) 
    r3 = requests.post(f"{use_address}/tts", json={"text": word, "type": 1}) 
    assert r1.json()['code'] == 200
    assert r2.json()['code'] == 200
    assert r3.status_code == 200


@time_use
def get_article2():
    json_data = {"words": [
        {'spell': 'term', 'meaning': '学期'}, {'spell': 'ordinary', 'meaning': '平常的, 普通的, 一般的'},
        {'spell': 'discussion', 'meaning': '讨论, 谈论'}, {'spell': 'shine', 'meaning': '照耀, 发光, 闪耀'},
        {'spell': 'million', 'meaning': '百万'}, {'spell': 'greet', 'meaning': '问候'}
    ], "take_count": 1, "student_stage": 3, "demo_name": "测试项目"}

    r = requests.post(f"{use_address}/article/reading-comprehension", json=json_data)
    r_json = r.json()
    try:
        assert len(r_json['articles']) == 1
        return r_json
    except Exception as e:
        print("春笋文章reading-comprehension错误", e)
        print("错误数据", r_json)


@time_use
def get_article2_1():
    """新的获取文章"""
    json_data = {'core_words': [{'spell': 'sudden', 'meaning': '突然的, 意外的', 'word_id': 1114468, 'meaning_id': 1734},
                                {'spell': 'frighten', 'meaning': '惊吓, 惊恐', 'word_id': 899278, 'meaning_id': 1735},
                                {'spell': 'relation', 'meaning': '关系, 联系, 亲戚, 亲属', 'word_id': 1061800, 'meaning_id': 1736},
                                {'spell': 'agreement', 'meaning': '协议,协定', 'word_id': 753401, 'meaning_id': 1743},
                                {'spell': 'pool', 'meaning': '游泳池, 池子', 'word_id': 1035634, 'meaning_id': 1747},
                                {'spell': 'risk', 'meaning': '冒险, 风险', 'word_id': 1069002, 'meaning_id': 1748},
                                {'spell': 'centre', 'meaning': '中心', 'word_id': 806629, 'meaning_id': 1749},
                                {'spell': 'shut', 'meaning': '关上, 关闭', 'word_id': 1088662, 'meaning_id': 1751},
                                {'spell': 'piano', 'meaning': '钢琴', 'word_id': 1027211, 'meaning_id': 1752},
                                {'spell': 'trust', 'meaning': '信任, 信赖', 'word_id': 1142977, 'meaning_id': 1753},
                                {'spell': 'camera', 'meaning': '照相机', 'word_id': 799656, 'meaning_id': 1754},
                                {'spell': 'course', 'meaning': '课程', 'word_id': 834016, 'meaning_id': 399},
                                {'spell': 'thought', 'meaning': '思想, 想法', 'word_id': 1130826, 'meaning_id': 685}],
                 'take_count': 4, 'student_stage': 3, 'demo_name': '春笋英语',"exercise_id":999,"article_length":200}

    r = requests.post(f"{use_address}/article/reading-comprehension", json=json_data)
    r_json = r.json()
    try:
        return r_json
    except Exception as e:
        print("春笋文章reading-comprehension错误", e)
        print("错误数据", r_json)


def download_word():
    from make_docx_demo.data import test_json2
    params = {"document_format": 2, "scanpage_format": 1}

    r = requests.post(f"{use_address}/make_word/vocabulary_assault", params=params, json=test_json2)
    r.raise_for_status()
    suffix = {1: "docx", 2: "pdf"}[params['document_format']]
    with open(f"test.{suffix}", "wb") as f:
        f.write(r.content)


def spoken_language():
    url = f"{use_address}/spoken_language"
    with open(r"1.mp3", "rb") as f:
        audio_data = {'file': ('file.bin', f, 'application/octet-stream')}
        data = {"text": "You must study to be frank with the world"}
        r1 = requests.post(url, data=data, files=audio_data)
    assert r1.json()['code'] == 0

    data = {"text": "apple Life is like a box of chocolates， you never know what you're gonna",
            "url": "https://yunzhixue.blob.core.chinacloudapi.cn/download/ttsmaker-file-2025-1-13-10-47-47.mp3"}
    r2 = requests.post(url, data=data)
    assert r2.json()['code'] == 0


@time_use
def get_article3():
    json_data = {"words": [
        {'spell': 'term', 'meaning': '学期'}, {'spell': 'ordinary', 'meaning': '平常的, 普通的, 一般的'},
        {'spell': 'discussion', 'meaning': '讨论, 谈论'}, {'spell': 'shine', 'meaning': '照耀, 发光, 闪耀'},
        {'spell': 'million', 'meaning': '百万'}, {'spell': 'greet', 'meaning': '问候'}
    ], "take_count": 1, "student_stage": 3, "demo_name": "测试项目"}

    r = requests.post(f"{use_address}/article/reading-comprehension/deepseek", json=json_data)
    r_json = r.json()
    try:
        assert r_json['articles'][0]["english"]
        assert r_json['articles'][0]["difficultSentences"]
        assert r_json['articles'][0]["questions"]
        assert r_json['articles'][0]["allWordAmount"] > 10
        return r_json
    except Exception as e:
        print("deepseek生成文章出错", e)
        print(r_json)


@time_use
def run_all_test_cese():
    test_connect() 

    get_audio() 
    spoken_language() 
    download_word() 
    print(get_article2_1()) 

   
   


if __name__ == '__main__':
    run_all_test_cese()