Python爬虫实战：Jimeng LoRA赋能智能数据采集与分析

做数据采集的朋友应该都有过这样的经历：辛辛苦苦写了个爬虫，跑了一晚上，第二天一看，要么被网站封了IP，要么抓回来的数据乱七八糟，清洗起来比重新抓还费劲。更头疼的是，有些数据光靠爬虫抓回来还不够，还得分析出里面的门道，比如电商评论里的情感倾向、新闻文章的关键信息提取，这些传统方法处理起来特别麻烦。

最近我在实际项目中尝试了一种新思路，把Python爬虫和Jimeng LoRA技术结合起来，效果出乎意料的好。简单来说，就是用爬虫负责“抓”，用LoRA负责“理”和“析”，整个数据采集分析流程变得智能多了。今天就跟大家分享一下这套方案的具体做法，特别是电商数据采集这个场景，我会用完整的代码示例来演示。

1. 为什么需要智能数据采集？

先说说我们平时做数据采集遇到的几个典型问题。

反爬虫越来越难对付。现在稍微有点规模的网站，反爬措施都做得相当完善。常见的像IP频率限制、请求头检测、验证码挑战，还有那些动态加载的页面，用传统的requests+BeautifulSoup组合经常碰壁。你可能遇到过这种情况：同一个爬虫脚本，白天能跑，晚上就被封；或者换个网络环境就不行了。

数据清洗是个体力活。抓回来的HTML页面，里面什么都有：广告代码、导航栏、页脚信息、无关的脚本标签。提取正文内容就像是在垃圾堆里找宝贝，写一堆正则表达式和XPath规则，稍微页面结构一变，整个规则就得重写。更别说那些非结构化的数据了，比如商品描述里的规格参数、用户评论里的情感表达，手动处理效率太低了。

数据分析深度不够。数据抓回来存到数据库，这只是第一步。真正的价值在于分析：这些评论是正面还是负面？这篇文章主要讲了什么？这个商品的热点卖点是什么？传统方法要么靠人工看，要么写复杂的规则引擎，覆盖的场景有限，准确率也不稳定。

我之前帮一个电商团队做竞品分析，他们需要监控十几个竞争对手的商品信息、价格变动、用户评价。最初用传统方法，光是维护爬虫规则就占了一个人大部分时间，数据分析更是滞后。后来我们引入了Jimeng LoRA来做智能处理，效率提升了不止一个档次。

2. Jimeng LoRA：轻量级智能处理核心

可能有些朋友对Jimeng LoRA还不太熟悉，我简单解释一下。

你可以把LoRA理解为一个“智能插件”。它本身不是完整的AI模型，而是一个轻量级的适配层，可以加载到现有的基础模型上，让模型具备特定的能力。比如我们用的这个Jimeng LoRA，就是专门针对文本理解和信息提取优化的。

为什么选LoRA而不是完整的AI模型？ 主要是三个原因：速度快、资源省、效果好。

完整的AI模型动辄几个GB，部署和推理都需要大量计算资源。而LoRA文件通常只有几十到几百MB，加载速度快，运行效率高。更重要的是，LoRA可以针对特定任务进行优化，比如我们做数据清洗和分析，就需要一个擅长理解网页结构、提取关键信息、分析文本情感的专用工具。

在实际使用中，Jimeng LoRA表现出了几个很实用的特性：

• 上下文理解能力强：能区分网页的主内容、导航、广告等部分，准确提取正文
• 结构化信息提取：从非结构化文本中识别出价格、日期、人名、地点等实体
• 情感和主题分析：判断文本的情感倾向，提取核心主题和关键词
• 多语言支持：对中文网页的处理特别友好，包括各种网络用语和行业术语

下面我们看看怎么把爬虫和LoRA结合起来，构建一个完整的智能数据采集系统。

3. 系统架构设计

整个系统可以分为三个主要模块：智能爬虫模块、数据处理模块、LoRA分析模块。它们之间的关系是这样的：

爬虫抓取原始数据 → 初步清洗和解析 → LoRA深度处理 → 结构化输出

我画了一个简单的架构图，大家看了就明白了：

