version: '3.8'

services:
  timescaledb:
    image: docker.1ms.run/timescale/timescaledb-ha:pg17.6-ts2.22.0-oss
    container_name: timescaledb-oss
    restart: unless-stopped
    ports:
      - "5438:5432"  # 已確認的未占用端口，避開沖突
    volumes:
      # 官網指定 PGDATA 路徑，宿主機目錄按實際權限調整
      - /volume5/docker5/timescale/pgdata:/home/postgres/pgdata/data
      - /volume5/docker5/timescale/conf:/etc/postgresql/17/main/conf.d
    #######################################
    # 核心修復：使用容器內默認 postgres 用戶（UID 999），而非宿主機 UID
    # 官網強調：TimescaleDB 鏡像內 postgres 用戶 UID 固定為 999，GID 固定為 999
    #######################################
    user: "999:999"  # 容器內 postgres 用戶的默認 UID/GID，無需修改
    environment:
      # 官網必填初始化配置
      POSTGRES_DB: mystocks
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: password #自己設置
      PGDATA: /home/postgres/pgdata/data  # 嚴格對齊官網 PGDATA 路徑
      
      # 官網單節點與擴展配置
      PATRONI_MODE: "standalone"
      PATRONI_SCOPE: "timescale-single-node"
      TIMESCALEDB_TELEMETRY: "off"
      POSTGRES_MULTIPLE_DATABASES: "mystocks"
      TIMESCALEDB_EXTENSION_CREATE_IF_NOT_EXISTS: "true"
      
      # 官網性能與時區配置
      TZ: "Asia/Shanghai"
      PATRONI_POSTGRESQL_PARAMETERS: |
        shared_buffers = 512MB
        work_mem = 32MB
        max_connections = 100
        log_directory = '/home/postgres/pgdata/data/logs'
    # 官網標準健康檢查
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER} -d ${POSTGRES_DB} -h 127.0.0.1 -p 5432"]
      interval: 10s
      timeout: 5s
      retries: 5
      start_period: 60s  # 適配 PostgreSQL 17 初始化耗時

networks:
  default:
    driver: bridge

import json
import logging
import os
from datetime import datetime
from enum import Enum
from typing import Dict, List, Any, Optional, Tuple
import sqlalchemy as sa
from sqlalchemy import create_engine, Column, Integer, String, DateTime, Text, JSON, Boolean, Enum as SQLEnum
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
import psycopg2
import taos
import redis
import pymysql
from dotenv import load_dotenv
import yaml

# 加載環境變量
load_dotenv()

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("DatabaseTableManager")

# 從環境變量獲取監控數據庫連接
MONITOR_DB_URL = os.getenv("MONITOR_DB_URL", "mysql+pymysql://user:password@localhost/db_monitor")

Base = declarative_base()

class DatabaseType(Enum):
    TDENGINE = "TDengine"
    POSTGRESQL = "PostgreSQL"
    REDIS = "Redis"
    MYSQL = "MySQL"
    MARIADB = "MariaDB"

# ORM模型
class TableCreationLog(Base):
    __tablename__ = 'table_creation_log'
    
    id = Column(Integer, primary_key=True)
    table_name = Column(String(255), nullable=False)
    database_type = Column(String(20), nullable=False)
    database_name = Column(String(255), nullable=False)
    creation_time = Column(DateTime, default=datetime.utcnow)
    modification_time = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    status = Column(String(10), nullable=False)
    table_parameters = Column(JSON, nullable=False)
    ddl_command = Column(Text, nullable=False)
    error_message = Column(Text)
    
    # 關系
    columns = relationship("ColumnDefinitionLog", backref="table_log", cascade="all, delete-orphan")

