feat: 知乎支持创作者主页数据爬取(回答、文章、视频)

media_platform/zhihu/help.py

@@ -1,8 +1,10 @@
 # -*- coding: utf-8 -*-
-from typing import Dict, List
+import json
+from typing import Dict, List, Optional
 from urllib.parse import parse_qs, urlparse
 
 import execjs
+from parsel import Selector
 
 from constant import zhihu as zhihu_constant
 from model.m_zhihu import ZhihuComment, ZhihuContent, ZhihuCreator
@@ -29,11 +31,11 @@ def sign(url: str, cookies: str) -> Dict:
     return ZHIHU_SGIN_JS.call("get_sign", url, cookies)
 
 
-class ZhiHuJsonExtractor:
+class ZhihuExtractor:
     def __init__(self):
         pass
 
-    def extract_contents(self, json_data: Dict) -> List[ZhihuContent]:
+    def extract_contents_from_search(self, json_data: Dict) -> List[ZhihuContent]:
         """
         extract zhihu contents
         Args:
@@ -45,21 +47,34 @@ class ZhiHuJsonExtractor:
         if not json_data:
             return []
 
-        result: List[ZhihuContent] = []
         search_result: List[Dict] = json_data.get("data", [])
         search_result = [s_item for s_item in search_result if s_item.get("type") in ['search_result', 'zvideo']]
-        for sr_item in search_result:
-            sr_object: Dict = sr_item.get("object", {})
-            if sr_object.get("type") == zhihu_constant.ANSWER_NAME:
-                result.append(self._extract_answer_content(sr_object))
-            elif sr_object.get("type") == zhihu_constant.ARTICLE_NAME:
-                result.append(self._extract_article_content(sr_object))
-            elif sr_object.get("type") == zhihu_constant.VIDEO_NAME:
-                result.append(self._extract_zvideo_content(sr_object))
+        return self._extract_content_list([sr_item.get("object") for sr_item in search_result if sr_item.get("object")])
+
+
+    def _extract_content_list(self, content_list: List[Dict]) -> List[ZhihuContent]:
+        """
+        extract zhihu content list
+        Args:
+            content_list:
+
+        Returns:
+
+        """
+        if not content_list:
+            return []
+
+        res: List[ZhihuContent] = []
+        for content in content_list:
+            if content.get("type") == zhihu_constant.ANSWER_NAME:
+                res.append(self._extract_answer_content(content))
+            elif content.get("type") == zhihu_constant.ARTICLE_NAME:
+                res.append(self._extract_article_content(content))
+            elif content.get("type") == zhihu_constant.VIDEO_NAME:
+                res.append(self._extract_zvideo_content(content))
             else:
                 continue
-
-        return result
+        return res
 
     def _extract_answer_content(self, answer: Dict) -> ZhihuContent:
         """
@@ -72,22 +87,23 @@ class ZhiHuJsonExtractor:
         res = ZhihuContent()
         res.content_id = answer.get("id")
         res.content_type = answer.get("type")
-        res.content_text = extract_text_from_html(answer.get("content"))
+        res.content_text = extract_text_from_html(answer.get("content", ""))
         res.question_id = answer.get("question").get("id")
         res.content_url = f"{zhihu_constant.ZHIHU_URL}/question/{res.question_id}/answer/{res.content_id}"
-        res.title = extract_text_from_html(answer.get("title"))
-        res.desc = extract_text_from_html(answer.get("description"))
+        res.title = extract_text_from_html(answer.get("title", ""))
+        res.desc = extract_text_from_html(answer.get("description", "") or answer.get("excerpt", ""))
         res.created_time = answer.get("created_time")
         res.updated_time = answer.get("updated_time")
-        res.voteup_count = answer.get("voteup_count")
-        res.comment_count = answer.get("comment_count")
+        res.voteup_count = answer.get("voteup_count", 0)
+        res.comment_count = answer.get("comment_count", 0)
 
         # extract author info
-        author_info = self._extract_author(answer.get("author"))
+        author_info = self._extract_content_or_comment_author(answer.get("author"))
         res.user_id = author_info.user_id
         res.user_link = author_info.user_link
         res.user_nickname = author_info.user_nickname
         res.user_avatar = author_info.user_avatar
+        res.user_url_token = author_info.url_token
         return res
 
     def _extract_article_content(self, article: Dict) -> ZhihuContent:
@@ -106,17 +122,18 @@ class ZhiHuJsonExtractor:
         res.content_url = f"{zhihu_constant.ZHIHU_URL}/p/{res.content_id}"
         res.title = extract_text_from_html(article.get("title"))
         res.desc = extract_text_from_html(article.get("excerpt"))
-        res.created_time = article.get("created_time")
-        res.updated_time = article.get("updated_time")
-        res.voteup_count = article.get("voteup_count")
-        res.comment_count = article.get("comment_count")
+        res.created_time = article.get("created_time", 0) or article.get("created", 0)
+        res.updated_time = article.get("updated_time", 0) or article.get("updated", 0)
+        res.voteup_count = article.get("voteup_count", 0)
+        res.comment_count = article.get("comment_count", 0)
 
         # extract author info
-        author_info = self._extract_author(article.get("author"))
+        author_info = self._extract_content_or_comment_author(article.get("author"))
         res.user_id = author_info.user_id
         res.user_link = author_info.user_link
         res.user_nickname = author_info.user_nickname
         res.user_avatar = author_info.user_avatar
