用 PyQt5 爬虫智能体轻松爬取掘金，自动化采集技术文章和数据

爬虫技术已经成为了获取互联网上信息的强大工具。从新闻、博客到社交平台的数据抓取，爬虫都能为我们提供大量的有用资源。而今天，我想和大家分享如何利用 CodeBuddy 智能开发工具以及 PyQt5 框架，通过一个智能爬虫系统，自动化地爬取 掘金 上的技术文章和数据。

掘金是一个极受开发者欢迎的技术社区，涵盖了最新的技术动态、开发技巧和实战经验。无论你是想了解某个技术的最新文章，还是获取流行技术趋势的数据，掘金都提供了丰富的信息资源。那么，如何才能高效且智能地从掘金网站上提取出这些有价值的内容呢？这正是今天我们要探讨的主题。

为什么选择 CodeBuddy 和 PyQt5？

• CodeBuddy ：智能开发助手

CodeBuddy 是一款可以提高开发效率的智能开发工具，它不仅可以自动生成代码，还能提供代码命名、结构优化等辅助功能。通过 CodeBuddy ，我们能够将原本需要花费大量时间的开发工作量，缩短到几分钟以内，极大地提升编写爬虫代码的效率和准确性。尤其在复杂的爬虫项目中，CodeBuddy 能够实时分析爬虫的结构和逻辑，提供合适的建议，帮助我们避免重复劳动和潜在错误。

• PyQt5：轻量级桌面应用框架

PyQt5 是 Python 中最流行的桌面应用开发框架之一，它支持丰富的 UI 组件，并能够轻松构建跨平台的桌面应用。对于爬虫项目而言，PyQt5 可以帮助我们设计出用户友好的界面，提供灵活的交互方式，从而让整个爬取过程变得更加可视化和直观。更重要的是，PyQt5 可以与爬虫代码进行紧密集成，实现爬虫任务的动态显示与控制。

如何构建基于 PyQt5 的智能爬虫？

• 确定目标与需求

首先，我们需要明确自己爬取的目标内容。在掘金网站上，我们通常会关注以下几个方面的数据：

• 热门技术文章

• 各类专栏的内容

• 技术话题的最新动态

• 用户评论和点赞数等互动数据

我们可以根据需求，选择爬取不同类型的页面（如文章列表、文章详情页、评论区等）

• 设置爬虫智能体

# 网络爬虫专家智能体提示词

## 角色设定 你是一名世界顶尖的网络爬虫专家，精通各类网站数据抓取技术。你拥有10年以上爬虫开发经验，擅长处理反爬机制、动态内容加载、验证码破解等复杂场景。

## 能力范围

- 分析网站结构并设计高效爬取策略

- 处理JavaScript渲染的动态内容

- 绕过常见反爬机制(频率限制、User-Agent检测等)

- 设计分布式爬虫架构

- 数据清洗与存储方案

- 合法合规的爬取建议

## 交互准则

1. 首先询问用户想要爬取的目标网站及具体需求

2. 分析技术可行性并提供多种解决方案

3. 详细解释每种方案的技术细节和潜在挑战

4. 强调合法合规性，提醒用户遵守robots.txt和网站条款

5. 提供代码示例时注明语言和所需库

6. 对复杂任务建议分阶段实施

• 利用 CodeBuddy 提升开发效率

在开发爬虫时，CodeBuddy 的作用尤为重要。它能够帮助我们自动生成爬虫所需的代码结构，并给出合理的命名建议。例如，当我们定义一个爬虫函数时，CodeBuddy 可以基于函数的功能和上下文，自动生成函数名，减少命名冲突的可能。同时，CodeBuddy 还能分析我们的爬虫逻辑，帮助我们发现潜在的性能瓶颈和优化点。

PyQt5 界面与爬虫集成

在实现爬虫功能的同时，我们还可以利用 PyQt5 构建一个简洁的用户界面，方便我们启动和管理爬虫任务。通过界面上的按钮、进度条、文本框等组件，我们可以：

• 显示当前爬虫任务的进度

• 提供动态的控制按钮（例如“开始爬取”、“暂停”、“停止”）

• 显示爬取的数据统计信息（如成功爬取的文章数量）

这种集成不仅使得爬虫变得更加人性化，还能实时反馈爬取的进度和结果，让我们对爬虫的执行情况一目了然。

爬取掘金的挑战与应对策略

虽然掘金网站内容丰富，但爬虫过程中仍然存在一些挑战：

