fuck15.py

#!/usr/bin/env python3

# -*- coding: utf-8 -*-

import requests
import bson
import json
import time
import random
import re
import hashlib
import os
import platform
import sys
import traceback
import getopt
import base64

VERSION_NAME = "FxxkSsxx 1.5 (Community Version)"

answer_dictionary = {}
hit_count = 0
mode_id_list = [
    {"id": "5f71e934bcdbf3a8c3ba51d5", "name": "英雄篇", "mode": 1},
    {"id": "5f71e934bcdbf3a8c3ba51d6", "name": "复兴篇", "mode": 1},
    {"id": "5f71e934bcdbf3a8c3ba51d7", "name": "创新篇", "mode": 1},
    {"id": "5f71e934bcdbf3a8c3ba51d8", "name": "信念篇", "mode": 1},
    {"id": "5f71e934bcdbf3a8c3ba51d9", "name": "限时赛", "mode": 2},
    {"id": "5f71e934bcdbf3a8c3ba51da", "name": "抢十赛", "mode": 3},
]
curr_mode = None
random_mode_enabled = True
inform_enabled = False
auto_refresh_token_enabled = False
expire_time = -1
way = ""


class MyError(Exception):
    def __init__(self, code, msg):
        self.code = code
        self.msg = msg

    def __str__(self):
        return "{}({})".format(self.msg, self.code)


def SubmitVerification(header, code=None):
    if code == None:
        if way == "1":
            code = "HD1bhUGI4d/FhRfIX4m972tZ0g3jRHIwH23ajyre9m1Jxyw4CQ1bMKeIG5T/voFOsKLmnazWkPe6yBbr+juVcMkPwqyafu4JCDePPsVEbVSjLt8OsiMgjloG1fPKANShQCHAX6BwpK33pEe8jSx55l3Ruz/HfcSjDLEHCATdKs4="
        elif way == "2":
            code = "BzPdXIQzu5Z2GHMkgWt9iBUQTbo2EWmw7tIuDaIEa4EQSXziGlUmS4aqrzw8Pzo51P+YtbUpAaqKLWF/rG4G41TnHGtItm7NkU90024LysxGUzh4jhy9YO0ivJZ3MSv+WcA8u/1TacORW4cvm9uBejdpTBOdWsDFOjJpTdCkYYE="

    submit_data = {
        "activity_id": "5f71e934bcdbf3a8c3ba5061",
        "mode_id": curr_mode['id'],
        "way": way,
        "code": code
    }
    url = "https://ssxx.univs.cn/cgi-bin/save/verification/code/"
    response = requests.post(url, json=submit_data, headers=header)
    result = json.loads(response.text)
    if result["code"] != 0:
        raise MyError(result["code"], "提交验证码失败: " + str(result))


def CheckVerification(header, code=None):
    if code == None:
        if way == "1":
            code = "E5ZKeoD8xezW4TVEn20JVHPFVJkBIfPg+zvMGW+kx1s29cUNFfNka1+1Fr7lUWsyUQhjiZXHDcUhbOYJLK4rS5MflFUvwSwd1B+1kul06t1z9x0mfxQZYggbnrJe3PKEk4etwG/rm3s3FFJd/EbFSdanfslt41aULzJzSIJ/HWI="
        elif way == "2":
            code = "caap/GndfVmhoudMEhf1C7vjG5Hyc1gMLPNZAT6yfdx2kt8jOTCzUYsbd1dyqUYRCv4/rLZGf++4HIK/7Q2/ZWlOkQDTV9gxOOHjwkKqkF/oZZIOg0714FXPZj8eet44bp481QRDyjqBzwGAM3xzQut5bLqeTLkayexsu2Q3P9g="

    submit_data = {
        "activity_id": "5f71e934bcdbf3a8c3ba5061",
        "mode_id": curr_mode['id'],
        "way": way,
        "code": code
    }
    url = "https://ssxx.univs.cn/cgi-bin/check/verification/code/"
    response = requests.post(url, json=submit_data, headers=header)
    result = json.loads(response.text)
    if result["code"] != 0:
        raise MyError(result["code"], "检查验证码失败: " + str(result))

    return result["status"]


def GetProfile(token):
    timestamp = int(time.time())
    url = "https://ssxx.univs.cn/cgi-bin/race/grade/?t={}&activity_id=5f71e934bcdbf3a8c3ba5061".format(
        timestamp)
    referer = "https://ssxx.univs.cn/client/detail/5f71e934bcdbf3a8c3ba5061/score"

    response = requests.get(url, headers=BuildHeader(token, referer))
    result = json.loads(response.text)
    if result["code"] != 0:
        raise MyError(result["code"], "获取信息失败: " + str(result))

    data = result["data"]
    return data["integral"]


