zjpb.net/test_news_feature.py

"""
测试v2.2新闻功能 - 完整流程测试
"""
import os
import sys
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# 添加项目路径
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from app import create_app
from models import db, Site, News
from utils.news_searcher import NewsSearcher

def test_news_feature():
    """测试新闻功能"""
    print("=" * 60)
    print("v2.2 新闻功能测试")
    print("=" * 60)

    # 创建应用上下文
    app = create_app('development')

    with app.app_context():
        # 1. 测试API配置
        print("\n[1/4] 检查API配置...")
        api_key = app.config.get('BOCHA_API_KEY')
        if not api_key:
            print(">>> 错误：BOCHA_API_KEY未配置")
            return False
        print(f">>> API Key: {api_key[:20]}...")

        # 2. 测试数据库连接
        print("\n[2/4] 检查数据库...")
        try:
            site_count = Site.query.filter_by(is_active=True).count()
            print(f">>> 找到 {site_count} 个启用的网站")

            if site_count == 0:
                print(">>> 警告：没有可用的网站")
                return False

        except Exception as e:
            print(f">>> 数据库错误：{e}")
            return False

        # 3. 测试新闻搜索
        print("\n[3/4] 测试新闻搜索...")
        searcher = NewsSearcher(api_key)

        # 获取第一个网站
        site = Site.query.filter_by(is_active=True).first()
        print(f">>> 测试网站：{site.name}")

        try:
            news_items = searcher.search_site_news(
                site_name=site.name,
                site_url=site.url,
                count=3,
                freshness='oneWeek'
            )

            print(f">>> 找到 {len(news_items)} 条新闻")

            if news_items:
                print("\n新闻列表：")
                for i, item in enumerate(news_items, 1):
                    print(f"  {i}. {item['title'][:50]}...")
                    print(f"     来源：{item.get('site_name', '未知')}")
                    print(f"     URL：{item['url'][:60]}...")

        except Exception as e:
            print(f">>> 搜索失败：{e}")
            return False

        # 4. 测试保存到数据库
        print(f"\n[4/4] 测试保存到数据库...")

        if not news_items:
            print(">>> 没有新闻可保存")
            return True

        try:
            saved_count = 0
            for item in news_items[:2]:  # 只保存前2条作为测试
                # 检查是否已存在
                existing = News.query.filter_by(
                    site_id=site.id,
                    url=item['url']
                ).first()

                if not existing:
                    news = News(
                        site_id=site.id,
                        title=item['title'],
                        content=item.get('summary') or item.get('snippet', ''),
                        url=item['url'],
                        source_name=item.get('site_name', ''),
                        source_icon=item.get('site_icon', ''),
                        published_at=item.get('published_at'),
                        news_type='Search Result',
                        is_active=True
                    )
                    db.session.add(news)
                    saved_count += 1

            db.session.commit()
            print(f">>> 成功保存 {saved_count} 条新闻")

            # 验证保存
            total_news = News.query.filter_by(site_id=site.id).count()
            print(f">>> 该网站共有 {total_news} 条新闻记录")

        except Exception as e:
            db.session.rollback()
            print(f">>> 保存失败：{e}")
            return False

        print("\n" + "=" * 60)
        print(">>> 所有测试通过！")
        print("=" * 60)

        # 提供下一步建议
        print("\n下一步操作：")
        print(f"1. 访问网站详情页查看新闻：http://localhost:5000/site/{site.code}")
        print(f"2. 访问后台新闻管理：http://localhost:5000/admin/newsadmin/")
        print(f"3. 运行定期任务脚本：python fetch_news_cron.py --limit 5")

        return True

if __name__ == '__main__':
    try:
        success = test_news_feature()
        sys.exit(0 if success else 1)
    except Exception as e:
        print(f"\n严重错误：{e}")
        import traceback
        traceback.print_exc()
        sys.exit(1)