donald/hbr-crawler
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Security: 移除硬編碼的資料庫帳密

Browse Source 
- database.py: 改從環境變數讀取 DB 設定，新增必要變數檢查
- settings.py: 改從環境變數讀取 DB 設定
- test_db_connection.py: 改從環境變數讀取 DB 設定

所有機敏資料現在必須透過 .env 檔案設定，
參考 .env.example 取得設定範本。

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
DonaldFang 方士碩
2025-12-03 19:03:50 +08:00
parent 5b4041afd5
commit 719ba8c133
3 changed files with 100 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
hbr_crawler/hbr_crawler/database.py
View File

@@ -200,33 +200,64 @@ class DatabaseManager:
def get_database_manager() -> DatabaseManager:
"""
從環境變數或設定檔取得資料庫連線資訊，建立 DatabaseManager 實例
優先順序：
1. 環境變數
2. Scrapy settings（如果可用）
3. 預設值
環境變數：
DB_HOST: 資料庫主機位址 (必要)
DB_PORT: 資料庫埠號 (預設: 3306)
DB_USER: 資料庫使用者名稱 (必要)
DB_PASSWORD: 資料庫密碼 (必要)
DB_NAME: 資料庫名稱 (必要)
Returns:
DatabaseManager: 資料庫管理物件
Raises:
ValueError: 當必要的環境變數未設定時
"""
import os
from dotenv import load_dotenv
# 載入 .env 檔案
load_dotenv()
# 嘗試從 Scrapy settings 取得設定
try:
from scrapy.utils.project import get_project_settings
settings = get_project_settings()
host = settings.get('DB_HOST', os.environ.get('DB_HOST', 'mysql.theaken.com'))
port = settings.getint('DB_PORT', int(os.environ.get('DB_PORT', 33306)))
user = settings.get('DB_USER', os.environ.get('DB_USER', 'A101'))
password = settings.get('DB_PASSWORD', os.environ.get('DB_PASSWORD', 'Aa123456'))
database = settings.get('DB_NAME', os.environ.get('DB_NAME', 'db_A101'))
host = settings.get('DB_HOST') or os.environ.get('DB_HOST')
port = settings.getint('DB_PORT', int(os.environ.get('DB_PORT', 3306)))
user = settings.get('DB_USER') or os.environ.get('DB_USER')
password = settings.get('DB_PASSWORD') or os.environ.get('DB_PASSWORD')
database = settings.get('DB_NAME') or os.environ.get('DB_NAME')
except:
# 如果無法取得 Scrapy settings，使用環境變數或預設值
host = os.environ.get('DB_HOST', 'mysql.theaken.com')
port = int(os.environ.get('DB_PORT', 33306))
user = os.environ.get('DB_USER', 'A101')
password = os.environ.get('DB_PASSWORD', 'Aa123456')
database = os.environ.get('DB_NAME', 'db_A101')
# 如果無法取得 Scrapy settings，使用環境變數
host = os.environ.get('DB_HOST')
port = int(os.environ.get('DB_PORT', 3306))
user = os.environ.get('DB_USER')
password = os.environ.get('DB_PASSWORD')
database = os.environ.get('DB_NAME')
# 檢查必要的設定
missing_vars = []
if not host:
missing_vars.append('DB_HOST')
if not user:
missing_vars.append('DB_USER')
if not password:
missing_vars.append('DB_PASSWORD')
if not database:
missing_vars.append('DB_NAME')
if missing_vars:
raise ValueError(
f"缺少必要的環境變數: {', '.join(missing_vars)}\n"
f"請在 .env 檔案或系統環境變數中設定這些值。\n"
f"參考 .env.example 檔案取得設定範本。"
)
return DatabaseManager(host, port, user, password, database)

16
hbr_crawler/hbr_crawler/settings.py
View File

@@ -60,12 +60,16 @@ ITEM_PIPELINES = {
'hbr_crawler.pipelines.DatabasePipeline': 400,
}
# 資料庫設定
DB_HOST = 'mysql.theaken.com'
DB_PORT = 33306
DB_USER = 'A101'
DB_PASSWORD = 'Aa123456'
DB_NAME = 'db_A101'
# 資料庫設定（從環境變數讀取）
import os
from dotenv import load_dotenv
load_dotenv()
DB_HOST = os.environ.get('DB_HOST')
DB_PORT = int(os.environ.get('DB_PORT', 3306))
DB_USER = os.environ.get('DB_USER')
DB_PASSWORD = os.environ.get('DB_PASSWORD')
DB_NAME = os.environ.get('DB_NAME')
# Enable and configure the AutoThrottle extension (disabled by default)
# See https://docs.scrapy.org/en/latest/topics/autothrottle.html

