import requests from bs4 import BeautifulSoup import json import time import re import pandas as pd import random from urllib.parse import quote, urljoin from datetime import datetime import logging # 设置日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class HuxiuSpider: def __init__(self): self.session = requests.Session() self.headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7', 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', 'Accept-Encoding': 'gzip, deflate, br', 'Connection': 'keep-alive', 'Upgrade-Insecure-Requests': '1', } self.base_url = "https://www.huxiu.com" def random_delay(self): """随机延迟""" time.sleep(random.uniform(2, 4)) def get_with_retry(self, url, retries=3): """带重试的请求""" for attempt in range(retries): try: response = self.session.get(url, headers=self.headers, timeout=15) if response.status_code == 200: return response else: logger.warning(f"请求返回状态码 {response.status_code}: {url}") except Exception as e: logger.warning(f"请求失败 {url} (尝试 {attempt + 1}/{retries}): {e}") if attempt < retries - 1: time.sleep(2 ** attempt) return None def get_articles_from_homepage(self): """从首页获取文章""" try: logger.info("从虎嗅网首页获取文章...") response = self.get_with_retry(self.base_url) if not response: return [] soup = BeautifulSoup(response.text, 'html.parser') articles = [] # 查找文章卡片 article_selectors = [ '.article-item', '.mod-art', '.vertical-article', '.article-card', '.newsfeed-item' ] for selector in article_selectors: items = soup.select(selector) if items: logger.info(f"找到 {len(items)} 个文章项目，使用选择器: {selector}") for item in items[:20]: # 限制数量 article = self.parse_article_card(item) if article: articles.append(article) break return articles except Exception as e: logger.error(f"从首页获取文章时出错: {e}") return [] def parse_article_card(self, item): """解析文章卡片""" try: # 查找标题 title_elem = item.find('a', href=re.compile(r'/article/\d+\.html')) if not title_elem: return None title = title_elem.get_text().strip() link = title_elem.get('href') if link and not link.startswith('http'): link = urljoin(self.base_url, link) # 查找摘要 summary_elem = item.find('p', class_=re.compile(r'brief|summary|desc')) if not summary_elem: summary_elem = item.find('div', class_=re.compile(r'brief|summary|desc')) summary = summary_elem.get_text().strip() if summary_elem else "" # 查找作者和时间 author_elem = item.find(['span', 'a'], class_=re.compile(r'author|writer')) author = author_elem.get_text().strip() if author_elem else "" time_elem = item.find('span', class_=re.compile(r'time|date')) publish_time = time_elem.get_text().strip() if time_elem else "" return { 'title': title, 'link': link, 'summary': summary, 'author': author, 'publish_time': publish_time, 'source': 'homepage' } except Exception as e: logger.error(f"解析文章卡片时出错: {e}") return None def search_articles_direct(self, keywords): """直接搜索文章""" all_articles = [] for keyword in keywords: logger.info(f"直接搜索关键词: {keyword}") try: # 使用虎嗅的搜索页面 search_url = f"https://www.huxiu.com/search.html?keyword={quote(keyword)}" response = self.get_with_retry(search_url) if not response: continue soup = BeautifulSoup(response.text, 'html.parser') # 解析搜索结果 search_results = soup.find_all('div', class_=re.compile(r'article-item|search-result')) for item in search_results: article = self.parse_search_result(item, keyword) if article: all_articles.append(article) logger.info(f"关键词 '{keyword}' 找到 {len(search_results)} 个结果") self.random_delay() except Exception as e: logger.error(f"搜索关键词 '{keyword}' 时出错: {e}") continue return all_articles def parse_search_result(self, item, keyword): """解析搜索结果""" try: title_elem = item.find('a', href=re.compile(r'/article/\d+\.html')) if not title_elem: return None title = title_elem.get_text().strip() link = title_elem.get('href') if link and not link.startswith('http'): link = urljoin(self.base_url, link) # 检查标题是否包含关键词 if not any(word in title for word in ['大模型', 'LLM', '语言模型', 'GPT', 'ChatGPT', 'AI', '人工智能']): return None summary_elem = item.find('p', class_=re.compile(r'brief|summary')) summary = summary_elem.get_text().strip() if summary_elem else "" return { 'title': title, 'link': link, 'summary': summary, 'keyword': keyword, 'source': 'direct_search' } except Exception as e: logger.error(f"解析搜索结果时出错: {e}") return None def get_article_content(self, article_url): """获取文章详细内容""" try: logger.info(f"获取文章内容: {article_url}") response = self.get_with_retry(article_url) if not response: return self.get_empty_content() soup = BeautifulSoup(response.text, 'html.parser') # 查找文章内容区域 content_selectors = [ '.article-content-wrap', '.article-content', '.article-detail-content', '.article-main-content', '.content' ] content = "" for selector in content_selectors: content_elem = soup.select_one(selector) if content_elem: # 清理脚本和样式 for tag in content_elem(['script', 'style', 'nav', 'footer', 'aside']): tag.decompose() content = content_elem.get_text().strip() content = re.sub(r'\s+', ' ', content) if len(content) > 200: break # 如果没找到内容，尝试其他选择器 if not content or len(content) < 200: # 尝试获取所有段落 paragraphs = soup.find_all('p') content = ' '.join([p.get_text().strip() for p in paragraphs if len(p.get_text().strip()) > 20]) content = re.sub(r'\s+', ' ', content) main_points = self.extract_main_points(content) return { 'full_content': content, 'main_points': main_points, 'content_length': len(content), 'source_method': 'web' } except Exception as e: logger.error(f"获取文章内容时出错 {article_url}: {e}") return self.get_empty_content() def get_empty_content(self): """返回空内容结构""" return { 'full_content': "无法获取内容", 'main_points': "无法提取主要观点", 'content_length': 0, 'source_method': 'failed' } def extract_main_points(self, content): """提取主要观点""" if not content or len(content) < 50: return "内容过短，无法提取主要观点" # 分割成句子 sentences = re.split(r'[。！？.!?]', content) sentences = [s.strip() for s in sentences if len(s.strip()) > 15] # 寻找包含关键词的句子 key_sentences = [] keyword_groups = [ ['大模型', 'LLM', '语言模型'], ['GPT', 'ChatGPT', 'OpenAI'], ['人工智能', 'AI', '智能'], ['认为', '观点', '应该', '需要', '重要', '关键'], ['趋势', '发展', '未来', '前景'] ] for sentence in sentences: for keywords in keyword_groups: if any(keyword in sentence for keyword in keywords): if sentence not in key_sentences: key_sentences.append(sentence) break # 如果关键词句子不够，取前几个有意义的句子 if len(key_sentences) < 3: key_sentences.extend([s for s in sentences if len(s) > 30][:5-len(key_sentences)]) # 去重并限制长度 unique_sentences = [] seen = set() for s in key_sentences: if s not in seen: seen.add(s) unique_sentences.append(s) main_points = "。".join(unique_sentences[:5]) + "。" # 如果内容较短，直接返回前300字符 if len(main_points) < 50: main_points = content[:300] + "..." if len(content) > 300 else content return main_points def analyze_article_themes(self, content, title): """分析文章主题""" themes = [] content_lower = content.lower() title_lower = title.lower() theme_keywords = { '技术发展': ['技术', '算法', '架构', '训练', '参数', '模型结构', 'transformer', '神经网络'], '商业应用': ['商业', '应用', '落地', '场景', '企业', '客户', '产品', '服务', '商业化'], '投资融资': ['投资', '融资', '资本', '基金', '估值', '融资', '市值', '投资'], '市场竞争': ['竞争', '市场', '对手', '领先', '优势', '市场份额', 'bat', '微软', '谷歌'], '政策监管': ['政策', '监管', '法规', '合规', '政府', '立法', '安全', '伦理'], '挑战风险': ['挑战', '风险', '问题', '困难', '局限', '不足', '缺陷', '担忧'], '未来趋势': ['趋势', '未来', '预测', '展望', '方向', '发展', '前景', '机会'], '开源生态': ['开源', '社区', '生态', '开放', '贡献', '协作', '开源'], '行业影响': ['行业', '影响', '变革', '革命', '颠覆', '创新', '改变'] } for theme, keywords in theme_keywords.items(): if any(keyword in content_lower or keyword in title_lower for keyword in keywords): themes.append(theme) # 如果没有找到主题，根据内容推断 if not themes: if len(content) > 500: themes.append('综合讨论') else: themes.append('简要报道') return themes def extract_key_data(self, content): """提取关键数据""" key_data = [] # 提取数字相关数据 patterns = [ (r'(\d+(?:\.\d+)?)[亿万]*(?:个|款|家|项|种)(?:大模型|模型)', '模型数量'), (r'参数[^\d]*(\d+(?:\.\d+)?)[亿万]*', '参数规模'), (r'投资[^\d]*(\d+(?:\.\d+)?)[亿万]*(?:元|美元)', '投资金额'), (r'市场[^\d]*(\d+(?:\.\d+)?)[亿万]*(?:元|美元)', '市场规模'), (r'增长[^\d]*(\d+(?:\.\d+)?)%', '增长率'), (r'(\d+(?:\.\d+)?)[亿万]*参数', '参数量'), (r'准确[^\d]*(\d+(?:\.\d+)?)%', '准确率') ] for pattern, label in patterns: matches = re.findall(pattern, content) for match in matches: key_data.append(f"{label}:{match}") return list(set(key_data))[:5] def run_comprehensive_crawl(self): """运行综合爬取""" logger.info("开始综合爬取虎嗅网大模型相关文章...") all_articles = [] # 方法1: 从首页获取 logger.info("方法1: 从首页获取文章") homepage_articles = self.get_articles_from_homepage() # 过滤出相关文章 relevant_articles = [] for article in homepage_articles: title = article.get('title', '').lower() if any(keyword in title for keyword in ['大模型', 'llm', '语言模型', 'gpt', 'chatgpt', 'ai', '人工智能']): relevant_articles.append(article) all_articles.extend(relevant_articles) logger.info(f"首页找到 {len(relevant_articles)} 篇相关文章") # 方法2: 直接搜索 logger.info("方法2: 直接搜索关键词") keywords = ['大模型', 'LLM', '语言模型', 'GPT', 'ChatGPT'] search_articles = self.search_articles_direct(keywords) all_articles.extend(search_articles) logger.info(f"搜索找到 {len(search_articles)} 篇文章") # 去重 unique_articles = [] seen_titles = set() for article in all_articles: title = article.get('title', '') if title and title not in seen_titles: seen_titles.add(title) unique_articles.append(article) logger.info(f"去重后剩余 {len(unique_articles)} 篇唯一文章") if not unique_articles: logger.warning("没有找到任何文章，尝试备选方案...") return self.run_fallback_crawl() # 获取详细内容 logger.info("开始获取文章详细内容...") detailed_articles = [] for i, article in enumerate(unique_articles): logger.info(f"处理第 {i+1}/{len(unique_articles)} 篇: {article['title'][:30]}...") if article.get('link'): detail = self.get_article_content(article['link']) article.update(detail) detailed_articles.append(article) self.random_delay() # 分析文章并生成洞察 logger.info("分析文章关键观点...") insights = self.generate_insights(detailed_articles) # 保存结果 self.save_results(insights) return insights def run_fallback_crawl(self): """备选爬取方案""" logger.info("启动备选爬取方案...") # 尝试获取虎嗅网的其他页面 urls_to_try = [ "https://www.huxiu.com/channel/107.html", # 科技频道 "https://www.huxiu.com/channel/101.html", # 商业频道 "https://www.huxiu.com/tag/267.html", # AI标签 "https://www.huxiu.com/tag/人工智能.html" ] all_articles = [] for url in urls_to_try: try: logger.info(f"尝试访问: {url}") response = self.get_with_retry(url) if response and response.status_code == 200: soup = BeautifulSoup(response.text, 'html.parser') # 查找文章链接 links = soup.find_all('a', href=re.compile(r'/article/\d+\.html')) for link in links[:10]: # 限制数量 title = link.get_text().strip() if any(keyword in title.lower() for keyword in ['大模型', 'llm', '语言模型', 'gpt']): article_url = urljoin(self.base_url, link.get('href')) article = { 'title': title, 'link': article_url, 'source': 'fallback' } all_articles.append(article) self.random_delay() except Exception as e: logger.error(f"访问 {url} 时出错: {e}") continue if all_articles: logger.info(f"备选方案找到 {len(all_articles)} 篇文章") # 获取内容 detailed_articles = [] for i, article in enumerate(all_articles): logger.info(f"处理备选文章 {i+1}/{len(all_articles)}: {article['title'][:30]}...") detail = self.get_article_content(article['link']) article.update(detail) detailed_articles.append(article) self.random_delay() insights = self.generate_insights(detailed_articles) self.save_results(insights) return insights return [] def generate_insights(self, articles): """生成洞察数据""" insights = [] for article in articles: if article.get('content_length', 0) > 100: content = article.get('full_content', '') title = article.get('title', '') themes = self.analyze_article_themes(content, title) key_data = self.extract_key_data(content) insight = { '标题': title, '发布时间': article.get('publish_time', '未知'), '作者': article.get('author', '未知'), '来源': article.get('source', '未知'), '主要观点': article.get('main_points', ''), '文章主题': '、'.join(themes), '关键数据': '、'.join(key_data), '内容长度': article.get('content_length', 0), '文章链接': article.get('link', ''), '获取方式': article.get('source_method', '未知'), '爬取时间': datetime.now().strftime('%Y-%m-%d %H:%M:%S') } insights.append(insight) return insights def save_results(self, insights): """保存结果到Excel""" if not insights: logger.warning("没有找到有效数据可保存") return timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"虎嗅网大模型观点分析_{timestamp}.xlsx" try: # 创建DataFrame df = pd.DataFrame(insights) # 设置列顺序 column_order = [ '标题', '主要观点', '文章主题', '关键数据', '作者', '发布时间', '来源', '内容长度', '获取方式', '文章链接', '爬取时间' ] # 重新排列列顺序 existing_columns = [col for col in column_order if col in df.columns] other_columns = [col for col in df.columns if col not in column_order] df = df[existing_columns + other_columns] # 保存Excel with pd.ExcelWriter(filename, engine='openpyxl') as writer: df.to_excel(writer, sheet_name='大模型观点', index=False) # 自动调整列宽 worksheet = writer.sheets['大模型观点'] for column in worksheet.columns: max_length = 0 column_letter = column[0].column_letter for cell in column: try: if len(str(cell.value)) > max_length: max_length = len(str(cell.value)) except: pass adjusted_width = min(max_length + 2, 50) worksheet.column_dimensions[column_letter].width = adjusted_width logger.info(f"数据已保存到: {filename}") # 生成报告 self.generate_report(insights, timestamp) except Exception as e: logger.error(f"保存结果时出错: {e}") def generate_report(self, insights, timestamp): """生成分析报告""" report_filename = f"虎嗅网大模型分析报告_{timestamp}.txt" total_articles = len(insights) if total_articles == 0: return avg_content_length = sum(insight['内容长度'] for insight in insights) / total_articles # 统计主题分布 theme_count = {} for insight in insights: themes = insight['文章主题'].split('、') if insight['文章主题'] else [] for theme in themes: theme_count[theme] = theme_count.get(theme, 0) + 1 # 生成报告 report = f"""虎嗅网大模型相关文章分析报告 生成时间: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")} ============================================ 基本统计: - 总文章数: {total_articles} 篇 - 平均内容长度: {avg_content_length:.0f} 字符 主题分布: """ for theme, count in sorted(theme_count.items(), key=lambda x: x[1], reverse=True): percentage = (count / total_articles) * 100 report += f"- {theme}: {count} 篇 ({percentage:.1f}%)\n" report += f"\n代表性观点摘要 (前5篇):\n" for i, insight in enumerate(insights[:5]): report += f"\n{i+1}. {insight['标题']}\n" report += f" 主题: {insight['文章主题']}\n" report += f" 关键数据: {insight['关键数据']}\n" report += f" 主要观点: {insight['主要观点'][:150]}...\n" try: with open(report_filename, 'w', encoding='utf-8') as f: f.write(report) logger.info(f"分析报告已保存到: {report_filename}") except Exception as e: logger.error(f"保存报告时出错: {e}") def main(): """主函数""" logger.info("启动虎嗅网大模型文章爬虫...") spider = HuxiuSpider() try: insights = spider.run_comprehensive_crawl() if insights: logger.info(f"\n爬取完成! 成功分析 {len(insights)} 篇文章") logger.info("前3篇文章标题:") for i, insight in enumerate(insights[:3]): logger.info(f"{i+1}. {insight['标题']}") logger.info(f" 主题: {insight['文章主题']}") logger.info(f" 观点摘要: {insight['主要观点'][:100]}...") else: logger.warning("没有找到相关文章") logger.warning("建议手动访问虎嗅网确认当前可用的文章列表") except Exception as e: logger.error(f"爬虫运行出错: {e}") import traceback logger.error(traceback.format_exc()) if __name__ == "__main__": main()