feat: Initialize project with core dependencies, Vultr API client, customer database service, and migration documentation.

2026-02-06 18:11:28 +08:00
parent 73366f4395
commit 41c2fa5fcc
2618 changed files with 852274 additions and 1449 deletions
--- a/check_billing_db.py
+++ b/check_billing_db.py
@@ -0,0 +1,107 @@
+#!/usr/bin/env python3
+"""
+Check Billing Database Connection - Python Version
+
+Replacement for check_billing_db.js with same functionality.
+
+Usage:
+    python check_billing_db.py
+
+Environment variables required:
+    BILLING_DB_HOST     - Database host address
+    BILLING_DB_USER     - Database username
+    BILLING_DB_PASS     - Database password
+    BILLING_DB_NAME     - Database name
+
+Example .env file:
+    BILLING_DB_HOST=localhost
+    BILLING_DB_USER=root
+    BILLING_DB_PASS=password
+    BILLING_DB_NAME=billing_db
+
+Install dependencies:
+    pip install -r requirements.txt
+"""
+
+import os
+import sys
+import pymysql
+from pymysql import MySQLError
+from dotenv import load_dotenv
+
+def check_billing():
+    """Test connection to BILLING Database"""
+    print('Testing connection to BILLING Database...')
+    
+    # Load environment variables
+    load_dotenv()
+    
+    # Get environment variables
+    host = os.getenv('BILLING_DB_HOST')
+    user = os.getenv('BILLING_DB_USER')
+    password = os.getenv('BILLING_DB_PASS')
+    database = os.getenv('BILLING_DB_NAME')
+    
+    print(f'Host: {host}')
+    print(f'User: {user}')
+    print(f'Database: {database}')
+    
+    if not all([host, user, password, database]):
+        print('ERROR: Missing required environment variables')
+        print('Required: BILLING_DB_HOST, BILLING_DB_USER, BILLING_DB_PASS, BILLING_DB_NAME')
+        sys.exit(1)
+    
+    try:
+        # Validate connection parameters
+        if not all([host, user, password, database]):
+            print('ERROR: Missing database connection parameters')
+            return False
+        
+        # Type assurance after validation
+        assert host is not None and user is not None and password is not None and database is not None
+        
+        # Establish connection with timeout
+        connection = pymysql.connect(
+            host=host,
+            user=user,
+            password=password,
+            database=database,
+            connect_timeout=10  # 10 seconds timeout
+        )
+        
+        print('Connection successful!')
+        
+        # Test SELECT query
+        with connection.cursor(pymysql.cursors.DictCursor) as cursor:
+            cursor.execute('SELECT COUNT(*) as count FROM customer')
+            rows = cursor.fetchall()
+            
+            if rows:
+                count = rows[0]['count']
+                print(f'Test SELECT successful. Customer Count: {count}')
+            else:
+                print('Test SELECT returned no results')
+        
+        connection.close()
+        
+        print('BILLING DB TEST PASSED')
+        return True
+        
+    except MySQLError as e:
+        print(f'BILLING DB Test Failed: {e}')
+        
+        # Provide helpful hints based on error
+        if 'timed out' in str(e).lower() or 'timeout' in str(e).lower():
+            print('Hint: Check firewall or IP whitelist on the remote server.')
+        elif 'access denied' in str(e).lower():
+            print('Hint: Check username/password or database permissions.')
+        elif 'unknown database' in str(e).lower():
+            print('Hint: Check database name or create the database first.')
+        elif 'cannot connect' in str(e).lower():
+            print('Hint: Check host/port or network connectivity.')
+        
+        return False
+
+if __name__ == '__main__':
+    success = check_billing()
+    sys.exit(0 if success else 1)