Basic Implementation
Extract post content from Baidu Tieba by URL and save comments localy.
URL Structure Analysis
Baidu Tieba post URLs follow a consistent pattern: https://tieba.baidu.com/p/[POST_ID]
- View only original poster's comments: append
?see_lz=1 - View all comments: appand
?see_lz=0 - Pagination: add
&pn=[PAGE_NUMBER]
Implementation Strategy
- Determine whether to extract only original poster's content
- Parse the main page to identify total page count
- Extract content from each page
- Save results to local storage
Core Components
Fetch Page Content
def fetch_page_content(self, page_number):
try:
target_url = self.base_url + self.see_lz_param + '&pn=' + str(page_number)
print("Accessing URL: " + target_url)
request = urllib2.Request(target_url)
response = urllib2.urlopen(request)
return response.read().decode('utf-8')
except urllib2.URLError as error:
if hasattr(error, 'reason'):
print("Connection failed: " + error.reason)
return None
Determine Page Count
def get_total_pages(self, html_content):
page_pattern = re.compile('<li class="l_reply_num.*?</span>.*?<span.*?>(.*?)</span>', re.S)
match_result = re.search(page_pattern, html_content)
if match_result:
return match_result.group(1).strip()
return '1' # Default to one page if pattern not found
Extract Post Content
def extract_posts(self, html_content):
content_pattern = re.compile('<div id="post_content_.*?>(.*?)</div>', re.S)
matches = re.findall(content_pattern, html_content)
extracted_content = []
for match in matches:
processed_content = "\n" + self.cleaner.process_content(match) + "\n"
extracted_content.append(processed_content.encode('utf-8'))
return extracted_content
Save to Local File
def save_content(self, content_list):
for content_item in content_list:
if self.include_floor_markers == '1':
floor_separator = '\n' + str(self.current_floor) + "-" * 60 + '\n'
self.output_file.write(floor_separator)
self.output_file.write(content_item)
self.current_floor += 1
Copmlete Implementation
#!/usr/bin/python
# -*- coding: utf-8 -*-
import urllib
import urllib2
import re
import sys
reload(sys)
sys.setdefaultencoding('utf-8')
class ContentCleaner:
def __init__(self):
self.remove_images = re.compile('<img.*?>| {7}|')
self.remove_links = re.compile('<a.*?>|</a>')
self.replace_line_breaks = re.compile('<tr>|<div>|</div>|</p>')
self.replace_table_cells = re.compile('<td>')
self.replace_paragraphs = re.compile('<p.*?>')
self.replace_break_tags = re.compile('<br><br>|<br>')
self.remove_remaining_tags = re.compile('<.*?>')
def process_content(self, content):
content = re.sub(self.remove_images, "", content)
content = re.sub(self.remove_links, "", content)
content = re.sub(self.replace_line_breaks, "\n", content)
content = re.sub(self.replace_table_cells, "\t", content)
content = re.sub(self.replace_paragraphs, "\n ", content)
content = re.sub(self.replace_break_tags, "\n", content)
content = re.sub(self.remove_remaining_tags, "", content)
return content.strip()
class TiebaScraper:
def __init__(self, base_url, see_lz_setting, floor_markers):
self.base_url = base_url
self.see_lz_param = '?see_lz=' + str(see_lz_setting)
self.cleaner = ContentCleaner()
self.output_file = None
self.current_floor = 1
self.default_filename = "tieba_content"
self.include_floor_markers = floor_markers
def fetch_page_content(self, page_number):
try:
target_url = self.base_url + self.see_lz_param + '&pn=' + str(page_number)
print("Accessing URL: " + target_url)
request = urllib2.Request(target_url)
response = urllib2.urlopen(request)
return response.read().decode('utf-8')
except urllib2.URLError as error:
if hasattr(error, 'reason'):
print("Connection failed: " + error.reason)
return None
def get_total_pages(self, html_content):
page_pattern = re.compile('<li class="l_reply_num.*?</span>.*?<span.*?>(.*?)</span>', re.S)
match_result = re.search(page_pattern, html_content)
if match_result:
return match_result.group(1).strip()
return None
def extract_title(self, html_content):
title_patterns = [
re.compile('<h1 class="core_title_txt.*?>(.*?)</h1>', re.S),
re.compile('<h2 class="core_title_txt.*?>(.*?)</h2>', re.S),
re.compile('<h3 class="core_title_txt.*?>(.*?)</h3>', re.S)
]
for pattern in title_patterns:
match_result = re.search(pattern, html_content)
if match_result:
title = match_result.group(1).strip()
print(title)
return title
return None
def initialize_output_file(self, title):
if title:
filename = title + ".txt"
else:
filename = self.default_filename + ".txt"
self.output_file = open(filename, "w+")
self.output_file.write(self.base_url)
def extract_posts(self, html_content):
content_pattern = re.compile('<div id="post_content_.*?>(.*?)</div>', re.S)
matches = re.findall(content_pattern, html_content)
extracted_content = []
for match in matches:
processed_content = "\n" + self.cleaner.process_content(match) + "\n"
extracted_content.append(processed_content.encode('utf-8'))
return extracted_content
def save_content(self, content_list):
for content_item in content_list:
if self.include_floor_markers == '1':
floor_separator = '\n' + str(self.current_floor) + "-" * 60 + '\n'
self.output_file.write(floor_separator)
self.output_file.write(content_item)
self.current_floor += 1
def execute_scraping(self):
first_page = self.fetch_page_content(1)
total_pages = self.get_total_pages(first_page)
post_title = self.extract_title(first_page)
self.initialize_output_file(post_title)
if total_pages is None:
print("Invalid URL or page structure changed")
return
try:
print("Total pages: " + str(total_pages))
for page_num in range(1, int(total_pages) + 1):
page_content = self.fetch_page_content(page_num)
posts = self.extract_posts(page_content)
print("Writing page " + str(page_num))
self.save_content(posts)
except IOError as error:
print("File write error: " + error.message)
finally:
print("Scraping completed")
if __name__ == '__main__':
print("*" * 40)
print("** Baidu Tieba Scraper **")
print("** Version 2017-04-25 **")
print("*" * 40)
post_id = raw_input("Enter post ID: ")
base_url = "http://tieba.baidu.com/p/" + str(post_id)
see_lz_setting = raw_input("Original poster only? (1=yes, 0=no): ")
floor_markers = raw_input("Include floor markers? (1=yes, 0=no): ")
scraper = TiebaScraper(base_url, see_lz_setting, floor_markers)
scraper.execute_scraping()
Advanced Implementation
Extended version with multi-threading and comprehensive data extraction.
#!/usr/bin/python
# -*- coding: utf-8 -*-
import urllib2
import re
import threading
import time
import os
from HTMLParser import HTMLParser
class AdvancedTiebaScraper:
def __init__(self, keyword):
self.keyword = keyword
self.base_url = 'http://tieba.baidu.com/f?kw=%s' % self.keyword
self.posts_data = []
self.create_directory(self.keyword)
def create_directory(self, path):
if not os.path.exists(path):
os.makedirs(path)
def fetch_html(self, url):
headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 6.2; rv:16.0) Gecko/20100101 Firefox/16.0'}
request = urllib2.Request(url=url, headers=headers)
try:
response = urllib2.urlopen(request)
html_content = response.read().decode('utf-8')
return HTMLParser().unescape(html_content)
except urllib2.URLError as error:
print("URL Error: " + str(error))
return None
def extract_post_list(self, html_content):
post_pattern = re.compile('<li class=" j_thread_list clearfix.*?</li>', re.S)
posts = re.findall(post_pattern, html_content)
extracted_data = []
for post in posts:
post_info = {}
# Extract reply count, title, and URL
reply_pattern = re.compile('<span class="threadlist_rep_num.*?>(.*?)</span>.*?<a href="(.*?)" title="(.*?)"', re.S)
reply_match = re.search(reply_pattern, post)
if reply_match:
post_info['reply_count'] = reply_match.group(1)
post_info['url'] = 'http://tieba.baidu.com' + reply_match.group(2)
post_info['title'] = reply_match.group(3).strip()
if int(post_info['reply_count']) > 10: # Filter low-activity posts
extracted_data.append(post_info)
return extracted_data
def scrape_post_content(self, post_info):
post_content = self.fetch_html(post_info['url'] + '?see_lz=0&pn=1')
if post_content:
# Extract and process post content
content_pattern = re.compile('<div id="post_content_.*?>(.*?)</div>', re.S)
content_matches = re.findall(content_pattern, post_content)
processed_content = []
for content in content_matches:
# Clean and process content
cleaned_content = self.clean_html(content)
processed_content.append(cleaned_content)
return processed_content
return []
def clean_html(self, content):
# Remove HTML tags and clean content
clean_patterns = [
(re.compile('<.*?>'), ''),
(re.compile('\\s+'), ' ')
]
for pattern, replacement in clean_patterns:
content = re.sub(pattern, replacement, content)
return content.strip()
def execute(self):
main_page = self.fetch_html(self.base_url)
if main_page:
post_list = self.extract_post_list(main_page)
# Use threading for concurrent scraping
threads = []
for post_info in post_list[:3]: # Limit to first 3 posts
thread = threading.Thread(target=self.process_single_post, args=(post_info,))
threads.append(thread)
thread.start()
for thread in threads:
thread.join()
def process_single_post(self, post_info):
content = self.scrape_post_content(post_info)
filename = self.keyword + '/' + post_info['title'] + '.txt'
with open(filename, 'w') as output_file:
for item in content:
output_file.write(item + '\n')
if __name__ == '__main__':
keyword = raw_input("Enter tieba name: ")
if not keyword:
keyword = 'python'
scraper = AdvancedTiebaScraper(keyword)
scraper.execute()