🔨任务点接口基类

cxapi/__init__.py

@@ -2,7 +2,4 @@
 from .chapters import ChapterContainer
 from .classes import ClassSelector
 from .exam import ExamDto
-from .exception import TaskPointError
-from .jobs.document import PointDocumentDto
-from .jobs.video import PointVideoDto
-from .jobs.work import PointWorkDto
+from .task_point import PointDocumentDto, PointVideoDto, PointWorkDto

cxapi/base.py

@@ -1,31 +1,43 @@
+import json
+import re
 from pathlib import Path
 from typing import Literal
 
+from bs4 import BeautifulSoup
+
+from logger import Logger
+
+from .exception import APIError, ChapterNotOpened
 from .schema import QuestionModel, QuestionsExportSchema
+from .session import SessionWraper
+
+# SSR页面-客户端章节任务卡片
+PAGE_MOBILE_CHAPTER_CARD = "https://mooc1-api.chaoxing.com/knowledge/cards"
 
 
 class QAQDtoBase:
     """答题接口基类
     用于定义 问题-回答-查询 类的各种 trait
     """
-    current_index: int              # 当前题目索引 用于迭代
-    title: str                      # 标题
-
+
+    current_index: int  # 当前题目索引 用于迭代
+    title: str  # 标题
+
     def __init__(self) -> None:
         self.current_index = 0
 
     def __iter__(self):
-        self.fetch_all()    # 刷新提单缓存
+        self.fetch_all()  # 刷新题单缓存
         self.current_index = 0
         return self
-    
+
     def __next__(self) -> tuple[int, QuestionModel]:
         """迭代返回
         Returns:
             int, QuestionModel: 题目索引 题目模型
         """
         raise NotImplemented
-        
+
     def fetch(self, index: int) -> QuestionModel:
         """拉取一道题
         Args:
@@ -34,45 +46,109 @@ def fetch(self, index: int) -> QuestionModel:
             QuestionModel: 题目模型
         """
         raise NotImplementedError
-    
+
     def fetch_all(self) -> list[QuestionModel]:
         """拉取全部题单
         Returns:
             list[QuestionModel]: 题目模型列表 (按题号顺序排列)
         """
         raise NotImplementedError
-
-    def submit(
-        self, *, 
-        index: int=0,
-        question: QuestionModel,
-        **kwargs
-    ) -> dict:
+
+    def submit(self, *, index: int = 0, question: QuestionModel, **kwargs) -> dict:
         """提交试题
         Args:
             index: 题目索引
             question: 题目数据模型
         """
         raise NotImplementedError
-    
+
     def final_submit(self) -> dict:
-        """直接交卷
-        """
+        """直接交卷"""
         raise NotImplementedError
-    
+
     def fallback_save(self) -> dict:
-        """临时保存试题
-        """
+        """临时保存试题"""
         raise NotImplementedError
-    
+
     def export(
         self,
-        format_or_path: Literal["schema", "dict", "json"] | Path = "schema"
+        format_or_path: Literal["schema", "dict", "json"] | Path = "schema",
     ) -> QuestionsExportSchema | str | dict | None:
         """导出当前试题
         Args:
             format_or_path: 导出格式或路径
         """
         raise NotImplementedError
 
-__all__ = ["QAQDtoBase"]
+
+class TaskPointBase:
+    """任务点基类"""
+
+    logger: Logger
+    session: SessionWraper
+    card_index: int  # 卡片索引位置
+    course_id: int
+    class_id: int
+    knowledge_id: int
+    cpi: int
+
+    attachment: dict
+
+    def __init__(
+        self,
+        session: SessionWraper,
+        card_index: int,
+        course_id: int,
+        class_id: int,
+        knowledge_id: int,
+        cpi: int,
+    ) -> None:
+        self.session = session
+        self.card_index = card_index
+        self.course_id = course_id
+        self.class_id = class_id
+        self.knowledge_id = knowledge_id
+        self.cpi = cpi
+
+    def fetch_attachment(self) -> None:
+        """拉取任务卡片 Attachment 数据"""
+        resp = self.session.get(
+            PAGE_MOBILE_CHAPTER_CARD,
+            params={
+                "clazzid": self.class_id,
+                "courseid": self.course_id,
+                "knowledgeid": self.knowledge_id,
+                "num": self.card_index,
+                "isPhone": 1,
+                "control": "true",
+                "cpi": self.cpi,
+            },
+        )
+        resp.raise_for_status()
+        html = BeautifulSoup(resp.text, "lxml")
+
+        if r := re.search(
+            r"window\.AttachmentSetting *= *(.+?);",
+            html.head.find("script", type="text/javascript").text,
+        ):
+            self.attachment = json.loads(r.group(1))
+            self.logger.debug(f"Attachment: {self.attachment}")
+        else:
+            if t := html.select_one("p.blankTips"):
+                if t.text == "章节未开放！":
+                    self.logger.error("章节未开放")
+                    raise ChapterNotOpened
+                else:
+                    raise APIError(t.text)
+            raise APIError
+        self.logger.info("Attachment拉取成功")
+
+    def parse_attachment(self) -> bool:
+        """解析任务点卡片 Attachment
+        Returns:
+            bool: 是否需要完成
+        """
+        raise NotImplementedError
+
+
+__all__ = ["QAQDtoBase", "TaskPointBase"]