def ParseToken(token):
    data = token.split('.')
    user_info = json.loads(str(base64.b64decode(data[1] + "=="), "utf-8"))
    try:
        token_info = {
            "name": user_info["name"],
            "uid": user_info["uid"],
            "time": int(user_info["iat"]),
            "expire": int(user_info["exp"])
        }
        return token_info
    except Exception as err:
        print(user_info)


def RefreshToken(header):
    url = "https://ssxx.univs.cn/cgi-bin/authorize/token/refresh/"
    response = requests.get(url, headers=header)

    result = json.loads(response.text)
    if result["code"] != 0:
        raise MyError(result["code"], "更新token失败: " + str(result["message"]))

    new_token = result["token"]

    detail = ParseToken(new_token)
    print("token更新完成，于 ", time.asctime(time.localtime(detail["time"])), " 生效")
    print("将在 ", time.asctime(time.localtime(detail["expire"])), " 失效")

    return new_token


def SendNotification(msg):
    if not inform_enabled:
        print("[info] ", msg)
        return
    url = "http://localhost:6666"
    payload = {"type": "fxxkSsxx", "msg": msg, "ver": VERSION_NAME}
    response = requests.post(url, json=payload)
    print(response.text)


def ReadAnswerFromFile():
    global answer_dictionary
    answer_file = open('ssxx.bson', 'rb')
    answer_dictionary = bson.loads(answer_file.read())
    answer_file.close()
    print("已读取", len(answer_dictionary['hashBank'].items()), "个答案！")
    print("learnt: ", len(answer_dictionary['questionBank'].items()))
    if answer_dictionary.__contains__("userBank"):
        print("已经和 {} 人PK过".format(len(answer_dictionary["userBank"].items())))
    else:
        answer_dictionary["userBank"] = {}


def SaveAnswerToFile():
    global answer_dictionary
    answer_file = open('ssxx.bson', 'wb')
    answer_file.write(bson.dumps(answer_dictionary))
    answer_file.close()
    print("已保存", len(answer_dictionary['hashBank'].items()), "个答案！")
    print("learnt: ", len(answer_dictionary['questionBank'].items()))
    if answer_dictionary.__contains__("userBank"):
        print("已经和 {} 人PK过".format(len(answer_dictionary["userBank"].items())))


def BuildHeader(token, referer):
    headers = {
        'Host': 'ssxx.univs.cn',
        'Connection': 'keep-alive',
        'Accept': 'application/json, text/plain, */*',
        'Authorization': 'Bearer ' + token,
        'Origin': 'https://ssxx.univs.cn',
        'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/39.0.2171.95 Safari/537.36 MicroMessenger/6.5.2.501 NetType/WIFI WindowsWechat QBCore/3.43.884.400 QQBrowser/9.0.2524.400',
        'Referer': referer,
        'Accept-Encoding': 'gzip, deflate',
        'Accept-Language': 'zh-CN,zh;q=0.8,en-us;q=0.6,en;q=0.5;q=0.4',
        'Cookie': 'tgw_l7_route=0076108fd10666c6fe8ef0b035e2048d; _ga=GA1.2.1492401396.1620982965; _gid=GA1.2.901374102.1620982965; _gat=1'
    }

    return headers


def PrintQuizObject(quiz_object):
    print("问题ID列表: ")
    for i in range(0, 20):
        print("问题", i, ": ", quiz_object["question_ids"][i])