class SmartCrawlerSystem:
    """智能数据采集系统核心架构"""
    
    def __init__(self):
        # 爬虫配置
        self.crawler_config = {
            'concurrent_requests': 5,      # 并发请求数
            'delay_range': (1, 3),         # 请求延迟范围（秒）
            'retry_times': 3,              # 失败重试次数
            'timeout': 30                  # 请求超时时间
        }
        
        # 数据处理管道
        self.processing_pipeline = [
            'html_cleaning',      # HTML清洗
            'content_extraction', # 内容提取
            'entity_recognition', # 实体识别
            'sentiment_analysis', # 情感分析
            'topic_modeling'      # 主题建模
        ]
        
        # LoRA模型配置
        self.lora_config = {
            'model_path': './models/jimeng_lora',
            'device': 'cuda' if torch.cuda.is_available() else 'cpu',
            'batch_size': 8
        }

这个架构的关键在于管道化处理。数据像流水线一样经过各个处理环节，每个环节专注做好一件事，最后输出高质量的结构化数据。

4. 智能爬虫模块实现

爬虫模块的核心任务是稳定、高效地获取网页数据，同时要应对各种反爬措施。我总结了一套比较实用的策略组合。

4.1 反爬应对策略

IP轮换和代理池是最基础的。单一IP频繁请求，被封是迟早的事。我一般会准备多个代理IP，按一定的策略轮换使用。

import random
import time
from typing import List, Optional
import requests
from fake_useragent import UserAgent

class SmartCrawler:
    """智能爬虫核心类"""
    
    def __init__(self, proxy_pool: Optional[List[str]] = None):
        self.proxy_pool = proxy_pool or []
        self.ua = UserAgent()
        self.session = requests.Session()
        self.request_count = 0
        
    def get_random_proxy(self) -> Optional[dict]:
        """从代理池中随机选择一个代理"""
        if not self.proxy_pool:
            return None
            
        proxy = random.choice(self.proxy_pool)
        return {
            'http': f'http://{proxy}',
            'https': f'http://{proxy}'
        }
    
    def get_random_headers(self) -> dict:
        """生成随机请求头"""
        return {
            'User-Agent': self.ua.random,
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
            'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
            'Accept-Encoding': 'gzip, deflate',
            'Connection': 'keep-alive',
            'Upgrade-Insecure-Requests': '1'
        }
    
    def smart_request(self, url: str, **kwargs) -> Optional[requests.Response]:
        """智能请求方法，包含反爬策略"""
        
        # 控制请求频率
        if self.request_count > 0:
            delay = random.uniform(1, 3)
            time.sleep(delay)
        
        # 准备请求参数
        headers = kwargs.pop('headers', self.get_random_headers())
        proxies = kwargs.pop('proxies', self.get_random_proxy())
        
        max_retries = 3
        for attempt in range(max_retries):
            try:
                response = self.session.get(
                    url,
                    headers=headers,
                    proxies=proxies,
                    timeout=30,
                    **kwargs
                )
                
                # 检查是否被反爬
                if self._check_anti_crawler(response):
                    print(f"检测到反爬，切换策略重试...")
                    self._rotate_strategy()
                    continue
                    
                self.request_count += 1
                return response
                
            except Exception as e:
                print(f"请求失败 (尝试 {attempt + 1}/{max_retries}): {e}")
                if attempt < max_retries - 1:
                    time.sleep(2 ** attempt)  # 指数退避
                else:
                    return None
    
    def _check_anti_crawler(self, response: requests.Response) -> bool:
        """检查是否触发反爬机制"""
        status_code = response.status_code
        
        # 常见反爬状态码
        if status_code in [403, 429, 503]:
            return True
            
        # 检查页面内容中的反爬提示
        anti_keywords = ['access denied', 'forbidden', 'robot', 'captcha', '验证码']
        content_lower = response.text.lower()
        
        for keyword in anti_keywords:
            if keyword in content_lower:
                return True
                
        return False
    
    def _rotate_strategy(self):
        """切换反爬策略"""
        # 更换User-Agent
        self.session.headers.update(self.get_random_headers())
        
        # 更换代理
        if self.proxy_pool:
            new_proxy = self.get_random_proxy()
            self.session.proxies.update(new_proxy)
        
        # 增加延迟
        time.sleep(random.uniform(5, 10))

动态页面处理是另一个难点。现在很多网站都用JavaScript动态加载内容，传统的requests抓不到。这时候就需要用到Selenium或者Playwright这样的浏览器自动化工具。

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.common.exceptions import TimeoutException

