diff --git a/.gitignore b/.gitignore index 4931c6f..f15ec3d 100644 --- a/.gitignore +++ b/.gitignore @@ -32,6 +32,7 @@ wheels/ # 输出文件 output/ +output_post/ # 临时文件 *.tmp diff --git a/1_零差云控官网爬虫方案.md b/1_零差云控官网爬虫方案.md index ecf0e9a..66172d0 100644 --- a/1_零差云控官网爬虫方案.md +++ b/1_零差云控官网爬虫方案.md @@ -7,14 +7,15 @@ ## 代码结构 ``` -crawl/ +crawl_0131(1)/ ├── main.py # 主入口 ├── requirements.txt # 依赖 ├── zeroerr_crawler/ # 爬虫模块 │ ├── __init__.py │ ├── config.py # 配置文件(所有任务配置) │ ├── base_crawler.py # 基础爬虫类 -│ ├── product_crawler.py # 产品页专用爬虫 +│ ├── product_crawler.py # 产品页专用爬虫(处理 eRob、eCoder、配件) +│ ├── extract_abstract.py # 摘要提取模块(使用大模型生成文档摘要) │ └── utils.py # 工具函数 └── output/ # 输出目录 ``` @@ -62,6 +63,17 @@ python main.py - 每个分类生成一个汇总文档(`xxx_汇总.md` 和 `xxx_汇总.docx`) - 图片保存到对应分类的 `images/` 目录 - Word 文档中嵌入本地图片 +- 支持表格、列表、标题等格式转换 + +## 技术说明 +- 使用 `StandardCrawler` 处理标准页面(新闻、案例、问题等) +- 使用 `ProductCrawler` 处理产品页面(机器人关节、编码器、配件) +- 支持多种页面布局和内容选择器 +- 自动去重标题,优化 Word 文档格式 +- **摘要提取**:`extract_abstract.py` 模块使用大模型(OpenAI API)为每个分类的文档集合生成摘要 + - 面向客户售前咨询场景,生成100-200字的简洁摘要 + - 自动生成相关链接列表 + - 摘要失败时自动降级为仅生成链接列表 ## 待处理项目 - [ ] 报价与交期页面:https://www.zeroerr.cn/inquiry/ diff --git a/zeroerr_crawler/base_crawler.py b/zeroerr_crawler/base_crawler.py index b844c12..7952e03 100644 --- a/zeroerr_crawler/base_crawler.py +++ b/zeroerr_crawler/base_crawler.py @@ -18,6 +18,7 @@ from abc import ABC, abstractmethod from .config import BASE_URL, HEADERS, REQUEST_DELAY, OUTPUT_DIR from .utils import ensure_dir, download_image, safe_filename, make_absolute_url +from .extract_abstract import generate_abstract class BaseCrawler(ABC): @@ -128,14 +129,28 @@ class BaseCrawler(ABC): selector = self.config.get("title_selector", "h1") index = self.config.get("title_index", 0) + # 支持多个选择器,用逗号分隔(类似 extract_content 的处理方式) + selectors = [s.strip() for s in selector.split(',')] + + # 收集所有匹配的标签 + all_tags = [] + for sel in selectors: + # 对于简单的标签名(如 "h1", "h2"),直接查找 + if sel in ['h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'title']: + found_tags = soup.find_all(sel) + all_tags.extend(found_tags) + else: + # 对于其他选择器,尝试查找 + found_tags = soup.find_all(sel) + all_tags.extend(found_tags) + # 优先从配置的选择器提取 - tags = soup.find_all(selector) - if tags and len(tags) > index: - title = tags[index].get_text(strip=True) + if all_tags and len(all_tags) > index: + title = all_tags[index].get_text(strip=True) if title: return title - elif tags: - title = tags[0].get_text(strip=True) + elif all_tags: + title = all_tags[0].get_text(strip=True) if title: return title @@ -328,19 +343,52 @@ class BaseCrawler(ABC): return images_info - def content_to_markdown(self, content: BeautifulSoup) -> str: + def content_to_markdown(self, content: BeautifulSoup, page_title: str = None) -> str: """ 将内容转换为 Markdown Args: content: 内容区域 + page_title: 页面标题(如果提供,会移除内容中与标题重复的第一个h1/h2标签) Returns: Markdown 文本 """ + # 如果提供了页面标题,检查并移除内容中与标题重复的标签 + if page_title: + # 创建内容的副本,避免修改原始内容 + content_copy = BeautifulSoup(str(content), 'html.parser') + + # 移除与标题完全相同的第一个h1 + first_h1 = content_copy.find('h1') + if first_h1: + h1_text = first_h1.get_text(strip=True) + if h1_text == page_title: + first_h1.decompose() + + # 移除与标题完全相同的第一个h2 + first_h2 = content_copy.find('h2') + if first_h2: + h2_text = first_h2.get_text(strip=True) + if h2_text == page_title: + first_h2.decompose() + + # 检查标题是否包含"型号:"前缀,如果是,也移除内容中只包含产品名称的h2 + # 例如:标题是"型号:eCoder11",内容中有"eCoder11"的h2 + if '型号:' in page_title or '型号:' in page_title: + product_name = page_title.replace('型号:', '').replace('型号:', '').strip() + if product_name: + # 查找第一个只包含产品名称的h2 + for h2 in content_copy.find_all('h2'): + h2_text = h2.get_text(strip=True) + if h2_text == product_name: + h2.decompose() + break # 只移除第一个匹配的 + + return markdownify.markdownify(str(content_copy), heading_style="ATX") return markdownify.markdownify(str(content), heading_style="ATX") - def add_content_to_docx(self, doc: Document, content: BeautifulSoup, output_dir: str): + def add_content_to_docx(self, doc: Document, content: BeautifulSoup, output_dir: str, page_title: str = None): """ 将内容添加到 Word 文档 @@ -348,7 +396,17 @@ class BaseCrawler(ABC): doc: Document 对象 content: 内容区域 output_dir: 输出目录(用于解析图片路径) + page_title: 页面标题(如果提供,会跳过内容中与标题重复的第一个h1标签) """ + # 如果提供了页面标题,创建内容副本并移除重复的h1 + if page_title: + content = BeautifulSoup(str(content), 'html.parser') + first_h1 = content.find('h1') + if first_h1: + h1_text = first_h1.get_text(strip=True) + if h1_text == page_title: + first_h1.decompose() # 移除该标签 + # 按文档顺序处理元素,保持列表的连续性 for element in content.find_all(['p', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'img', 'li', 'ul', 'ol', 'table']): if element.name == 'img': @@ -444,8 +502,8 @@ class BaseCrawler(ABC): # 处理图片 images = self.process_images(content, url) - # 转换为 Markdown - markdown = self.content_to_markdown(content) + # 转换为 Markdown(传入标题,用于去除重复的h1标签) + markdown = self.content_to_markdown(content, title) return { "url": url, @@ -481,7 +539,7 @@ class BaseCrawler(ABC): p = doc.add_paragraph() p.add_run(f"原文链接: {page_data['url']}").italic = True - self.add_content_to_docx(doc, page_data["content"], self.output_dir) + self.add_content_to_docx(doc, page_data["content"], self.output_dir, title) doc.save(docx_path) def save_combined_documents(self, all_pages: list[dict]): @@ -505,14 +563,37 @@ class BaseCrawler(ABC): # === 处理 Markdown === existing_urls = set() existing_content = "" + existing_pages = [] # 存储已存在的页面信息(用于重新生成摘要) - # 如果文件已存在,读取现有内容并提取已存在的URL + # 如果文件已存在,读取现有内容并提取已存在的URL和页面信息 if os.path.exists(md_path): with open(md_path, "r", encoding="utf-8") as f: existing_content = f.read() # 提取已存在的URL(用于去重) url_pattern = r'\*\*原文链接\*\*: (https?://[^\s\n]+)' existing_urls = set(re.findall(url_pattern, existing_content)) + + # 提取已存在的页面信息(标题、URL和部分内容),用于重新生成摘要 + # 匹配格式:## 标题\n\n**原文链接**: URL\n\n内容... + # 使用更复杂的正则来匹配每个页面的完整内容块 + page_blocks = re.split(r'\n\n---\n\n', existing_content) + for block in page_blocks: + # 匹配页面标题和URL + title_match = re.search(r'^##\s+([^\n]+)', block, re.MULTILINE) + url_match = re.search(r'\*\*原文链接\*\*:\s+(https?://[^\s\n]+)', block) + if title_match and url_match: + title = title_match.group(1).strip() + url = url_match.group(1).strip() + # 提取内容部分(跳过标题和URL行) + content_start = url_match.end() + markdown_content = block[content_start:].strip() + # 只取前500字符作为预览 + markdown_preview = markdown_content[:500] if len(markdown_content) > 500 else markdown_content + existing_pages.append({ + 'title': title, + 'url': url, + 'markdown': markdown_preview + }) # 过滤掉已存在的页面(基于URL去重) new_pages = [page for page in all_pages if page['url'] not in existing_urls] @@ -529,14 +610,63 @@ class BaseCrawler(ABC): new_md_content += page["markdown"] new_md_content += "\n\n---\n\n" + # 合并所有页面(已存在的 + 新添加的),用于生成摘要 + all_pages_for_abstract = existing_pages + all_pages + + # 生成摘要(新建文档时生成,追加新内容时也重新生成,确保包含所有URL) + abstract = None + if not existing_content: + # 新建文档:使用当前爬取的页面生成摘要 + print(f" 正在生成文档摘要...") + abstract = generate_abstract(all_pages, output_dir_name) + else: + # 追加模式:重新生成摘要,包含所有页面(已存在的 + 新添加的) + print(f" 正在重新生成文档摘要(包含所有 {len(all_pages_for_abstract)} 篇)...") + abstract = generate_abstract(all_pages_for_abstract, output_dir_name) + # 追加或创建文件 if existing_content: - # 追加模式:在现有内容后追加新内容 - combined_md = existing_content.rstrip() + "\n\n" + new_md_content - print(f" 追加 {len(new_pages)} 篇新内容到现有文档") + # 追加模式:更新摘要部分,然后在现有内容后追加新内容 + # 使用正则表达式替换摘要部分(从标题后到第一个"---"分隔符之间的内容) + # 匹配模式:标题行 + 摘要内容 + 分隔符 + title_pattern = r'^#\s+.*?全集\s*\n\n' + separator_pattern = r'\n\n---\n\n' + + # 查找标题后的第一个分隔符位置 + title_match = re.search(title_pattern, existing_content, re.MULTILINE) + if title_match: + title_end = title_match.end() + # 查找第一个分隔符 + separator_match = re.search(separator_pattern, existing_content[title_end:]) + if separator_match: + # 替换摘要部分 + separator_start = title_end + separator_match.start() + separator_end = title_end + separator_match.end() + # 保留标题和分隔符,替换中间的内容 + combined_md = existing_content[:title_end] + if abstract: + combined_md += abstract + combined_md += existing_content[separator_end:] + # 追加新内容 + combined_md = combined_md.rstrip() + "\n\n" + new_md_content + else: + # 如果没有找到分隔符,说明可能没有摘要,直接添加摘要和新内容 + combined_md = existing_content[:title_end] + if abstract: + combined_md += abstract + "\n\n---\n\n" + combined_md += existing_content[title_end:].lstrip() + combined_md = combined_md.rstrip() + "\n\n" + new_md_content + else: + # 如果没有找到标题,说明格式异常,直接追加 + combined_md = existing_content.rstrip() + "\n\n" + new_md_content + print(f" 追加 {len(new_pages)} 篇新内容到现有文档,并更新摘要") else: # 新建模式:创建新文档 - combined_md = f"# {output_dir_name}全集\n\n" + new_md_content + # 构建文档内容:标题 + 摘要 + 正文 + combined_md = f"# {output_dir_name}全集\n\n" + if abstract: + combined_md += f"{abstract}\n\n---\n\n" + combined_md += new_md_content with open(md_path, "w", encoding="utf-8") as f: f.write(combined_md) @@ -563,7 +693,7 @@ class BaseCrawler(ABC): doc.add_heading(page["title"], level=1) p = doc.add_paragraph() p.add_run(f"原文链接: {page['url']}").italic = True - self.add_content_to_docx(doc, page["content"], self.output_dir) + self.add_content_to_docx(doc, page["content"], self.output_dir, page["title"]) doc.add_page_break() doc.save(docx_path) print(f" 追加 {len(new_pages_for_doc)} 篇新内容到 Word 文档") @@ -574,11 +704,28 @@ class BaseCrawler(ABC): doc = Document() doc.add_heading(f'{output_dir_name}全集', level=1) + # 添加摘要(只在新建时生成,复用Markdown部分生成的摘要) + if not existing_content and abstract: + # 将Markdown格式的摘要转换为Word格式 + # 处理Markdown链接:将 [文本](URL) 转换为 "文本 (URL)" 格式 + abstract_text = re.sub(r'\[([^\]]+)\]\(([^\)]+)\)', r'\1 (\2)', abstract) + # 移除Markdown加粗标记 + abstract_text = abstract_text.replace('**', '') + # 添加摘要段落 + for line in abstract_text.split('\n'): + if line.strip(): + doc.add_paragraph(line.strip()) + else: + doc.add_paragraph() # 空行 + doc.add_paragraph() # 空行 + doc.add_paragraph("─" * 50) # 分隔线 + doc.add_paragraph() # 空行 + for page in all_pages: doc.add_heading(page["title"], level=1) p = doc.add_paragraph() p.add_run(f"原文链接: {page['url']}").italic = True - self.add_content_to_docx(doc, page["content"], self.output_dir) + self.add_content_to_docx(doc, page["content"], self.output_dir, page["title"]) doc.add_page_break() doc.save(docx_path) diff --git a/zeroerr_crawler/extract_abstract.py b/zeroerr_crawler/extract_abstract.py new file mode 100644 index 0000000..be3be0d --- /dev/null +++ b/zeroerr_crawler/extract_abstract.py @@ -0,0 +1,91 @@ +""" +摘要提取模块 +使用大模型生成文档摘要 +""" + +from openai import OpenAI + +# API 配置 +API_BASE_URL = "https://yiming.zeroerr.team/v1" +API_KEY = "sk-LX1g8KkG61S6eUaVD567C0C187D4452c90F9E6985cDf3586" +MODEL = "Yiming" + + +def generate_abstract(all_pages: list[dict], category_name: str) -> str: + """ + 使用大模型生成文档摘要 + + Args: + all_pages: 所有页面数据列表,每个元素包含 'title', 'url', 'markdown' 等字段 + category_name: 文档类别名称(如"应用案例") + + Returns: + 摘要文本(Markdown格式),包含摘要内容和链接列表 + """ + if not all_pages: + return "" + + try: + # 构建文档内容(用于生成摘要) + # 只使用标题和部分内容,避免内容过长 + content_parts = [] + for page in all_pages: + title = page.get('title', '') + markdown = page.get('markdown', '') + # 只取前500字符的内容,避免输入过长 + content_preview = markdown[:500] if len(markdown) > 500 else markdown + content_parts.append(f"标题:{title}\n内容预览:{content_preview}") + + document_content = "\n\n".join(content_parts) + + # 构建提示词 + prompt = f"""面向客户售前咨询,请为以下"{category_name}"类别的文档集合生成一个简洁的摘要。 + +文档内容: +{document_content} + +要求: +1. 摘要应概括该页面的主题和主要内容 +2. 摘要长度控制在100-200字之间 +3. 使用简洁、专业的语言 +4. 突出该页面主题的价值和特点 + +请直接输出摘要内容,不要包含其他说明文字。""" + + # 调用大模型API + client = OpenAI( + base_url=API_BASE_URL, + api_key=API_KEY + ) + + response = client.chat.completions.create( + model=MODEL, + temperature=0.3, # 使用较低的温度值,保证摘要的准确性 + messages=[ + {"role": "user", "content": prompt} + ] + ) + + abstract_text = response.choices[0].message.content.strip() + + # 构建链接列表 + links_section = "\n\n**相关链接:**\n\n" + for i, page in enumerate(all_pages, 1): + title = page.get('title', '未命名') + url = page.get('url', '') + links_section += f"{i}. [{title}]({url})\n" + + # 组合摘要和链接 + result = f"{abstract_text}{links_section}" + + return result + + except Exception as e: + print(f" 警告: 生成摘要失败: {e}") + # 如果生成摘要失败,至少返回链接列表 + links_section = "\n\n**相关链接:**\n\n" + for i, page in enumerate(all_pages, 1): + title = page.get('title', '未命名') + url = page.get('url', '') + links_section += f"{i}. [{title}]({url})\n" + return links_section \ No newline at end of file diff --git a/zeroerr_crawler/product_crawler.py b/zeroerr_crawler/product_crawler.py index 3518b65..79d7405 100644 --- a/zeroerr_crawler/product_crawler.py +++ b/zeroerr_crawler/product_crawler.py @@ -45,24 +45,97 @@ class ProductCrawler(BaseCrawler): 提取产品页面标题 产品页面标题可能在不同位置 """ - # 尝试从面包屑导航后的第一个 h1 - h1_tags = soup.find_all('h1') - for h1 in h1_tags: - text = h1.get_text(strip=True) - # 跳过网站名称 - if '零差云控' in text or '零误差' in text: - continue - if text: - return text + # 优先使用配置中的选择器(支持 h1, h2 等) + selector = self.config.get("title_selector", "h1") + index = self.config.get("title_index", 0) - # 从 URL 提取 - return url.split('/')[-1].replace('.html', '') + # 支持多个选择器,用逗号分隔 + selectors = [s.strip() for s in selector.split(',')] + + # 收集所有匹配的标签 + all_tags = [] + for sel in selectors: + # 对于简单的标签名(如 "h1", "h2"),直接查找 + if sel in ['h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'title']: + found_tags = soup.find_all(sel) + all_tags.extend(found_tags) + else: + # 对于其他选择器,尝试查找 + found_tags = soup.find_all(sel) + all_tags.extend(found_tags) + + # 优先从配置的选择器提取 + if all_tags and len(all_tags) > index: + title = all_tags[index].get_text(strip=True) + # 跳过网站名称 + if title and '零差云控' not in title and '零误差' not in title: + return title + elif all_tags: + # 如果指定索引的标签被跳过,尝试其他标签 + for tag in all_tags: + title = tag.get_text(strip=True) + # 跳过网站名称 + if title and '零差云控' not in title and '零误差' not in title: + return title + + # 尝试从页面 title 标签提取 + title_tag = soup.find('title') + if title_tag: + title = title_tag.get_text(strip=True) + # 移除网站名称后缀(如 " - 零差云控") + if ' - ' in title: + title = title.split(' - ')[0].strip() + if title and title.lower() not in ['about-us', 'contact-us', 'join-us']: + return title + + # 最后从 URL 提取 + url_part = url.split('/')[-1].replace('.html', '') + # 将连字符替换为空格,并首字母大写 + if '-' in url_part: + url_part = ' '.join(word.capitalize() for word in url_part.split('-')) + return url_part - def add_content_to_docx(self, doc: Document, content: BeautifulSoup, output_dir: str): + def add_content_to_docx(self, doc: Document, content: BeautifulSoup, output_dir: str, page_title: str = None): """ 将产品内容添加到 Word 文档 针对产品页面的表格等进行优化处理 + + Args: + doc: Document 对象 + content: 内容区域 + output_dir: 输出目录(用于解析图片路径) + page_title: 页面标题(如果提供,会跳过内容中与标题重复的h1/h2标签或包含标题的段落) """ + # 如果提供了页面标题,创建内容副本并移除重复的标题元素 + if page_title: + content = BeautifulSoup(str(content), 'html.parser') + + # 移除与标题完全相同的第一个h1 + first_h1 = content.find('h1') + if first_h1: + h1_text = first_h1.get_text(strip=True) + if h1_text == page_title: + first_h1.decompose() + + # 移除与标题完全相同的第一个h2 + first_h2 = content.find('h2') + if first_h2: + h2_text = first_h2.get_text(strip=True) + if h2_text == page_title: + first_h2.decompose() + + # 检查标题是否包含"型号:"前缀,如果是,也移除内容中只包含产品名称的h2 + # 例如:标题是"型号:eCoder11",内容中有"eCoder11"的h2 + if '型号:' in page_title or '型号:' in page_title: + product_name = page_title.replace('型号:', '').replace('型号:', '').strip() + if product_name: + # 查找第一个只包含产品名称的h2 + for h2 in content.find_all('h2'): + h2_text = h2.get_text(strip=True) + if h2_text == product_name: + h2.decompose() + break # 只移除第一个匹配的 + for element in content.find_all(['p', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'img', 'li', 'table', 'div']): # 跳过嵌套元素 if element.find_parent(['table', 'li']):