def StartQuiz(header, ignoreNewBee=True):
    global hit_count
    hit_count = 0
    super_speed = False

    print("尝试开始考试……")

    url = "https://ssxx.univs.cn/cgi-bin/race/beginning/?t={}&activity_id=5f71e934bcdbf3a8c3ba5061&mode_id={}&way={}".format(
        int(time.time()),
        curr_mode['id'],
        way
    )

    for fail in [0, 1, 2]:    # 最多尝试等待3次
        response = requests.request("GET", url, headers=header)
        quiz_object = json.loads(response.text)
        status_code = quiz_object["code"]

        if status_code == 0:
            print("开始考试成功，模式: ", quiz_object["mode_title"])
            if quiz_object.__contains__("opponent") and quiz_object["opponent"]:
                opponent = quiz_object["opponent"]
                print("对手: {} [{}]".format(
                    opponent["name"], opponent["univ_name"]))
                if answer_dictionary["userBank"].__contains__(opponent["id"]):
                    record = answer_dictionary["userBank"][opponent["id"]]["record"]
                    print("历史比分 {}:{}".format(record["lose"], record["win"]))
                    if record["win"] - record["lose"] > 2:
                        super_speed = True
                    if record["win"] - record["lose"] > 4:
                        if ignoreNewBee:
                            raise MyError(666, "对方开挂了吧")
                        else:
                            print("对方开挂了吧")
                # if opponent["univ_name"] == "武汉工商学院":
                #    super_speed = True

            # print("race_code", quiz_object["race_code"])

            return quiz_object["question_ids"], quiz_object["race_code"], super_speed

        elif status_code == 4832:
            print("答题次数用完，等待10分钟")
            if fail == 0:   # 在开始等待时发送通知
                SendNotification("...")
            time.sleep(600)

        else:
            raise MyError(status_code, "开始考试失败: " + str(quiz_object))


def GetTitleAndMd5(title):
    title = re.sub(
        r"<(\w+)[^>]+?(?:display: {0,}none;).*?>.*?<\/\1>", "", title)
    title = re.sub("<.*?>", "", title)
    result = hashlib.md5(title.encode(encoding='UTF-8')).hexdigest()
    return title, result


def GetQuestionDetail(question_id, header):
    global answer_dictionary

    url = "https://ssxx.univs.cn/cgi-bin/race/question/?t={}&activity_id=5f71e934bcdbf3a8c3ba5061&question_id={}&mode_id={}&way={}".format(
        int(time.time()),
        question_id,
        curr_mode['id'],
        way
    )

    response = requests.request("GET", url, headers=header)

    if response.status_code != 200:
        raise MyError(response.status_code, "获取题目信息失败")

    question_detail_object = json.loads(response.text)
    if question_detail_object["code"] != 0:
        raise MyError(question_detail_object["code"], "获取题目信息失败。问题ID: " +
                      question_id + "错误信息: " + str(question_detail_object))

    print("获取题目信息成功。")

    question = {}
    question["id"] = question_detail_object["data"]["id"]
    question["title"], question["hash"] = GetTitleAndMd5(
        question_detail_object["data"]["title"])
    question["answer_list"] = []
    for i in question_detail_object["data"]["options"]:
        option_title, option_hash = GetTitleAndMd5(i["title"])
        question["answer_list"].append((i["id"], option_hash, option_title))
        i["title"] = option_title

    print()
    print(question["title"])
    # learn
    if question["hash"] not in answer_dictionary['questionBank']:
        question_detail_object["data"]["title"] = question["title"]
        answer_dictionary['questionBank'][question["hash"]
                                          ] = question_detail_object["data"]
        print(answer_dictionary['questionBank'][question["hash"]])
        print("已加入习题集")

    return question


def BuildAnswerObject(question):
    global answer_dictionary
    global hit_count

    # print("正在尝试寻找……")

    answer_object = {
        "activity_id": "5f71e934bcdbf3a8c3ba5061",
        "question_id": question["id"],
        "answer": None,
        "mode_id": curr_mode['id'],
        "way": way
    }

    if answer_dictionary['hashBank'].__contains__(question["hash"]):
        hit_count += 1
        answer_object["answer"] = []
        for i in question["answer_list"]:
            if i[1] in answer_dictionary['hashBank'][question["hash"]]:
                answer_object["answer"].append(i[0])
                print(i[2], " 正确")
    else:
        print("答案库中不存在该题答案，蒙一个A选项吧")
        answer_object["answer"] = [question["answer_list"][0][0]]

    return answer_object, question