class DynamicPageCrawler:
    """动态页面爬虫"""
    
    def __init__(self, headless: bool = True):
        options = webdriver.ChromeOptions()
        if headless:
            options.add_argument('--headless')
        options.add_argument('--disable-blink-features=AutomationControlled')
        options.add_experimental_option("excludeSwitches", ["enable-automation"])
        options.add_experimental_option('useAutomationExtension', False)
        
        self.driver = webdriver.Chrome(options=options)
        self.wait = WebDriverWait(self.driver, 10)
    
    def crawl_dynamic_page(self, url: str, wait_for: str = None):
        """爬取动态页面"""
        try:
            self.driver.get(url)
            
            # 等待页面加载
            if wait_for:
                self.wait.until(
                    EC.presence_of_element_located((By.CSS_SELECTOR, wait_for))
                )
            
            # 等待动态内容加载
            time.sleep(2)
            
            # 获取完整页面源码
            page_source = self.driver.page_source
            
            return page_source
            
        except TimeoutException:
            print(f"等待元素超时: {wait_for}")
            return self.driver.page_source
        except Exception as e:
            print(f"动态页面爬取失败: {e}")
            return None
    
    def close(self):
        """关闭浏览器"""
        self.driver.quit()

4.2 电商数据采集实战

以电商商品页面采集为例，我们需要抓取商品标题、价格、销量、评价、详情等信息。不同电商平台的页面结构差异很大，这就需要我们的爬虫足够灵活。

class EcommerceCrawler:
    """电商数据采集专用爬虫"""
    
    def __init__(self):
        self.crawler = SmartCrawler()
        self.product_selectors = {
            'title': [
                'h1.product-title', 
                '.product-name',
                '[class*="title"]'
            ],
            'price': [
                '.product-price',
                '.price',
                '[class*="price"]'
            ],
            'sales': [
                '.sales-count',
                '.sold',
                '[class*="sale"]'
            ],
            'rating': [
                '.product-rating',
                '.review-score',
                '[class*="rating"]'
            ]
        }
    
    def extract_product_info(self, html: str, url: str) -> dict:
        """提取商品信息"""
        from bs4 import BeautifulSoup
        
        soup = BeautifulSoup(html, 'html.parser')
        product_info = {
            'url': url,
            'title': self._extract_by_selectors(soup, self.product_selectors['title']),
            'price': self._extract_price(soup),
            'sales': self._extract_sales(soup),
            'rating': self._extract_rating(soup),
            'description': self._extract_description(soup),
            'reviews': self._extract_reviews(soup),
            'images': self._extract_images(soup)
        }
        
        return product_info
    
    def _extract_by_selectors(self, soup, selectors):
        """尝试多个选择器提取内容"""
        for selector in selectors:
            element = soup.select_one(selector)
            if element:
                return element.get_text(strip=True)
        return None
    
    def _extract_price(self, soup):
        """提取价格信息"""
        price_text = self._extract_by_selectors(soup, self.product_selectors['price'])
        if price_text:
            # 清理价格文本，提取数字
            import re
            numbers = re.findall(r'\d+\.?\d*', price_text)
            return float(numbers[0]) if numbers else None
        return None
    
    def crawl_product_page(self, url: str) -> dict:
        """爬取商品页面"""
        response = self.crawler.smart_request(url)
        if response and response.status_code == 200:
            product_info = self.extract_product_info(response.text, url)
            return product_info
        return None

这个电商爬虫的特点是选择器备选机制。同一个信息可能有多种CSS选择器可以匹配，我们按优先级尝试，提高提取的成功率。

5. 数据清洗与LoRA智能处理

数据抓回来之后，真正的挑战才开始。原始HTML包含大量噪音，我们需要先进行清洗，然后用LoRA进行深度处理。

5.1 智能数据清洗

传统的数据清洗主要靠正则表达式和规则，但网页结构千变万化，规则很难覆盖所有情况。我这里用了一个结合规则和机器学习的方法。

import re
from bs4 import BeautifulSoup, Comment
import justext

