cqcst/CKWK
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 新增考试，可选（kimi，豆包，deepseek）模型进行答题，自己设置 API_KEY, 请不要相信 AI 的结果，最后交卷会提示你

Browse Source
This commit is contained in:
zhilv
2025-12-12 19:17:36 +08:00
parent 6dfeac0d24
commit 14d4bc014d
4 changed files with 284 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

221
main.py
View File

@@ -3,7 +3,7 @@
文件名称: ckwk.py
作者: zhilv
邮箱: zhilv666@qq.com
版本: 1.2
版本: 1.3
--------------------------------------------
说明:
本脚本仅用于学习与技术研究，禁止将其用于任何非法用途。
@@ -28,6 +28,7 @@ import warnings
from deepseek import DeepSeek
from doubao import DOUBAO
from kimi import KIMI
# 修复 ANTIALIAS 错误 - 添加猴子补丁
import PIL.Image
@@ -82,7 +83,7 @@ def replace_html_entities(text, replace_mode="keep_char"):
class CKWK:
def __init__(self, un: str, pw: str, host: str, ai_type="doubao") -> None:
def __init__(self, un: str, pw: str, host: str, ai_type="kimi") -> None:
"""
un: 用户名
pw: 密码
@@ -93,6 +94,8 @@ class CKWK:
self.ai = DOUBAO()
elif ai_type == "deepseek":
self.ai = DeepSeek()
elif ai_type == "kimi":
self.ai = KIMI()
else:
self.ai = DOUBAO()
@@ -109,19 +112,20 @@ class CKWK:
adapter = HTTPAdapter(max_retries=retry_strategy)
self.session.mount("http://", adapter)
self.session.mount("https://", adapter)
# self.session.verify = False
# self.session.proxies.update(
# {
# "http": "http://127.0.0.1:9000",
# "https": "http://127.0.0.1:9000",
# }
# )
self.session.verify = False
self.session.proxies.update(
{
"http": "http://127.0.0.1:9000",
"https": "http://127.0.0.1:9000",
}
)
self.username = un
self.host = host
self.password = pw
self.user = dict()
self.courses = []
self.works = []
self.exams = []
self.qas = []
self.ocr = ddddocr.DdddOcr()
self.session.headers.update(
@@ -233,7 +237,7 @@ class CKWK:
finally:
return video, state
def parse_q_a(self, html: str) -> None:
def parse_q_a_work(self, html: str) -> None:
tree = etree.HTML(html)
qas = tree.xpath('//div[@class="topic-item"]/form')
for qa in qas:
@@ -262,7 +266,36 @@ class CKWK:
}
self.qas.append(items)
def get_question_answer(self, courseID: str, nodeID: str, workID: str):
def parse_q_a_exam(self, html: str) -> None:
tree = etree.HTML(html)
qas = tree.xpath("//div/form")
for qa in qas:
items = {
"id": qa.xpath(
'./div[@class="courseexamcon-submit"]/input[@name="answerId"]/@value'
)[0],
"index": qa.xpath(
'./div[@class="courseexamcon-main"]/div/div[@class="num"]/span/text()'
)[0],
"type": qa.xpath(
'./div[@class="courseexamcon-main"]/div/div[@class="type"]/text()'
)[0],
"title": qa.xpath(
'./div[@class="courseexamcon-main"]/div[@class="name"]/text()'
)[0].strip(),
"chooies": [
{
"num": a.xpath('.//span[@class="num"]/text()')[0],
"txt": a.xpath('.//span[@class="txt"]/text()')[0],
}
for a in qa.xpath(
'./div[@class="courseexamcon-main"]/div[@class="list"]//li'
)
],
}
self.qas.append(items)
def get_question_answer_work(self, courseID: str, nodeID: str, workID: str):
headers = self.session.headers
headers["x-requested-with"] = "XMLHttpRequest"
resp = self.session.get(
@@ -272,22 +305,47 @@ class CKWK:
)
if resp.status_code == 200 and resp.json().get("status", False):
qa_path = resp.json().get("url", "")
self.log(f'开始作业: {resp.json().get("msg", "")} url -> {qa_path}')
resp2 = self.session.get("https://{}{}".format(self.host, qa_path))
self.start_url = "https://{}{}".format(self.host, qa_path)
self.log(f'开始作业: {resp.json().get("msg", "")} url -> {self.start_url}')
resp2 = self.session.get(self.start_url)
self.qas = []
if resp.status_code == 200:
self.parse_q_a(resp2.text)
self.parse_q_a_work(resp2.text)
else:
self.log(f'开始作业: {resp.json().get("msg", "")}')
return False
return True
def work_submit(self, answerID: str, workID: str, answer: str, final: bool):
def get_question_answer_exam(self, courseID: str, nodeID: str, examId: str):
headers = self.session.headers
headers["x-requested-with"] = "XMLHttpRequest"
resp = self.session.get(
"https://{}/user/exam/start".format(self.host),
params={"courseId": courseID, "nodeId": nodeID, "examId": examId},
headers=headers,
)
if resp.status_code == 200 and resp.json().get("status", False):
qa_path = resp.json().get("url", "")
self.start_url = "https://{}{}".format(self.host, qa_path)
self.log(f'开始考试: {resp.json().get("msg", "")} url -> {self.start_url}')
resp2 = self.session.get(self.start_url)
self.qas = []
if resp.status_code == 200:
self.parse_q_a_exam(resp2.text)
else:
self.log(f'开始考试: {resp.json().get("msg", "")}')
return False
return True
def work_submit(
self, answerID: str, workID: str, answer: str, final: bool, index: int
):
headers = self.session.headers
headers["x-requested-with"] = "XMLHttpRequest"
data = {
"answerId": answerID,
"workId": workID,
"finish": "0" if final else "1",
"finish": "0",
}
if "," in answer:
answers = answer.split(",")
@@ -296,7 +354,7 @@ class CKWK:
# data.setdefault("answer[]", []).append(a)
else:
if answer == "":
data["answer"] = "C"
data["answer"] = "A"
data["answer"] = answer
resp = self.session.post(
@@ -304,9 +362,69 @@ class CKWK:
data=data,
headers=headers,
)
if not final:
aa = 1
while aa == 1:
print(f"作业链接: {self.start_url}")
aa = input("现在是最后一道题，请你去检查后输入 0 后自动提交：")
if aa == 0:
data["finish"] = "1"
resp = self.session.post(
f"https://{self.host}/user/work/submit",
data=data,
headers=headers,
)
break
if resp.status_code == 200 and resp.json().get("status", False):
self.log(
f'提交答案: {resp.json().get("msg", "")} {answerID} => {answer} {final}'
f'{index}/{len(self.qas)} 提交答案: {resp.json().get("msg", "")} {answerID} => {answer} {final}'
)
def exam_submit(
self, answerID: str, workID: str, answer: str, final: bool, index: int
):
headers = self.session.headers
headers["x-requested-with"] = "XMLHttpRequest"
data = {
"answerId": answerID,
"examId": workID,
"finish": "0",
}
if "," in answer:
answers = answer.split(",")
data["answer[]"] = answers
# for a in answers:
# data.setdefault("answer[]", []).append(a)
else:
if answer == "":
data["answer"] = "A"
data["answer"] = answer
resp = self.session.post(
f"https://{self.host}/user/exam/submit",
data=data,
headers=headers,
)
if not final:
aa = 1
while aa == 1:
print(f"考试链接: {self.start_url}")
aa = input("现在是最后一道，请你去检查后输入 0 后自动提交：")
if aa == 0:
data["finish"] = "1"
resp = self.session.post(
f"https://{self.host}/user/exam/submit",
data=data,
headers=headers,
)
break
if resp.status_code == 200 and resp.json().get("status", False):
self.log(
f'{index}/{len(self.qas)} 提交答案: {resp.json().get("msg", "")} {answerID} => {answer} {final}'
)
def get_course(self, _id: str) -> tuple[int, int]:
@@ -419,6 +537,21 @@ class CKWK:
if page < resp.json().get("pageInfo", {}).get("pageCount", 0):
self.get_study_record(course_id, page + 1)
def get_exam_record(self, course_id: str, page=1):
resp = self.session.get(
f"https://{self.host}/user/study_record/exam.json",
params={
"courseId": course_id,
"page": page,
"_": str(int(time.time() * 1000)),
},
)
if resp.status_code == 200 and resp.json().get("status", False):
self.log(f'{resp.json().get("msg", "")} page -> {page}')
self.exams.extend(resp.json().get("list", []))
if page < resp.json().get("pageInfo", {}).get("pageCount", 0):
self.get_exam_record(course_id, page + 1)
def log(self, *args, **kwargs) -> None:
exit_flag = False
time_str = datetime.now().strftime("[%Y-%m-%d %H:%M:%S]")
@@ -607,7 +740,7 @@ class CKWK:
f'{work.get("title", "")} 已阅 {work.get("finalScore", "")}'
)
continue
if not self.get_question_answer(
if not self.get_question_answer_work(
c.get("id", ""), work.get("nodeId", ""), work.get("id", "")
):
continue
@@ -620,6 +753,51 @@ class CKWK:
work.get("id", ""),
chooise,
int(q.get("index", 1)) < len(self.qas),
q.get("index", 1),
)
# break
break
def exam(self):
while self.login():
self.get_user()
thread = Thread(target=self.start_loop, daemon=True)
thread.start()
cs = self.user.get("courses", [])
print(
f'姓名: {self.user.get("name", "")}\n\t互评: {self.user.get("huping", 0)}\n\t乐学园: {self.user.get("lexueyuan", 0)}\n\t讨论主题: {self.user.get("taolunzhuti", 0)}\n\t学习时长: {self.user.get("study_time", "")}'
)
if len(cs) > 1:
for i, _c in enumerate(cs):
print(f'{i+1}: {_c.get("name", "")} {_c.get("progress", "")}')
cs_order = input("请输入刷课顺序(用英文逗号分隔,不需要耍的不用加): ")
cs_list = [cs[int(_c) - 1] for _c in cs_order.split(",")]
cs_list = cs
for c in cs_list:
self.get_exam_record(c.get("id", ""))
print(self.exams)
for exam in self.exams:
self.qas = []
if "线下阅" in exam.get("state"):
self.log(
f'{exam.get("title", "")} 已阅 {exam.get("finalScore", "")}'
)
continue
print(exam)
if not self.get_question_answer_exam(
c.get("id", ""), exam.get("nodeId", ""), exam.get("id", "")
):
continue
for q in self.qas:
# print(q)
chooise = self.ai.chat(q)
chooise = chooise if chooise else "A"
self.exam_submit(
q.get("id", ""),
exam.get("id", ""),
chooise,
int(q.get("index", 1)) < len(self.qas),
q.get("index", 1),
)
# break
break
@@ -655,4 +833,7 @@ if __name__ == "__main__":
# CKWK(f"{u}", f"{p}", host_server[c][0]).run()
# 刷作业，不需要用的时候可以直接注释
CKWK(f"{u}", f"{p}", host_server[c][0]).work()
# CKWK(f"{u}", f"{p}", host_server[c][0]).work()
# 考试，不需要用的时候可以直接注释
CKWK(f"{u}", f"{p}", host_server[c][0]).exam()