Python爬虫实战Jimeng LoRA赋能智能数据采集与分析做数据采集的朋友应该都有过这样的经历辛辛苦苦写了个爬虫跑了一晚上第二天一看要么被网站封了IP要么抓回来的数据乱七八糟清洗起来比重新抓还费劲。更头疼的是有些数据光靠爬虫抓回来还不够还得分析出里面的门道比如电商评论里的情感倾向、新闻文章的关键信息提取这些传统方法处理起来特别麻烦。最近我在实际项目中尝试了一种新思路把Python爬虫和Jimeng LoRA技术结合起来效果出乎意料的好。简单来说就是用爬虫负责“抓”用LoRA负责“理”和“析”整个数据采集分析流程变得智能多了。今天就跟大家分享一下这套方案的具体做法特别是电商数据采集这个场景我会用完整的代码示例来演示。1. 为什么需要智能数据采集先说说我们平时做数据采集遇到的几个典型问题。反爬虫越来越难对付。现在稍微有点规模的网站反爬措施都做得相当完善。常见的像IP频率限制、请求头检测、验证码挑战还有那些动态加载的页面用传统的requestsBeautifulSoup组合经常碰壁。你可能遇到过这种情况同一个爬虫脚本白天能跑晚上就被封或者换个网络环境就不行了。数据清洗是个体力活。抓回来的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: fhttp://{proxy}, https: fhttp://{proxy} } def get_random_headers(self) - dict: 生成随机请求头 return { User-Agent: self.ua.random, Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.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, headersheaders, proxiesproxies, timeout30, **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-featuresAutomationControlled) options.add_experimental_option(excludeSwitches, [enable-automation]) options.add_experimental_option(useAutomationExtension, False) self.driver webdriver.Chrome(optionsoptions) 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(stripTrue) 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(stringlambda text: isinstance(text, Comment)): comment.extract() # 移除空白和短文本节点 for element in soup.find_all([div, p, span]): if element.get_text(stripTrue) : element.decompose() elif len(element.get_text(stripTrue)) 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(stripTrue) break if not main_content: # 如果没有找到特定容器提取body内容 body soup.find(body) if body: main_content body.get_text(stripTrue) # 进一步清理 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_dtypetorch.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_locationself.device) self.model.load_state_dict(lora_weights, strictFalse) print(LoRA权重加载成功) except Exception as e: print(fLoRA权重加载失败使用基础模型: {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_tensorspt, truncationTrue, max_length512).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*元, rRMB\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_featurestop_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 results5.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(), keylambda 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()], keylambda x: x[1], reverseTrue )[:10], top_topics: sorted( [(k, v) for k, v in topic_freq.items()], keylambda x: x[1], reverseTrue )[: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_okTrue) 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, encodingutf-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, encodingutf-8) as f: json.dump(price_history, f, ensure_asciiFalse, indent2) 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, encodingutf-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, encodingutf-8) as f: json.dump(result, f, ensure_asciiFalse, indent2) 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, encodingutf-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, encodingutf-8) as f: json.dump(report, f, ensure_asciiFalse, indent2) 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_hours6) # 或者单次执行 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星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。