+        res.user_url_token = author_info.url_token
         return res
 
     def _extract_zvideo_content(self, zvideo: Dict) -> ZhihuContent:
@@ -129,25 +146,34 @@ class ZhiHuJsonExtractor:
 
         """
         res = ZhihuContent()
-        res.content_id = zvideo.get("zvideo_id")
+
+        if "video" in zvideo and isinstance(zvideo.get("video"), dict): # 说明是从创作者主页的视频列表接口来的
+            res.content_id = zvideo.get("video").get("video_id")
+            res.content_url = f"{zhihu_constant.ZHIHU_URL}/zvideo/{res.content_id}"
+            res.created_time = zvideo.get("published_at")
+            res.updated_time = zvideo.get("updated_at")
+        else:
+            res.content_id = zvideo.get("zvideo_id")
+            res.content_url = zvideo.get("video_url")
+            res.created_time = zvideo.get("created_at")
+
         res.content_type = zvideo.get("type")
-        res.content_url = zvideo.get("video_url")
         res.title = extract_text_from_html(zvideo.get("title"))
         res.desc = extract_text_from_html(zvideo.get("description"))
-        res.created_time = zvideo.get("created_at")
         res.voteup_count = zvideo.get("voteup_count")
         res.comment_count = zvideo.get("comment_count")
 
         # extract author info
-        author_info = self._extract_author(zvideo.get("author"))
+        author_info = self._extract_content_or_comment_author(zvideo.get("author"))
         res.user_id = author_info.user_id
         res.user_link = author_info.user_link
         res.user_nickname = author_info.user_nickname
         res.user_avatar = author_info.user_avatar
+        res.user_url_token = author_info.url_token
         return res
 
     @staticmethod
-    def _extract_author(author: Dict) -> ZhihuCreator:
+    def _extract_content_or_comment_author(author: Dict) -> ZhihuCreator:
         """
         extract zhihu author
         Args:
@@ -165,6 +191,7 @@ class ZhiHuJsonExtractor:
         res.user_link = f"{zhihu_constant.ZHIHU_URL}/people/{author.get('url_token')}"
         res.user_nickname = author.get("name")
         res.user_avatar = author.get("avatar_url")
+        res.url_token = author.get("url_token")
         return res
 
     def extract_comments(self, page_content: ZhihuContent, comments: List[Dict]) -> List[ZhihuComment]:
@@ -209,7 +236,7 @@ class ZhiHuJsonExtractor:
         res.content_type = page_content.content_type
 
         # extract author info
-        author_info = self._extract_author(comment.get("author"))
+        author_info = self._extract_content_or_comment_author(comment.get("author"))
         res.user_id = author_info.user_id
         res.user_link = author_info.user_link
         res.user_nickname = author_info.user_nickname
@@ -254,3 +281,80 @@ class ZhiHuJsonExtractor:
         query_params = parse_qs(parsed_url.query)
         offset = query_params.get('offset', [""])[0]
         return offset
+
+    @staticmethod
+    def _foramt_gender_text(gender: int) -> str:
+        """
+        format gender text
+        Args:
+            gender:
+
+        Returns:
+
+        """
+        if gender == 1:
+            return "男"
+        elif gender == 0:
+            return "女"
+        else:
+            return "未知"
+
+
+    def extract_creator(self, user_url_token: str, html_content: str) -> Optional[ZhihuCreator]:
+        """
+        extract zhihu creator
+        Args:
+            user_url_token : zhihu creator url token
+            html_content: zhihu creator html content
+
+        Returns:
+
+        """
+        if not html_content:
+            return None
+
+        js_init_data = Selector(text=html_content).xpath("//script[@id='js-initialData']/text()").get(default="").strip()
+        if not js_init_data:
+            return None
+
+        js_init_data_dict: Dict = json.loads(js_init_data)
+        users_info: Dict = js_init_data_dict.get("initialState", {}).get("entities", {}).get("users", {})
+        if not users_info:
+            return None
+
+        creator_info: Dict = users_info.get(user_url_token)
+        if not creator_info:
+            return None
+
+        res = ZhihuCreator()
+        res.user_id = creator_info.get("id")
+        res.user_link = f"{zhihu_constant.ZHIHU_URL}/people/{user_url_token}"
+        res.user_nickname = creator_info.get("name")
+        res.user_avatar = creator_info.get("avatarUrl")
+        res.url_token = creator_info.get("urlToken") or user_url_token
+        res.gender = self._foramt_gender_text(creator_info.get("gender"))
+        res.ip_location = creator_info.get("ipInfo")
+        res.follows = creator_info.get("followingCount")
+        res.fans = creator_info.get("followerCount")
+        res.anwser_count = creator_info.get("answerCount")
+        res.video_count = creator_info.get("zvideoCount")
+        res.question_count = creator_info.get("questionCount")
+        res.article_count = creator_info.get("articlesCount")
+        res.column_count = creator_info.get("columnsCount")
+        res.get_voteup_count = creator_info.get("voteupCount")
+        return res
+
+
+    def extract_content_list_from_creator(self, anwser_list: List[Dict]) -> List[ZhihuContent]:
+        """
+        extract content list from creator
+        Args:
+            anwser_list:
+
+        Returns:
+
+        """
+        if not anwser_list:
+            return []
+
+        return self._extract_content_list(anwser_list)