jiazhizhong/MediaCrawler
1
0
Fork 0
mirror of https://github.com/NanmiCoder/MediaCrawler.git synced 2026-06-09 03:17:25 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: 百度贴吧一级评论done

Browse Source
This commit is contained in:
Relakkes
2024-08-07 02:34:56 +08:00
parent 3c98808409
commit 026d81e131
8 changed files with 1042 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files

94
media_platform/tieba/help.py
View File

@@ -1,10 +1,12 @@
# -*- coding: utf-8 -*-
import re
import json
import html
from typing import List, Dict, Tuple
from parsel import Selector
from model.m_baidu_tieba import TiebaNote
from model.m_baidu_tieba import TiebaNote, TiebaComment
from constant import baidu_tieba as const
@@ -40,7 +42,6 @@ class TieBaExtractor:
result.append(tieba_note)
return result
def extract_note_detail(self, page_content: str) -> TiebaNote:
"""
提取贴吧帖子详情
@@ -66,8 +67,10 @@ class TieBaExtractor:
title=content_selector.xpath("//title/text()").get(default='').strip(),
desc=content_selector.xpath("//meta[@name='description']/@content").get(default='').strip(),
note_url=const.TIEBA_URL + f"/p/{note_id}",
user_link=const.TIEBA_URL + first_floor_selector.xpath(".//a[@class='p_author_face ']/@href").get(default='').strip(),
user_nickname=first_floor_selector.xpath(".//a[@class='p_author_name j_user_card']/text()").get(default='').strip(),
user_link=const.TIEBA_URL + first_floor_selector.xpath(".//a[@class='p_author_face ']/@href").get(
default='').strip(),
user_nickname=first_floor_selector.xpath(".//a[@class='p_author_name j_user_card']/text()").get(
default='').strip(),
user_avatar=first_floor_selector.xpath(".//a[@class='p_author_face ']/img/@src").get(default='').strip(),
tieba_name=content_selector.xpath("//a[@class='card_title_fname']/text()").get(default='').strip(),
tieba_link=const.TIEBA_URL + content_selector.xpath("//a[@class='card_title_fname']/@href").get(default=''),
@@ -79,33 +82,44 @@ class TieBaExtractor:
note.title = note.title.replace(f"【{note.tieba_name}】_百度贴吧", "")
return note
@staticmethod
def extract_tieba_note_comments(page_content: str) -> List[Dict]:
def extract_tieba_note_parment_comments(self, page_content: str, note_id: str) -> List[TiebaComment]:
"""
提取贴吧帖子评论
提取贴吧帖子一级评论
Args:
page_content:
note_id:
Returns:
"""
xpath_selector = "//div[@id='j_p_postlist']/div[@class='l_post l_post_bright j_l_post clearfix']"
xpath_selector = "//div[@class='l_post l_post_bright j_l_post clearfix ']"
comment_list = Selector(text=page_content).xpath(xpath_selector)
result = []
for comment in comment_list:
comment_id = comment.xpath(".//@data-pid").get(default='').strip()
author = comment.xpath(".//a[@data-field]/text()").get(default='').strip()
author_link = comment.xpath(".//a[@data-field]/@href").get(default='')
content = comment.xpath(".//div[@class='d_post_content j_d_post_content ']/text()").get(default='').strip()
date = comment.xpath(".//span[@class='tail-info']/text()").get(default='').strip()
result: List[TiebaComment] = []
for comment_selector in comment_list:
comment_field_value: Dict = self.extract_data_field_value(comment_selector)
if not comment_field_value:
continue
result.append({
"comment_id": comment_id,
"author": author,
"author_link": author_link,
"content": content,
"time": date,
})
other_info_content = comment_selector.xpath(".//div[@class='post-tail-wrap']").get(default="").strip()
ip_location, publish_time = self.extract_ip_and_pub_time(other_info_content)
tieba_comment = TiebaComment(
comment_id=str(comment_field_value.get("content").get("post_id")),
sub_comment_count=comment_field_value.get("content").get("comment_num"),
content=comment_field_value.get("content").get("content"),
note_url=const.TIEBA_URL + f"/p/{note_id}",
user_link=const.TIEBA_URL + comment_selector.xpath(".//a[@class='p_author_face ']/@href").get(default='').strip(),
user_nickname=comment_selector.xpath(".//a[@class='p_author_name j_user_card']/text()").get(
default='').strip(),
user_avatar=comment_selector.xpath(".//a[@class='p_author_face ']/img/@src").get(
default='').strip(),
tieba_name=comment_selector.xpath("//a[@class='card_title_fname']/text()").get(default='').strip(),
ip_location=ip_location,
publish_time=publish_time,
note_id=note_id,
)
print(tieba_comment.model_dump())
result.append(tieba_comment)
return result
@staticmethod
def extract_ip_and_pub_time(html_content: str) -> Tuple[str, str]:
@@ -125,6 +139,31 @@ class TieBaExtractor:
pub_time = time_match.group(1) if time_match else ""
return ip, pub_time
@staticmethod
def extract_data_field_value(selector: Selector) -> Dict:
"""
提取data-field的值
Args:
selector:
Returns:
"""
data_field_value = selector.xpath("./@data-field").get(default='').strip()
if not data_field_value or data_field_value == "{}":
return {}
try:
# 先使用 html.unescape 处理转义字符 再json.loads 将 JSON 字符串转换为 Python 字典
unescaped_json_str = html.unescape(data_field_value)
data_field_dict_value = json.loads(unescaped_json_str)
except Exception as ex:
print(f"extract_data_field_value，错误信息：{ex}, 尝试使用其他方式解析")
data_field_dict_value = {}
return data_field_dict_value
def test_extract_search_note_list():
with open("test_data/search_keyword_notes.html", "r", encoding="utf-8") as f:
content = f.read()
@@ -140,7 +179,14 @@ def test_extract_note_detail():
result = extractor.extract_note_detail(content)
print(result.model_dump())
def test_extract_tieba_note_parment_comments():
with open("test_data/note_comments.html", "r", encoding="utf-8") as f:
content = f.read()
extractor = TieBaExtractor()
result = extractor.extract_tieba_note_parment_comments(content, "123456")
print(result)
if __name__ == '__main__':
test_extract_search_note_list()
test_extract_note_detail()
# test_extract_search_note_list()
# test_extract_note_detail()
test_extract_tieba_note_parment_comments()