upload sources

This commit is contained in:
xtaodada 2022-06-01 19:06:17 +08:00
parent da73f20c0e
commit 9edb7e89bb
Signed by: xtaodada
GPG Key ID: 4CBB3F4FA8C85659
15 changed files with 1937 additions and 1 deletions

4
.gitignore vendored
View File

@ -150,5 +150,7 @@ cython_debug/
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/
.idea/
*.xlsx
config.py

3
config.gen.py Normal file
View File

@ -0,0 +1,3 @@
username = "111"
password = "xxx"
sid = "xxx"

90
defs/api_url.py Normal file
View File

@ -0,0 +1,90 @@
# -*- coding: utf-8 -*-
import time
class ApiUrl:
@staticmethod
def __get_ts() -> int:
return int(time.time() * 1000)
def id_api(self, xsid: str) -> str:
return f"http://www.cqooc.com/user/session?xsid={xsid}&ts={self.__get_ts()}"
def info_api(self) -> str:
return f"http://www.cqooc.com/account/session/api/profile/get?ts={self.__get_ts()}"
def get_nonce_api(self):
return f"http://www.cqooc.com/user/login?ts={self.__get_ts()}"
@staticmethod
def login_api(username: str, hash_str: str, nonce: str, cn: str) -> str:
return (
"http://www.cqooc.com/user/login"
+ f"?username={username}&password={hash_str}"
+ f"&nonce={nonce}&cnonce={cn}"
)
def course_api(self, ownerId: str, limit: int) -> str:
return (
"http://www.cqooc.com/json/mcs?sortby=id&reverse=true&del=2"
+ f"&courseType=2&ownerId={ownerId}&limit={limit}"
+ f"&ts={self.__get_ts()}"
)
def lessons_api(
self, course_id: str, start: int = 0, limit: int = 100
) -> str:
return (
"http://www.cqooc.com/json/mooc/lessons"
+ f"?limit={limit}&start={start}&sortby=selfId&reverse=false"
+ f"&courseId={course_id}&ts={self.__get_ts()}"
)
def lessons_status_api(
self, course_id: str, username: str, start: int = 0, limit: int = 100
) -> str:
return (
"http://www.cqooc.com/json/learnLogs"
+ f"?limit={limit}&start={start}&courseId={course_id}"
+ f"&select=sectionId&username={username}&ts={self.__get_ts()}"
)
def mcs_id_api(self, owner_id: str, course_id: str) -> str:
return (
"http://www.cqooc.com/json/mcs"
+ f"?ownerId={owner_id}&courseId={course_id}"
+ f"&ts={self.__get_ts()}"
)
def learn_log_api(self, section_id: str, username: str) -> str:
return (
"http://www.cqooc.com/json/learnLogs"
+ f"?sectionId={section_id}&username={username}"
+ f"&ts={self.__get_ts()}"
)
def exam_list_api(self, course_id: str) -> str:
return (
"http://www.cqooc.com/json/exam/papers"
+ f"?limit=200&start=1&courseId={course_id}&select=id,title,parentId,submitEnd"
f"&ts={self.__get_ts()}"
)
def task_list_api(self, course_id: str) -> str:
return (
"http://www.cqooc.com/json/tasks"
+ f"?limit=200&start=1&isPublish=1&courseId={course_id}&select=id,title,pubClass,submitEnd"
+ f"&ts={self.__get_ts()}"
)
def chapters_api(self, course_id: str) -> str:
return (
"http://www.cqooc.com/json/chapters"
+ f"?limit=200&start=1&isPublish=1&courseId={course_id}&select=id,title,parentId"
+ f"&ts={self.__get_ts()}"
)
@staticmethod
def skip_section_api() -> str:
return "http://www.cqooc.com/learnLog/api/add"

184
defs/core.py Normal file
View File

