hz4th_coder
b7b09251f8
功能: - 文章内容库管理 - 待处理产品列表管理 - 自动处理流程 - 智能搜索和数据提取 - ParamHub API集成 - 定时任务调度 部署端口: 16043
99 lines
3.3 KiB
Python
99 lines
3.3 KiB
Python
"""
|
|
搜索服务 - 从内容库和互联网搜索数据
|
|
"""
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
import json
|
|
from datetime import datetime
|
|
from config import Config
|
|
from models.database import db
|
|
|
|
class SearchService:
|
|
def __init__(self):
|
|
self.timeout = Config.SEARCH_TIMEOUT
|
|
self.max_results = Config.SEARCH_MAX_RESULTS
|
|
|
|
def search_internet(self, keyword, max_results=None):
|
|
"""
|
|
从互联网搜索（使用搜索API或爬虫）
|
|
这里暂时使用简单的搜索模拟
|
|
"""
|
|
max_results = max_results or self.max_results
|
|
|
|
# TODO: 接入真实的搜索API（如Google Custom Search、Bing等）
|
|
# 这里先返回空列表，等待后续接入真实API
|
|
results = []
|
|
|
|
return results
|
|
|
|
def fetch_url_content(self, url):
|
|
"""抓取网页内容"""
|
|
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()
|
|
|
|
soup = BeautifulSoup(response.text, 'lxml')
|
|
|
|
# 提取标题
|
|
title = soup.find('title')
|
|
title = title.text.strip() if title else ''
|
|
|
|
# 提取正文（简单提取，可优化）
|
|
# 移除脚本和样式
|
|
for script in soup(['script', 'style']):
|
|
script.decompose()
|
|
|
|
# 提取文本
|
|
text = soup.get_text(separator='\n', strip=True)
|
|
|
|
# 提取元数据
|
|
meta_desc = soup.find('meta', attrs={'name': 'description'})
|
|
description = meta_desc['content'] if meta_desc else ''
|
|
|
|
return {
|
|
'title': title,
|
|
'description': description,
|
|
'content': text,
|
|
'url': url,
|
|
'fetch_date': datetime.now().isoformat()
|
|
}
|
|
except Exception as e:
|
|
print(f"抓取URL失败: {url}, 错误: {str(e)}")
|
|
return None
|
|
|
|
def search_articles(self, keyword, category=None):
|
|
"""从内容库搜索"""
|
|
return db.search_articles(keyword, category)
|
|
|
|
def search_all(self, keyword, category=None, include_internet=True):
|
|
"""
|
|
综合搜索：内容库 + 互联网
|
|
"""
|
|
results = {
|
|
'articles': [],
|
|
'internet': [],
|
|
'total': 0
|
|
}
|
|
|
|
# 1. 从内容库搜索
|
|
articles = self.search_articles(keyword, category)
|
|
results['articles'] = articles
|
|
|
|
# 2. 从互联网搜索（如果启用）
|
|
if include_internet:
|
|
internet_results = self.search_internet(keyword)
|
|
results['internet'] = internet_results
|
|
|
|
results['total'] = len(articles) + len(results['internet'])
|
|
|
|
return results
|
|
|
|
def save_to_articles(self, product_names, category, keywords, summary, content, source, url=None):
|
|
"""保存搜索结果到内容库"""
|
|
return db.add_article(product_names, category, keywords, summary, content, source, url)
|
|
|
|
# 全局搜索服务实例
|
|
search_service = SearchService() |