feat: 新增360/搜狗/微信搜索提供商 + PG15→16升级 & 项目文档更新
- search_utils.py: 新增 _call_360/_call_sogou/_call_wechat HTML爬取函数 - initial_data.py: 种子数据新增三个搜索提供商（priority 3/4/5） - models.py: provider_type 注释补充新类型 - admin.html: 搜索提供商类型下拉框新增三个选项 - AGENTS.md/PROGRESS.md/README.md: PostgreSQL 15→16 - README.md: 移除硬编码数据库密码
This commit is contained in:
Yuzhiran Dev
+106
-1
@@ -205,12 +205,116 @@ def _call_mcp(api_key: str, api_url: str, query: str, max_results: int) -> List[
|
||||
return []
|
||||
|
||||
|
||||
def _call_360(api_key: str, api_url: str, query: str, max_results: int) -> List[Dict]:
|
||||
"""360搜索（HTML爬取，无需 API Key）"""
|
||||
from bs4 import BeautifulSoup
|
||||
import requests
|
||||
try:
|
||||
resp = requests.get(
|
||||
api_url or "https://www.so.com/s",
|
||||
params={"q": query},
|
||||
headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"},
|
||||
timeout=15,
|
||||
)
|
||||
if resp.status_code != 200:
|
||||
logger.warning("360搜索返回 %s", resp.status_code)
|
||||
return []
|
||||
soup = BeautifulSoup(resp.text, "html.parser")
|
||||
results = []
|
||||
for item in soup.select("li.res-list, li[class*=result], .rb"):
|
||||
title_el = item.select_one("h3.res-title a, h3[class*=title] a, .res-title a")
|
||||
if not title_el:
|
||||
continue
|
||||
title = title_el.get_text(strip=True)[:120]
|
||||
url = title_el.get("href", "")
|
||||
snippet_el = item.select_one("p.res-desc, p[class*=desc], .res-desc")
|
||||
snippet = snippet_el.get_text(strip=True)[:300] if snippet_el else ""
|
||||
results.append({"title": title, "url": url, "content": snippet, "source": "360"})
|
||||
if len(results) >= max_results:
|
||||
break
|
||||
return results
|
||||
except Exception as e:
|
||||
logger.warning("360搜索失败: %s", e)
|
||||
return []
|
||||
|
||||
|
||||
def _call_sogou(api_key: str, api_url: str, query: str, max_results: int) -> List[Dict]:
|
||||
"""搜狗搜索（HTML爬取，无需 API Key）"""
|
||||
from bs4 import BeautifulSoup
|
||||
import requests
|
||||
try:
|
||||
resp = requests.get(
|
||||
api_url or "https://sogou.com/web",
|
||||
params={"query": query},
|
||||
headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"},
|
||||
timeout=15,
|
||||
)
|
||||
if resp.status_code != 200:
|
||||
logger.warning("搜狗搜索返回 %s", resp.status_code)
|
||||
return []
|
||||
soup = BeautifulSoup(resp.text, "html.parser")
|
||||
results = []
|
||||
for item in soup.select("div.vrwrap, div[class*=vr], .rb"):
|
||||
title_el = item.select_one("h3.vr-title a, h3[class*=title] a, .vr-title a")
|
||||
if not title_el:
|
||||
continue
|
||||
title = title_el.get_text(strip=True)[:120]
|
||||
url = title_el.get("href", "")
|
||||
snippet_el = item.select_one("p.str-text, div.str-text, p[class*=str]")
|
||||
snippet = snippet_el.get_text(strip=True)[:300] if snippet_el else ""
|
||||
results.append({"title": title, "url": url, "content": snippet, "source": "sogou"})
|
||||
if len(results) >= max_results:
|
||||
break
|
||||
return results
|
||||
except Exception as e:
|
||||
logger.warning("搜狗搜索失败: %s", e)
|
||||
return []
|
||||
|
||||
|
||||
def _call_wechat(api_key: str, api_url: str, query: str, max_results: int) -> List[Dict]:
|
||||
"""微信搜一搜（通过搜狗抓取，无需 API Key）"""
|
||||
from bs4 import BeautifulSoup
|
||||
import requests
|
||||
try:
|
||||
resp = requests.get(
|
||||
api_url or "https://wx.sogou.com/weixin",
|
||||
params={"type": 2, "query": query},
|
||||
headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"},
|
||||
timeout=15,
|
||||
)
|
||||
if resp.status_code != 200:
|
||||
logger.warning("微信搜索返回 %s", resp.status_code)
|
||||
return []
|
||||
soup = BeautifulSoup(resp.text, "html.parser")
|
||||
results = []
|
||||
for item in soup.select("div.news-box, li.news-list, div[class*=news]"):
|
||||
title_el = item.select_one("h3 a, .txt-box h3 a")
|
||||
if not title_el:
|
||||
continue
|
||||
title = title_el.get_text(strip=True)[:120]
|
||||
url = title_el.get("href", "")
|
||||
if url and not url.startswith("http"):
|
||||
url = "https://wx.sogou.com" + url
|
||||
snippet_el = item.select_one("p.txt-info, div.txt-info, .txt-info")
|
||||
snippet = snippet_el.get_text(strip=True)[:300] if snippet_el else ""
|
||||
results.append({"title": title, "url": url, "content": snippet, "source": "wechat"})
|
||||
if len(results) >= max_results:
|
||||
break
|
||||
return results
|
||||
except Exception as e:
|
||||
logger.warning("微信搜索失败: %s", e)
|
||||
return []
|
||||
|
||||
|
||||
_PROVIDER_CALLS = {
|
||||
"baidu": _call_baidu,
|
||||
"qiniu": _call_qiniu,
|
||||
"tinyfish": _call_tinyfish,
|
||||
"bing": _call_bing,
|
||||
"mcp": _call_mcp,
|
||||
"360": _call_360,
|
||||
"sogou": _call_sogou,
|
||||
"wechat": _call_wechat,
|
||||
}
|
||||
|
||||
|
||||
@@ -221,7 +325,8 @@ def search(query: str, max_results: int = 5) -> List[Dict]:
|
||||
if (p.get("usage_today") or 0) >= (p.get("daily_limit") or 99999):
|
||||
logger.info("提供商 %s 已达日限 %s，跳过", p.get("name"), p.get("daily_limit"))
|
||||
continue
|
||||
if not p.get("api_key") and p.get("provider_type") != "mcp":
|
||||
no_key_types = {"mcp", "360", "sogou", "wechat"}
|
||||
if not p.get("api_key") and p.get("provider_type") not in no_key_types:
|
||||
logger.info("提供商 %s 未配置 API Key，跳过", p.get("name"))
|
||||
continue
|
||||
call_fn = _PROVIDER_CALLS.get(p.get("provider_type"))
|
||||
|
||||
Reference in New Issue
Block a user