nanocode/it_automation_test/scripts/test_crawler.py

#!/usr/bin/env python3
"""
简化版网页抓取测试脚本
只测试InfoQ网站，避免过多请求
"""

import sys
import os
import sqlite3
import datetime
import time
import random
from urllib.parse import urljoin
import requests
from bs4 import BeautifulSoup

# 测试配置
TEST_CONFIG = {
    'database_path': '/home/ubuntu/nanocode/it_automation_test/data/test_articles.db',
    'base_dir': '/home/ubuntu/nanocode/it_automation_test',
    'test_website': {
        'name': 'InfoQ',
        'url': 'https://www.infoq.com',
        'category': 'tech'
    },
    'user_agents': [
        'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
        'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36',
    ],
    'request_delay': 3,  # 更长的延迟，避免被封
    'timeout': 30,
    'max_articles': 5,   # 只抓取5篇文章用于测试
}

def init_database():
    """初始化测试数据库"""
    db_path = TEST_CONFIG['database_path']
    os.makedirs(os.path.dirname(db_path), exist_ok=True)
    
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS test_articles (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        title TEXT NOT NULL,
        url TEXT UNIQUE NOT NULL,
        source TEXT NOT NULL,
        category TEXT,
        publish_date TEXT,
        word_count INTEGER DEFAULT 0,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
    ''')
    
    conn.commit()
    conn.close()
    print(f"✅ 测试数据库初始化完成: {db_path}")

def save_test_article(title, url, source, category):
    """保存测试文章"""
    try:
        conn = sqlite3.connect(TEST_CONFIG['database_path'])
        cursor = conn.cursor()
        
        cursor.execute('''
        INSERT OR IGNORE INTO test_articles 
        (title, url, source, category, publish_date)
        VALUES (?, ?, ?, ?, ?)
        ''', (title, url, source, category, datetime.datetime.now().strftime('%Y-%m-%d')))
        
        conn.commit()
        conn.close()
        return True
    except Exception as e:
        print(f"❌ 保存文章失败: {e}")
        return False

def fetch_with_retry(url, max_retries=2):
    """带重试的网页抓取"""
    headers = {
        'User-Agent': random.choice(TEST_CONFIG['user_agents']),
        'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
        'Accept-Language': 'en-US,en;q=0.5',
    }
    
    for attempt in range(max_retries):
        try:
            time.sleep(TEST_CONFIG['request_delay'])
            print(f"  尝试抓取 ({attempt+1}/{max_retries}): {url[:60]}...")
            response = requests.get(url, headers=headers, timeout=TEST_CONFIG['timeout'])
            response.raise_for_status()
            return response
        except Exception as e:
            if attempt == max_retries - 1:
                print(f"❌ 抓取失败: {e}")
                return None
            time.sleep(2 ** attempt)

def extract_infoq_articles(html, base_url):
    """从InfoQ首页提取文章链接"""
    articles = []
    soup = BeautifulSoup(html, 'html.parser')
    
    # InfoQ的文章通常在带有特定class的链接中
    # 这里使用通用选择器，避免网站结构变化
    article_links = []
    
    # 尝试多种选择器
    selectors = [
        'a[href*="/news/"]',
        'a[href*="/articles/"]',
        '.card__title a',
        '.article-card a',
        'h2 a',
        'h3 a'
    ]
    
    for selector in selectors:
        links = soup.select(selector)
        if links:
            article_links.extend(links)
            break
    
    # 去重
    seen_urls = set()
    for link in article_links:
        if len(articles) >= TEST_CONFIG['max_articles']:
            break
            
        href = link.get('href', '')
        title = link.get_text(strip=True)
        
        if not href or not title or len(title) < 10:
            continue
            
        # 构建完整URL
        if href.startswith('http'):
            full_url = href
        else:
            full_url = urljoin(base_url, href)
            
        # 去重
        if full_url in seen_urls:
            continue
        seen_urls.add(full_url)
        
        articles.append({
            'title': title,
            'url': full_url,
            'source': TEST_CONFIG['test_website']['name'],
            'category': TEST_CONFIG['test_website']['category'],
        })
    
    return articles

def main():
    """主测试函数"""
    print("=" * 60)
    print("🚀 开始网页抓取测试")
    print("=" * 60)
    print(f"📁 测试目录: {TEST_CONFIG['base_dir']}")
    print(f"🗄️ 数据库: {TEST_CONFIG['database_path']}")
    print(f"🌐 测试网站: {TEST_CONFIG['test_website']['name']}")
    print(f"🔗 网站URL: {TEST_CONFIG['test_website']['url']}")
    print("=" * 60)
    
    # 初始化数据库
    init_database()
    
    # 抓取首页
    site = TEST_CONFIG['test_website']
    print(f"\n📡 抓取网站: {site['name']}")
    
    response = fetch_with_retry(site['url'])
    if not response:
        print("❌ 无法抓取网站首页")
        return 0
    
    # 提取文章
    articles = extract_infoq_articles(response.text, site['url'])
    print(f"📰 找到 {len(articles)} 篇文章")
    
    # 保存文章
    saved_count = 0
    for i, article in enumerate(articles, 1):
        print(f"  {i}. {article['title'][:50]}...")
        if save_test_article(article['title'], article['url'], 
                           article['source'], article['category']):
            saved_count += 1
    
    # 验证结果
    print(f"\n✅ 测试完成，共保存 {saved_count} 篇文章")
    
    # 显示保存的文章
    if saved_count > 0:
        print("\n📋 保存的文章列表:")
        conn = sqlite3.connect(TEST_CONFIG['database_path'])
        cursor = conn.cursor()
        cursor.execute('SELECT title, url FROM test_articles ORDER BY id DESC LIMIT 5')
        for row in cursor.fetchall():
            print(f"  • {row[0][:40]}...")
        conn.close()
    
    return saved_count

if __name__ == '__main__':
    try:
        saved = main()
        if saved > 0:
            print("\n🎉 网页抓取测试成功！")
            sys.exit(0)
        else:
            print("\n⚠️ 网页抓取测试完成，但未保存文章")
            sys.exit(1)
    except Exception as e:
        print(f"\n❌ 测试过程中出现错误: {e}")
        import traceback
        traceback.print_exc()
        sys.exit(1)