@ -0,0 +1,184 @@
# -*- coding: utf-8 -*-
from defs.request import Request
from defs.user import User
from defs.msg import Msg
from defs.test import test
from defs.processer import Processer
from defs.api_url import ApiUrl
class Core:
def __init__(self, username: str, pwd: str) -> None:
self.__processer = Processer()
self.__request = Request()
self.__api_url = ApiUrl()
self.__user = User(username, pwd)
def __process_user_info(self) -> None:
id_res = self.__request.do_get(
self.__api_url.id_api(self.__user.get_xsid())
)
id_data = id_res.json()
self.__user.set_id(id_data["id"])
info_res = self.__request.do_get(self.__api_url.info_api())
info_data = info_res.json()
self.__user.set_name(info_data["name"])
self.__user.set_avatar(
self.__request.get_host() + info_data["headimgurl"]
)
def login(self) -> dict:
api = self.__api_url.get_nonce_api()
nonce_res = self.__request.do_get(
api,
{
"Referer": "http://www.cqooc.com/login",
},
)
data = nonce_res.json()
cn = test.cnonce()
hash_str = test.getEncodePwd(
data["nonce"] + test.getEncodePwd(self.__user.get_pwd()) + cn
)
login_res = self.__request.do_post(
self.__api_url.login_api(
self.__user.get_username(), hash_str, data["nonce"], cn
),
headers={
"Referer": "http://www.cqooc.com/login",
},
)
data = login_res.json()
login_success = data["code"] == 0
if login_success:
self.__user.set_xsid(data["xsid"])
self.__request.set_headers("Cookie", f'xsid={data["xsid"]}')
self.__process_user_info()
return Msg().processing("登录成功", 200, data)
else:
return Msg().processing("登录失败,可能需要官网登录后重试", 400, data)
def login_use_sid(self, sid: str):
self.__user.set_xsid(sid)
self.__request.set_headers("Cookie", f'xsid={sid}')
self.__process_user_info()
return Msg().processing("设置成功", 200)
def get_user_info(self) -> dict:
return Msg().processing("登录成功", 200, self.__user.get_info())
def get_course(self, limit: int = 20) -> dict:
course_res = self.__request.do_get(
self.__api_url.course_api(str(self.__user.get_id()), limit),
headers={
"Referer": "http://www.cqooc.com/my/learn",
"Host": "www.cqooc.com",
},
)
course_data = self.__processer.process_course_data(course_res)
self.__user.set_course_data(course_data.copy())
return Msg().processing("获取课程成功", 200, self.__user.get_course_data())
def get_course_lessons(self, course_id: str) -> dict:
mcs_id_res = self.__request.do_get(
self.__api_url.mcs_id_api(str(self.__user.get_id()), course_id),
headers={
"Referer": "http://www.cqooc.com/my/learn",
"Host": "www.cqooc.com",
},
)
mcs_id_data = mcs_id_res.json()
self.__user.set_mcs_id(mcs_id_data["data"][0]["id"])
lessons_res = self.__request.do_get(
self.__api_url.lessons_api(course_id),
headers={
"Referer": "http://www.cqooc.com/learn"
+ f"/mooc/structure?id={course_id}",
"host": "www.cqooc.com",
},
)
lessons_status_res = self.__request.do_get(
self.__api_url.lessons_status_api(
course_id, self.__user.get_username()
),
headers={
"Referer": (
"http://www.cqooc.com/learn/mooc/progress"
+ f"?id={course_id}"
),
"host": "www.cqooc.com",
},
)
lessons_data = self.__processer.process_lessons_data(
self.__user.get_username(), lessons_res, lessons_status_res
)
self.__user.set_lessons_data(lessons_data.copy())
return Msg().processing(
"获取课程课时成功", 200, self.__user.get_lessons_data()
)
def skip_section(self, section_id: str) -> dict:
section_data = list(
filter(
lambda d: d["id"] == section_id,
self.__user.get_lessons_data()["data"],
)
)[0]
post_data = self.__processer.process_section_data(
section_data, self.__user.get_mcs_id()
)
skip_res = self.__request.do_post(
self.__api_url.skip_section_api(),
data=post_data,
headers={
"Referer": "http://www.cqooc.com/learn/mooc/structure?id="
+ section_data["courseId"],
"Host": "www.cqooc.com",
},
)
status_code = skip_res.json()["code"]
if status_code == 2:
return Msg().processing("已经跳过该课程", 200)
elif status_code == 0:
return Msg().processing("跳过课程成功", 200)
elif status_code == 1:
return Msg().processing("非法操作", 400)
else:
return Msg().processing("跳过课程失败", 400)
def get_exam_info(self, course_id: str) -> dict:
exam_list_res = self.__request.do_get(
self.__api_url.exam_list_api(course_id),
headers={
"Referer": "http://www.cqooc.com/my/learn",
"Host": "www.cqooc.com",
},
)
return Msg().processing(
"获取测验列表成功", 200, exam_list_res.json()
)
def get_task_info(self, course_id: str) -> dict:
task_list_res = self.__request.do_get(
self.__api_url.task_list_api(course_id),
headers={
"Referer": "http://www.cqooc.com/my/learn",
"Host": "www.cqooc.com",
},
)
return Msg().processing(
"获取作业列表成功", 200, task_list_res.json()
)
def get_chapters_info(self, course_id: str) -> dict:
chapter_list_res = self.__request.do_get(
self.__api_url.chapters_api(course_id),
headers={
"Referer": f"http://www.cqooc.com/learn/mooc/progress?id={course_id}",
"Host": "www.cqooc.com",
},
)
return Msg().processing(
"获取章节列表成功", 200, chapter_list_res.json()
)