class SmartDataCleaner:
    """智能数据清洗器"""
    
    def __init__(self):
        # 常见噪音标签
        self.noise_tags = ['script', 'style', 'nav', 'footer', 'header', 'aside']
        
        # 广告和无关内容关键词
        self.ad_keywords = ['广告', '推荐', '热门', '相关', 'sponsor', 'advertisement']
        
    def clean_html(self, html: str) -> str:
        """清洗HTML，提取主要内容"""
        
        # 方法1：使用justext进行智能提取
        try:
            paragraphs = justext.justext(html, justext.get_stoplist("Chinese"))
            main_content = "\n".join([p.text for p in paragraphs if not p.is_boilerplate])
            if main_content and len(main_content) > 100:
                return main_content
        except:
            pass
        
        # 方法2：基于规则的清洗
        soup = BeautifulSoup(html, 'html.parser')
        
        # 移除噪音标签
        for tag in self.noise_tags:
            for element in soup.find_all(tag):
                element.decompose()
        
        # 移除注释
        for comment in soup.find_all(string=lambda text: isinstance(text, Comment)):
            comment.extract()
        
        # 移除空白和短文本节点
        for element in soup.find_all(['div', 'p', 'span']):
            if element.get_text(strip=True) == '':
                element.decompose()
            elif len(element.get_text(strip=True)) < 10:
                # 可能是导航或按钮文字
                element.decompose()
        
        # 提取正文（通常位于main、article或特定class中）
        main_content = ''
        for selector in ['main', 'article', '.content', '.main-content', '#content']:
            element = soup.select_one(selector)
            if element:
                main_content = element.get_text(strip=True)
                break
        
        if not main_content:
            # 如果没有找到特定容器，提取body内容
            body = soup.find('body')
            if body:
                main_content = body.get_text(strip=True)
        
        # 进一步清理
        main_content = self._post_clean(main_content)
        
        return main_content
    
    def _post_clean(self, text: str) -> str:
        """后处理清洗"""
        if not text:
            return text
        
        # 移除多余空白
        text = re.sub(r'\s+', ' ', text)
        
        # 移除常见页眉页脚内容
        lines = text.split('\n')
        cleaned_lines = []
        for line in lines:
            line_stripped = line.strip()
            if len(line_stripped) < 5:
                continue
                
            # 检查是否包含广告关键词
            is_ad = any(keyword in line_stripped for keyword in self.ad_keywords)
            if is_ad:
                continue
                
            cleaned_lines.append(line_stripped)
        
        return '\n'.join(cleaned_lines)

5.2 LoRA集成与智能分析

这是整个系统的核心部分。我们使用Jimeng LoRA来对清洗后的文本进行深度分析。

import torch
from transformers import AutoModelForSequenceClassification, AutoTokenizer
from typing import List, Dict, Any