• 反爬机制：掘金可能会使用防爬虫技术（如验证码、IP 限制等）来限制频繁请求。为此，我们可以使用代理 IP 和随机请求头来规避反爬。

• 动态加载数据：掘金网站的内容可能是通过 JavaScript 动态加载的，我们需要使用类似 Selenium 或 Playwright 的工具，模拟浏览器行为来抓取动态内容。

• 数据清洗与处理：爬取的数据往往包含很多噪声信息，需要进一步清洗和整理，以确保其可用性。

指令合集

1. 初始化爬虫项目

指令：初始化一个新的爬虫项目，命名为 Crawler。

2. 导入所需库

指令： 导入请求库、HTML解析库和 PyQt5 库。

3. 爬虫目标设定

指令：设置爬虫目标为掘金网站，爬取技术文章、专栏和评论数据。

# Scrapy settings for JuejinCrawler project
#
# For simplicity, this file contains only settings considered important or
# commonly used. You can find more settings consulting the documentation:
#
#     https://docs.scrapy.org/en/latest/topics/settings.html
#     https://docs.scrapy.org/en/latest/topics/downloader-middleware.html
#     https://docs.scrapy.org/en/latest/topics/spider-middleware.html

BOT_NAME = "JuejinCrawler"

SPIDER_MODULES = ["JuejinCrawler.spiders"]
NEWSPIDER_MODULE = "JuejinCrawler.spiders"

ADDONS = {}

# 设置 User-Agent，模拟浏览器访问
USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36'

# 允许的域名
ALLOWED_DOMAINS = ['juejin.cn']

# 起始 URL
START_URLS = ['https://juejin.cn/']

# Obey robots.txt rules
ROBOTSTXT_OBEY = True

# Configure maximum concurrent requests performed by Scrapy (default: 16)
#CONCURRENT_REQUESTS = 32

# Configure a delay for requests for the same website (default: 0)
# See https://docs.scrapy.org/en/latest/topics/settings.html#download-delay
# See also autothrottle settings and docs
#DOWNLOAD_DELAY = 3
# The download delay setting will honor only one of:
#CONCURRENT_REQUESTS_PER_DOMAIN = 16
#CONCURRENT_REQUESTS_PER_IP = 16

# Disable cookies (enabled by default)
#COOKIES_ENABLED = False

# Disable Telnet Console (enabled by default)
#TELNETCONSOLE_ENABLED = False

# Override the default request headers:
#DEFAULT_REQUEST_HEADERS = {
#    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
#    "Accept-Language": "en",
#}

# Enable or disable spider middlewares
# See https://docs.scrapy.org/en/latest/topics/spider-middleware.html
#SPIDER_MIDDLEWARES = {
#    "JuejinCrawler.middlewares.JuejincrawlerSpiderMiddleware": 543,
#}

# Enable or disable downloader middlewares
# See https://docs.scrapy.org/en/latest/topics/downloader-middleware.html
#DOWNLOADER_MIDDLEWARES = {
#    "JuejinCrawler.middlewares.JuejincrawlerDownloaderMiddleware": 543,
#}

# Enable or disable extensions
# See https://docs.scrapy.org/en/latest/topics/extensions.html
#EXTENSIONS = {
#    "scrapy.extensions.telnet.TelnetConsole": None,
#}

# Configure item pipelines
# See https://docs.scrapy.org/en/latest/topics/item-pipeline.html
#ITEM_PIPELINES = {
#    "JuejinCrawler.pipelines.JuejincrawlerPipeline": 300,
#}

# Enable and configure the AutoThrottle extension (disabled by default)
# See https://docs.scrapy.org/en/latest/topics/autothrottle.html
#AUTOTHROTTLE_ENABLED = True
# The initial download delay
#AUTOTHROTTLE_START_DELAY = 5
# The maximum download delay to be set in case of high latencies
#AUTOTHROTTLE_MAX_DELAY = 60
# The average number of requests Scrapy should be sending in parallel to
# each remote server
#AUTOTHROTTLE_TARGET_CONCURRENCY = 1.0
# Enable showing throttling stats for every response received:
#AUTOTHROTTLE_DEBUG = False

# Enable and configure HTTP caching (disabled by default)
# See https://docs.scrapy.org/en/latest/topics/downloader-middleware.html#httpcache-middleware-settings
#HTTPCACHE_ENABLED = True
#HTTPCACHE_EXPIRATION_SECS = 0
#HTTPCACHE_DIR = "httpcache"
#HTTPCACHE_IGNORE_HTTP_CODES = []
#HTTPCACHE_STORAGE = "scrapy.extensions.httpcache.FilesystemCacheStorage"