44
defs/export.py Normal file
View File

@ -0,0 +1,44 @@
from typing import List, Optional
from openpyxl import Workbook, load_workbook
from os.path import isfile
from model.course import Course
from model.exam import Exam
from model.task import Task
def export_exam_list(course: Course, exam_list: List[Exam], task_list: List[Task]) -> None:
if isfile("exams.xlsx"):
wb = load_workbook('exams.xlsx')
else:
# 创建一个工作簿对象
wb = Workbook()
# 创建一个名为 课程名称 的 sheet 页
ws = wb.create_sheet(course.title)
if "Sheet" in wb.get_sheet_names():
wb.remove_sheet(wb["Sheet"])
# 标题
ws["A1"] = "课程名称"
ws["B1"] = "章节名称"
ws["C1"] = "测验名称"
ws["D1"] = "测验结束时间"
ws["E1"] = "测验结束时间戳"
# 写入 exam_list
for i, exam in enumerate(exam_list):
ws.cell(row=i + 2, column=1, value=course.title)
ws.cell(row=i + 2, column=2, value=exam.chapter)
ws.cell(row=i + 2, column=3, value=exam.title)
ws.cell(row=i + 2, column=4, value=exam.end_time)
ws.cell(row=i + 2, column=5, value=exam.submitEnd)
# 写入 task_list
for i, task in enumerate(task_list):
ws.cell(row=i + 2 + len(exam_list), column=1, value=course.title)
ws.cell(row=i + 2 + len(exam_list), column=2, value=task.chapter)
ws.cell(row=i + 2 + len(exam_list), column=3, value=task.title)
ws.cell(row=i + 2 + len(exam_list), column=4, value=task.end_time)
ws.cell(row=i + 2 + len(exam_list), column=5, value=task.submitEnd)
# 将创建的工作簿保存为 exams.xlsx
wb.save("exams.xlsx")
# 最后关闭文件
wb.close()

26
defs/msg.py Normal file
View File

@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
class Msg:
def __init__(self):
self.__sussess = {
"code": 200,
"status": "ok",
}
self.__fail = {
"code": 400,
"status": "fail",
}
def processing(self, msg: str, code: int, res=None) -> dict:
if res is None:
res = {}
if code == 200:
res["code"] = self.__sussess["code"]
res["status"] = self.__sussess["status"]
res["msg"] = msg
else:
res["code"] = self.__fail["code"]
res["status"] = self.__fail["status"]
res["msg"] = msg
res["data"] = None
return res

71
defs/processer.py Normal file
View File

@ -0,0 +1,71 @@
# -*- coding: utf-8 -*-
import json
import httpx
class Processer:
@staticmethod
def process_course_data(course_res: httpx.Response) -> dict:
res_course_data = json.loads(course_res.text)
course_data = {"meta": res_course_data["meta"], "data": []}
for course in res_course_data["data"]:
course_data["data"].append(
{
"courseId": course["courseId"],
"ownerId": course["ownerId"],
"title": course["course"]["title"],
}
)
return course_data
@staticmethod
def process_lessons_data(
username: str,
lessons_res: httpx.Response,
lessons_status_res: httpx.Response,
) -> dict:
lessons_res_data = json.loads(lessons_res.text)
lessons_status_res_data = json.loads(lessons_status_res.text)
lessons_data = {"meta": lessons_res_data["meta"], "data": []}
for lesson in lessons_res_data["data"]:
lessons_data["data"].append(
{
"title": lesson["title"],
"sectionId": lesson["id"],
"category": lesson["category"],
"chapterId": lesson["chapterId"],
"courseId": lesson["courseId"],
"ownerId": lesson["ownerId"],
"parentId": lesson["parentId"],
"id": lesson["id"],
"username": username,
}
)
# add status
lesson_status = [
i["sectionId"] for i in lessons_status_res_data["data"]
]
for lesson in lessons_data["data"]:
if lesson["sectionId"] in lesson_status:
lesson["status"] = 1
else:
lesson["status"] = 0
# sort by sectionId
lessons_data["data"] = sorted(
lessons_data["data"], key=lambda x: x["sectionId"]
)
return lessons_data
@staticmethod
def process_section_data(section_data: dict, mcs_id: str) -> dict:
post_data = {}
post_data["action"] = 0
post_data["category"] = 2
post_data["chapterId"] = section_data["chapterId"]
post_data["courseId"] = section_data["courseId"]
post_data["ownerId"] = int(section_data["ownerId"])
post_data["parentId"] = mcs_id
post_data["sectionId"] = section_data["sectionId"]
post_data["username"] = section_data["username"]
return post_data