class ColumnDefinitionLog(Base):
    __tablename__ = 'column_definition_log'
    
    id = Column(Integer, primary_key=True)
    table_log_id = Column(Integer, sa.ForeignKey('table_creation_log.id'))
    column_name = Column(String(255), nullable=False)
    data_type = Column(String(100), nullable=False)
    col_length = Column(Integer)  # 避免使用關鍵字 length
    col_precision = Column(Integer)  # 避免使用關鍵字 precision
    col_scale = Column(Integer)  # 避免使用關鍵字 scale
    is_nullable = Column(Boolean, default=True)
    is_primary_key = Column(Boolean, default=False)
    default_value = Column(String(255))
    comment = Column(Text)

class TableOperationLog(Base):
    __tablename__ = 'table_operation_log'
    
    id = Column(Integer, primary_key=True)
    table_name = Column(String(255), nullable=False)
    database_type = Column(String(20), nullable=False)
    database_name = Column(String(255), nullable=False)
    operation_type = Column(SQLEnum('CREATE', 'ALTER', 'DROP', 'VALIDATE'), nullable=False)
    operation_time = Column(DateTime, default=datetime.utcnow)
    operation_status = Column(String(10), nullable=False)
    operation_details = Column(JSON, nullable=False)
    ddl_command = Column(Text)
    error_message = Column(Text)

class TableValidationLog(Base):
    __tablename__ = 'table_validation_log'
    
    id = Column(Integer, primary_key=True)
    table_name = Column(String(255), nullable=False)
    database_type = Column(String(20), nullable=False)
    database_name = Column(String(255), nullable=False)
    validation_time = Column(DateTime, default=datetime.utcnow)
    validation_status = Column(String(10), nullable=False)
    validation_details = Column(JSON, nullable=False)
    issues_found = Column(Text)