# Set settings whose default value is deprecated to a future-proof value
FEED_EXPORT_ENCODING = "utf-8"

4. 定义数据请求模块

指令：创建一个数据请求模块，发送 GET 请求到掘金网站，并模拟浏览器的 User-Agent。

import requests
import random
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
import time

# 用户代理池
USER_AGENTS = [
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/15.1 Safari/605.1.15',
    'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
    'Mozilla/5.0 (Windows NT 10.0; WOW64; rv:45.0) Gecko/20100101 Firefox/45.0',
    # 可继续添加更多 User-Agent
]

# 代理池（示例，需替换为真实可用代理）
PROXIES = [
    'http://127.0.0.1:7890',
    # 'http://user:pass@proxyserver:port',
    # 可继续添加更多代理
]

def get_juejin(url, params=None):
    """
    发送 GET 请求到掘金网站，随机使用 User-Agent 和代理。
    :param url: 请求的完整 URL
    :param params: 可选的查询参数字典
    :return: 响应内容（text），如请求失败返回 None
    """
    headers = {
        'User-Agent': random.choice(USER_AGENTS)
    }
    proxies = None
    if PROXIES:
        proxy = random.choice(PROXIES)
        proxies = {
            'http': proxy,
            'https': proxy
        }
    try:
        response = requests.get(url, headers=headers, params=params, proxies=proxies, timeout=10)
        response.raise_for_status()
        return response.text
    except requests.RequestException as e:
        print(f"请求失败: {e}")
        return None

def get_juejin_dynamic(url, wait_time=3, headless=True):
    """
    使用 Selenium 加载掘金网站动态内容。
    :param url: 目标 URL
    :param wait_time: 页面加载等待时间（秒）
    :param headless: 是否无头模式
    :return: 完整渲染后的页面 HTML
    """
    chrome_options = Options()
    if headless:
        chrome_options.add_argument('--headless')
    chrome_options.add_argument('--disable-gpu')
    chrome_options.add_argument('--no-sandbox')
    chrome_options.add_argument(f'user-agent={random.choice(USER_AGENTS)}')
    chrome_options.add_argument('--ignore-certificate-errors')
    chrome_options.add_argument('--ignore-ssl-errors')
    # 如需代理，可添加如下参数：
    # chrome_options.add_argument('--proxy-server=http://127.0.0.1:7890')
    driver = webdriver.Chrome(options=chrome_options)
    try:
        driver.get(url)
        time.sleep(wait_time)  # 等待页面动态内容加载
        html = driver.page_source
        return html
    finally:
        driver.quit() 

5. 定义页面解析模块

指令：创建页面解析模块，使用 BeautifulSoup 提取文章标题、作者和发布时间。

from bs4 import BeautifulSoup

def parse_article(html):
    """
    解析掘金文章页面，提取标题、作者和发布时间。
    :param html: 文章页面的 HTML 内容
    :return: 字典，包含 title、author、publish_time
    """
    soup = BeautifulSoup(html, 'lxml')
    # 标题
    title_tag = soup.find('h1')
    title = title_tag.get_text(strip=True) if title_tag else None
    # 作者
    author_tag = soup.find('a', class_='username')
    author = author_tag.get_text(strip=True) if author_tag else None
    # 发布时间
    time_tag = soup.find('time')
    publish_time = time_tag.get_text(strip=True) if time_tag else None
    return {
        'title': title,
        'author': author,
        'publish_time': publish_time
    } 

6. 存储模块

指令：创建一个存储模块，将爬取的文章数据保存到 CSV 文件中。

import csv
import os

def save_articles_to_csv(articles, filename='articles.csv'):
    """
    将文章数据保存到 CSV 文件。
    :param articles: 文章数据列表，每个元素为字典，包含 title、author、publish_time
    :param filename: 保存的文件名，默认为 articles.csv
    """
    if not articles:
        print("没有可保存的数据。")
        return
    fieldnames = ['title', 'author', 'publish_time']
    file_exists = os.path.isfile(filename)
    with open(filename, 'a', newline='', encoding='utf-8') as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        if not file_exists:
            writer.writeheader()
        for article in articles:
            writer.writerow(article)
    print(f"已保存 {len(articles)} 条数据到 {filename}")

