feat!: 通过selenium接入了统一身份认证 (#239)

* 接入清华大学统一身份认证(selenium)

* 避免文件名出现无效字符

* 记录下载文件时的错误信息

* 优化进度条显示

* 增加-all功能，下载全部学期

Author: jolchmo

pyproject.toml

@@ -18,6 +18,7 @@ python = "^3.10"
 python-dateutil = "^2.8.2"
 requests = "^2.31.0"
 rich = "^13.7.1"
+selenium = "^4.15.0"
 tenacity = "^9.0.0"
 typer = "^0.16.0"
 

thu_learn_downloader/client/client.py

@@ -26,5 +26,13 @@ def get_with_token(
         return self.get(url=url, params={**params, "_csrf": self.token})
 
     @property
-    def token(self) -> None:
-        return self.cookies["XSRF-TOKEN"]
+    def token(self) -> str:
+        try:
+            token = self.cookies.get("XSRF-TOKEN")
+            if token is None:
+                raise KeyError("XSRF-TOKEN not found in cookies")
+            return token
+        except KeyError as e:
+            print(f"无法获取CSRF token: {e}")
+            print(f"当前cookies: {list(self.cookies.keys())}")
+            raise Exception("登录状态可能已失效，请重新登录") from e

thu_learn_downloader/client/course.py

@@ -10,6 +10,26 @@
 from .homework import Homework
 from .model import BaseModel
 
+# 全局变量用于收集课程问题信息
+_course_issues = {
+    'missing_documents': [],
+    'missing_document_classes': [],
+    'missing_homeworks': []
+}
+
+def get_course_issues():
+    """获取课程问题汇总"""
+    return _course_issues.copy()
+
+def clear_course_issues():
+    """清空课程问题记录"""
+    global _course_issues
+    _course_issues = {
+        'missing_documents': [],
+        'missing_document_classes': [],
+        'missing_homeworks': []
+    }
+
 
 class Course(BaseModel):
     client: Client = Field(exclude=True)
@@ -23,32 +43,55 @@ class Course(BaseModel):
 
     @property
     def document_classes(self) -> Sequence[DocumentClass]:
-        return [
-            DocumentClass(client=self.client, **result)
-            for result in self.client.get_with_token(
+        # 异常处理，记录问题
+        try:
+            response = self.client.get_with_token(
                 url=url.make_url(path="/b/wlxt/kj/wlkc_kjflb/student/pageList"),
                 params={"wlkcid": self.id},
-            ).json()["object"]["rows"]
-        ]
+            )
+            json_data = response.json()["object"]["rows"]
+
+            return [
+                DocumentClass(client=self.client, **result)
+                for result in response.json()["object"]["rows"]
+            ]
+            
+        except Exception as e:
+            _course_issues['missing_document_classes'].append({
+                'course': self.name,
+                'course_id': self.id,
+                'reason': f'异常: {str(e)}'
+            })
+            return []
 
     @property
     def documents(self) -> Sequence[Document]:
-        documents: Sequence[Document] = [
-            Document(client=self.client, **result)
-            for result in self.client.get_with_token(
+        try:
+            response = self.client.get_with_token(
                 url=url.make_url(
                     path="/b/wlxt/kj/wlkc_kjxxb/student/kjxxbByWlkcidAndSizeForStudent"
                 ),
                 params={"wlkcid": self.id, "size": MAX_SIZE},
-            ).json()["object"]
-        ]
-        documents.sort(key=lambda document: document.title)
-        documents.sort(key=lambda document: document.upload_time)
-        return documents
+            )
+            documents: Sequence[Document] = [
+                Document(client=self.client, **result)
+                for result in response.json()["object"]
+            ]
+            documents.sort(key=lambda document: document.title)
+            documents.sort(key=lambda document: document.upload_time)
+            return documents
+            
+        except Exception as e:
+            _course_issues['missing_documents'].append({
+                'course': self.name,
+                'course_id': self.id,
+                'reason': f'异常: {str(e)}'
+            })
+            return []
 
     @property
     def homeworks(self) -> Sequence[Homework]:
-        return [
+        all_homeworks = [
             *self._homeworks_at_url(
                 url=url.make_url(
                     path="/b/wlxt/kczy/zy/student/index/zyListWj",
@@ -68,11 +111,42 @@ def homeworks(self) -> Sequence[Homework]:
                 ),
             ),
         ]
+        
+        # 如果所有作业API都没有返回数据，记录该课程
+        if not all_homeworks:
+            _course_issues['missing_homeworks'].append({
+                'course': self.name,
+                'course_id': self.id,
+                'reason': '该课程无作业'
+            })
+        
+        return all_homeworks
 
     def _homeworks_at_url(self, url: str) -> Sequence[Homework]:
-        resp: Response = self.client.get_with_token(url=url)
-        json: dict[str, Any] = resp.json()
-        results: Sequence[dict[str, Any]] = json["object"]["aaData"] or []
-        return [
-            Homework.from_json(client=self.client, json=result) for result in results
-        ]
+        try:
+            resp: Response = self.client.get_with_token(url=url)
+            
+            # 检查响应状态
+            if resp.status_code != 200:
+                # 作业API失败时，只在所有作业API都失败时才记录
+                return []
+            
+            # 尝试解析JSON
+            json_data: dict[str, Any] = resp.json()
+            if json_data is None:
+                return []
+            
+            # 检查数据结构
+            if "object" not in json_data:
+                return []
+            
+            if "aaData" not in json_data["object"]:
+                return []
+            
+            results: Sequence[dict[str, Any]] = json_data["object"]["aaData"] or []
+            return [
+                Homework.from_json(client=self.client, json=result) for result in results
+            ]
+            
+        except Exception as e:
+            return []

thu_learn_downloader/client/learn.py

@@ -18,32 +18,67 @@ class Learn:
 
     def __init__(self, language: Language = Language.ENGLISH, *args, **kwargs) -> None:
         self.client = Client(language, *args, **kwargs)
+        
+    """使用浏览器进行统一用户登录"""    
+    def login(self) -> None:
+        from ..login.browser import login_with_browser
+        
+        try:
+            cookies = login_with_browser()
+            
+            # 将cookies设置到client中
+            for name, value in cookies.items():
+                self.client.cookies[name] = value
+            
+            # 验证登录是否成功，尝试访问课程页面
+            response = self.client.get(url=url.make_url(path="/f/wlxt/index/course/student/"))
+            if response.status_code == 200:
+                print("浏览器登录成功！")
+            else:
+                raise Exception("登录验证失败，请重试")
+                
+        except Exception as e:
+            print(f"浏览器登录失败: {e}")
 
-    def login(self, username: str, password: str) -> None:
-        response: Response = self.client.get(url=url.make_url())
-        soup: BeautifulSoup = BeautifulSoup(
-            markup=response.text, features="html.parser"
-        )
-        login_form: Tag = cast(Tag, soup.select_one(selector="#loginForm"))
-        action: str = cast(str, login_form["action"])
-        response: Response = self.client.post(
-            url=action, data={"i_user": username, "i_pass": password, "atOnce": True}
-        )
-        soup: BeautifulSoup = BeautifulSoup(
-            markup=response.text, features="html.parser"
-        )
-        a: Tag = cast(Tag, soup.select_one(selector="a"))
-        href: str = cast(str, a["href"])
-        parse_result: ParseResult = urllib.parse.urlparse(url=href)
-        query: dict[str, list[str]] = urllib.parse.parse_qs(qs=parse_result.query)
-        status, ticket = query["status"][0], query["ticket"][0]
-        self.client.get(url=href)
-        self.client.get(
-            url=url.make_url(path="/b/j_spring_security_thauth_roaming_entry"),
-            params={"ticket": ticket},
-        )
-        self.client.get(url=url.make_url(path="/f/wlxt/index/course/student/"))
-        assert status == "SUCCESS"
+    # def login(self, username: str = None, password: str = None) -> None:
+    #     """
+    #     登录方法
+    #     如果提供了用户名和密码，使用传统登录方式（保留兼容性）
+    #     否则使用浏览器登录方式
+    #     """
+    #     if username and password:
+    #         # 使用传统的用户名密码登录方式
+    #         self._login_with_credentials(username, password)
+    #     else:
+    #         # 使用新的浏览器登录方式
+    #         self.login_with_browser()
+    
+    # def _login_with_credentials(self, username: str, password: str) -> None:
+    #     """传统的用户名密码登录方式（保留作为备用）"""
+    #     response: Response = self.client.get(url=url.make_url())
+    #     soup: BeautifulSoup = BeautifulSoup(
+    #         markup=response.text, features="html.parser"
+    #     )
+    #     login_form: Tag = cast(Tag, soup.select_one(selector="#loginForm"))
+    #     action: str = cast(str, login_form["action"])
+    #     response: Response = self.client.post(
+    #         url=action, data={"i_user": username, "i_pass": password, "atOnce": True}
+    #     )
+    #     soup: BeautifulSoup = BeautifulSoup(
+    #         markup=response.text, features="html.parser"
+    #     )
+    #     a: Tag = cast(Tag, soup.select_one(selector="a"))
+    #     href: str = cast(str, a["href"])
+    #     parse_result: ParseResult = urllib.parse.urlparse(url=href)
+    #     query: dict[str, list[str]] = urllib.parse.parse_qs(qs=parse_result.query)
+    #     status, ticket = query["status"][0], query["ticket"][0]
+    #     self.client.get(url=href)
+    #     self.client.get(
+    #         url=url.make_url(path="/b/j_spring_security_thauth_roaming_entry"),
+    #         params={"ticket": ticket},
+    #     )
+    #     self.client.get(url=url.make_url(path="/f/wlxt/index/course/student/"))
+    #     assert status == "SUCCESS"
 
     @functools.cached_property
     def semesters(self) -> Sequence[Semester]:

thu_learn_downloader/download/downloader.py

@@ -46,6 +46,7 @@ class Downloader:
     semesters_task_id: TaskID
     courses_task_id: TaskID
     documents_task_id: TaskID
+    homeworks_task_id: TaskID
 
     def __init__(
         self,
@@ -75,6 +76,7 @@ def __init__(
         self.semesters_task_id = self.progress_prepare.add_task(description="Semesters")
         self.courses_task_id = self.progress_prepare.add_task(description="Courses")
         self.documents_task_id = self.progress_prepare.add_task(description="Documents")
+        self.homeworks_task_id = self.progress_prepare.add_task(description="Homeworks")
         self.live = Live(
             Group(
                 Panel(self.progress_download, height=jobs + 2),
@@ -187,13 +189,10 @@ def sync_semesters(self, semesters: Sequence[Semester]) -> None:
                 if semester.id in self.selector.semesters
             ]
         self.progress_prepare.reset(task_id=self.semesters_task_id)
-        for semester in self.progress_prepare.track(
-            sequence=semesters,
-            total=len(semesters),
-            task_id=self.semesters_task_id,
-            description="Semesters",
-        ):
+        self.progress_prepare.update(task_id=self.semesters_task_id, total=len(semesters))
+        for semester in semesters:
             self.sync_semester(semester=semester)
+            self.progress_prepare.advance(task_id=self.semesters_task_id, advance=1)
 
     def sync_semester(self, semester: Semester) -> None:
         self.sync_courses(semester=semester, courses=semester.courses)
@@ -204,13 +203,10 @@ def sync_courses(self, semester: Semester, courses: Sequence[Course]) -> None:
                 course for course in courses if course.id in self.selector.courses
             ]
         self.progress_prepare.reset(task_id=self.courses_task_id)
-        for course in self.progress_prepare.track(
-            sequence=courses,
-            total=len(courses),
-            task_id=self.courses_task_id,
-            description="Courses",
-        ):
+        self.progress_prepare.update(task_id=self.courses_task_id, total=len(courses))
+        for course in courses:
             self.sync_course(semester=semester, course=course)
+            self.progress_prepare.advance(task_id=self.courses_task_id, advance=1)
 
     def sync_course(self, semester: Semester, course: Course) -> None:
         if self.selector.document:
@@ -236,22 +232,16 @@ def sync_documents(
             document_class.id: document_class for document_class in document_classes
         }
         self.progress_prepare.reset(task_id=self.documents_task_id)
-        for index, document in enumerate(
-            self.progress_prepare.track(
-                sequence=documents,
-                total=len(documents),
-                task_id=self.documents_task_id,
-                description="Documents",
-            ),
-            start=1,
-        ):
+        self.progress_prepare.update(task_id=self.documents_task_id, total=len(documents))
+        for index, document in enumerate(documents, start=1):
             self.sync_document(
                 semester=semester,
                 course=course,
                 document_class=document_class_map[document.class_id],
                 document=document,
                 index=index,
             )
+            self.progress_prepare.advance(task_id=self.documents_task_id, advance=1)
 
     def sync_document(
         self,
@@ -287,14 +277,11 @@ def sync_document(
     def sync_homeworks(
         self, semester: Semester, course: Course, homeworks: Sequence[Homework]
     ) -> None:
-        self.progress_prepare.reset(task_id=self.documents_task_id)
-        for homework in self.progress_prepare.track(
-            sequence=homeworks,
-            total=len(homeworks),
-            task_id=self.documents_task_id,
-            description="Homeworks",
-        ):
+        self.progress_prepare.reset(task_id=self.homeworks_task_id)
+        self.progress_prepare.update(task_id=self.homeworks_task_id, total=len(homeworks))
+        for homework in homeworks:
             self.sync_homework(semester=semester, course=course, homework=homework)
+            self.progress_prepare.advance(task_id=self.homeworks_task_id, advance=1)
 
     def sync_homework(
         self, semester: Semester, course: Course, homework: Homework