def SubmitAnswer(answer_question_tuple, header):
    global answer_dictionary
    answer_object, question = answer_question_tuple

    url = "https://ssxx.univs.cn/cgi-bin/race/answer/"

    header["Content-Type"] = "application/json;charset=utf-8"

    response = requests.request(
        "POST", url, headers=header, data=json.dumps(answer_object))

    if response.status_code != 200:
        raise MyError(response.status_code, "提交答案失败")

    result_object = {}
    try:
        result_object = json.loads(response.text)
    except:
        SendNotification(response.text)
        Pause()

    if result_object["code"] != 0:
        raise MyError(
            result_object["code"], "Failed to submit answer: " + str(result_object))

    question_hash = question["hash"]
    found_hash = answer_dictionary['hashBank'].__contains__(question_hash)
    correct = result_object["data"]["correct"]
    finished = result_object["data"]["finished"]

    if correct and found_hash:
        return True, finished
    elif not correct and found_hash:
        print("答案库中已有答案但不正确！")
    elif correct and not found_hash:
        # SaveAnswerToFile()
        print("运气不错，居然蒙对了，保存答案")
    elif not correct and not found_hash:
        # SaveAnswerToFile()
        print("答案错误，更新答案")

    if not found_hash or not correct:
        answer_dictionary['hashBank'][question_hash] = []

    for i in result_object["data"]["correct_ids"]:
        print("Right answer id: ", i)
        for j in question["answer_list"]:
            if i == j[0]:
                print("已在问题列表中找到该答案，元组为", j)
                answer_dictionary['hashBank'][question_hash].append(j[1])
                break

    if finished:
        print("本轮答题结束")

    return correct, finished


def FinishQuiz(race_code, header):
    global answer_dictionary

    url = "https://ssxx.univs.cn/cgi-bin/race/finish/"

    header["Content-Type"] = "application/json;charset=utf-8"
    payload = "{\"race_code\":\"" + race_code + "\"}"

    result = json.loads(requests.request(
        "POST", url, headers=header, data=payload).text)
    fail = 0
    while result["code"] != 0:
        print("完成考试时出错，错误代码: ", result)
        err_code = result["code"]
        if err_code == 1001 or err_code == 1002:
            raise MyError(err_code, "请重新登录")
        fail += 1
        if fail > 5:
            raise MyError(err_code, "完成时出错: " + str(result))
        time.sleep(0.5)
        result = json.loads(requests.request(
            "POST", url, headers=header, data=payload).text)

    print("回答完毕，本次得分: ", result["data"]["owner"]
          ["correct_amount"], "答案库命中数: ", hit_count)

    if result["data"].__contains__("opponent") and result["data"]["opponent"]:
        opponent = result["data"]["opponent"]
        print("对手: {} [{}]".format(opponent["name"], opponent["univ_name"]))
        if answer_dictionary["userBank"].__contains__(opponent["id"]):
            pass
        else:
            answer_dictionary["userBank"][opponent["id"]] = {
                "name": opponent["name"],
                "univ_name": opponent["univ_name"],
                "record": {
                    "win": 0,
                    "lose": 0,
                },
            }

        if result["data"].__contains__("badge"):
            badge = result["data"]["badge"]
            if badge == True:
                print("获得徽章")
                answer_dictionary["userBank"][opponent["id"]
                                              ]["record"]["lose"] += 1
            elif badge == False:
                print("未获得徽章")
                answer_dictionary["userBank"][opponent["id"]
                                              ]["record"]["win"] += 1


def Pause():
    if platform.system() == "Windows":
        os.system("pause")