class DatabaseTableManager:
    def __init__(self):
        # 初始化監控數據庫連接
        self.monitor_engine = create_engine(MONITOR_DB_URL)
        Base.metadata.create_all(self.monitor_engine)
        Session = sessionmaker(bind=self.monitor_engine)
        self.monitor_session = Session()
        
        # 從環境變量加載各數據庫連接配置
        self.db_configs = {
            DatabaseType.TDENGINE: {
                'host': os.getenv('TDENGINE_HOST', 'localhost'),
                'user': os.getenv('TDENGINE_USER', 'root'),
                'password': os.getenv('TDENGINE_PASSWORD', 'taosdata'),
                'port': int(os.getenv('TDENGINE_PORT', '6030'))
            },
            DatabaseType.POSTGRESQL: {
                'host': os.getenv('POSTGRESQL_HOST', 'localhost'),
                'user': os.getenv('POSTGRESQL_USER', 'postgres'),
                'password': os.getenv('POSTGRESQL_PASSWORD', ''),
                'port': int(os.getenv('POSTGRESQL_PORT', '5432'))
            },
            DatabaseType.REDIS: {
                'host': os.getenv('REDIS_HOST', 'localhost'),
                'port': int(os.getenv('REDIS_PORT', '6379')),
                'password': os.getenv('REDIS_PASSWORD', None),
                'db': int(os.getenv('REDIS_DB', '0'))
            },
            DatabaseType.MYSQL: {
                'host': os.getenv('MYSQL_HOST', 'localhost'),
                'user': os.getenv('MYSQL_USER', 'root'),
                'password': os.getenv('MYSQL_PASSWORD', ''),
                'port': int(os.getenv('MYSQL_PORT', '3306'))
            },
            DatabaseType.MARIADB: {
                'host': os.getenv('MARIADB_HOST', 'localhost'),
                'user': os.getenv('MARIADB_USER', 'root'),
                'password': os.getenv('MARIADB_PASSWORD', ''),
                'port': int(os.getenv('MARIADB_PORT', '3306'))
            }
        }
        
        # 各數據庫連接池
        self.db_connections = {}
        
    def get_connection(self, db_type: DatabaseType, db_name: str, **kwargs):
        """獲取數據庫連接"""
        conn_key = f"{db_type.value}_{db_name}"
        if conn_key in self.db_connections:
            return self.db_connections[conn_key]
            
        # 獲取默認配置并更新用戶提供的參數
        config = self.db_configs[db_type].copy()
        config.update(kwargs)
        
        try:
            if db_type == DatabaseType.TDENGINE:
                # TDengine連接
                conn = taos.connect(
                    host=config['host'],
                    user=config['user'],
                    password=config['password'],
                    port=config['port'],
                    database=db_name
                )
            elif db_type == DatabaseType.POSTGRESQL:
                # PostgreSQL連接
                conn = psycopg2.connect(
                    host=config['host'],
                    user=config['user'],
                    password=config['password'],
                    port=config['port'],
                    database=db_name
                )
            elif db_type == DatabaseType.REDIS:
                # Redis連接
                conn = redis.Redis(
                    host=config['host'],
                    port=config['port'],
                    db=config['db'],
                    password=config['password'],
                    decode_responses=True
                )
            elif db_type in [DatabaseType.MYSQL, DatabaseType.MARIADB]:
                # MySQL/MariaDB連接
                conn = pymysql.connect(
                    host=config['host'],
                    user=config['user'],
                    password=config['password'],
                    port=config['port'],
                    database=db_name,
                    charset='utf8mb4'
                )
            else:
                raise ValueError(f"Unsupported database type: {db_type}")
                
            self.db_connections[conn_key] = conn
            return conn
            
        except Exception as e:
            logger.error(f"Failed to connect to {db_type.value} database {db_name}: {str(e)}")
            raise
    
    def _log_operation(self, table_name: str, db_type: DatabaseType, db_name: str, 
                      operation_type: str, operation_details: Dict, ddl_command: str = "", 
                      status: str = "success", error_message: str = ""):
        """記錄操作日志到監控數據庫"""
        log_entry = TableOperationLog(
            table_name=table_name,
            database_type=db_type.value,
            database_name=db_name,
            operation_type=operation_type,
            operation_status=status,
            operation_details=operation_details,
            ddl_command=ddl_command,
            error_message=error_message
        )
        self.monitor_session.add(log_entry)
        self.monitor_session.commit()
        return log_entry.id
    
    def create_table(self, db_type: DatabaseType, db_name: str, table_name: str, 
                    columns: List[Dict], **kwargs) -> bool:
        """在指定數據庫中創建表"""
        # 記錄開始信息到監控表
        operation_id = self._log_operation(
            table_name, db_type, db_name, "CREATE", 
            {"columns": columns, "kwargs": kwargs}, 
            status="processing"
        )
        
        try:
            # 獲取數據庫連接
            conn = self.get_connection(db_type, db_name, **kwargs)
            
            # 生成DDL語句
            if db_type == DatabaseType.TDENGINE:
                ddl = self._generate_tdengine_ddl(table_name, columns, kwargs.get('tags', []))
                cursor = conn.cursor()
                cursor.execute(ddl)
            elif db_type == DatabaseType.POSTGRESQL:
                ddl = self._generate_postgresql_ddl(table_name, columns, **kwargs)
                cursor = conn.cursor()
                cursor.execute(ddl)
                conn.commit()
            elif db_type in [DatabaseType.MYSQL, DatabaseType.MARIADB]:
                ddl = self._generate_mysql_ddl(table_name, columns, **kwargs)
                cursor = conn.cursor()
                cursor.execute(ddl)
                conn.commit()
            elif db_type == DatabaseType.REDIS:
                # Redis沒有表結構，這里可以創建一些初始結構
                ddl = f"REDIS INIT for {table_name}"
                self._initialize_redis_structure(conn, table_name, columns)
            else:
                raise ValueError(f"Unsupported database type: {db_type}")
            
            # 記錄表創建日志
            log_entry = TableCreationLog(
                table_name=table_name,
                database_type=db_type.value,
                database_name=db_name,
                status='success',
                table_parameters=json.dumps({"columns": columns, "kwargs": kwargs}),
                ddl_command=ddl
            )
            self.monitor_session.add(log_entry)
            self.monitor_session.flush()  # 獲取ID但不提交
            
            # 記錄列定義
            for col_def in columns:
                col_log = ColumnDefinitionLog(
                    table_log_id=log_entry.id,
                    column_name=col_def['name'],
                    data_type=col_def['type'],
                    col_length=col_def.get('length'),
                    col_precision=col_def.get('precision'),
                    col_scale=col_def.get('scale'),
                    is_nullable=col_def.get('nullable', True),
                    is_primary_key=col_def.get('primary_key', False),
                    default_value=col_def.get('default'),
                    comment=col_def.get('comment')
                )
                self.monitor_session.add(col_log)
            
            # 提交所有更改
            self.monitor_session.commit()
            
            # 更新操作日志
            self._log_operation(
                table_name, db_type, db_name, "CREATE", 
                {"columns": columns, "kwargs": kwargs}, ddl, "success"
            )
            
            # 驗證表結構
            self.validate_table_structure(db_type, db_name, table_name, columns)
            
            return True
            
        except Exception as e:
            # 記錄錯誤信息
            error_msg = str(e)
            logger.error(f"Failed to create table {table_name}: {error_msg}")
            
            # 更新操作日志
            self._log_operation(
                table_name, db_type, db_name, "CREATE", 
                {"columns": columns, "kwargs": kwargs}, "", "failed", error_msg
            )
            
            self.monitor_session.rollback()
            return False
    
    def alter_table(self, db_type: DatabaseType, db_name: str, table_name: str, 
                   alterations: List[Dict], **kwargs) -> bool:
        """修改表結構"""
        # 記錄開始信息
        operation_id = self._log_operation(
            table_name, db_type, db_name, "ALTER", 
            {"alterations": alterations, "kwargs": kwargs}, 
            status="processing"
        )
        
        try:
            conn = self.get_connection(db_type, db_name, **kwargs)
            cursor = conn.cursor()
            
            # 生成ALTER語句
            ddl = self._generate_alter_ddl(db_type, table_name, alterations)
            
            # 執行ALTER語句
            cursor.execute(ddl)
            if db_type != DatabaseType.REDIS:  # Redis不需要commit
                conn.commit()
            
            # 更新操作日志
            self._log_operation(
                table_name, db_type, db_name, "ALTER", 
                {"alterations": alterations, "kwargs": kwargs}, ddl, "success"
            )
            
            # 更新表創建日志中的修改時間
            table_log = self.monitor_session.query(TableCreationLog).filter_by(
                table_name=table_name, 
                database_type=db_type.value,
                database_name=db_name
            ).first()
            
            if table_log:
                table_log.modification_time = datetime.utcnow()
                self.monitor_session.commit()
            
            return True
            
        except Exception as e:
            error_msg = str(e)
            logger.error(f"Failed to alter table {table_name}: {error_msg}")
            
            self._log_operation(
                table_name, db_type, db_name, "ALTER", 
                {"alterations": alterations, "kwargs": kwargs}, "", "failed", error_msg
            )
            
            return False
    
    def drop_table(self, db_type: DatabaseType, db_name: str, table_name: str, **kwargs) -> bool:
        """刪除表"""
        # 記錄開始信息
        operation_id = self._log_operation(
            table_name, db_type, db_name, "DROP", 
            {"kwargs": kwargs}, 
            status="processing"
        )
        
        try:
            conn = self.get_connection(db_type, db_name, **kwargs)
            cursor = conn.cursor()
            
            # 生成DROP語句
            if db_type == DatabaseType.TDENGINE:
                ddl = f"DROP TABLE IF EXISTS {table_name}"
            elif db_type == DatabaseType.POSTGRESQL:
                ddl = f"DROP TABLE IF EXISTS {table_name} CASCADE"
            elif db_type in [DatabaseType.MYSQL, DatabaseType.MARIADB]:
                ddl = f"DROP TABLE IF EXISTS {table_name}"
            elif db_type == DatabaseType.REDIS:
                # Redis沒有表的概念，刪除相關的所有鍵
                ddl = f"Redis keys deletion for pattern: {table_name}:*"
                keys = conn.keys(f"{table_name}:*")
                if keys:
                    conn.delete(*keys)
            else:
                raise ValueError(f"Unsupported database type: {db_type}")
            
            # 執行DROP語句（Redis除外）
            if db_type != DatabaseType.REDIS:
                cursor.execute(ddl)
                conn.commit()
            
            # 更新操作日志
            self._log_operation(
                table_name, db_type, db_name, "DROP", 
                {"kwargs": kwargs}, ddl, "success"
            )
            
            # 從監控表中刪除相關記錄
            table_log = self.monitor_session.query(TableCreationLog).filter_by(
                table_name=table_name, 
                database_type=db_type.value,
                database_name=db_name
            ).first()
            
            if table_log:
                # 刪除相關的列定義記錄（由于外鍵約束，會級聯刪除）
                self.monitor_session.delete(table_log)
                self.monitor_session.commit()
            
            return True
            
        except Exception as e:
            error_msg = str(e)
            logger.error(f"Failed to drop table {table_name}: {error_msg}")
            
            self._log_operation(
                table_name, db_type, db_name, "DROP", 
                {"kwargs": kwargs}, "", "failed", error_msg
            )
            
            return False
    
    def validate_table_structure(self, db_type: DatabaseType, db_name: str, 
                               table_name: str, expected_columns: List[Dict], **kwargs) -> Dict:
        """驗證表結構是否符合預期"""
        validation_details = {
            "expected_columns": expected_columns,
            "actual_columns": [],
            "matches": False,
            "issues": []
        }
        
        try:
            # 獲取實際表結構
            actual_structure = self.get_table_info(db_type, db_name, table_name, **kwargs)
            
            if not actual_structure:
                validation_details["issues"].append("Table does not exist or cannot be accessed")
                validation_status = "fail"
            else:
                validation_details["actual_columns"] = actual_structure.get("columns", [])
                
                # 驗證列匹配
                expected_cols = {col["name"]: col for col in expected_columns}
                actual_cols = {col["name"]: col for col in actual_structure.get("columns", [])}
                
                # 檢查缺失的列
                for col_name in expected_cols:
                    if col_name not in actual_cols:
                        validation_details["issues"].append(f"Missing column: {col_name}")
                
                # 檢查多余的列
                for col_name in actual_cols:
                    if col_name not in expected_cols:
                        validation_details["issues"].append(f"Extra column: {col_name}")
                
                # 檢查列屬性
                for col_name in expected_cols:
                    if col_name in actual_cols:
                        expected = expected_cols[col_name]
                        actual = actual_cols[col_name]
                        
                        # 檢查數據類型
                        if expected.get("type") and actual.get("type"):
                            if expected["type"].lower() != actual["type"].lower():
                                validation_details["issues"].append(
                                    f"Column {col_name} type mismatch: expected {expected['type']}, got {actual['type']}"
                                )
                        
                        # 檢查是否允許為空
                        if "nullable" in expected and "nullable" in actual:
                            if expected["nullable"] != actual["nullable"]:
                                validation_details["issues"].append(
                                    f"Column {col_name} nullable mismatch: expected {expected['nullable']}, got {actual['nullable']}"
                                )
                
                # 確定驗證狀態
                validation_details["matches"] = len(validation_details["issues"]) == 0
                validation_status = "pass" if validation_details["matches"] else "fail"
            
        except Exception as e:
            error_msg = str(e)
            logger.error(f"Failed to validate table {table_name}: {error_msg}")
            validation_details["issues"].append(f"Validation error: {error_msg}")
            validation_status = "fail"
        
        # 記錄驗證結果
        validation_log = TableValidationLog(
            table_name=table_name,
            database_type=db_type.value,
            database_name=db_name,
            validation_status=validation_status,
            validation_details=validation_details,
            issues_found="; ".join(validation_details["issues"]) if validation_details["issues"] else None
        )
        self.monitor_session.add(validation_log)
        self.monitor_session.commit()
        
        return validation_details
    
    def batch_create_tables(self, config_file: str):
        """通過配置文件批量創建表"""
        try:
            with open(config_file, 'r') as f:
                config = yaml.safe_load(f)
            
            results = {}
            for table_config in config.get('tables', []):
                db_type = DatabaseType[table_config['database_type'].upper()]
                db_name = table_config['database_name']
                table_name = table_config['table_name']
                columns = table_config['columns']
                kwargs = table_config.get('kwargs', {})
                
                result = self.create_table(db_type, db_name, table_name, columns, **kwargs)
                results[table_name] = result
            
            return results
            
        except Exception as e:
            logger.error(f"Failed to batch create tables: {str(e)}")
            return {"error": str(e)}
    
    def _generate_alter_ddl(self, db_type: DatabaseType, table_name: str, alterations: List[Dict]) -> str:
        """生成ALTER TABLE語句"""
        ddl_parts = []
        
        for alteration in alterations:
            operation = alteration.get('operation')  # ADD, DROP, MODIFY, RENAME
            
            if operation == 'ADD':
                col_def = self._generate_column_definition(alteration)
                ddl_parts.append(f"ADD COLUMN {col_def}")
            elif operation == 'DROP':
                ddl_parts.append(f"DROP COLUMN {alteration['column_name']}")
            elif operation == 'MODIFY':
                col_def = self._generate_column_definition(alteration)
                ddl_parts.append(f"MODIFY COLUMN {col_def}")
            elif operation == 'RENAME':
                ddl_parts.append(f"RENAME COLUMN {alteration['old_name']} TO {alteration['new_name']}")
        
        return f"ALTER TABLE {table_name} {', '.join(ddl_parts)}"
    
    def _generate_column_definition(self, col_def: Dict) -> str:
        """生成列定義字符串"""
        definition = f"{col_def['name']} {col_def['type']}"
        
        # 處理長度/精度
        if col_def.get('length') and col_def['type'].lower() in ['varchar', 'char']:
            definition += f"({col_def['length']})"
        elif col_def.get('precision') and col_def['type'].lower() in ['numeric', 'decimal']:
            if col_def.get('scale'):
                definition += f"({col_def['precision']}, {col_def['scale']})"
            else:
                definition += f"({col_def['precision']})"
        
        # 處理約束
        if not col_def.get('nullable', True):
            definition += " NOT NULL"
            
        if col_def.get('default') is not None:
            definition += f" DEFAULT {col_def['default']}"
            
        if col_def.get('comment'):
            definition += f" COMMENT '{col_def['comment']}'"
        
        return definition
    
    # 其他方法 (_generate_tdengine_ddl, _generate_postgresql_ddl, _generate_mysql_ddl, 
    # _initialize_redis_structure, get_table_info, close_all_connections) 保持不變
    # 但需要更新以使用新的列名 (col_length, col_precision, col_scale)

# 使用示例
if __name__ == "__main__":
    manager = DatabaseTableManager()
    
    # 批量創建表示例
    results = manager.batch_create_tables("tables_config.yaml")
    print("Batch create results:", results)
    
    # 修改表示例
    alterations = [
        {
            "operation": "ADD",
            "name": "new_column",
            "type": "VARCHAR",
            "length": 100,
            "nullable": True,
            "comment": "New column added by alter operation"
        }
    ]
    
    success = manager.alter_table(
        DatabaseType.MYSQL,
        'test_db',
        'test_table',
        alterations,
        host='localhost',
        user='root',
        password='password'
    )
    
    print(f"Alter table {'成功' if success else '失敗'}")
    
    # 刪除表示例
    success = manager.drop_table(
        DatabaseType.MYSQL,
        'test_db',
        'test_table',
        host='localhost',
        user='root',
        password='password'
    )
    
    print(f"Drop table {'成功' if success else '失敗'}")
    
    manager.close_all_connections()