65
defs/request.py Normal file
View File

@ -0,0 +1,65 @@
# -*- coding: utf-8 -*-
import httpx
class Request:
__host = "http://www.cqooc.com"
def __init__(self):
self.__headers = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)"
+ " AppleWebKit/537.36 (KHTML, like Gecko)"
+ " Chrome/101.0.0.0 Safari/537.36",
}
self.__proxies = {}
def set_headers(self, name: str, value: str) -> None:
self.__headers[name] = value
def del_headers(self, name: str) -> None:
del self.__headers[name]
def get_headers(self) -> dict:
return self.__headers
def set_proxies(self, name: str, value: str) -> None:
self.__proxies[name] = value
def get_proxies(self) -> dict:
return self.__proxies
def get_host(self) -> str:
return self.__host
def do_get(
self, url: str, headers: dict = None, proxies: dict = None
) -> httpx.Response:
self_headers = self.__headers.copy()
if headers is not None:
for key in headers:
self_headers[key] = headers[key]
self_proxies = self.__proxies.copy()
if proxies is not None:
for key in proxies:
self_proxies[key] = proxies[key]
return httpx.get(url, headers=self_headers, proxies=self_proxies)
def do_post(
self,
url: str,
data: dict = None,
headers: dict = None,
proxies: dict = None,
) -> httpx.Response:
self_headers = self.__headers.copy()
if headers is not None:
for key in headers:
self_headers[key] = headers[key]
self_proxies = self.__proxies.copy()
if proxies is not None:
for key in proxies:
self_proxies[key] = proxies[key]
return httpx.post(
url, data=data, headers=self_headers, proxies=self_proxies
)

1207
defs/test.py Normal file

File diff suppressed because it is too large Load Diff

83
defs/user.py Normal file
View File

@ -0,0 +1,83 @@
# -*- coding: utf-8 -*-
class User:
__xsid = None
__id = None
__username = None
__pwd = None
__name = None
__avatar = None
__course_data = None
__lessons_data = None
__mcs_id = None
def __init__(self, username: str, pwd: str) -> None:
self.__username = username
self.__pwd = pwd
def set_xsid(self, xsid: str) -> None:
self.__xsid = xsid
def get_xsid(self) -> str:
return self.__xsid
def set_id(self, id_: int) -> None:
self.__id = id_
def get_id(self) -> int:
return self.__id
def set_username(self, username: str) -> None:
self.__username = username
def get_username(self) -> str:
return self.__username
def set_pwd(self, pwd: str) -> None:
self.__pwd = pwd
def get_pwd(self) -> str:
return self.__pwd
def set_name(self, name: str) -> None:
self.__name = name
def get_name(self) -> str:
return self.__name
def set_avatar(self, avatar: str) -> None:
self.__avatar = avatar
def get_avatar(self) -> str:
return self.__avatar
def set_course_data(self, course_data: dict) -> None:
self.__course_data = course_data
def get_course_data(self) -> dict:
return self.__course_data
def set_lessons_data(self, lessons_data: dict) -> None:
self.__lessons_data = lessons_data
def get_lessons_data(self) -> dict:
return self.__lessons_data
def set_mcs_id(self, mcs_id: str) -> None:
self.__mcs_id = mcs_id
def get_mcs_id(self) -> str:
return self.__mcs_id
def get_info(self) -> dict:
return {
"xsid": self.__xsid,
"id": self.__id,
"username": self.__username,
"pwd": self.__pwd,
"name": self.__name,
"avatar": self.__avatar,
"mcs_id": self.__mcs_id,
"course_data": self.__course_data,
"lessons_data": self.__lessons_data,
}

58
main.py Normal file
View File

