Initial commit: HBR 文章爬蟲專案
- Scrapy 爬蟲框架，爬取 HBR 繁體中文文章 - Flask Web 應用程式，提供文章查詢介面 - SQL Server 資料庫整合 - 自動化排程與郵件通知功能 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
212
test_db_connection.py
Normal file
212
test_db_connection.py
Normal file
@@ -0,0 +1,212 @@
|
||||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
資料庫連線測試腳本
|
||||
用於測試資料庫連線、建立資料庫和資料表
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import logging
|
||||
from pathlib import Path
|
||||
|
||||
# 設定日誌
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# 加入專案路徑
|
||||
project_root = Path(__file__).parent
|
||||
sys.path.insert(0, str(project_root))
|
||||
|
||||
from hbr_crawler.hbr_crawler.database import DatabaseManager, get_database_manager
|
||||
|
||||
# 資料庫連線資訊
|
||||
DB_CONFIG = {
|
||||
'host': 'mysql.theaken.com',
|
||||
'port': 33306,
|
||||
'user': 'A101',
|
||||
'password': 'Aa123456',
|
||||
'database': 'db_A101'
|
||||
}
|
||||
|
||||
|
||||
def test_basic_connection():
|
||||
"""測試基本連線（不指定資料庫）"""
|
||||
print("\n" + "="*50)
|
||||
print("測試 1: 基本資料庫連線（不指定資料庫）")
|
||||
print("="*50)
|
||||
|
||||
db_manager = DatabaseManager(
|
||||
host=DB_CONFIG['host'],
|
||||
port=DB_CONFIG['port'],
|
||||
user=DB_CONFIG['user'],
|
||||
password=DB_CONFIG['password'],
|
||||
database=None
|
||||
)
|
||||
|
||||
if db_manager.test_connection():
|
||||
print("✓ 基本連線測試成功")
|
||||
return True
|
||||
else:
|
||||
print("✗ 基本連線測試失敗")
|
||||
return False
|
||||
|
||||
|
||||
def create_database():
|
||||
"""建立 HBR_scraper 資料庫（如果需要）"""
|
||||
print("\n" + "="*50)
|
||||
print("測試 2: 檢查資料庫連線（使用現有資料庫 db_A101）")
|
||||
print("="*50)
|
||||
|
||||
db_manager = DatabaseManager(
|
||||
host=DB_CONFIG['host'],
|
||||
port=DB_CONFIG['port'],
|
||||
user=DB_CONFIG['user'],
|
||||
password=DB_CONFIG['password'],
|
||||
database=None
|
||||
)
|
||||
|
||||
# 嘗試建立資料庫（可能需要管理員權限）
|
||||
try:
|
||||
if db_manager.create_database('HBR_scraper'):
|
||||
print("✓ 資料庫建立成功（或已存在）")
|
||||
return True
|
||||
else:
|
||||
print("✗ 資料庫建立失敗")
|
||||
print("提示：如果沒有 CREATE DATABASE 權限，請請管理員協助建立資料庫")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"✗ 資料庫建立失敗: {e}")
|
||||
print("提示：如果沒有 CREATE DATABASE 權限，請請管理員協助建立資料庫")
|
||||
print("或者使用現有的資料庫，修改 DB_NAME 設定")
|
||||
return False
|
||||
|
||||
|
||||
def test_database_connection():
|
||||
"""測試連接到 db_A101 資料庫"""
|
||||
print("\n" + "="*50)
|
||||
print("測試 3: 連接到 db_A101 資料庫")
|
||||
print("="*50)
|
||||
|
||||
db_manager = DatabaseManager(
|
||||
host=DB_CONFIG['host'],
|
||||
port=DB_CONFIG['port'],
|
||||
user=DB_CONFIG['user'],
|
||||
password=DB_CONFIG['password'],
|
||||
database=DB_CONFIG['database']
|
||||
)
|
||||
|
||||
if db_manager.test_connection(DB_CONFIG['database']):
|
||||
print("✓ 資料庫連線測試成功")
|
||||
return True
|
||||
else:
|
||||
print("✗ 資料庫連線測試失敗")
|
||||
return False
|
||||
|
||||
|
||||
def create_tables():
|
||||
"""建立資料表"""
|
||||
print("\n" + "="*50)
|
||||
print("測試 4: 建立資料表")
|
||||
print("="*50)
|
||||
|
||||
db_manager = DatabaseManager(
|
||||
host=DB_CONFIG['host'],
|
||||
port=DB_CONFIG['port'],
|
||||
user=DB_CONFIG['user'],
|
||||
password=DB_CONFIG['password'],
|
||||
database=DB_CONFIG['database']
|
||||
)
|
||||
|
||||
sql_file = project_root / 'create_tables.sql'
|
||||
|
||||
if not sql_file.exists():
|
||||
print(f"✗ SQL 檔案不存在: {sql_file}")
|
||||
return False
|
||||
|
||||
if db_manager.execute_sql_file(str(sql_file), DB_CONFIG['database']):
|
||||
print("✓ 資料表建立成功")
|
||||
return True
|
||||
else:
|
||||
print("✗ 資料表建立失敗")
|
||||
return False
|
||||
|
||||
|
||||
def verify_tables():
|
||||
"""驗證資料表是否建立成功"""
|
||||
print("\n" + "="*50)
|
||||
print("測試 5: 驗證資料表")
|
||||
print("="*50)
|
||||
|
||||
db_manager = DatabaseManager(
|
||||
host=DB_CONFIG['host'],
|
||||
port=DB_CONFIG['port'],
|
||||
user=DB_CONFIG['user'],
|
||||
password=DB_CONFIG['password'],
|
||||
database=DB_CONFIG['database']
|
||||
)
|
||||
|
||||
expected_tables = ['articles', 'tags', 'article_tags']
|
||||
|
||||
try:
|
||||
tables = db_manager.execute_query(
|
||||
"SHOW TABLES",
|
||||
database=DB_CONFIG['database']
|
||||
)
|
||||
|
||||
# 取得資料表名稱列表
|
||||
table_names = [list(table.values())[0] for table in tables]
|
||||
|
||||
print(f"找到 {len(table_names)} 個資料表: {', '.join(table_names)}")
|
||||
|
||||
for table in expected_tables:
|
||||
if table in table_names:
|
||||
print(f"✓ 資料表 {table} 存在")
|
||||
else:
|
||||
print(f"✗ 資料表 {table} 不存在")
|
||||
|
||||
return all(table in table_names for table in expected_tables)
|
||||
except Exception as e:
|
||||
print(f"✗ 驗證資料表失敗: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def main():
|
||||
"""主函數"""
|
||||
print("\n" + "="*60)
|
||||
print("HBR 爬蟲系統 - 資料庫連線測試")
|
||||
print("="*60)
|
||||
|
||||
results = []
|
||||
|
||||
# 執行測試
|
||||
results.append(("基本連線", test_basic_connection()))
|
||||
results.append(("建立資料庫", create_database()))
|
||||
results.append(("資料庫連線", test_database_connection()))
|
||||
results.append(("建立資料表", create_tables()))
|
||||
results.append(("驗證資料表", verify_tables()))
|
||||
|
||||
# 顯示測試結果摘要
|
||||
print("\n" + "="*60)
|
||||
print("測試結果摘要")
|
||||
print("="*60)
|
||||
|
||||
for test_name, result in results:
|
||||
status = "✓ 通過" if result else "✗ 失敗"
|
||||
print(f"{test_name}: {status}")
|
||||
|
||||
all_passed = all(result for _, result in results)
|
||||
|
||||
if all_passed:
|
||||
print("\n✓ 所有測試通過！資料庫設定完成。")
|
||||
return 0
|
||||
else:
|
||||
print("\n✗ 部分測試失敗，請檢查錯誤訊息。")
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main())
|
||||
|
||||
Reference in New Issue
Block a user