import requests from lxml import html,etree import re #百度网页搜索 def baidu_web(keywords): # 百度搜索的URL模板 url = "https://www.baidu.com/s" # 搜索关键词 params = { "wd": keywords } headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0 Safari/537.36" } response = requests.get(url, params=params, headers=headers) # 设置编码格式（自动检测） response.encoding = response.apparent_encoding pages=[] # 确保响应正常 if response.status_code == 200: #print(response.content.decode("utf-8")) # 使用 lxml 解析 HTML tree = html.fromstring(response.content) # 使用 XPath 提取所有搜索结果标题和链接 results = tree.xpath('//div[@id="content_left"]/div[contains(@class, "result")]') for result in results: title = result.xpath('.//h3/a//text()') link = result.xpath('.//h3/a/@href') title = ''.join(title).strip() if title else '' link = link[0] if link else '' # 使用 XPath 查找 data-module="abstract" 的 div desc = result.xpath('.//div[@data-module="abstract"]//text()') desc = ''.join(desc).strip() if title else '' # print("标题:", title) # print("链接:", link) # print("摘要:", desc) # print("-" * 80) if title and link: pages.append({"title":title,"link":link,"desc":desc}) else: print(f"请求失败，状态码：{response.status_code}") return pages """ 单独的网页 """ def get_single_page(url,size=2048): headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0 Safari/537.36" } try: response = requests.get(url, headers=headers) # 设置编码格式（自动检测） response.encoding = response.apparent_encoding # 确保响应正常 if response.status_code == 200: #print(response.text) tree = html.fromstring(response.text) all_text="" # 获取标签下的所有文本 body = tree.find('.//body') if body is not None: for tag in body.xpath('.//script | .//style | .//noscript'): parent = tag.getparent() if parent is not None: parent.remove(tag) # 可选：移除内联事件（如 onclick） for elem in body.xpath('.//*[@onclick or @onload or @onerror]'): # 移除特定属性 if 'onclick' in elem.attrib: del elem.attrib['onclick'] if 'onload' in elem.attrib: del elem.attrib['onload'] all_text = body.text_content() all_text = all_text.strip() #print(all_text) # 提取 title title = tree.xpath('//title/text()') title = title[0] if title else "" # 提取 keywords keywords = tree.xpath('//meta[@name="keywords"]/@content') keywords = keywords[0] if keywords else "" result={"code":200,"url":url,"title":title,"keywords":keywords,"content":all_text[0:size]} except Exception as e : result={"code":500,"url":url,"content":f'{e}'},None return result,response.text #百度网页搜索 def cn_bing_web(keywords): # 百度搜索的URL模板 url = "https://cn.bing.com/search" # 搜索关键词 params = { "q": keywords } headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0 Safari/537.36" } response = requests.get(url, params=params, headers=headers) # 设置编码格式（自动检测） response.encoding = response.apparent_encoding pages=[] # 确保响应正常 if response.status_code == 200: #print(response.content.decode("utf-8")) # 使用 lxml 解析 HTML tree = html.fromstring(response.text) search_items = tree.xpath('//li[@class="b_algo"]') for item in search_items: title = item.xpath('.//h2/a/text()') link = item.xpath('.//h2/a/@href') title = ''.join(title).strip() if title else '' link = link[0] if link else '' if title and link: pages.append({"title":title,"link":link,"desc":title}) else: print(f"请求失败，状态码：{response.status_code}") return pages #联网搜索的主入口 def search_web(keywords): r0 = sogou_search(keywords) r1 = baidu_web(keywords) r2 = cn_bing_web(keywords) print(f"搜狗[{len(r0)}],百度[{len(r1)}],必应[{len(r2)}]===={keywords}") return r0+r1+r2 """ 页面的详细内容 """ def get_detail_page(pages,size=2048): result=[] for page in pages: url = page["link"] headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0 Safari/537.36" } try: response = requests.get(url, headers=headers) # 设置编码格式（自动检测） response.encoding = response.apparent_encoding # 确保响应正常 if response.status_code == 200: tree = html.fromstring(response.text) # 获取标签下的所有文本 body_element = tree.find('.//body') if body_element is not None: all_text = body_element.text_content() result.append({"url":page["link"],"title":page["title"],"desc":page["desc"],"content":all_text[0:size]}) except: pass #end for return result import asyncio import aiohttp #单个页面 async def fetch(session, url): try: async with session.get(url) as resp: if resp.status == 200: text = await resp.text() body = html.fromstring(text) # 增加对sogo重定向页面的处理 noscript_meta = body.xpath('//noscript/meta[@http-equiv="refresh"]/@content') if noscript_meta: content = noscript_meta[0] # 匹配 URL='xxx' 或 URL=xxx match = re.search(r'URL\s*=\s*["\']?([^"\'>\s]+)', content, re.IGNORECASE) if match: url = match.group(1) return await fetch(session,url) # 获取标签下的所有文本 for tag in body.xpath('.//script | .//style | .//noscript'): parent = tag.getparent() if parent is not None: parent.remove(tag) # 可选：移除内联事件（如 onclick） for elem in body.xpath('.//*[@onclick or @onload or @onerror]'): # 移除特定属性 if 'onclick' in elem.attrib: del elem.attrib['onclick'] if 'onload' in elem.attrib: del elem.attrib['onload'] all_text = body.text_content() all_text = all_text.strip() return {"url": url, "content": all_text[0:6000]} except: pass #任务池 async def task_pool(urls): # 设置全局 headers headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0 Safari/537.36" } async with aiohttp.ClientSession(headers=headers) as session: tasks = [fetch(session, url) for url in urls] results = await asyncio.gather(*tasks) # 并发执行，收集结果 return results # 返回所有结果 #异步请求 def get_detail_page2(pages): result = asyncio.run(task_pool([page["link"] for page in pages])) return [r for r in result if r ] # def test_web(keywords): # 百度搜索的URL模板 url = "https://www.sogou.com/web" # 搜索关键词 params = { "query": keywords } headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0 Safari/537.36" } response = requests.get(url, params=params, headers=headers) print(response.text) def sogou_search(keyword, timeout=10): """ 模拟搜狗网页搜索，返回解析后的结果列表 :param keyword: 搜索关键词 :param page: 页码 (从1开始) :param timeout: 请求超时时间 :return: 搜索结果列表 [{title, url, abstract}, ...] """ # 搜狗搜索URL构造 url = f"https://www.sogou.com/web" # 搜索关键词 params = { "query": keyword } headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', 'Accept-Encoding': 'gzip, deflate, br', 'Connection': 'keep-alive', 'Referer': 'https://www.sogou.com/', } try: response = requests.get(url, params=params,headers=headers, timeout=timeout) response.raise_for_status() response.encoding = 'utf-8' html = etree.HTML(response.text) results = [] # 搜狗搜索结果容器（可能随前端改版而变化） result_nodes = html.xpath('//div[@class="vrwrap"] | //div[@class="rb"]') for node in result_nodes: try: # 标题 title_node = node.xpath('.//h3/a') if not title_node: continue title = ''.join(title_node[0].xpath('.//text()')).strip() # 链接 href = title_node[0].get('href', '') # 搜狗的链接是跳转链接，需要处理 if href.startswith('/link?url='): full_url = 'https://www.sogou.com' + href # 可选：请求一次获取真实URL（会增加延迟和请求量） real_url = get_real_url(full_url, headers, timeout) else: real_url = href # 摘要/描述 abstract_parts = node.xpath('.//div[contains(@class, "fz-mid")]//text() | .//p[contains(@class, "str_info")]//text()') abstract = ''.join(abstract_parts).strip().replace('\n', '').replace('\r', '').replace(' ', '') if title and real_url: results.append({ 'title': title, 'link': real_url, 'desc': abstract }) except Exception as e: print(f"[解析单条结果出错]: {e}") continue return results except Exception as e: print(f"[请求失败]: {e}") return [] def get_real_url(sogou_redirect_url, headers, timeout=10): """ （可选）访问搜狗跳转链接，获取真实目标URL 注意：频繁请求可能触发反爬 """ try: r = requests.get(sogou_redirect_url, headers=headers, timeout=timeout, allow_redirects=True) return r.url except: return sogou_redirect_url # 失败则返回原跳转链接 if __name__ == "__main__": # keyword = "4b模型" # print(f"正在搜索: {keyword}") # results = sogou_search(keyword) # for i, res in enumerate(results, 1): # print(f"\n{i}. {res['title']}") # print(f" URL: {res['url']}") # print(f" 摘要: {res['abstract'][:100]}...") # print(f"\n共找到 {len(results)} 条结果") import sys question = "云智信安" print("=========================Search=============================") #r = cn_bing_web(question) r = search_web(sys.argv[1]) print(len(r),r) print("=========================Detail=============================") pages = get_detail_page2(r) for page in pages: print(page) print(len(pages)) """ # r2 = get_detail_page(r) # print(r2) print("===========================BAIDU===========================") r1 = baidu_web(question) print(len(r1),r1) # r2 = get_detail_page(r) # print(r2) print("===========================R2===========================") r4 = search_web(question) print(len(r4),r4) print("===========================get_detail_page2===========================") r5 = get_detail_page2(r4) print(len(r5),r5) """