山西恒安刷课以及考试

软件正常无法抓包,简单绕过,无加密,整个流程可也说是设计得比较好,就是有点费服务器,斻多API,都是一条一条拉满了的,以下是python代码

考试

import requests, json



header={
    'Host': 'nazx.yunkeonline.cn',
    'content-type': 'application/json',
    'x-access-token': '8ECDCD1C5FCC04DE8E35C5255185385F7CFE78459FAE318A432E95140FC1CFFC',
    'app-version': '2.1.0',
    'app-os': 'ANDROID',
    'code-version': '2.1.1',
    'user-agent': 'okhttp/3.14.9',
}
def login(uname,pwd):

    headers = {
        'Host': 'nazx.yunkeonline.cn',
        'app-version': '2.1.0',
        'app-os': 'ANDROID',
        'code-version': '2.1.1',
        'content-type': 'application/json;charset=UTF-8',
        'user-agent': 'okhttp/3.14.9',
    }
    data = {"bindMobile":uname,"imageCode":"","password":pwd,"username":"","verifyCode":""}

    response = requests.post('https://nazx.yunkeonline.cn/c/user/login', headers=headers, json=data)
    print(response.text)
    return response.json()['result']['token']


def get_courseId():
    params = {
        'statusList': '1,2',
        'pageNo': '1',
        'pageSize': '3',
    }

    response = requests.get('https://nazx.yunkeonline.cn/c/trainPlan/getMyPlan', params=params, headers=header)
    classId=response.json()['result']['records'][0]['trainPlanId']
    params = {
        'classId': classId,
    }
    response = requests.get('https://nazx.yunkeonline.cn/c/trainPlan/getAppClassTaskList', params=params, headers=header)

    return response.json()['result']['trainPlanStageDetailDtos'][1]['planAppTaskListDtos'][0]['taskId']


def get_question_list(id):

    response = requests.get(f'https://nazx.yunkeonline.cn/c/exam/getQuestionList/{id}', headers=header)
    for i in response.json()['result']:

        with open('answer.json', 'r') as f:
            answers=json.load(f)

        for j in answers['result']:
            
            if j['id']==i['id']:
                print(j['analysis'],' ==> ',answer)
                answer=j['answer']
                answerQuestion(answer,id,j['id'])

                
    return response.json()['result']


def start_exam(id):

    response = requests.post(f'https://nazx.yunkeonline.cn/c/exam/startExam/{id}', headers=header)
    print(response.text)


def answerQuestion(answer,id,qid):
    data = {'answer':answer,'examinationId':id,'questionId':qid}
    response = requests.post('https://nazx.yunkeonline.cn/c/exam/answerQuestion', headers=header, json=data)


def submit_exam(id):

    response = requests.post(f'https://nazx.yunkeonline.cn/c/exam/submit/{id}/0', headers=header)
    print(response.text)
    print("考试完成")

def run():
    header['x-access-token']=login(phone,pwd)
    id=get_courseId()
    start_exam(id)
    get_question_list(id)

    submit_exam(id)


phone="150343xxxx"
pwd="xx"
run()

刷课, 只想到了多线程,异步没能实现,理论上可,但是失败了 开了36线程的线程池,10 ==>360秒,1分钟看36分钟,算总时常,异步理论上更佳,奈何实力不到

import requests,time
from concurrent.futures import ThreadPoolExecutor


header={
    'Host': 'nazx.yunkeonline.cn',
    'content-type': 'application/json',
    'x-access-token': '',
    'app-version': '2.1.0',
    'app-os': 'ANDROID',
    'code-version': '2.1.1',
    'user-agent': 'okhttp/3.14.9',
}
def login(uname,pwd):

    headers = {
        'Host': 'nazx.yunkeonline.cn',
        'app-version': '2.1.0',
        'app-os': 'ANDROID',
        'code-version': '2.1.1',
        'content-type': 'application/json;charset=UTF-8',
        'user-agent': 'okhttp/3.14.9',
    }
    data = {"bindMobile":uname,"imageCode":"","password":pwd,"username":"","verifyCode":""}

    response = requests.post('https://nazx.yunkeonline.cn/c/user/login', headers=headers, json=data)
    print(response.text)
    return response.json()['result']['token']


def get_courseId():
    params = {
        'statusList': '1,2',
        'pageNo': '1',
        'pageSize': '3',
    }

    response = requests.get('https://nazx.yunkeonline.cn/c/trainPlan/getMyPlan', params=params, headers=header)
    classId=response.json()['result']['records'][0]['trainPlanId']
    params = {
        'classId': classId,
    }
    response = requests.get('https://nazx.yunkeonline.cn/c/trainPlan/getAppClassTaskList', params=params, headers=header)

    return response.json()['result']['trainPlanStageDetailDtos'][0]['planAppTaskListDtos'][0]['taskId']
def get_all_lesson():

    header['x-access-token']=login(phone,pwd)
    params = {
        'courseId': get_courseId(),
    }
    print(params)
    response = requests.get('https://nazx.yunkeonline.cn/c/courseApi/getLessonList', params=params, headers=header)
    ids=[]
    for i in response.json()['result']:
        for _ in i['lessons']:
            if _['progress']!=2:
                ids.append({'id': _['id'],'title':_['title'],"duration":_['duration']})

    return ids

def queryLessonPlaytime(x):
    
    _id=x['id']
    response = requests.get(f'https://nazx.yunkeonline.cn/course/api/monitor/queryLessonPlaytime/{_id}', headers=header)
    result=response.json()
    playTime=0
    duration=x['duration']
    if result=="200":
        playTime=result['info']['playtime']
        duration=result['info']['maxPlaytime']
    else:
        startPlay(x['id'])
    print(response.text)

    addPlaytime(x['id'],duration,playTime)

def startPlay(id):
    response = requests.get(
        f'https://nazx.yunkeonline.cn/course/api/CourseLesson/course/5479/lesson/{id}/start',
        headers=header,
    )
    print(response.text)

def addPlaytime(id,duration,playTime=0):
    params = {
            'lessonId': id,
            'playTime': playTime,
            'lessonType': 'video',
        }
    while playTime<duration:
        
        playTime+=10
        if playTime>=duration:
            finsih()
            playTime+=100
            return
        
        params['playTime']=playTime
        response = requests.post('https://nazx.yunkeonline.cn/course/api/monitor/addPlaytime', params=params, headers=header)
        print(response.text)
        time.sleep(10)

def finsih(id):
    params = {
        'userId': '107000',
    }
    response = requests.get(
        f'https://nazx.yunkeonline.cn/course/api/CourseLesson/course/5479/lesson/{id}/finish',
        params=params,
        headers=header,
    )
    print(response.text)


phone="1503435xxx"
pwd="xx"
get_all_lesson()


def run():
    with ThreadPoolExecutor(max_workers=36) as t:
        test=get_all_lesson()
        print(test)
        for x in test:
            print(x)
            t.submit(queryLessonPlaytime,x)


# run()