DonaldFang 方士碩
29c1633e49
- Database schema with MySQL support - LLM API integration (Gemini 2.5 Flash, DeepSeek, OpenAI) - Error handling with copyable error messages - CORS fix for API calls - Complete setup documentation 🤖 Generated with Claude Code https://claude.com/claude-code Co-Authored-By: Claude <noreply@anthropic.com>
131 lines
3.9 KiB
Python
131 lines
3.9 KiB
Python
"""
|
|
Database initialization script for HR Position System
|
|
Connects to MySQL and executes the schema creation script
|
|
"""
|
|
|
|
import os
|
|
import pymysql
|
|
from dotenv import load_dotenv
|
|
|
|
# Load environment variables
|
|
load_dotenv()
|
|
|
|
def init_database():
|
|
"""Initialize the database schema"""
|
|
|
|
# Database connection parameters
|
|
db_config = {
|
|
'host': os.getenv('DB_HOST'),
|
|
'port': int(os.getenv('DB_PORT')),
|
|
'user': os.getenv('DB_USER'),
|
|
'password': os.getenv('DB_PASSWORD'),
|
|
'database': os.getenv('DB_NAME'),
|
|
'charset': 'utf8mb4',
|
|
'cursorclass': pymysql.cursors.DictCursor
|
|
}
|
|
|
|
print(f"Connecting to database: {db_config['host']}:{db_config['port']}")
|
|
print(f"Database: {db_config['database']}")
|
|
|
|
try:
|
|
# Connect to MySQL
|
|
connection = pymysql.connect(**db_config)
|
|
print("✓ Successfully connected to database")
|
|
|
|
# Read SQL schema file
|
|
schema_file = os.path.join(os.path.dirname(__file__), 'database_schema.sql')
|
|
with open(schema_file, 'r', encoding='utf-8') as f:
|
|
sql_script = f.read()
|
|
|
|
# Split SQL script by semicolon and execute each statement
|
|
cursor = connection.cursor()
|
|
|
|
# Split by semicolon and filter out empty statements
|
|
statements = [stmt.strip() for stmt in sql_script.split(';') if stmt.strip()]
|
|
|
|
print(f"\nExecuting {len(statements)} SQL statements...")
|
|
|
|
executed = 0
|
|
for i, statement in enumerate(statements, 1):
|
|
try:
|
|
# Skip comments and empty lines
|
|
if not statement or statement.startswith('--'):
|
|
continue
|
|
|
|
cursor.execute(statement)
|
|
executed += 1
|
|
|
|
# Show progress every 10 statements
|
|
if executed % 10 == 0:
|
|
print(f" Executed {executed} statements...")
|
|
|
|
except Exception as e:
|
|
print(f" Warning on statement {i}: {str(e)[:100]}")
|
|
continue
|
|
|
|
# Commit changes
|
|
connection.commit()
|
|
print(f"\n✓ Successfully executed {executed} SQL statements")
|
|
|
|
# Verify tables were created
|
|
cursor.execute("""
|
|
SELECT table_name
|
|
FROM information_schema.tables
|
|
WHERE table_schema = 'hr_position_system'
|
|
ORDER BY table_name
|
|
""")
|
|
tables = cursor.fetchall()
|
|
|
|
print(f"\n✓ Created {len(tables)} tables:")
|
|
for table in tables:
|
|
print(f" - {table['table_name']}")
|
|
|
|
# Close connection
|
|
cursor.close()
|
|
connection.close()
|
|
|
|
print("\n✓ Database initialization completed successfully!")
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"\n✗ Error initializing database: {str(e)}")
|
|
return False
|
|
|
|
def test_connection():
|
|
"""Test database connection"""
|
|
db_config = {
|
|
'host': os.getenv('DB_HOST'),
|
|
'port': int(os.getenv('DB_PORT')),
|
|
'user': os.getenv('DB_USER'),
|
|
'password': os.getenv('DB_PASSWORD'),
|
|
'database': os.getenv('DB_NAME')
|
|
}
|
|
|
|
try:
|
|
connection = pymysql.connect(**db_config)
|
|
print("✓ Database connection test successful")
|
|
connection.close()
|
|
return True
|
|
except Exception as e:
|
|
print(f"✗ Database connection test failed: {str(e)}")
|
|
return False
|
|
|
|
if __name__ == '__main__':
|
|
print("=" * 60)
|
|
print("HR Position System - Database Initialization")
|
|
print("=" * 60)
|
|
print()
|
|
|
|
# Test connection first
|
|
print("Step 1: Testing database connection...")
|
|
if not test_connection():
|
|
print("\nPlease check your database configuration in .env file")
|
|
exit(1)
|
|
|
|
print("\nStep 2: Initializing database schema...")
|
|
if init_database():
|
|
print("\nDatabase is ready to use!")
|
|
else:
|
|
print("\nDatabase initialization failed. Please check the errors above.")
|
|
exit(1)
|