Automated Blog Metrics Aggregation and Export with Python

Extracting and analyzing publication metrics from a personal technical blog requires handling dynamic pagination, parsing structured HTML, normalizing extracted text, and persisting the results. A modular Python approach separates network requests, DOM traversal, data transformation, and file export into distinct components.

Network Request and Session Management

Maintaining a persistent session reduces connection overhead and preserves cookies across paginated requests. Configuring standard browser headers prevents basic bot detection mechanisms from blocking the scraper.

import requests
from bs4 import BeautifulSoup
import pandas as pd
import os
import re
from datetime import datetime
from typing import List, Dict, Optional

class BlogMetricsCollector:
    def __init__(self, base_profile_url: str):
        self.profile_url = base_profile_url
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/115.0.0.0 Safari/537.36",
            "Accept-Language": "en-US,en;q=0.9"
        })
        self.collected_records: List[Dict] = []

    def fetch_html(self, target_url: str) -> Optional[str]:
        try:
            response = self.session.get(target_url, timeout=10)
            response.raise_for_status()
            return response.text
        except requests.RequestException as err:
            print(f"[Network Error] Failed to retrieve {target_url}: {err}")
            return None

Pagination Detection and DOM Parsing

Blog archives typically split entries across multiple pages. The scraper must first identify the total page count by inspecting the pagination container, then iterate through each page to extract article metadata. Regular expresssions clean non-numeric characters from view counts.

    def discover_pagination(self, html_content: str) -> int:
        soup = BeautifulSoup(html_content, "lxml")
        pager_container = soup.select_one(".pagination-wrapper, .page-nav, #papelist")
        if not pager_container:
            return 1
        
        page_links = pager_container.find_all("a")
        if not page_links:
            return 1
        
        last_page_text = page_links[-1].get_text(strip=True)
        match = re.search(r"\d+", last_page_text)
        return int(match.group()) if match else 1

    def extract_page_entries(self, html_content: str, base_domain: str) -> None:
        soup = BeautifulSoup(html_content, "lxml")
        article_blocks = soup.select(".article-list-item, .list-item, .blog-list-box")
        
        for block in article_blocks:
            title_tag = block.select_one("h4 a, .title a, a[href*='details']")
            views_tag = block.select_one(".read-num, .link_view, .view-count")
            comments_tag = block.select_one(".link_comments, .comment-num")
            
            if not title_tag:
                continue
                
            raw_views = views_tag.get_text(strip=True) if views_tag else "0"
            view_count = int(re.sub(r"\D", "", raw_views)) or 0
            
            raw_comments = comments_tag.get_text(strip=True) if comments_tag else "0"
            comment_count = int(re.sub(r"\D", "", raw_comments)) or 0
            
            relative_link = title_tag.get("href", "")
            full_link = relative_link if relative_link.startswith("http") else f"{base_domain}{relative_link}"
            
            self.collected_records.append({
                "title": title_tag.get_text(strip=True),
                "views": view_count,
                "comments": comment_count,
                "url": full_link
            })

Data Sorting and Export Pipeline

Once all pages are processed, the dataset requires sorting based on engagement metrics. Exporting to both plain text and spreadsheet formats ensures compatibility with different enalysis workflows. Using pandas simplifies DataFrame creation, sorting, and Excel generation with automatic index handling.

    def process_and_export(self, output_dir: str = "blog_metrics") -> None:
        if not self.collected_records:
            print("[Warning] No data collected. Exiting export pipeline.")
            return

        os.makedirs(output_dir, exist_ok=True)
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        
        df = pd.DataFrame(self.collected_records)
        df_sorted = df.sort_values(by="views", ascending=True).reset_index(drop=True)
        df_sorted.index += 1  # 1-based ranking
        
        txt_path = os.path.join(output_dir, f"metrics_{timestamp}.txt")
        with open(txt_path, "w", encoding="utf-8") as f:
            f.write(f"Total Articles: {len(df_sorted)}\n")
            f.write("-" * 80 + "\n")
            for idx, row in df_sorted.iterrows():
                line = f"Rank {idx} | {row['title']} | Views: {row['views']} | Comments: {row['comments']} | {row['url']}\n"
                f.write(line)
                
        excel_path = os.path.join(output_dir, f"metrics_{timestamp}.xlsx")
        df_sorted.to_excel(excel_path, index_label="Rank")
        print(f"[Success] Exported {len(df_sorted)} records to {excel_path}")

Execution Workflow

The main routine orchestrates the collection process. It initializes the collector, resolves the pagination boundary, iterates through the archive URLs, and triggers the export sequence. Error handling ensures partial failures do not halt the entire batch.

    def run_collection(self, base_domain: str = "https://blog.csdn.net") -> None:
        print(f"[Init] Starting metrics collection for {self.profile_url}")
        initial_html = self.fetch_html(self.profile_url)
        if not initial_html:
            return

        total_pages = self.discover_pagination(initial_html)
        print(f"[Pagination] Detected {total_pages} archive pages.")

        for page_num in range(1, total_pages + 1):
            page_url = f"{self.profile_url.rstrip('/')}/article/list/{page_num}"
            print(f"[Fetching] Processing page {page_num}/{total_pages}...")
            page_html = self.fetch_html(page_url)
            if page_html:
                self.extract_page_entries(page_html, base_domain)
                
        self.process_and_export()

if __name__ == "__main__":
    TARGET_BLOG = "https://blog.csdn.net/your_username"
    collector = BlogMetricsCollector(TARGET_BLOG)
    collector.run_collection()

The architecture isolates network I/O from parsing logic, alowing easy substitution of HTTP clients or HTML parsers. DataFrame operations replace manual list sorting and dictionary manipulation, reducing boilerplate while improving performance on large datasets. Timestamped output directories prevent accidental overwrites during repeated execution cycles.

Tags: python web-scraping data-analysis automation Pandas

Posted on Tue, 08 Sep 2026 16:14:16 +0000 by Yaak