76
test_db_connection.py
View File

@@ -20,24 +20,35 @@ logger = logging.getLogger(__name__)
project_root = Path(__file__).parent
sys.path.insert(0, str(project_root))
from dotenv import load_dotenv
from hbr_crawler.hbr_crawler.database import DatabaseManager, get_database_manager
# 資料庫連線資訊
# 載入 .env 檔案
load_dotenv()
# 資料庫連線資訊（從環境變數讀取）
DB_CONFIG = {
'host': 'mysql.theaken.com',
'port': 33306,
'user': 'A101',
'password': 'Aa123456',
'database': 'db_A101'
'host': os.environ.get('DB_HOST'),
'port': int(os.environ.get('DB_PORT', 3306)),
'user': os.environ.get('DB_USER'),
'password': os.environ.get('DB_PASSWORD'),
'database': os.environ.get('DB_NAME')
}
# 檢查必要的環境變數
missing_vars = [k for k, v in DB_CONFIG.items() if v is None and k != 'port']
if missing_vars:
print(f"錯誤: 缺少必要的環境變數: {', '.join(['DB_' + k.upper() for k in missing_vars])}")
print("請在 .env 檔案中設定這些值，參考 .env.example")
sys.exit(1)
def test_basic_connection():
"""測試基本連線（不指定資料庫）"""
print("\n" + "="*50)
print("測試 1: 基本資料庫連線（不指定資料庫）")
print("="*50)
db_manager = DatabaseManager(
host=DB_CONFIG['host'],
port=DB_CONFIG['port'],
@@ -45,7 +56,7 @@ def test_basic_connection():
password=DB_CONFIG['password'],
database=None
)
if db_manager.test_connection():
print("✓ 基本連線測試成功")
return True
@@ -57,9 +68,9 @@ def test_basic_connection():
def create_database():
"""建立 HBR_scraper 資料庫（如果需要）"""
print("\n" + "="*50)
print("測試 2: 檢查資料庫連線（使用現有資料庫 db_A101）")
print("測試 2: 檢查資料庫連線（使用現有資料庫）")
print("="*50)
db_manager = DatabaseManager(
host=DB_CONFIG['host'],
port=DB_CONFIG['port'],
@@ -67,7 +78,7 @@ def create_database():
password=DB_CONFIG['password'],
database=None
)
# 嘗試建立資料庫（可能需要管理員權限）
try:
if db_manager.create_database('HBR_scraper'):
@@ -85,11 +96,11 @@ def create_database():
def test_database_connection():
"""測試連接到 db_A101 資料庫"""
"""測試連接到指定資料庫"""
print("\n" + "="*50)
print("測試 3: 連接到 db_A101 資料庫")
print(f"測試 3: 連接到 {DB_CONFIG['database']} 資料庫")
print("="*50)
db_manager = DatabaseManager(
host=DB_CONFIG['host'],
port=DB_CONFIG['port'],
@@ -97,7 +108,7 @@ def test_database_connection():
password=DB_CONFIG['password'],
database=DB_CONFIG['database']
)
if db_manager.test_connection(DB_CONFIG['database']):
print("✓ 資料庫連線測試成功")
return True
@@ -111,7 +122,7 @@ def create_tables():
print("\n" + "="*50)
print("測試 4: 建立資料表")
print("="*50)
db_manager = DatabaseManager(
host=DB_CONFIG['host'],
port=DB_CONFIG['port'],
@@ -119,13 +130,13 @@ def create_tables():
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
@@ -139,7 +150,7 @@ def verify_tables():
print("\n" + "="*50)
print("測試 5: 驗證資料表")
print("="*50)
db_manager = DatabaseManager(
host=DB_CONFIG['host'],
port=DB_CONFIG['port'],
@@ -147,26 +158,26 @@ def verify_tables():
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}")
@@ -178,27 +189,27 @@ 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
@@ -209,4 +220,3 @@ def main():
if __name__ == '__main__':
sys.exit(main())