def Start(token):
    global curr_mode
    global expire_time

    ReadAnswerFromFile()

    try:
        rounds = 0
        while True:
            if random_mode_enabled:
                curr_mode = mode_id_list[random.randrange(0, 4)]
                print("这一轮，这一轮是", curr_mode["name"])

            if rounds % 10 == 0:
                SaveAnswerToFile()
                print("Profile score: ", GetProfile(token))
            else:
                time.sleep(0.5)
            rounds += 1

            referer = "https://ssxx.univs.cn/client/exam/5f71e934bcdbf3a8c3ba5061/{}/1/{}".format(
                curr_mode['mode'], curr_mode['id'])
            header = BuildHeader(token, referer)

            question_list = None
            race_code = ""
            super_speed = False

            try:
                question_list, race_code, super_speed = StartQuiz(header)
            except MyError as err:
                if err.code == 666:
                    print("对方太强，不跟ta玩了")
                    # time.sleep(float(random.randint(1000, 2000)) / 1000)
                    curr_mode = mode_id_list[random.randrange(0, 4)]
                    print("这一轮，这一轮是", curr_mode["name"])
                    question_list, race_code, super_speed = StartQuiz(
                        header, False)
                else:
                    raise err

            question_list_len = len(question_list)
            for i in range(0, question_list_len):
                try:
                    correct, finished = SubmitAnswer(BuildAnswerObject(
                        GetQuestionDetail(question_list[i], header)), header)
                    if correct:
                        print("第", i + 1, "题回答正确！")
                        if not super_speed:
                            if curr_mode['mode'] == 1:
                                # Personal
                                time.sleep(
                                    float(random.randint(500, 900)) / 1000)
                            else:
                                # PK
                                time.sleep(
                                    float(random.randint(600, 1500)) / 1000)
                        else:
                            # Crusade
                            time.sleep(float(random.randint(290, 350)) / 1000)
                            print("加油！！")
                    else:
                        print("第", i + 1, "题回答错误，答案已更新！")
                        time.sleep(float(random.randrange(1500, 3000)) / 1000)

                    if finished:
                        break
                except MyError as err:
                    if err.code == 2104:  # 2104: Question not exist
                        SendNotification("问题不存在: " + question_list[i])
                        pass
                    else:
                        raise err

                if i == 1:
                    if CheckVerification(header):
                        print("验证码已通过")
                    else:
                        SubmitVerification(header)
                        print("验证码状态: ", CheckVerification(header))

            print("==============================================================")
            FinishQuiz(race_code, header)
            print("==============================================================")
            print()
            print()
            time.sleep(float(random.randrange(700, 1500)) / 1000)

            if auto_refresh_token_enabled and expire_time - time.time() < 500:
                new_token = RefreshToken(header)
                expire_time = ParseToken(new_token)["expire"]
                token = new_token
                SendNotification(new_token)
                SendNotification(
                    "token已更新至 " + time.asctime(time.localtime(expire_time)))
                time.sleep(5)

    except MyError as err:
        tag = "[{}] ".format(time.asctime(time.localtime(time.time())))

        if err.code == 1001 or err.code == 1002:
            print(tag, "登录无效，通知重新登录")
            SendNotification("请重新登录（代码: {}）".format(err.code))
        elif err.code == 1005:
            print(tag, "当前token已退出登录，请重新获取")
        else:
            msg = "已停止，原因: " + str(err)
            print(tag, msg)
            SendNotification(msg)
    except Exception as err:
        tag = "[{}] ".format(time.asctime(time.localtime(time.time())))
        print(tag, traceback.format_exc())
    finally:
        SaveAnswerToFile()
        SendNotification("awsl")   # 程序退出时发送通知
        Pause()


def PrintHelp():
    print("FxxkSsxx (Community Version)")
    print(sys.argv[0])
    print()
    print("    -a, --auto         Auto refresh the token (Off by default)")
    print("    -h, --help         Display this information")
    print("    -i, --inform       Enable webhook notification (Off by default)")
    print("    -m, --mode [index] Category selection (Default 0:random)")
    print("                       1:英雄篇, 2:复兴篇, 3:创新篇, 4:信念篇, 5:限时赛, 6:抢十赛")
    print("    -v, --version      Show version name")
    print()


if __name__ == "__main__":
    argv = sys.argv[1:]

    try:
        opts, args = getopt.getopt(
            argv, "ahim:v", ["auto", "help", "inform", "mode=", "version"])
    except getopt.GetoptError:
        PrintHelp()
        Pause()
        sys.exit(2)

    for opt, arg in opts:
        if opt in ['-a', '--auto']:
            auto_refresh_token_enabled = True
        elif opt in ['-h', '--help']:
            PrintHelp()
            Pause()
            sys.exit()
        elif opt in ['-i', '--inform']:
            inform_enabled = True
        elif opt in ['-m', '--mode']:
            try:
                mode = int(arg)
                if mode not in range(0, 7):
                    raise ValueError()
                if mode == 0:
                    random_mode_enabled = True
                else:
                    random_mode_enabled = False
                    curr_mode = mode_id_list[mode - 1]
            except ValueError:
                PrintHelp()
                Pause()
                sys.exit()
        elif opt in ['-v', '--version']:
            print(VERSION_NAME)
            Pause()
            sys.exit()

    print("Current Version: " + VERSION_NAME)

    token = input("Input token: ").strip()
    try:
        if token.find("token:") == 0:
            token = token[6:]
        token = token.strip("\" ")

        token_info = ParseToken(token)
        expire_time = token_info["expire"]
        print("Hi there!")
        print("Name: ", token_info["name"])
        print("UID: ", token_info["uid"])
        print("Token will expire in: ", time.strftime(
            "%Hh %Mm %Ss", time.gmtime(expire_time - time.time())))

        # print("Choose a mode: ")
        # print("1: Standard Mode")
        # print("2: Team Mode (Use only if you joined a team)")
        way = "1"  # input()
    except Exception as err:
        print(err, traceback.format_exc())
        Pause()
        sys.exit(1)

    time.sleep(2.5)
    Start(token)