class LoRAAnalyzer:
    """LoRA智能分析器"""
    
    def __init__(self, model_path: str = './models/jimeng_lora'):
        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
        
        # 加载基础模型和tokenizer
        self.tokenizer = AutoTokenizer.from_pretrained(model_path)
        self.model = AutoModelForSequenceClassification.from_pretrained(
            model_path,
            torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32
        ).to(self.device)
        
        # 加载LoRA权重
        self._load_lora_weights(model_path)
        
        self.model.eval()
    
    def _load_lora_weights(self, model_path: str):
        """加载LoRA权重"""
        # 这里简化了LoRA加载过程，实际使用时需要根据具体格式调整
        try:
            lora_weights = torch.load(f'{model_path}/lora_weights.bin', map_location=self.device)
            self.model.load_state_dict(lora_weights, strict=False)
            print("LoRA权重加载成功")
        except Exception as e:
            print(f"LoRA权重加载失败，使用基础模型: {e}")
    
    def analyze_text(self, text: str, tasks: List[str] = None) -> Dict[str, Any]:
        """分析文本内容"""
        if tasks is None:
            tasks = ['sentiment', 'entities', 'topics', 'summary']
        
        results = {}
        
        # 情感分析
        if 'sentiment' in tasks:
            results['sentiment'] = self._analyze_sentiment(text)
        
        # 实体识别
        if 'entities' in tasks:
            results['entities'] = self._extract_entities(text)
        
        # 主题提取
        if 'topics' in tasks:
            results['topics'] = self._extract_topics(text)
        
        # 文本摘要
        if 'summary' in tasks:
            results['summary'] = self._generate_summary(text)
        
        return results
    
    def _analyze_sentiment(self, text: str) -> Dict[str, float]:
        """情感分析"""
        inputs = self.tokenizer(text, return_tensors='pt', truncation=True, max_length=512).to(self.device)
        
        with torch.no_grad():
            outputs = self.model(**inputs)
            probabilities = torch.softmax(outputs.logits, dim=-1)
        
        # 假设模型输出为: [负面, 中性, 正面]
        sentiment_scores = {
            'negative': float(probabilities[0][0]),
            'neutral': float(probabilities[0][1]),
            'positive': float(probabilities[0][2])
        }
        
        return sentiment_scores
    
    def _extract_entities(self, text: str) -> List[Dict[str, str]]:
        """提取命名实体"""
        # 这里简化了实体识别过程
        # 实际使用时，Jimeng LoRA应该能识别多种实体类型
        entities = []
        
        # 价格实体
        price_patterns = [
            r'¥\s*(\d+(?:\.\d{1,2})?)',
            r'(\d+(?:\.\d{1,2})?)\s*元',
            r'RMB\s*(\d+(?:\.\d{1,2})?)'
        ]
        
        for pattern in price_patterns:
            matches = re.finditer(pattern, text)
            for match in matches:
                entities.append({
                    'text': match.group(0),
                    'type': 'PRICE',
                    'value': match.group(1)
                })
        
        # 日期实体
        date_pattern = r'\d{4}年\d{1,2}月\d{1,2}日|\d{4}-\d{1,2}-\d{1,2}'
        dates = re.findall(date_pattern, text)
        for date in dates:
            entities.append({
                'text': date,
                'type': 'DATE',
                'value': date
            })
        
        return entities
    
    def _extract_topics(self, text: str, top_k: int = 5) -> List[str]:
        """提取主题关键词"""
        # 这里使用简单的TF-IDF方法，实际LoRA应该有更好的主题建模能力
        from sklearn.feature_extraction.text import TfidfVectorizer
        import jieba
        
        # 中文分词
        words = jieba.lcut(text)
        filtered_words = [word for word in words if len(word) > 1]
        processed_text = ' '.join(filtered_words)
        
        # 计算TF-IDF
        vectorizer = TfidfVectorizer(max_features=top_k)
        tfidf_matrix = vectorizer.fit_transform([processed_text])
        
        # 获取关键词
        feature_names = vectorizer.get_feature_names_out()
        scores = tfidf_matrix.toarray()[0]
        
        topics = []
        for idx in scores.argsort()[-top_k:][::-1]:
            topics.append(feature_names[idx])
        
        return topics
    
    def _generate_summary(self, text: str, max_length: int = 150) -> str:
        """生成文本摘要"""
        # 这里简化了摘要生成，实际LoRA应该有摘要生成能力
        if len(text) <= max_length:
            return text
        
        # 简单的基于句子的摘要
        sentences = re.split(r'[。！？!?]', text)
        sentences = [s.strip() for s in sentences if s.strip()]
        
        if len(sentences) <= 3:
            return text[:max_length] + '...'
        
        # 选择前几个句子作为摘要
        summary_sentences = sentences[:3]
        summary = '。'.join(summary_sentences) + '。'
        
        if len(summary) > max_length:
            summary = summary[:max_length] + '...'
        
        return summary
    
    def batch_analyze(self, texts: List[str], tasks: List[str] = None) -> List[Dict[str, Any]]:
        """批量分析文本"""
        results = []
        for text in texts:
            result = self.analyze_text(text, tasks)
            results.append(result)
        return results

5.3 电商评论分析实战

电商场景下，用户评论分析特别有价值。我们来看一个完整的例子：

