diff --git a/services/search_service.py b/services/search_service.py index b5ed215..99dd5cf 100644 --- a/services/search_service.py +++ b/services/search_service.py @@ -155,43 +155,71 @@ class SearchService: return None def fetch_url_content(self, url): - """抓取网页内容""" + """抓取网页内容(使用 agent-browser 浏览器方式,绑过反爬虫)""" try: - headers = { - 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' - } - response = requests.get(url, headers=headers, timeout=self.timeout) - response.raise_for_status() + # 使用浏览器方式抓取 + stdout, stderr, code = self._run_browser('open', url, '--timeout', '20000') + if code != 0: + print(f"打开页面失败: {stderr}") + return None - soup = BeautifulSoup(response.text, 'lxml') + # 等待页面加载 + self._run_browser('wait', '5000') - # 提取标题 - title = soup.find('title') - title = title.text.strip() if title else '' + # 获取页面标题 + stdout, stderr, code = self._run_browser('get', 'title', '--timeout', '5000') + title = stdout.strip().replace('[agent-browser] ', '').strip() if code == 0 else '' - # 提取正文(简单提取,可优化) - # 移除脚本和样式 - for script in soup(['script', 'style']): - script.decompose() + # 获取页面内容(通过 snapshot 获取 accessibility tree) + stdout, stderr, code = self._run_browser('snapshot', '--json', '--timeout', '15000') + text = '' + if code == 0 and stdout: + try: + data = json.loads(stdout) + snapshot = data.get('data', {}).get('snapshot', '') + # 从 snapshot 中提取所有 StaticText + text = self._extract_text_from_snapshot(snapshot) + except: + pass - # 提取文本 - text = soup.get_text(separator='\n', strip=True) + # 获取 URL(可能被重定向) + stdout, stderr, code = self._run_browser('get', 'url', '--timeout', '5000') + actual_url = stdout.strip() if code == 0 else url - # 提取元数据 - meta_desc = soup.find('meta', attrs={'name': 'description'}) - description = meta_desc['content'] if meta_desc else '' + # 关闭浏览器 + self._run_browser('close') + + # 提取描述(从页面内容的前200字符) + description = text[:200].strip() if text else '' return { 'title': title, 'description': description, 'content': text, - 'url': url, + 'url': actual_url, 'fetch_date': datetime.now().isoformat() } except Exception as e: print(f"抓取URL失败: {url}, 错误: {str(e)}") + # 尝试关闭浏览器 + try: + self._run_browser('close') + except: + pass return None + def _extract_text_from_snapshot(self, snapshot): + """从 accessibility tree snapshot 中提取文本内容""" + # 提取所有 StaticText 行 + texts = [] + for line in snapshot.split('\n'): + if 'StaticText' in line: + # 格式: - StaticText "文本内容" + match = re.search(r'StaticText "([^"]+)"', line) + if match: + texts.append(match.group(1)) + return '\n'.join(texts) + def search_articles(self, keyword, category=None): """从内容库搜索""" return db.search_articles(keyword, category)