b00e52e1e0
核心功能： - 集成博查Web Search API自动获取网站相关新闻 - 智能新闻更新机制（每日首次访问触发） - 精确新闻搜索（使用引号强制匹配网站名称） - News模型扩展（source_name, source_icon字段） - 网站详情页新闻展示模块 - 新闻来源网站信息展示 - 自动去重防止重复新闻 技术实现： - NewsSearcher工具类封装博查API - 数据库迁移脚本migrate_news_fields.py - 测试脚本test_news_feature.py - 定期任务脚本fetch_news_cron.py - API路由：/api/fetch-site-news, /api/fetch-all-news 配置优化： - 修复manage.sh路径和启动命令 - 博查API配置（BOCHA_API_KEY, BOCHA_BASE_URL） - 新闻搜索参数配置 界面优化： - 详情页新闻模块（左侧主栏） - 相似推荐模块（右侧边栏） - 首页标签图标修复 - 后台添加修改密码功能 - 登录页面优化 部署信息： - 部署日期: 2025-12-30 - 部署方式: 手动上传文件 - 数据库: 已迁移（添加source_name和source_icon字段）
143 lines
4.4 KiB
Python
143 lines
4.4 KiB
Python
"""
|
||
测试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)
|