class EcommerceReviewAnalyzer:
    """电商评论分析器"""
    
    def __init__(self):
        self.cleaner = SmartDataCleaner()
        self.analyzer = LoRAAnalyzer()
    
    def analyze_product_reviews(self, reviews: List[str]) -> Dict[str, Any]:
        """分析商品评论"""
        if not reviews:
            return {}
        
        # 清洗评论
        cleaned_reviews = []
        for review in reviews:
            cleaned = self.cleaner.clean_html(review) if '<' in review else review
            cleaned_reviews.append(cleaned)
        
        # 批量分析
        analysis_results = self.analyzer.batch_analyze(
            cleaned_reviews,
            tasks=['sentiment', 'entities', 'topics']
        )
        
        # 汇总分析结果
        summary = self._summarize_analysis(analysis_results, cleaned_reviews)
        
        return summary
    
    def _summarize_analysis(self, results: List[Dict], reviews: List[str]) -> Dict[str, Any]:
        """汇总分析结果"""
        total_reviews = len(results)
        
        # 情感分布
        sentiment_counts = {'positive': 0, 'neutral': 0, 'negative': 0}
        sentiment_scores = []
        
        for result in results:
            sentiment = result.get('sentiment', {})
            if sentiment:
                # 确定主要情感
                max_sentiment = max(sentiment.items(), key=lambda x: x[1])
                sentiment_counts[max_sentiment[0]] += 1
                sentiment_scores.append(sentiment)
        
        # 提取高频实体
        all_entities = []
        for result in results:
            entities = result.get('entities', [])
            all_entities.extend(entities)
        
        # 统计实体频率
        entity_freq = {}
        for entity in all_entities:
            entity_type = entity['type']
            entity_text = entity['text']
            key = f"{entity_type}:{entity_text}"
            entity_freq[key] = entity_freq.get(key, 0) + 1
        
        # 提取高频主题
        all_topics = []
        for result in results:
            topics = result.get('topics', [])
            all_topics.extend(topics)
        
        topic_freq = {}
        for topic in all_topics:
            topic_freq[topic] = topic_freq.get(topic, 0) + 1
        
        # 生成分析报告
        report = {
            'total_reviews': total_reviews,
            'sentiment_distribution': {
                'positive': sentiment_counts['positive'],
                'neutral': sentiment_counts['neutral'],
                'negative': sentiment_counts['negative'],
                'positive_rate': sentiment_counts['positive'] / total_reviews if total_reviews > 0 else 0
            },
            'top_entities': sorted(
                [(k, v) for k, v in entity_freq.items()],
                key=lambda x: x[1],
                reverse=True
            )[:10],
            'top_topics': sorted(
                [(k, v) for k, v in topic_freq.items()],
                key=lambda x: x[1],
                reverse=True
            )[:10],
            'avg_sentiment_score': self._calculate_avg_sentiment(sentiment_scores)
        }
        
        return report
    
    def _calculate_avg_sentiment(self, sentiment_scores: List[Dict]) -> float:
        """计算平均情感得分"""
        if not sentiment_scores:
            return 0.5  # 中性
        
        total_score = 0
        for scores in sentiment_scores:
            # 计算综合情感得分：正面-负面
            sentiment_value = scores.get('positive', 0) - scores.get('negative', 0)
            total_score += (sentiment_value + 1) / 2  # 归一化到0-1
        
        return total_score / len(sentiment_scores)

6. 完整案例：电商竞品监控系统

最后，我把所有这些技术整合到一个完整的电商竞品监控系统中。这个系统可以自动监控竞争对手的商品信息、价格变动、用户评价，并生成分析报告。

import json
import pandas as pd
from datetime import datetime
import schedule
import time