@ -0,0 +1,58 @@
from coloredlogs import ColoredFormatter
from logging import getLogger, StreamHandler, CRITICAL, INFO, basicConfig
from sys import exit
from defs.core import Core
from defs.export import export_exam_list
from model.course import get_course_list, print_course_list
from model.exam import get_exam_list
from model.task import get_task_list
from config import username, password, sid
logs = getLogger(__name__)
logging_format = "%(levelname)s [%(asctime)s] [%(name)s] %(message)s"
logging_handler = StreamHandler()
logging_handler.setFormatter(ColoredFormatter(logging_format))
root_logger = getLogger()
root_logger.setLevel(CRITICAL)
root_logger.addHandler(logging_handler)
basicConfig(level=INFO)
logs.setLevel(INFO)
core = Core(username, password)
core.login_use_sid(sid)
# 获取课程列表
courses = core.get_course()
if courses.get("code", 0) != 200:
logs.error("获取课程列表失败")
exit(1)
course_list = get_course_list(courses.get("data", []))
print_course_list(course_list)
if not course_list:
logs.warning("无课程")
exit(0)
# 输入需要输出的课程
while 1:
try:
course_id = input("请输入课程ID")
except KeyboardInterrupt:
break
course = None
for i in course_list:
if course_id == i.course_id:
course = i
break
if not course:
logs.error("课程 id 错误,请重新输入")
else:
# 获取章节列表
chapters = core.get_chapters_info(course_id)
# 获取测验列表
exams = core.get_exam_info(course_id)
exam_list = get_exam_list(exams.get("data", []), chapters.get("data", []))
tasks = core.get_task_info(course_id)
task_list = get_task_list(tasks.get("data", []))
# 导出
export_exam_list(course, exam_list, task_list)
logs.info(f"{course.title} | 导出成功")

28
model/course.py Normal file
View File

@ -0,0 +1,28 @@
from typing import List
class Course:
course_id: str = "0"
title: str = ""
owner_id: str = "0"
def __init__(self, data: dict):
self.course_id = data.get("courseId", "0")
self.title = data.get("title", "")
self.owner_id = data.get("ownerId", "0")
def get_course_list(courses: List[dict]) -> List[Course]:
course_list = []
for course in courses:
try:
course_list.append(Course(course))
except Exception as e:
print(e)
continue
return course_list
def print_course_list(course_list: List[Course]) -> None:
for course in course_list:
print(f"{course.title} | {course.course_id}")

42
model/exam.py Normal file
View File

@ -0,0 +1,42 @@
import traceback
from datetime import datetime
from typing import List
class Exam:
id: str = "0"
chapter: str = ""
title: str = ""
parentId: str = "0"
submitEnd: int = 0
end_time: str = ""
def __init__(self, data: dict):
self.id = data.get("id", "0")
self.title = data.get("title", "")
self.parentId = data.get("parentId", "0")
self.submitEnd = data.get("submitEnd", 0) / 1000
self.end_time = datetime.strftime(datetime.fromtimestamp(self.submitEnd), '%Y-%m-%d %H:%M:%S')
def get_title(chapter_id: str, chapters: List[dict]) -> str:
for chapter in chapters:
if chapter.get("id", "0") == chapter_id and chapter.get("title", None) is not None:
if chapter.get("parentId", None) is None:
return chapter.get("title", "")
else:
return get_title(chapter.get("parentId"), chapters)
return ""
def get_exam_list(exams: List[dict], chapters: List[dict]) -> List[Exam]:
exam_list = []
for exam in exams:
try:
exam_ = Exam(exam)
exam_.chapter = get_title(exam_.parentId, chapters)
exam_list.append(exam_)
except Exception as e: # noqa
print(traceback.format_exc())
continue
return exam_list

29
model/task.py Normal file
View File

@ -0,0 +1,29 @@
import traceback
from datetime import datetime
from typing import List
class Task:
id: str = "0"
title: str = ""
chapter: str = ""
submitEnd: int = 0
end_time: str = ""
def __init__(self, data: dict):
self.id = data.get("id", "0")
self.chapter = data.get("chapter", {}).get("title", "")
self.title = data.get("title", "")
self.submitEnd = data.get("submitEnd", 0) / 1000
self.end_time = datetime.strftime(datetime.fromtimestamp(self.submitEnd), '%Y-%m-%d %H:%M:%S')
def get_task_list(tasks: List[dict], ) -> List[Task]:
task_list = []
for task in tasks:
try:
task_list.append(Task(task))
except Exception as e: # noqa
print(traceback.format_exc())
continue
return task_list

4
requirements.txt Normal file
View File

@ -0,0 +1,4 @@
httpx==0.23.0
coloredlogs==15.0.1
Js2Py==0.71
openpyxl==3.0.10