98 lines
2.9 KiB
Python
Executable File
98 lines
2.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
测试定时爬取脚本配置
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
|
|
# 添加项目路径
|
|
sys.path.insert(0, os.path.dirname(__file__))
|
|
|
|
try:
|
|
from gx_gp_monitor.core.config_manager import load_config
|
|
from gx_gp_monitor.core.logger import init_logger
|
|
from gx_gp_monitor.storage.postgresql import init_storage
|
|
|
|
def test_config():
|
|
"""测试配置加载"""
|
|
print("🔍 测试配置加载...")
|
|
|
|
config = load_config()
|
|
if not config:
|
|
print("❌ 配置加载失败")
|
|
return False
|
|
|
|
print("✅ 配置加载成功")
|
|
print(f" 关键词: {config.crawler.keyword}")
|
|
print(f" 企业微信启用: {config.wechat_app.enabled}")
|
|
print(f" 数据库启用: {config.database.enabled}")
|
|
|
|
return True
|
|
|
|
def test_logger():
|
|
"""测试日志系统"""
|
|
print("\n🔍 测试日志系统...")
|
|
|
|
try:
|
|
init_logger()
|
|
print("✅ 日志系统初始化成功")
|
|
return True
|
|
except Exception as e:
|
|
print(f"❌ 日志系统初始化失败: {e}")
|
|
return False
|
|
|
|
def test_database():
|
|
"""测试数据库连接"""
|
|
print("\n🔍 测试数据库连接...")
|
|
|
|
try:
|
|
init_storage()
|
|
print("✅ 数据库连接成功")
|
|
return True
|
|
except Exception as e:
|
|
print(f"❌ 数据库连接失败: {e}")
|
|
return False
|
|
|
|
def main():
|
|
"""主测试函数"""
|
|
print("🚀 开始测试定时爬取脚本配置\n")
|
|
|
|
results = []
|
|
results.append(("配置加载", test_config()))
|
|
results.append(("日志系统", test_logger()))
|
|
results.append(("数据库连接", test_database()))
|
|
|
|
print("\n" + "="*50)
|
|
print("📊 测试结果:")
|
|
|
|
all_passed = True
|
|
for test_name, passed in results:
|
|
status = "✅ 通过" if passed else "❌ 失败"
|
|
print(f" {test_name}: {status}")
|
|
if not passed:
|
|
all_passed = False
|
|
|
|
print("\n" + "="*50)
|
|
if all_passed:
|
|
print("🎉 所有测试通过！可以安全使用定时爬取脚本。")
|
|
print("\n💡 使用方法:")
|
|
print(" 直接运行: python cron_crawl.py")
|
|
print(" 定时运行: 参考 CRON_README.md 或 cron_example.txt")
|
|
else:
|
|
print("⚠️ 部分测试失败，请检查配置后再使用脚本。")
|
|
|
|
return all_passed
|
|
|
|
if __name__ == "__main__":
|
|
success = main()
|
|
sys.exit(0 if success else 1)
|
|
|
|
except ImportError as e:
|
|
print(f"❌ 导入失败: {e}", file=sys.stderr)
|
|
print("请确保已安装所有依赖: pip install -r gx_gp_monitor/requirements.txt", file=sys.stderr)
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(f"❌ 测试失败: {e}", file=sys.stderr)
|
|
sys.exit(1)
|