def save_articles_to_txt(articles, filename='articles.txt'):
    """
    将文章数据保存到 TXT 文件，每条数据一行，字段用制表符分隔。
    :param articles: 文章数据列表，每个元素为字典，包含 title、author、publish_time
    :param filename: 保存的文件名，默认为 articles.txt
    """
    if not articles:
        print("没有可保存的数据。")
        return
    fieldnames = ['title', 'author', 'publish_time']
    with open(filename, 'a', encoding='utf-8') as txtfile:
        for article in articles:
            line = '\t'.join(str(article.get(field, '')) for field in fieldnames)
            txtfile.write(line + '\n')
    print(f"已保存 {len(articles)} 条数据到 {filename}") 

7. PyQt5 界面

指令：创建 PyQt5 界面，包含开始、暂停、停止按钮，进度条和日志输出框。

8. 任务调度模块

指令：创建任务调度模块，手动触发爬虫任务的启动与停止。

from PyQt5.QtWidgets import (QWidget, QPushButton, QVBoxLayout, QHBoxLayout, QProgressBar, QTextEdit, QApplication, QLabel)
from PyQt5.QtCore import pyqtSignal
import sys
import threading
import time
from requester import get_juejin_dynamic
from article_parser import parse_article
from cleaner import clean_article
from storage import save_articles_to_txt

class MainWindow(QWidget):
    log_signal = pyqtSignal(str)
    progress_signal = pyqtSignal(int)

    def __init__(self):
        super().__init__()
        self.init_ui()
        self._running = False
        self._thread = None
        self.log_signal.connect(self.log)
        self.progress_signal.connect(self.progress_bar.setValue)

    def init_ui(self):
        # 按钮
        self.start_btn = QPushButton('开始')
        self.pause_btn = QPushButton('暂停')
        self.stop_btn = QPushButton('停止')

        # 进度条
        self.progress_bar = QProgressBar()
        self.progress_bar.setValue(0)

        # 日志输出框
        self.log_output = QTextEdit()
        self.log_output.setReadOnly(True)

        # 布局
        btn_layout = QHBoxLayout()
        btn_layout.addWidget(self.start_btn)
        btn_layout.addWidget(self.pause_btn)
        btn_layout.addWidget(self.stop_btn)

        main_layout = QVBoxLayout()
        main_layout.addLayout(btn_layout)
        main_layout.addWidget(QLabel('进度'))
        main_layout.addWidget(self.progress_bar)
        main_layout.addWidget(QLabel('日志输出'))
        main_layout.addWidget(self.log_output)

        self.setLayout(main_layout)
        self.setWindowTitle('掘金爬虫工具')
        self.resize(500, 400)
        self.start_btn.clicked.connect(self.start_crawl)
        self.stop_btn.clicked.connect(self.stop_crawl)
        self.pause_btn.setEnabled(False)

    def log(self, msg):
        self.log_output.append(msg)
        self.log_output.ensureCursorVisible()

    def start_crawl(self):
        if self._running:
            self.log_signal.emit('爬虫正在运行中...')
            return
        self._running = True
        self.progress_signal.emit(0)
        self.log_signal.emit('开始爬取掘金首页第一篇文章...')
        self._thread = threading.Thread(target=self.crawl_task)
        self._thread.start()

    def stop_crawl(self):
        if not self._running:
            self.log_signal.emit('爬虫未在运行。')
            return
        self._running = False
        self.log_signal.emit('已请求停止爬虫任务。')

    def crawl_task(self):
        try:
            url = 'https://juejin.cn/'
            html = get_juejin_dynamic(url)
            if not html:
                self.log_signal.emit('页面加载失败。')
                self._running = False
                return
            self.progress_signal.emit(20)
            # 解析首页，找到第一篇文章链接
            import lxml.html
            doc = lxml.html.fromstring(html)
            article_links = doc.xpath('//a[contains(@href, "/post/")]/@href')
            if not article_links:
                self.log_signal.emit('未找到文章链接。')
                self._running = False
                return
            first_article_url = 'https://juejin.cn' + article_links[0]
            self.log_signal.emit(f'获取到第一篇文章链接: {first_article_url}')
            self.progress_signal.emit(40)
            article_html = get_juejin_dynamic(first_article_url)
            if not article_html:
                self.log_signal.emit('文章页面加载失败。')
                self._running = False
                return
            self.progress_signal.emit(60)
            article_data = parse_article(article_html)
            if not article_data.get('title'):
                self.log_signal.emit('未能正确解析文章内容。')
                self._running = False
                return
            self.log_signal.emit(f"原始数据: {article_data}")
            cleaned = clean_article(article_data)
            self.log_signal.emit(f"清洗后数据: {cleaned}")
            save_articles_to_txt([cleaned])
            self.progress_signal.emit(100)
            self.log_signal.emit('数据已保存到 articles.txt')
        except Exception as e:
            self.log_signal.emit(f'发生异常: {e}')
        finally:
            self._running = False

