feat: 添加数据库配置模块，支持动态获取数据库连接信息

database_test_python/db_config.py

@@ -0,0 +1,63 @@
+import os
+from getpass import getpass
+
+# Global cache for database passwords to avoid asking multiple times for the same database
+_password_cache = {}
+
+def get_db_password(server_name: str, db_name: str) -> str:
+    conn_name = f"{server_name}.{db_name}"
+    
+    # Only ask once per target database per script run
+    if conn_name in _password_cache:
+        return _password_cache[conn_name]
+        
+    # Hardcoded fallback per user request
+    if server_name == "109portal":
+        password = "YourActualStrongPasswordForPostgres!"
+    elif server_name == "80portal":
+        password = "H3IkLUt8K!x"
+    else:
+        env_var_name = f"{server_name.upper()}_PG_PASSWORD"
+        password = os.getenv(env_var_name)
+    
+    if not password:
+        # Prompt specific to 'name'.'database'
+        password = getpass(f"{conn_name} password: ")
+    
+    _password_cache[conn_name] = password
+    return password
+
+def get_pg_conn_kwargs(prefix: str) -> dict:
+    if prefix == "PORTAL":
+        server_name = "80portal"
+        default_port = 15434
+        default_db = "log_platform"
+    elif prefix == "TEST" or prefix == "TARGET":
+        server_name = "109portal"
+        default_port = 15433
+        default_db = "test"
+    else: # SOURCE
+        server_name = "80portal"
+        default_port = 15434
+        default_db = "log_platform"
+
+    dbname = os.getenv(f"{prefix}_PG_DB", default_db)
+    
+    return {
+        "host": os.getenv(f"{prefix}_PG_HOST", "blv-rd.tech"),
+        "port": int(os.getenv(f"{prefix}_PG_PORT", default_port)),
+        "dbname": dbname,
+        "user": os.getenv(f"{prefix}_PG_USER", "log_admin"),
+        "password": get_db_password(server_name, dbname)
+    }
+
+def get_db_url(prefix: str) -> str:
+    kwargs = get_pg_conn_kwargs(prefix)
+    return f"postgresql+psycopg2://{kwargs['user']}:{kwargs['password']}@{kwargs['host']}:{kwargs['port']}/{kwargs['dbname']}"
+
+DB_CONFIG = {
+    "SOURCE": lambda: get_pg_conn_kwargs("SOURCE"),
+    "TARGET": lambda: get_pg_conn_kwargs("TARGET"),
+    "PORTAL": lambda: get_pg_conn_kwargs("PORTAL"),
+    "TEST": lambda: get_pg_conn_kwargs("TEST"),
+}