mirosociety/skills/searxng_skills/searxng_search.py

#!/usr/bin/env python3
"""
SearXNG API 客户端 - 快速搜索工具
"""

import requests
import json
import sys
import argparse
from urllib.parse import urlencode

# 配置
SEARXNG_URL = "https://searxng.erhe.top"
API_TOKEN = "fGWEFWGrgasegtjt1jUJG"

def search(query, 
           format_type="json",
           categories=None,
           language="auto",
           safesearch=0,
           time_range=None,
           max_results=10):
    """
    搜索函数
    
    参数:
        query: 搜索关键词
        format_type: 返回格式 (json/html/csv/rss)
        categories: 分类 (images/videos/news/music/general)
        language: 语言代码 (auto/zh/en/ja/de/fr)
        safesearch: 安全搜索级别 (0=关闭, 1=适中, 2=严格)
        time_range: 时间范围 (day/week/month/year)
        max_results: 最大结果数
    
    返回:
        dict: 搜索结果
    """
    params = {
        "q": query,
        "format": format_type,
        "token": API_TOKEN,
        "language": language,
        "safesearch": safesearch
    }
    
    if categories:
        params["categories"] = categories
    
    if time_range:
        params["time_range"] = time_range
    
    try:
        response = requests.get(
            f"{SEARXNG_URL}/search",
            params=params,
            timeout=15
        )
        
        if response.status_code == 200:
            if format_type == "json":
                return response.json()
            else:
                return {"html": response.text}
        elif response.status_code == 403:
            return {"error": "API token 无效或缺失"}
        else:
            return {"error": f"请求失败: {response.status_code}"}
            
    except requests.exceptions.Timeout:
        return {"error": "请求超时"}
    except requests.exceptions.ConnectionError:
        return {"error": "连接失败，请检查网络"}
    except Exception as e:
        return {"error": str(e)}

def print_results(results, max_results=10, verbose=False):
    """格式化输出结果"""
    if "error" in results:
        print(f"❌ 错误: {results['error']}")
        return
    
    query = results.get('query', '未知')
    results_list = results.get('results', [])
    unresponsive = results.get('unresponsive_engines', [])
    
    print(f"\n{'='*70}")
    print(f"🔍 查询: {query}")
    print(f"📊 找到 {len(results_list)} 个结果")
    if unresponsive:
        print(f"⚠️  部分搜索引擎失败: {', '.join([u[0] for u in unresponsive[:3]])}")
    print(f"{'='*70}\n")
    
    for i, result in enumerate(results_list[:max_results], 1):
        title = result.get('title', '无标题')
        url = result.get('url', '无链接')
        content = result.get('content', '')
        engine = result.get('engine', '未知')
        published = result.get('publishedDate', '')
        
        print(f"{i:2d}. {title}")
        print(f"    🔗 {url}")
        
        if content:
            # 截断内容
            if len(content) > 150:
                content = content[:150] + "..."
            print(f"    📝 {content}")
        
        print(f"    ⚙️  来源: {engine}", end="")
        if published:
            print(f" | 📅 {published}", end="")
        print("\n")

def main():
    parser = argparse.ArgumentParser(
        description="SearXNG 搜索工具",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例:
  %(prog)s python 教程
  %(prog)s "machine learning" --category videos
  %(prog)s linux --language en --time-range week
  %(prog)s python --max-results 20 --verbose
        """
    )
    
    parser.add_argument('query', nargs='+', help='搜索关键词')
    parser.add_argument('-c', '--category', 
                       choices=['general', 'images', 'videos', 'news', 'music', 'files', 'it', 'science'],
                       default='general',
                       help='搜索分类 (默认: general)')
    parser.add_argument('-l', '--language',
                       default='auto',
                       help='语言代码 (默认: auto)')
    parser.add_argument('-s', '--safesearch',
                       type=int, choices=[0, 1, 2], default=0,
                       help='安全搜索: 0=关闭, 1=适中, 2=严格 (默认: 0)')
    parser.add_argument('-t', '--time-range',
                       choices=['day', 'week', 'month', 'year'],
                       help='时间范围')
    parser.add_argument('-m', '--max-results',
                       type=int, default=10,
                       help='最大结果数 (默认: 10)')
    parser.add_argument('-v', '--verbose',
                       action='store_true',
                       help='详细输出')
    parser.add_argument('--format',
                       choices=['json', 'html', 'csv', 'rss'],
                       default='json',
                       help='返回格式 (默认: json)')
    parser.add_argument('--raw',
                       action='store_true',
                       help='输出原始 JSON')
    
    args = parser.parse_args()
    
    # 组合查询词
    query = ' '.join(args.query)
    
    # 执行搜索
    results = search(
        query,
        format_type=args.format,
        categories=args.category if args.category != 'general' else None,
        language=args.language,
        safesearch=args.safesearch,
        time_range=args.time_range,
        max_results=args.max_results
    )
    
    # 输出结果
    if args.raw:
        print(json.dumps(results, indent=2, ensure_ascii=False))
    else:
        print_results(results, max_results=args.max_results, verbose=args.verbose)
        
        # 如果有建议搜索词
        if 'suggestions' in results and results['suggestions']:
            print(f"💡 你可能想搜索:")
            for suggestion in results['suggestions']:
                print(f"   - {suggestion}")

if __name__ == "__main__":
    main()