cxapi/chapters.py

@@ -10,11 +10,9 @@
 from logger import Logger
 
 from .exception import APIError
-from .jobs.document import PointDocumentDto
-from .jobs.video import PointVideoDto
-from .jobs.work import PointWorkDto
 from .schema import AccountInfo, ChapterModel
 from .session import SessionWraper
+from .task_point import PointDocumentDto, PointVideoDto, PointWorkDto
 from .utils import calc_infenc, get_ts
 
 TaskPointType = PointWorkDto | PointVideoDto | PointDocumentDto
@@ -38,7 +36,7 @@ class ChapterContainer:
     # 课程参数
     course_id: int  # 课程 id
     name: str  # 课程名
-    clazz_id: int  # 班级 id
+    class_id: int  # 班级 id
     cpi: int
 
     tui_index: int  # TUI 列表指针索引值
@@ -49,15 +47,15 @@ def __init__(
         acc: AccountInfo,
         courseid: int,
         name: str,
-        clazzid: int,
+        classid: int,
         cpi: int,
         chapters: list[ChapterModel],
     ) -> None:
         self.logger = Logger("Chapters")
         self.session = session
         self.acc = acc
         self.course_id = courseid
-        self.clazz_id = clazzid
+        self.class_id = classid
         self.name = name
         self.cpi = cpi
         self.chapters = chapters
@@ -132,7 +130,7 @@ def fetch_point_status(self) -> None:
             data={
                 "view": "json",
                 "nodes": ",".join(str(c.chapter_id) for c in self.chapters),
-                "clazzid": self.clazz_id,
+                "clazzid": self.class_id,
                 "time": get_ts(),
                 "userid": self.acc.puid,
                 "cpi": self.cpi,
@@ -209,13 +207,13 @@ def fetch_points_by_index(self, index: int) -> list[TaskPointType]:
                         point_objs.append(
                             PointVideoDto(
                                 session=self.session,
-                                acc=self.acc,
                                 card_index=card_index,
                                 course_id=self.course_id,
+                                class_id=self.class_id,
                                 knowledge_id=self.chapters[index].chapter_id,
-                                object_id=json_data["objectid"],
-                                clazz_id=self.clazz_id,
                                 cpi=self.cpi,
+                                # 任务点附加
+                                object_id=json_data["objectid"],
                             )
                         )
                         self.logger.debug(
@@ -226,15 +224,15 @@ def fetch_points_by_index(self, index: int) -> list[TaskPointType]:
                         point_objs.append(
                             PointWorkDto(
                                 session=self.session,
-                                acc=self.acc,
                                 card_index=card_index,
                                 course_id=self.course_id,
+                                class_id=self.class_id,
+                                knowledge_id=self.chapters[index].chapter_id,
+                                cpi=self.cpi,
+                                # 任务点附加
                                 work_id=json_data["workid"],
                                 school_id=json_data.get("schoolid"),
                                 job_id=json_data["_jobid"],
-                                knowledge_id=self.chapters[index].chapter_id,
-                                clazz_id=self.clazz_id,
-                                cpi=self.cpi,
                             )
                         )
                         self.logger.debug(
@@ -245,12 +243,12 @@ def fetch_points_by_index(self, index: int) -> list[TaskPointType]:
                         point_objs.append(
                             PointDocumentDto(
                                 session=self.session,
-                                acc=self.acc,
                                 card_index=card_index,
                                 course_id=self.course_id,
+                                class_id=self.class_id,
                                 knowledge_id=self.chapters[index].chapter_id,
-                                clazz_id=self.clazz_id,
                                 cpi=self.cpi,
+                                # 任务点附加
                                 object_id=json_data["objectid"],
                             )
                         )

cxapi/classes.py

@@ -251,7 +251,7 @@ def __next__(self) -> ChapterContainer | ExamDto | list[ClassExamModule]:
                 session=self.__classes.session,
                 acc=self.__classes.acc,
                 courseid=class_meta.course_id,
-                clazzid=class_meta.class_id,
+                classid=class_meta.class_id,
                 name=class_meta.name,
                 cpi=class_meta.cpi,
                 chapters=chapters,

cxapi/exception.py

@@ -7,6 +7,9 @@ class HandleCaptchaError(Exception):
 class FaceDetectionError(Exception):
     "人脸识别错误"
 
+class ChapterNotOpened(APIError):
+    "章节未开放"
+
 # 任务点相关
 class TaskPointError(APIError):
     "任务点错误"