class EcommerceCompetitorMonitor:
    """电商竞品监控系统"""
    
    def __init__(self, competitors: List[Dict]):
        self.competitors = competitors
        self.crawler = EcommerceCrawler()
        self.review_analyzer = EcommerceReviewAnalyzer()
        self.data_storage = './data/monitor_results'
        
        # 创建数据存储目录
        import os
        os.makedirs(self.data_storage, exist_ok=True)
    
    def monitor_product(self, product_url: str, competitor_name: str) -> Dict:
        """监控单个商品"""
        print(f"开始监控商品: {product_url}")
        
        # 爬取商品信息
        product_info = self.crawler.crawl_product_page(product_url)
        if not product_info:
            print(f"商品爬取失败: {product_url}")
            return None
        
        # 分析评论
        reviews = product_info.get('reviews', [])
        review_analysis = self.review_analyzer.analyze_product_reviews(reviews)
        
        # 整合监控结果
        monitor_result = {
            'competitor': competitor_name,
            'product_url': product_url,
            'timestamp': datetime.now().isoformat(),
            'product_info': product_info,
            'review_analysis': review_analysis,
            'price_history': self._update_price_history(product_url, product_info.get('price'))
        }
        
        # 保存结果
        self._save_monitor_result(monitor_result)
        
        # 检查价格变动
        price_alert = self._check_price_alert(product_url, product_info.get('price'))
        if price_alert:
            self._send_price_alert(price_alert)
        
        return monitor_result
    
    def _update_price_history(self, product_url: str, current_price: float) -> List[Dict]:
        """更新价格历史"""
        history_file = f'{self.data_storage}/price_history.json'
        
        try:
            with open(history_file, 'r', encoding='utf-8') as f:
                price_history = json.load(f)
        except:
            price_history = {}
        
        product_history = price_history.get(product_url, [])
        product_history.append({
            'timestamp': datetime.now().isoformat(),
            'price': current_price
        })
        
        # 只保留最近30天的记录
        product_history = product_history[-30:]
        price_history[product_url] = product_history
        
        with open(history_file, 'w', encoding='utf-8') as f:
            json.dump(price_history, f, ensure_ascii=False, indent=2)
        
        return product_history
    
    def _check_price_alert(self, product_url: str, current_price: float) -> Optional[Dict]:
        """检查价格变动是否需要告警"""
        history_file = f'{self.data_storage}/price_history.json'
        
        try:
            with open(history_file, 'r', encoding='utf-8') as f:
                price_history = json.load(f)
        except:
            return None
        
        product_history = price_history.get(product_url, [])
        if len(product_history) < 2:
            return None
        
        # 获取上次价格
        previous_price = product_history[-2]['price']
        
        if previous_price and current_price:
            price_change = (current_price - previous_price) / previous_price
            
            # 价格变动超过5%时告警
            if abs(price_change) > 0.05:
                return {
                    'product_url': product_url,
                    'previous_price': previous_price,
                    'current_price': current_price,
                    'change_percent': price_change * 100,
                    'direction': '上涨' if price_change > 0 else '下跌'
                }
        
        return None
    
    def _send_price_alert(self, alert_info: Dict):
        """发送价格变动告警"""
        # 这里可以集成邮件、钉钉、企业微信等通知方式
        message = f"价格变动告警！\n商品: {alert_info['product_url']}\n"
        message += f"原价: {alert_info['previous_price']}元\n"
        message += f"现价: {alert_info['current_price']}元\n"
        message += f"变动: {alert_info['direction']} {abs(alert_info['change_percent']):.1f}%"
        
        print(f"价格告警: {message}")
        # 实际项目中这里调用通知接口
    
    def _save_monitor_result(self, result: Dict):
        """保存监控结果"""
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        competitor = result['competitor']
        
        filename = f'{self.data_storage}/{competitor}_{timestamp}.json'
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(result, f, ensure_ascii=False, indent=2)
        
        print(f"监控结果已保存: {filename}")
    
    def generate_daily_report(self):
        """生成日报"""
        import glob
        
        # 获取当天的监控结果
        today = datetime.now().strftime('%Y%m%d')
        pattern = f'{self.data_storage}/*_{today}*.json'
        today_files = glob.glob(pattern)
        
        if not today_files:
            print("今天没有监控数据")
            return
        
        # 汇总数据
        all_results = []
        for file in today_files:
            with open(file, 'r', encoding='utf-8') as f:
                result = json.load(f)
                all_results.append(result)
        
        # 生成报告
        report = self._generate_summary_report(all_results)
        
        # 保存报告
        report_file = f'{self.data_storage}/daily_report_{today}.json'
        with open(report_file, 'w', encoding='utf-8') as f:
            json.dump(report, f, ensure_ascii=False, indent=2)
        
        print(f"日报已生成: {report_file}")
        return report
    
    def _generate_summary_report(self, results: List[Dict]) -> Dict:
        """生成汇总报告"""
        summary = {
            'report_date': datetime.now().strftime('%Y-%m-%d'),
            'total_products_monitored': len(results),
            'competitors': {},
            'price_trends': [],
            'sentiment_overview': {
                'avg_positive_rate': 0,
                'best_rated_products': [],
                'worst_rated_products': []
            }
        }
        
        # 按竞争对手汇总
        for result in results:
            competitor = result['competitor']
            if competitor not in summary['competitors']:
                summary['competitors'][competitor] = {
                    'product_count': 0,
                    'avg_price': 0,
                    'total_reviews': 0
                }
            
            comp_data = summary['competitors'][competitor]
            comp_data['product_count'] += 1
            
            # 价格信息
            price = result['product_info'].get('price')
            if price:
                comp_data['avg_price'] = (comp_data['avg_price'] * (comp_data['product_count'] - 1) + price) / comp_data['product_count']
            
            # 评论信息
            review_analysis = result.get('review_analysis', {})
            total_reviews = review_analysis.get('total_reviews', 0)
            comp_data['total_reviews'] += total_reviews
            
            # 情感分析
            sentiment = review_analysis.get('sentiment_distribution', {})
            positive_rate = sentiment.get('positive_rate', 0)
            
            # 记录最佳和最差评价商品
            product_title = result['product_info'].get('title', '未知商品')
            if positive_rate > 0.8:  # 正面评价超过80%
                summary['sentiment_overview']['best_rated_products'].append({
                    'product': product_title,
                    'positive_rate': positive_rate,
                    'competitor': competitor
                })
            elif positive_rate < 0.3:  # 正面评价低于30%
                summary['sentiment_overview']['worst_rated_products'].append({
                    'product': product_title,
                    'positive_rate': positive_rate,
                    'competitor': competitor
                })
        
        # 计算平均正面评价率
        if results:
            total_positive_rate = sum(
                r.get('review_analysis', {}).get('sentiment_distribution', {}).get('positive_rate', 0)
                for r in results
            )
            summary['sentiment_overview']['avg_positive_rate'] = total_positive_rate / len(results)
        
        return summary
    
    def start_monitoring(self, interval_hours: int = 6):
        """启动定时监控"""
        print(f"启动电商竞品监控，每{interval_hours}小时执行一次")
        
        def monitoring_job():
            print(f"开始执行监控任务: {datetime.now()}")
            for competitor in self.competitors:
                competitor_name = competitor['name']
                product_urls = competitor['product_urls']
                
                for url in product_urls:
                    try:
                        self.monitor_product(url, competitor_name)
                        time.sleep(random.uniform(2, 5))  # 避免请求过快
                    except Exception as e:
                        print(f"监控失败 {url}: {e}")
            
            # 生成日报（如果是当天的最后一次监控）
            current_hour = datetime.now().hour
            if current_hour >= 22:  # 晚上10点后生成日报
                self.generate_daily_report()
        
        # 定时执行
        schedule.every(interval_hours).hours.do(monitoring_job)
        
        # 立即执行一次
        monitoring_job()
        
        # 保持运行
        while True:
            schedule.run_pending()
            time.sleep(60)