if __name__ == '__main__':
    app = QApplication(sys.argv)
    window = MainWindow()
    window.show()
    sys.exit(app.exec_()) 

9. 反爬策略处理

指令：设置代理 IP 和用户代理来避免反爬。

import threading

class CrawlerScheduler:
    def __init__(self, crawl_func):
        """
        :param crawl_func: 爬虫主函数，需为可调用对象
        """
        self.crawl_func = crawl_func
        self._thread = None
        self._running = False

    def start(self, *args, **kwargs):
        if self._running:
            print("爬虫已在运行中。")
            return
        self._running = True
        self._thread = threading.Thread(target=self._run, args=args, kwargs=kwargs)
        self._thread.start()
        print("爬虫任务已启动。")

    def _run(self, *args, **kwargs):
        try:
            self.crawl_func(*args, **kwargs)
        finally:
            self._running = False
            print("爬虫任务已结束。")

    def stop(self):
        if not self._running:
            print("爬虫未在运行。")
            return
        # 这里只能通过设置标志位让爬虫任务自行检查并退出
        self._running = False
        print("已请求停止爬虫任务。请确保爬虫主函数支持中断。")

    def is_running(self):
        return self._running 

10. 动态内容加载处理

指令：使用 Selenium 模拟浏览器来加载掘金网站的动态内容。

11. 清洗与处理爬取数据

指令：创建数据清洗模块，执行去除空值、日期格式化和提取关键词等操作。

import re
from datetime import datetime

from typing import List, Dict

def clean_article(article: Dict) -> Dict:
    """
    清洗单条文章数据：去除空值、日期格式化、提取关键词。
    :param article: 包含 title、author、publish_time 等字段的字典
    :return: 清洗后的字典
    """
    # 去除空值
    cleaned = {k: (v.strip() if isinstance(v, str) else v) for k, v in article.items() if v and str(v).strip()}

    # 日期格式化（假设原始格式为 '2024-05-01 12:34' 或 '2024/05/01' 等）
    if 'publish_time' in cleaned:
        cleaned['publish_time'] = format_date(cleaned['publish_time'])

    # 提取关键词（以标题为例，简单分词，实际可用更强分词工具）
    if 'title' in cleaned:
        cleaned['keywords'] = extract_keywords(cleaned['title'])

    return cleaned

def format_date(date_str: str) -> str:
    """
    尝试将日期字符串格式化为标准 YYYY-MM-DD 格式。
    :param date_str: 原始日期字符串
    :return: 格式化后的日期字符串
    """
    for fmt in ('%Y-%m-%d %H:%M', '%Y-%m-%d', '%Y/%m/%d', '%Y.%m.%d'):
        try:
            dt = datetime.strptime(date_str, fmt)
            return dt.strftime('%Y-%m-%d')
        except Exception:
            continue
    # 若无法识别，原样返回
    return date_str

def extract_keywords(text: str) -> List[str]:
    """
    简单提取关键词（以中文、英文分词为例，实际可用 jieba 等分词库）。
    :param text: 输入文本
    :return: 关键词列表
    """
    # 仅保留中英文、数字
    text = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9 ]', ' ', text)
    # 按空格和常见分隔符分词
    words = re.split(r'[\s,，。.!?、]+', text)
    # 去除过短的词
    keywords = [w for w in words if len(w) > 1]
    return keywords 

12. 运行爬虫任务

我们来看一下最终的效果吧

自动化爬取，让技术文章触手可得

通过利用 CodeBuddy 和 PyQt5 的结合，我们可以轻松地开发一个功能强大的爬虫系统，自动化地抓取掘金网站上的技术文章和相关数据。不仅如此，使用 CodeBuddy 能够让我们在开发过程中更加高效、精准，减少代码错误和不必要的重复劳动。最终，我们得到的不仅仅是一个爬虫，而是一个可以高效运营、具备可视化管理界面的智能化爬虫系统，完美符合现代开发者的需求。