diff --git a/downloader.py b/downloader.py index 6d201d9..5f62ee5 100644 --- a/downloader.py +++ b/downloader.py @@ -1,113 +1,234 @@ +import dataclasses import os -from tqdm import tqdm + +import tqdm +import tqdm.contrib +import tqdm.contrib.concurrent + +import thu_learn_lib +import thu_learn_lib.ty +import thu_learn_lib.utils -from thu_learn_lib import LearnHelper -from thu_learn_lib import ty as types -from thu_learn_lib.utils import slugify +@dataclasses.dataclass +class DownloadTask: + session: "Downloader" + url: str + filename: str + prefix: str = "." -class Downloader(LearnHelper): - prefix: str = "" - file_size_limit: int = None # MB +class Downloader: + helper: thu_learn_lib.LearnHelper + prefix: str + file_size_limit: float = None # MB sync_docs: bool = True sync_work: bool = True sync_submit: bool = True + download_tasks: list[DownloadTask] = None def __init__( self, - prefix: str = "", - file_size_limit: int = None, + username: str, + password: str, + prefix: str = "thu-learn", + file_size_limit: float = None, sync_docs: bool = True, sync_work: bool = True, sync_submit: bool = True, ) -> None: - super().__init__() - if prefix: - self.prefix = prefix - else: - self.prefix = os.path.join(os.getcwd(), slugify("learn")) + self.helper = thu_learn_lib.LearnHelper( + username=username, + password=password, + ) + self.prefix = prefix self.file_size_limit = file_size_limit self.sync_docs = sync_docs self.sync_work = sync_work self.sync_submit = sync_submit - def Download(self, url: str, prefix: str, filename: str) -> bool: - os.makedirs(prefix, exist_ok=True) - response = self.get(url=url, stream=True) + assert self.helper.login() + + @staticmethod + def download( + self: "Downloader", + url: str, + filename: str, + prefix: str = ".", + position: int = 0, + ) -> bool: + response = self.helper.get(url=url, stream=True) file_size = int(response.headers.get("content-length", 0)) if self.file_size_limit: if file_size > self.file_size_limit * 1024 * 1024: - print(f"Skipping file {filename}") + print(f"Skip file {filename}") return False - filename = slugify(filename) + filename = thu_learn_lib.utils.slugify(filename) + path = os.path.join(prefix, filename) + if os.path.exists(path): + if os.path.getsize(path) == file_size: + # print(f"file {filename} is already synced") + return True + os.makedirs(prefix, exist_ok=True) chunk_size = 8192 # 8KB - with tqdm( - desc=filename, - total=file_size, - unit="B", - ascii=True, - unit_scale=True, - dynamic_ncols=True, - ) as bar: - with open(os.path.join(prefix, filename), "wb") as file: - for chunck in response.iter_content(chunk_size): - file.write(chunck) - bar.update(len(chunck)) + try: + with open( + file=os.path.join(prefix, filename), + mode="wb", + ) as file: + list( + tqdm.contrib.tmap( + file.write, + response.iter_content(chunk_size), + desc=f"{position - 6} {filename}", + total=file_size, + leave=False, + unit="B", + unit_scale=True, + dynamic_ncols=True, + position=position, + ) + ) + pass + except KeyboardInterrupt: + raise KeyboardInterrupt() + except: + return False return True - def SyncSemester( - self, semester_id: str, course_type: types.CourseType = types.CourseType.STUDENT + @staticmethod + def retry_download( + task: DownloadTask, + position: int = 0, + max_retries: int = 5, ) -> bool: - print(f"Syncing Semester {semester_id} ......") - course_list = self.get_course_list( - semester_id=semester_id, course_type=course_type - ) - for course in course_list: - self.SyncCourse( - course=course, semester_id=semester_id, - ) + for i in range(max_retries): + if Downloader.download( + self=task.session, + url=task.url, + filename=task.filename, + prefix=task.prefix, + position=position, + ): + return True + print(f"Failed to download file {task.filename}") + return False - def SyncCourse(self, course: types.CourseInfo, semester_id: str) -> bool: - file_list = self.get_file_list( - course_id=course.id, course_type=course.course_type + def schedule_download( + self, + url: str, + filename: str, + prefix: str = ".", + ) -> None: + if not self.download_tasks: + self.download_tasks = [] + self.download_tasks.append( + DownloadTask( + session=self, + url=url, + filename=filename, + prefix=prefix, + ) ) - print( - f"Syncing Course {course.course_number} {course.name} {course.english_name} ......" + + def finish_download(self, desc: str = "download") -> bool: + if self.download_tasks: + success = all( + tqdm.contrib.concurrent.process_map( + Downloader.retry_download, + self.download_tasks, + range(6, 6 + len(self.download_tasks)), + desc=desc, + leave=False, + dynamic_ncols=True, + position=4, + ) + ) + self.download_tasks.clear() + return success + else: + return True + + def sync_semester( + self, + semester_id: str, + course_type: thu_learn_lib.ty.CourseType = thu_learn_lib.ty.CourseType.STUDENT, + ) -> bool: + course_list = self.helper.get_course_list( + semester_id=semester_id, + course_type=course_type, ) + for course in tqdm.tqdm( + iterable=course_list, + desc=semester_id, + leave=False, + dynamic_ncols=True, + position=2, + ): + self.sync_course(course=course, semester_id=semester_id) + + def sync_course( + self, + course: thu_learn_lib.ty.CourseInfo, + semester_id: str, + ) -> bool: + file_list = self.helper.get_file_list( + course_id=course.id, + course_type=course.course_type, + ) + # print( + # f"Syncing Course {course.course_number} {course.name} {course.english_name} ......" + # ) + if self.sync_docs: + pass if self.sync_docs: for file in file_list: - self.SyncFile(file, semester_id=semester_id, course=course) + self.sync_file(file, semester_id=semester_id, course=course) + self.finish_download(desc=course.english_name) if self.sync_work: - homework_list = self.get_homework_list(course_id=course.id) + homework_list = self.helper.get_homework_list(course_id=course.id) for homework in homework_list: - self.SyncHomework( + self.sync_homework( homework=homework, semester_id=semester_id, course=course ) + self.finish_download(desc=course.english_name) - def SyncFile( - self, file: types.File, semester_id: str, course: types.CourseInfo + def sync_file( + self, + file: thu_learn_lib.ty.File, + semester_id: str, + course: thu_learn_lib.ty.CourseInfo, ) -> bool: prefix = os.path.join( self.prefix, - slugify(course.english_name), - slugify("documents"), - slugify(file.clazz), + thu_learn_lib.utils.slugify(course.english_name), + thu_learn_lib.utils.slugify("documents"), + thu_learn_lib.utils.slugify(file.clazz), ) - filename = slugify(file.title) + slugify( - f".{file.file_type}" if file.file_type else "" + filename = ( + thu_learn_lib.utils.slugify(file.title) + + f".{thu_learn_lib.utils.slugify(file.file_type)}" + if file.file_type + else "" ) - self.Download(url=file.download_url, prefix=prefix, filename=filename) + self.schedule_download( + url=file.download_url, + filename=filename, + prefix=prefix, + ) return True - def SyncHomework( - self, homework: types.Homework, semester_id: str, course: types.CourseInfo + def sync_homework( + self, + homework: thu_learn_lib.ty.Homework, + semester_id: str, + course: thu_learn_lib.ty.CourseInfo, ) -> bool: prefix = os.path.join( - self.prefix, - slugify(course.english_name), - slugify("work"), - slugify(homework.title), + thu_learn_lib.utils.slugify(self.prefix), + thu_learn_lib.utils.slugify(course.english_name), + thu_learn_lib.utils.slugify("work"), + thu_learn_lib.utils.slugify(homework.title), ) os.makedirs(prefix, exist_ok=True) lines = [] @@ -122,11 +243,13 @@ lines.append(f"{homework.description}") lines.append(f"") if homework.attachment: - filename = slugify( + filename = thu_learn_lib.utils.slugify( f"attach-{homework.title}{os.path.splitext(homework.attachment.name)[-1]}" ) - self.Download( - url=homework.attachment.download_url, prefix=prefix, filename=filename, + self.schedule_download( + url=homework.attachment.download_url, + prefix=prefix, + filename=filename, ) lines.append(f"### Attach.") lines.append(f"") @@ -137,10 +260,10 @@ lines.append(f"{homework.answer_content}") lines.append(f"") if homework.answer_attachment: - filename = slugify( + filename = thu_learn_lib.utils.slugify( f"ans-{homework.title}{os.path.splitext(homework.answer_attachment.name)[-1]}" ) - self.Download( + self.schedule_download( url=homework.answer_attachment.download_url, prefix=prefix, filename=filename, @@ -161,10 +284,10 @@ lines.append(f"{homework.submitted_content}") lines.append(f"") if homework.submitted_attachment: - filename = slugify( + filename = thu_learn_lib.utils.slugify( f"submit-{homework.title}{os.path.splitext(homework.submitted_attachment.name)[-1]}" ) - self.Download( + self.schedule_download( url=homework.submitted_attachment.download_url, prefix=prefix, filename=filename, @@ -194,10 +317,10 @@ lines.append(f"{homework.grade_content}") lines.append(f"") if homework.grade_attachment: - filename = slugify( + filename = thu_learn_lib.utils.slugify( f"comment-{homework.title}{os.path.splitext(homework.grade_attachment.name)[-1]}" ) - self.Download( + self.schedule_download( url=homework.grade_attachment.download_url, prefix=prefix, filename=filename, @@ -207,6 +330,6 @@ lines.append(f"[{homework.grade_attachment.name}]({filename})") lines.append(f"") lines = [line + "\n" for line in lines] - filename = slugify("README.md") + filename = thu_learn_lib.utils.slugify("README.md") with open(os.path.join(prefix, filename), "w") as file: file.writelines(lines) diff --git a/main.py b/main.py index f4e7ea2..c0df3a2 100644 --- a/main.py +++ b/main.py @@ -1,5 +1,6 @@ import argparse +import tqdm from downloader import Downloader @@ -18,7 +19,7 @@ ) parser.add_argument( "--prefix", - default=None, + default="thu-learn", required=False, help='location to save downloaded files, default to "learn/"', ) @@ -54,18 +55,25 @@ def main(args: argparse.Namespace): downloader = Downloader( + username=args.username, + password=args.password, prefix=args.prefix, file_size_limit=args.file_size_limit, sync_docs=(not args.no_sync_docs), sync_work=(not args.no_sync_work), sync_submit=(not args.no_sync_submit), ) - downloader.login(username=args.username, password=args.password) - semester_id_list = downloader.get_semester_id_list() - semesters = args.semesters if args.semesters else semester_id_list - for semester in semesters: + semester_id_list: list[str] = downloader.helper.get_semester_id_list() + semesters: list[str] = args.semesters if args.semesters else semester_id_list + for semester in tqdm.tqdm( + iterable=semesters, + desc="semesters", + leave=False, + dynamic_ncols=True, + position=0, + ): if semester in semester_id_list: - downloader.SyncSemester(semester_id=semester) + downloader.sync_semester(semester_id=semester) else: print(f"{semester} not found") diff --git a/thu_learn_lib/helper.py b/thu_learn_lib/helper.py index b78d92e..ce7b882 100644 --- a/thu_learn_lib/helper.py +++ b/thu_learn_lib/helper.py @@ -1,81 +1,93 @@ -import urllib -import bs4 import datetime -import re +import typing +import urllib.parse + +import bs4 import requests +import requests.adapters - -from . import ty as types +from . import ty from . import urls +from . import parser class LearnHelper(requests.Session): - def __init__(self) -> None: + username: str + password: str + + def __init__(self, username: str, password: str) -> None: super().__init__() + self.username = username + self.password = password + + self.mount( + prefix="https://", + adapter=requests.adapters.HTTPAdapter( + max_retries=5, + ), + ) @property def token(self) -> str: - return self.cookies["XSRF-TOKEN"] if "XSRF-TOKEN" in self.cookies else None + return self.cookies.get(name="XSRF-TOKEN", default="") - def get_with_token(self, url: str, params: dict = None) -> requests.Response: - if params is None: - params = {"_csrf": self.token} + def get_with_token(self, url: ty.URL) -> requests.Response: + if url.query: + url.query["_csrf"] = self.token else: - params["_csrf"] = self.token - return self.get(url=url, params=params) + url.query = { + "_csrf": self.token, + } + return self.get(url) - def login(self, username: str, password: str) -> bool: + def login(self) -> bool: response = self.get(urls.LEARN_PREFIX) soup = bs4.BeautifulSoup(response.text, features="html.parser") form = soup.find( name="form", attrs={"class": "w", "id": "loginForm", "method": "post"} ) - payload = urls.id_login_form_data(username=username, password=password).params + payload = urls.id_login_form_data( + username=self.username, + password=self.password, + ).query response = self.post(url=form["action"], data=payload) soup = bs4.BeautifulSoup(response.text, features="html.parser") a = soup.find(name="a") - pattern = rf"{urls.LEARN_PREFIX}/f/login.do\?status=(\w*)&ticket=(\w*)" - match = re.match(pattern=pattern, string=a["href"]) - self.success = match.group(1) == "SUCCESS" - self.ticket = match.group(2) + href = a["href"] + parse_result: urllib.parse.ParseResult = urllib.parse.urlparse(url=href) + query = urllib.parse.parse_qs(qs=parse_result.query) + self.status = query["status"][0] + self.ticket = query["ticket"][0] response = self.get(a["href"]) - soup = bs4.BeautifulSoup(response.text, features="html.parser") - script = soup.find(name="script", type="text/javaScript") - pattern = r"\s*window.location=\"/b/j_spring_security_thauth_roaming_entry\?ticket=(\w*)\";\s*" - match = re.match(pattern=pattern, string=script.text) - self.ticket = match.group(1) - request = urls.learn_auth_roam(ticket=self.ticket) - response = self.get(url=request.url, params=request.params) + url = urls.learn_auth_roam(ticket=self.ticket) + response = self.get(url) - request = urls.learn_student_course_list_page() - response = self.get(url=request.url, params=request.params) + url = urls.learn_student_course_list_page() + response = self.get(url) - print( - f"User {username} login successfully" - if self.success - else f"User {username} login failed!" - ) - return self.success + print(f"User {self.username} login {self.status}!") + + return self.status == "SUCCESS" def logout(self) -> None: - request = urls.learn_logout() - response = self.post(url=request.url) + url = urls.learn_logout() + response = self.post(url) - def get_semester_id_list(self) -> list: - request = urls.learn_semester_list() - response = self.get_with_token(url=request.url, params=request.params) + def get_semester_id_list(self) -> list[str]: + url = urls.learn_semester_list() + response = self.get_with_token(url) json = response.json() - return [x for x in json if x] + return list(filter(bool, json)) - def get_current_semester(self) -> types.SemesterInfo: - request = urls.learn_current_semester() - response = self.get_with_token(url=request.url, params=request.params) + def get_current_semester(self) -> ty.SemesterInfo: + url = urls.learn_current_semester() + response = self.get_with_token(url) json = response.json() result = json["result"] - return types.SemesterInfo( + return ty.SemesterInfo( id=result["id"], start_date=datetime.datetime.strptime(result["kssj"], r"%Y-%m-%d").date(), end_date=datetime.datetime.strptime(result["jssj"], r"%Y-%m-%d").date(), @@ -85,129 +97,99 @@ ) def get_course_list( - self, semester_id: str, course_type: types.CourseType = types.CourseType.STUDENT - ) -> list[types.CourseInfo]: - request = urls.learn_course_list(semester_id, course_type) - response = self.get_with_token(url=request.url, params=request.params) + self, + semester_id: str, + course_type: ty.CourseType = ty.CourseType.STUDENT, + ) -> list[ty.CourseInfo]: + url = urls.learn_course_list(semester_id, course_type) + response = self.get_with_token(url) json = response.json() result = json["resultList"] - - def mapper(course: dict): - request = urls.learn_course_time_location(course["wlkcid"]) - response = self.get_with_token(request.url, request.params) - return types.CourseInfo( - id=course["wlkcid"], - name=course["kcm"], - english_name=course["ywkcm"], - time_and_location=response.json(), - url=str(urls.learn_course_url(course["wlkcid"], course_type)), - teacher_name=course["jsm"] - if course["jsm"] - else "", # teacher can not fetch this - teacher_number=course["jsh"], - course_number=course["kch"], - course_index=int( - course["kxh"] - ), # course["kxh"] could be string (teacher mode) or number (student mode) - course_type=course_type, + courses = list( + map( + parser.parse_course_info, + result, + [course_type] * len(result), ) - - courses = list(map(mapper, result)) + ) return courses def get_file_list( - self, course_id: str, course_type: types.CourseType = types.CourseType.STUDENT - ) -> list[types.File]: - request = urls.learn_file_classify(course_id=course_id) - response = self.get_with_token(url=request.url, params=request.params) + self, + course_id: str, + course_type: ty.CourseType = ty.CourseType.STUDENT, + ) -> list[ty.File]: + url = urls.learn_file_classify( + course_id=course_id, + ) + response = self.get_with_token(url) json = response.json() records = json["object"]["rows"] clazz = {} for record in records: clazz[record["kjflid"]] = record["bt"] # 课件分类 ID, 标题 - request = urls.learn_file_list(course_id=course_id, course_type=course_type) - response = self.get_with_token(request.url, request.params) + url = urls.learn_file_list( + course_id=course_id, + course_type=course_type, + ) + response = self.get_with_token(url) json = response.json() result = [] - if course_type == types.CourseType.STUDENT: + if course_type == ty.CourseType.STUDENT: result = json["object"] else: # teacher result = json["object"]["resultList"] - def mapper(file: dict) -> types.File: - title: str = file["bt"] - download_url: str = urls.learn_file_download( - file_id=file["wjid"] - if course_type == types.CourseType.STUDENT - else file["id"], - course_type=course_type, - course_id=course_id, + files = list( + map( + parser.parse_file, + result, + [course_id] * len(result), + [clazz] * len(result), + [course_type] * len(result), ) - preview_url = None - return types.File( - id=file["wjid"], # 文件 ID - title=file["bt"], # 标题 - description=file["ms"], # 描述 - raw_size=file["wjdx"], # 文件大小 - size=file["fileSize"], - upload_time=datetime.datetime.strptime( - file["scsj"], r"%Y-%m-%d %H:%M" - ), # 上传时间 - download_url=str(download_url), - preview_url=str(preview_url), - is_new=file["isNew"], - marked_important=file["sfqd"] == 1, # 是否强调 - visit_count=file["llcs"] or 0, # 浏览次数 - download_count=file["xzcs"] or 0, # 下载次数 - file_type=file["wjlx"], # 文件类型 - remote_file=types.RemoteFile( - id=file["wjid"], # 文件 ID - name=title, - download_url=str(download_url), - preview_url=str(preview_url), - size=file["fileSize"], - ), - clazz=clazz[file["kjflid"]], # 课件分类 ID - ) - - return list(map(mapper, result)) + ) + return files def get_homework_list( - self, course_id: str, course_type: types.CourseType = types.CourseType.STUDENT - ) -> list[types.Homework]: + self, course_id: str, course_type: ty.CourseType = ty.CourseType.STUDENT + ) -> list[ty.Homework]: result = [] - request = urls.learn_homework_list_new(course_id=course_id) + url = urls.learn_homework_list_new(course_id=course_id) result += self.get_homework_list_at_url( - request=request, status=types.HomeworkStatus(submitted=False, graded=False) + url=url, + status=ty.HomeworkStatus(submitted=False, graded=False), ) - request = urls.learn_homework_list_submitted(course_id=course_id) + url = urls.learn_homework_list_submitted(course_id=course_id) result += self.get_homework_list_at_url( - request=request, status=types.HomeworkStatus(submitted=True, graded=False) + url=url, + status=ty.HomeworkStatus(submitted=True, graded=False), ) - request = urls.learn_homework_list_graded(course_id=course_id) + url = urls.learn_homework_list_graded(course_id=course_id) result += self.get_homework_list_at_url( - request=request, status=types.HomeworkStatus(submitted=True, graded=True) + url=url, + status=ty.HomeworkStatus(submitted=True, graded=True), ) return result def get_homework_list_at_url( - self, request: urls.URL, status: types.HomeworkStatus - ) -> list[types.Homework]: - response = self.get_with_token(url=request.url, params=request.params) + self, url: ty.URL, status: ty.HomeworkStatus + ) -> list[ty.Homework]: + response = self.get_with_token(url) json = response.json() result = json["object"]["aaData"] - def mapper(work: dict) -> types.Homework: + def mapper(work: dict) -> ty.Homework: detail = self.parse_homework_detail( course_id=work["wlkcid"], # 课程 ID homework_id=work["zyid"], # 作业 ID student_homework_id=work["xszyid"], # 学生作业 ID ) - return types.Homework( + return ty.Homework( id=work["zyid"], # 作业 ID student_homework_id=work["xszyid"], # 学生作业 ID title=work["bt"], # 标题 @@ -248,76 +230,80 @@ def parse_homework_detail( self, course_id: str, homework_id: str, student_homework_id: str - ) -> types.HomeworkDetail: - request = urls.learn_homework_detail( + ) -> ty.HomeworkDetail: + url = urls.learn_homework_detail( course_id=course_id, homework_id=homework_id, student_homework_id=student_homework_id, ) - response = self.get_with_token(request.url, request.params) + response = self.get_with_token(url) text = response.text soup = bs4.BeautifulSoup(markup=text, features="html.parser") - div_list_calendar_clearfix = soup.find_all( + div_list_calendar_clearfix: typing.Iterable[bs4.element.Tag] = soup.find_all( name="div", attrs={"class": "list calendar clearfix"} ) - div_fl_right = sum( + div_fl_right: list[bs4.element.Tag] = sum( [ - div.find_all(name="div", attrs={"class": "fl right"}) + div.find_all( + name="div", + attrs={ + "class": "fl right", + }, + ) for div in div_list_calendar_clearfix ], [], ) - div_c55 = sum( - [div.find_all(name="div", attrs={"class": "c55"}) for div in div_fl_right], + div_c55: list[bs4.element.Tag] = sum( + [ + div.find_all( + name="div", + attrs={ + "class": "c55", + }, + ) + for div in div_fl_right + ], [], ) - description = div_c55[0].getText() - answer_content = div_c55[1].getText() - div_box = soup.find_all(name="div", attrs={"class": "boxbox"}) - div_box = div_box[1] - div_right = div_box.find_all(name="div", attrs={"class": "right"}) - submitted_content = div_right = div_right[2].getText() - - div_list_fujian_clearfix = soup.find_all( - name="div", attrs={"class": "list fujian clearfix"} + description: str = div_c55[0].getText() + answer_content: str = div_c55[1].getText() + div_box: typing.Iterable[bs4.element.Tag] = soup.find_all( + name="div", + attrs={ + "class": "boxbox", + }, ) - return types.HomeworkDetail( + div_box: bs4.element.Tag = div_box[1] + div_right: typing.Iterable[bs4.element.Tag] = div_box.find_all( + name="div", + attrs={ + "class": "right", + }, + ) + submitted_content: str = div_right[2].getText() + + div_list_fujian_clearfix: typing.Iterable[bs4.element.Tag] = soup.find_all( + name="div", + attrs={ + "class": "list fujian clearfix", + }, + ) + return ty.HomeworkDetail( description=description.strip(), answer_content=answer_content.strip(), submitted_content=submitted_content.strip(), - attachment=self.parse_homework_file(div_list_fujian_clearfix[0]) + attachment=parser.parse_homework_file(div_list_fujian_clearfix[0]) if len(div_list_fujian_clearfix) > 0 else None, - answer_attachment=self.parse_homework_file(div_list_fujian_clearfix[1]) + answer_attachment=parser.parse_homework_file(div_list_fujian_clearfix[1]) if len(div_list_fujian_clearfix) > 1 else None, - submitted_attachment=self.parse_homework_file(div_list_fujian_clearfix[2]) + submitted_attachment=parser.parse_homework_file(div_list_fujian_clearfix[2]) if len(div_list_fujian_clearfix) > 2 else None, - grade_attachment=self.parse_homework_file(div_list_fujian_clearfix[3]) + grade_attachment=parser.parse_homework_file(div_list_fujian_clearfix[3]) if len(div_list_fujian_clearfix) > 3 else None, ) - - def parse_homework_file(self, div) -> types.RemoteFile: - ftitle = div.find(name="span", attrs={"class": "ftitle"}) or div.find( - name="span", attrs={"class", "ft"} - ) - if not ftitle: - return None - a = ftitle.find(name="a") - size = div.find(name="span", attrs={"class": "color_999"}) - size = size.getText() - params = dict(urllib.parse.parse_qsl(urllib.parse.urlsplit(a["href"]).query)) - attachment_id = params["fileId"] - download_url = urls.LEARN_PREFIX + ( - params["downloadUrl"] if "downloadUrl" in params else a["href"] - ) - return types.RemoteFile( - id=attachment_id, - name=a.getText(), - download_url=download_url, - preview_url=None, - size=size.strip(), - ) diff --git a/thu_learn_lib/parser.py b/thu_learn_lib/parser.py new file mode 100644 index 0000000..5bbd8fe --- /dev/null +++ b/thu_learn_lib/parser.py @@ -0,0 +1,107 @@ +import datetime +import urllib.parse + +import bs4 + +from . import ty +from . import urls + + +def parse_course_info( + course: dict, + course_type: ty.CourseType = ty.CourseType.STUDENT, +): + # url = urls.learn_course_time_location(course["wlkcid"]) + # response = self.get_with_token(url) + return ty.CourseInfo( + id=course["wlkcid"], + name=course["kcm"], + english_name=course["ywkcm"], + # time_and_location=response.json(), + url=str(urls.learn_course_url(course["wlkcid"], course_type)), + teacher_name=course["jsm"] + if course["jsm"] + else "", # teacher can not fetch this + teacher_number=course["jsh"], + course_number=course["kch"], + course_index=int( + course["kxh"] + ), # course["kxh"] could be string (teacher mode) or number (student mode) + course_type=course_type, + ) + + +def parse_file( + file: dict, + course_id: str, + clazz: dict[str, str], + course_type: ty.CourseType = ty.CourseType.STUDENT, +) -> ty.File: + title: str = file["bt"] + download_url: str = urls.learn_file_download( + file_id=file["wjid"] if course_type == ty.CourseType.STUDENT else file["id"], + course_type=course_type, + course_id=course_id, + ) + preview_url = None + return ty.File( + id=file["wjid"], # 文件 ID + title=file["bt"], # 标题 + description=file["ms"], # 描述 + raw_size=file["wjdx"], # 文件大小 + size=file["fileSize"], + upload_time=datetime.datetime.strptime(file["scsj"], r"%Y-%m-%d %H:%M"), # 上传时间 + download_url=str(download_url), + preview_url=str(preview_url), + is_new=file["isNew"], + marked_important=file["sfqd"] == 1, # 是否强调 + visit_count=file["llcs"] or 0, # 浏览次数 + download_count=file["xzcs"] or 0, # 下载次数 + file_type=file["wjlx"], # 文件类型 + remote_file=ty.RemoteFile( + id=file["wjid"], # 文件 ID + name=title, + download_url=str(download_url), + preview_url=str(preview_url), + size=file["fileSize"], + ), + clazz=clazz[file["kjflid"]], # 课件分类 ID + ) + + +def parse_homework_file(div: bs4.element.Tag) -> ty.RemoteFile: + ftitle: bs4.element.Tag = div.find( + name="span", + attrs={ + "class": "ftitle", + }, + ) or div.find( + name="span", + attrs={ + "class", + "ft", + }, + ) + if not ftitle: + return None + a: bs4.element.Tag = ftitle.find(name="a") + size: bs4.element.Tag = div.find( + name="span", + attrs={ + "class": "color_999", + }, + ) + size: str = size.getText() + params = dict(urllib.parse.parse_qsl(urllib.parse.urlparse(a["href"]).query)) + attachment_id = params["fileId"] + download_url = ty.URL( + netloc=urls.LEARN_PREFIX.netloc, + path=params["downloadUrl"] if "downloadUrl" in params else a["href"], + ) + return ty.RemoteFile( + id=attachment_id, + name=a.getText(), + download_url=download_url, + preview_url=None, + size=size.strip(), + ) diff --git a/thu_learn_lib/ty.py b/thu_learn_lib/ty.py index 20dfd26..60e713c 100644 --- a/thu_learn_lib/ty.py +++ b/thu_learn_lib/ty.py @@ -1,7 +1,33 @@ import dataclasses import datetime import enum -import typing +import os +import urllib.parse + + +@dataclasses.dataclass +class URL: + scheme: str = "https" + netloc: str = "" + path: str = "" + params: str = "" + query: str | dict[str, str | list[str]] | list[tuple[str, str | list[str]]] = "" + fragment: str = "" + + def __str__(self) -> str: + return urllib.parse.urlunparse(self.astuple()) + + def astuple(self) -> urllib.parse.ParseResult: + return urllib.parse.ParseResult( + scheme=self.scheme, + netloc=self.netloc, + path=os.path.join(self.path) if isinstance(self.path, list) else self.path, + params=self.params, + query=self.query + if isinstance(self.query, str) + else urllib.parse.urlencode(self.query, doseq=True), + fragment=self.fragment, + ) @dataclasses.dataclass @@ -21,11 +47,6 @@ UNEXPECTED_STATUS = "unexpected status" -class ApiError(RuntimeError): - reason: FailReason - extra: typing.Any = None - - class SemesterType(enum.Enum): FALL = "Autumn Term" SPRING = "Spring Term" @@ -58,16 +79,16 @@ @dataclasses.dataclass class CourseInfo: - id: str - name: str - english_name: str - time_and_location: list[str] - url: str - teacher_name: str - teacher_number: str - course_number: str - course_index: int - course_type: CourseType + id: str = None + name: str = None + english_name: str = None + time_and_location: list[str] = None + url: str = None + teacher_name: str = None + teacher_number: str = None + course_number: str = None + course_index: int = None + course_type: CourseType = None @dataclasses.dataclass @@ -89,6 +110,7 @@ marked_important: bool publish_time: datetime.datetime publisher: str + # notification detail attachment: RemoteFile = None @@ -135,6 +157,7 @@ # status # submitted: bool # graded: bool + # homework id: str = None student_homework_id: str = None @@ -148,6 +171,7 @@ grade_time: datetime.datetime = None grader_name: str = None grade_content: str = None + # detail # description: str = None # attachment: RemoteFile = None # attachment from teacher @@ -169,6 +193,7 @@ last_reply_time: datetime.datetime visit_count: int reply_count: int + # discussion url: str board_id: str @@ -185,6 +210,7 @@ last_reply_time: datetime.datetime visit_count: int reply_count: int + # question url: str question: str diff --git a/thu_learn_lib/urls.py b/thu_learn_lib/urls.py index 1cdf5e7..708eeb8 100644 --- a/thu_learn_lib/urls.py +++ b/thu_learn_lib/urls.py @@ -1,140 +1,175 @@ -import dataclasses - -from . import ty as types +from . import ty from . import utils -LEARN_PREFIX = "https://learn.tsinghua.edu.cn" -REGISTRAR_PREFIX = "https://zhjw.cic.tsinghua.edu.cn" +LEARN_PREFIX = ty.URL(netloc="learn.tsinghua.edu.cn") +REGISTRAR_PREFIX = ty.URL(netloc="zhjw.cic.tsinghua.edu.cn") MAX_SIZE = 200 -@dataclasses.dataclass -class URL: - url: str = None - params: dict = None - - def __str__(self) -> str: - if self.params: - return ( - self.url - + "?" - + "&".join([f"{key}={value}" for key, value in self.params.items()]) - ) - - -def id_login() -> URL: - return URL( - url="https://id.tsinghua.edu.cn/do/off/ui/auth/login/post/bb5df85216504820be7bba2b0ae1535b/0?/login.do" +def id_login() -> ty.URL: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path="/do/off/ui/auth/login/post/bb5df85216504820be7bba2b0ae1535b/0", + query="/login.do", ) -def id_login_form_data(username: str, password: str) -> URL: - credential = {} - credential["i_user"] = username - credential["i_pass"] = password - credential["atOnce"] = True - return URL(params=credential) - - -def learn_auth_roam(ticket: str) -> URL: - return URL( - url=f"{LEARN_PREFIX}/b/j_spring_security_thauth_roaming_entry", - params={"ticket": ticket}, +def id_login_form_data(username: str, password: str) -> ty.URL: + return ty.URL( + query={ + "i_user": username, + "i_pass": password, + "atOnce": "true", + }, ) -def learn_logout() -> URL: - return URL(url=f"{LEARN_PREFIX}/f/j_spring_security_logout") +def learn_auth_roam(ticket: str) -> ty.URL: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path="/b/j_spring_security_thauth_roaming_entry", + query={ + "ticket": ticket, + }, + ) -def learn_student_course_list_page() -> URL: - return URL(url=f"{LEARN_PREFIX}/f/wlxt/index/course/student/") +def learn_logout() -> ty.URL: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path="/f/j_spring_security_logout", + ) -def learn_semester_list() -> URL: - return URL(url=f"{LEARN_PREFIX}/b/wlxt/kc/v_wlkc_xs_xktjb_coassb/queryxnxq") +def learn_student_course_list_page() -> ty.URL: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path="/f/wlxt/index/course/student/", + ) -def learn_current_semester() -> URL: - return URL(url=f"{LEARN_PREFIX}/b/kc/zhjw_v_code_xnxq/getCurrentAndNextSemester") +def learn_semester_list() -> ty.URL: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path="/b/wlxt/kc/v_wlkc_xs_xktjb_coassb/queryxnxq", + ) -def learn_course_list(semester: str, course_type: types.CourseType) -> URL: - if course_type == types.CourseType.STUDENT: - return URL( - url=f"{LEARN_PREFIX}/b/wlxt/kc/v_wlkc_xs_xkb_kcb_extend/student/loadCourseBySemesterId/{semester}" +def learn_current_semester() -> ty.URL: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path="/b/kc/zhjw_v_code_xnxq/getCurrentAndNextSemester", + ) + + +def learn_course_list(semester: str, course_type: ty.CourseType) -> ty.URL: + if course_type == ty.CourseType.STUDENT: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path=f"/b/wlxt/kc/v_wlkc_xs_xkb_kcb_extend/student/loadCourseBySemesterId/{semester}", ) else: - return URL( - url=f"{LEARN_PREFIX}/b/kc/v_wlkc_kcb/queryAsorCoCourseList/{semester}/0" + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path=f"/b/kc/v_wlkc_kcb/queryAsorCoCourseList/{semester}/0", ) -def learn_course_url(course_id: str, course_type: str) -> URL: - return URL( - url=f"{LEARN_PREFIX}/f/wlxt/index/course/{course_type}/course", - params={"wlkcid": course_id}, +def learn_course_url(course_id: str, course_type: str) -> ty.URL: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path=f"/f/wlxt/index/course/{course_type}/course", + query={ + "wlkcid": course_id, + }, ) -def learn_course_time_location(course_id: str) -> URL: - return URL( - url=f"{LEARN_PREFIX}/b/kc/v_wlkc_xk_sjddb/detail", params={"id": course_id} +def learn_course_time_location(course_id: str) -> ty.URL: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path="/b/kc/v_wlkc_xk_sjddb/detail", + query={ + "id": course_id, + }, ) -def learn_teacher_course_url(course_id: str) -> URL: - return URL( - url=f"{LEARN_PREFIX}/f/wlxt/index/course/teacher/course", - params={"wlkcid": course_id}, +def learn_teacher_course_url(course_id: str) -> ty.URL: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path="/f/wlxt/index/course/teacher/course", + query={ + "wlkcid": course_id, + }, ) -def learn_file_list(course_id: str, course_type: types.CourseType) -> URL: - if course_type == types.CourseType.STUDENT: - return URL( - url=f"{LEARN_PREFIX}/b/wlxt/kj/wlkc_kjxxb/student/kjxxbByWlkcidAndSizeForStudent", - params={"wlkcid": course_id, "size": MAX_SIZE}, +def learn_file_list(course_id: str, course_type: ty.CourseType) -> ty.URL: + if course_type == ty.CourseType.STUDENT: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path="/b/wlxt/kj/wlkc_kjxxb/student/kjxxbByWlkcidAndSizeForStudent", + query={ + "wlkcid": course_id, + "size": MAX_SIZE, + }, ) else: - return URL( - url=f"{LEARN_PREFIX}/b/wlxt/kj/v_kjxxb_wjwjb/teacher/queryByWlkcid", - params={"wlkcid": course_id, "size": MAX_SIZE}, + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path="/b/wlxt/kj/v_kjxxb_wjwjb/teacher/queryByWlkcid", + query={ + "wlkcid": course_id, + "size": MAX_SIZE, + }, ) -def learn_file_classify(course_id: str) -> URL: - return URL( - url=f"{LEARN_PREFIX}/b/wlxt/kj/wlkc_kjflb/student/pageList", - params={"wlkcid": course_id}, +def learn_file_classify(course_id: str) -> ty.URL: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path="/b/wlxt/kj/wlkc_kjflb/student/pageList", + query={ + "wlkcid": course_id, + }, ) -def learn_file_download(file_id: str, course_type: str, course_id: str) -> URL: - if course_type == types.CourseType.STUDENT: - return URL( - url=f"{LEARN_PREFIX}/b/wlxt/kj/wlkc_kjxxb/student/downloadFile", - params={"sfgk": 0, "wjid": file_id}, +def learn_file_download(file_id: str, course_type: str, course_id: str) -> ty.URL: + if course_type == ty.CourseType.STUDENT: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path="/b/wlxt/kj/wlkc_kjxxb/student/downloadFile", + query={ + "sfgk": 0, + "wjid": file_id, + }, ) else: - return URL( - url=f"{LEARN_PREFIX}/f/wlxt/kj/wlkc_kjxxb/teacher/beforeView", - params={"id": file_id, "wlkcid": course_id}, + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path="/f/wlxt/kj/wlkc_kjxxb/teacher/beforeView", + query={ + "id": file_id, + "wlkcid": course_id, + }, ) def learn_file_preview( - type: types.ContentType, + type: ty.ContentType, file_id: str, - course_type: types.CourseType, + course_type: ty.CourseType, first_page_only: bool = False, -) -> URL: - return URL( - url=f"{LEARN_PREFIX}/f/wlxt/kc/wj_wjb/{course_type}/beforePlay", - params={ +) -> ty.URL: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path=f"/f/wlxt/kc/wj_wjb/{course_type}/beforePlay", + query={ "wjid": file_id, "mk": utils.get_mk_from_type(type), "browser": -1, @@ -144,82 +179,114 @@ ) -def learn_notification_list(course_id: str, course_type: types.CourseType) -> URL: - if course_type == types.CourseType.STUDENT: - return URL( - url=f"{LEARN_PREFIX}/b/wlxt/kcgg/wlkc_ggb/student/kcggListXs", - params={"wlkcid": course_id, "size": MAX_SIZE}, +def learn_notification_list(course_id: str, course_type: ty.CourseType) -> ty.URL: + if course_type == ty.CourseType.STUDENT: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path="/b/wlxt/kcgg/wlkc_ggb/student/kcggListXs", + query={ + "wlkcid": course_id, + "size": MAX_SIZE, + }, ) else: - return URL( - url=f"{LEARN_PREFIX}/b/wlxt/kcgg/wlkc_ggb/teacher/kcggList", - params={"wlkcid": course_id, "size": MAX_SIZE}, + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path="/b/wlxt/kcgg/wlkc_ggb/teacher/kcggList", + query={ + "wlkcid": course_id, + "size": MAX_SIZE, + }, ) def learn_notification_detail( - course_id: str, notification_id: str, course_type: types.CourseType -) -> URL: - if course_type == types.CourseType.STUDENT: - return URL( - url=f"{LEARN_PREFIX}/f/wlxt/kcgg/wlkc_ggb/student/beforeViewXs", - params={"wlkcid": course_id, "id": notification_id}, + course_id: str, notification_id: str, course_type: ty.CourseType +) -> ty.URL: + if course_type == ty.CourseType.STUDENT: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path="/f/wlxt/kcgg/wlkc_ggb/student/beforeViewXs", + query={ + "wlkcid": course_id, + "id": notification_id, + }, ) else: - return URL( - url=f"{LEARN_PREFIX}/f/wlxt/kcgg/wlkc_ggb/teacher/beforeViewJs", - params={"wlkcid": course_id, "id": notification_id}, + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path="/f/wlxt/kcgg/wlkc_ggb/teacher/beforeViewJs", + query={ + "wlkcid": course_id, + "id": notification_id, + }, ) -def learn_notification_edit(course_type: types.CourseType) -> URL: - return URL(url=f"{LEARN_PREFIX}/b/wlxt/kcgg/wlkc_ggb/{course_type}/editKcgg") +def learn_notification_edit(course_type: ty.CourseType) -> ty.URL: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path=f"/b/wlxt/kcgg/wlkc_ggb/{course_type}/editKcgg", + ) -def learn_homework_list_source(course_id: str) -> dict[str]: +def learn_homework_list_source(course_id: str) -> list[dict[str]]: return [ { "url": learn_homework_list_new(course_id), - "status": {"submitted": False, "graded": False,}, + "status": ty.HomeworkStatus(submitted=False, graded=False), }, { "url": learn_homework_list_submitted(course_id), - "status": {"submitted": True, "graded": False,}, + "status": ty.HomeworkStatus(submitted=True, graded=False), }, { "url": learn_homework_list_graded(course_id), - "status": {"submitted": True, "graded": True,}, + "status": ty.HomeworkStatus(submitted=True, graded=True), }, ] -def learn_homework_list_new(course_id: str) -> URL: - return URL( - url=f"{LEARN_PREFIX}/b/wlxt/kczy/zy/student/index/zyListWj", - params={"wlkcid": course_id, "size": MAX_SIZE}, +def learn_homework_list_new(course_id: str) -> ty.URL: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path="/b/wlxt/kczy/zy/student/index/zyListWj", + query={ + "wlkcid": course_id, + "size": MAX_SIZE, + }, ) -def learn_homework_list_submitted(course_id: str) -> URL: - return URL( - url=f"{LEARN_PREFIX}/b/wlxt/kczy/zy/student/index/zyListYjwg", - params={"wlkcid": course_id, "size": MAX_SIZE}, +def learn_homework_list_submitted(course_id: str) -> ty.URL: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path="/b/wlxt/kczy/zy/student/index/zyListYjwg", + query={ + "wlkcid": course_id, + "size": MAX_SIZE, + }, ) -def learn_homework_list_graded(course_id: str) -> URL: - return URL( - url=f"{LEARN_PREFIX}/b/wlxt/kczy/zy/student/index/zyListYpg", - params={"wlkcid": course_id, "size": MAX_SIZE}, +def learn_homework_list_graded(course_id: str) -> ty.URL: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path="/b/wlxt/kczy/zy/student/index/zyListYpg", + query={ + "wlkcid": course_id, + "size": MAX_SIZE, + }, ) def learn_homework_detail( course_id: str, homework_id: str, student_homework_id: str -) -> URL: - return URL( - url=f"{LEARN_PREFIX}/f/wlxt/kczy/zy/student/viewCj", - params={ +) -> ty.URL: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path="/f/wlxt/kczy/zy/student/viewCj", + query={ "wlkcid": course_id, "zyid": homework_id, "xszyid": student_homework_id, @@ -227,23 +294,32 @@ ) -def learn_homework_download(course_id: str, attachment_id: str) -> URL: - return URL( - url=f"{LEARN_PREFIX}/b/wlxt/kczy/zy/student/downloadFile/{course_id}/{attachment_id}" +def learn_homework_download(course_id: str, attachment_id: str) -> ty.URL: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path=f"/b/wlxt/kczy/zy/student/downloadFile/{course_id}/{attachment_id}", ) -def learn_homework_submit(course_id: str, student_homework_id: str) -> URL: - return URL( - url=f"{LEARN_PREFIX}/f/wlxt/kczy/zy/student/tijiao", - params={"wlkcid": course_id, "xszyid": student_homework_id}, +def learn_homework_submit(course_id: str, student_homework_id: str) -> ty.URL: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path="/f/wlxt/kczy/zy/student/tijiao", + query={ + "wlkcid": course_id, + "xszyid": student_homework_id, + }, ) -def learn_discussion_list(course_id: str, course_type: types.CourseType) -> URL: - return URL( - url=f"{LEARN_PREFIX}/b/wlxt/bbs/bbs_tltb/{course_type}/kctlList", - params={"wlkcid": course_id, "size": MAX_SIZE}, +def learn_discussion_list(course_id: str, course_type: ty.CourseType) -> ty.URL: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path=f"/b/wlxt/bbs/bbs_tltb/{course_type}/kctlList", + query={ + "wlkcid": course_id, + "size": MAX_SIZE, + }, ) @@ -251,12 +327,13 @@ course_id: str, board_id: str, discussion_id: str, - course_type: types.CourseType, + course_type: ty.CourseType, tab_id=1, -) -> URL: - return URL( - url=f"{LEARN_PREFIX}/f/wlxt/bbs/bbs_tltb/{course_type}/viewTlById", - params={ +) -> ty.URL: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path=f"/f/wlxt/bbs/bbs_tltb/{course_type}/viewTlById", + query={ "wlkcid": course_id, "id": discussion_id, "tabbh": tab_id, @@ -265,49 +342,73 @@ ) -def learn_question_list_answered(course_id: str, course_type: types.CourseType) -> URL: - return URL( - url=f"{LEARN_PREFIX}/b/wlxt/bbs/bbs_tltb/{course_type}/kcdyList", - params={"wlkcid": course_id, "size": MAX_SIZE}, +def learn_question_list_answered(course_id: str, course_type: ty.CourseType) -> ty.URL: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path=f"/b/wlxt/bbs/bbs_tltb/{course_type}/kcdyList", + query={ + "wlkcid": course_id, + "size": MAX_SIZE, + }, ) def learn_question_detail( - course_id: str, question_id: str, course_type: types.CourseType -) -> URL: - if course_type == types.CourseType.STUDENT: - return URL( - url=f"{LEARN_PREFIX}/f/wlxt/bbs/bbs_kcdy/student/viewDyById", - params={"wlkcid": course_id, "id": question_id}, + course_id: str, question_id: str, course_type: ty.CourseType +) -> ty.URL: + if course_type == ty.CourseType.STUDENT: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path="/f/wlxt/bbs/bbs_kcdy/student/viewDyById", + query={ + "wlkcid": course_id, + "id": question_id, + }, ) else: - return URL( - url=f"{LEARN_PREFIX}/f/wlxt/bbs/bbs_kcdy/teacher/beforeEditDy", - params={"wlkcid": course_id, "id": question_id}, + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path="/f/wlxt/bbs/bbs_kcdy/teacher/beforeEditDy", + query={ + "wlkcid": course_id, + "id": question_id, + }, ) -def registrar_ticket_form_data() -> URL: - return URL(params={"appId": "ALL_ZHJW"}) +def registrar_ticket_form_data() -> ty.URL: + return ty.URL( + query={ + "appId": "ALL_ZHJW", + }, + ) -def registrar_ticket() -> URL: - return URL(url=f"{LEARN_PREFIX}/b/wlxt/common/auth/gnt") +def registrar_ticket() -> ty.URL: + return ty.URL( + netloc=LEARN_PREFIX.netloc, + path="/b/wlxt/common/auth/gnt", + ) -def registrar_auth(ticket: str) -> URL: - return URL( - url=f"{REGISTRAR_PREFIX}/j_acegi_login.do", - params={"url": "/", "ticket": ticket}, +def registrar_auth(ticket: str) -> ty.URL: + return ty.URL( + netloc=REGISTRAR_PREFIX.netloc, + path="/j_acegi_login.do", + query={ + "url": "/", + "ticket": ticket, + }, ) def registrar_calendar( start_date: str, end_date: str, graduate: bool = False, callback_name="unknown" -) -> URL: - return URL( - url=f"{REGISTRAR_PREFIX}/jxmh_out.do", - params={ +) -> ty.URL: + return ty.URL( + netloc=REGISTRAR_PREFIX.netloc, + path="/jxmh_out.do", + query={ "m": ("yjs" if graduate else "bks") + "_jxrl_all", "p_start_date": start_date, "p_end_date": end_date, diff --git a/thu_learn_lib/utils.py b/thu_learn_lib/utils.py index 6e1df37..66baba4 100644 --- a/thu_learn_lib/utils.py +++ b/thu_learn_lib/utils.py @@ -1,6 +1,7 @@ -from . import ty as types import slugify as slug +from . import ty + def slugify(text: str) -> str: return ".".join( @@ -11,25 +12,25 @@ ) -def parse_semester_type(n: int) -> types.SemesterType: +def parse_semester_type(n: int) -> ty.SemesterType: if n == 1: - return types.SemesterType.FALL + return ty.SemesterType.FALL elif n == 2: - return types.SemesterType.SPRING + return ty.SemesterType.SPRING elif n == 3: - return types.SemesterType.SUMMER + return ty.SemesterType.SUMMER else: - return types.SemesterType.UNKNOWN + return ty.SemesterType.UNKNOWN -CONTENT_TYPE_MK_MAP: dict[types.ContentType, str] = { - types.ContentType.NOTIFICATION: "kcgg", # 课程公告 - types.ContentType.FILE: "kcwj", # 课程文件 - types.ContentType.HOMEWORK: "kczy", # 课程作业 - types.ContentType.DISCUSSION: "", - types.ContentType.QUESTION: "", +CONTENT_TYPE_MK_MAP: dict[ty.ContentType, str] = { + ty.ContentType.NOTIFICATION: "kcgg", # 课程公告 + ty.ContentType.FILE: "kcwj", # 课程文件 + ty.ContentType.HOMEWORK: "kczy", # 课程作业 + ty.ContentType.DISCUSSION: "", + ty.ContentType.QUESTION: "", } -def get_mk_from_type(type: types.ContentType) -> str: +def get_mk_from_type(type: ty.ContentType) -> str: return "mk_" + CONTENT_TYPE_MK_MAP.get(type, "UNKNOWN")