# 使用示例
if __name__ == "__main__":
    # 配置竞争对手和商品
    competitors = [
        {
            'name': '竞品A',
            'product_urls': [
                'https://example.com/product/123',
                'https://example.com/product/456'
            ]
        },
        {
            'name': '竞品B', 
            'product_urls': [
                'https://competitor-b.com/item/789',
                'https://competitor-b.com/item/012'
            ]
        }
    ]
    
    # 创建监控系统
    monitor = EcommerceCompetitorMonitor(competitors)
    
    # 启动监控（在实际项目中，可以改为后台服务）
    # monitor.start_monitoring(interval_hours=6)
    
    # 或者单次执行
    for competitor in competitors:
        for url in competitor['product_urls']:
            result = monitor.monitor_product(url, competitor['name'])
            if result:
                print(f"监控完成: {result['product_info'].get('title', '未知商品')}")

7. 总结

这套结合Python爬虫和Jimeng LoRA的智能数据采集方案，在实际项目中跑了一段时间，效果确实不错。爬虫负责稳定高效地获取数据，LoRA负责深度理解和分析，两者配合起来，整个数据采集分析的流程都顺畅了很多。

从技术实现上看，有几个关键点值得注意。反爬策略要灵活组合，不能只依赖单一方法。数据清洗要兼顾规则和智能，既要有基本的过滤规则，也要有机器学习的方法来识别主要内容。LoRA的集成要考虑到实际部署的便利性，模型大小、推理速度、资源占用这些都要平衡。

在电商数据采集这个具体场景里，最大的价值在于能够实时监控竞品动态，自动分析用户反馈，及时发现市场变化。价格变动告警、评论情感分析、热点话题提取这些功能，对电商运营决策很有帮助。

当然，这套方案还有可以优化的地方。比如可以加入更多的数据源，不只是电商网站，还可以包括社交媒体、行业论坛、新闻网站等。分析维度也可以更丰富，比如竞品的营销策略分析、用户画像构建、趋势预测等。

技术总是在不断发展的，爬虫技术和AI模型的结合还有很多可能性。我觉得未来会有更多专门针对数据采集和分析优化的AI工具出现，让数据工作变得更智能、更高效。对于从事数据相关工作的朋友来说，掌握这些新技术，把传统的数据采集和现代的AI分析结合起来，会是一个很有价值的方向。

---

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。