#!/usr/bin/env php
<?php
/**
 * MySQL 테이블을 Elasticsearch에 업로드하는 스크립트
 *
 * 사용법:
 * ./ptyElasticUploadFromMysql --table=테이블명 --index=인덱스명 [옵션]
 *
 * 설정 파일:
 * - MySQL: ~/.ptyMysqlConfig.ini
 * - Elasticsearch: ~/.ptyElasticConfig.ini
 *
 * 예시:
 * ./ptyElasticUploadFromMysql --table=new_law_items --index=law_items --batch=500 --recreate
 * ./ptyElasticUploadFromMysql --table=users --index=users --mysql-section=production --elastic-section=production
 */

namespace platyFramework;

require_once __DIR__ . "/ptyLibrary_PHP/cli/ptyCliLog.php";
require_once __DIR__ . "/ptyLibrary_PHP/elastic/Elastic.php";

/**
 * INI 설정 파일 로더
 */
class ConfigLoader
{
    private static function getConfigPath($filename)
    {
        return $_SERVER['HOME'] . '/' . $filename;
    }

    /**
     * MySQL 설정 로드
     */
    public static function loadMysqlConfig($section = 'default')
    {
        $configPath = self::getConfigPath('.ptyMysqlConfig.ini');

        if (!file_exists($configPath)) {
            throw new \Exception("MySQL 설정 파일을 찾을 수 없습니다: {$configPath}");
        }

        $config = parse_ini_file($configPath, true);

        if ($config === false) {
            throw new \Exception("MySQL 설정 파일을 파싱할 수 없습니다: {$configPath}");
        }

        if (!isset($config[$section])) {
            throw new \Exception("MySQL 설정에서 [{$section}] 섹션을 찾을 수 없습니다. 사용 가능한 섹션: " . implode(', ', array_keys($config)));
        }

        $sectionConfig = $config[$section];

        // 필수 필드 검증
        $required = ['host', 'username', 'password', 'database'];
        foreach ($required as $field) {
            if (!isset($sectionConfig[$field]) || empty($sectionConfig[$field])) {
                throw new \Exception("MySQL 설정 [{$section}] 섹션에 필수 필드 '{$field}'가 없습니다.");
            }
        }

        return [
            'host' => $sectionConfig['host'],
            'username' => $sectionConfig['username'],
            'password' => trim($sectionConfig['password'], '"\''),
            'database' => $sectionConfig['database'],
            'charset' => $sectionConfig['charset'] ?? 'utf8mb4',
        ];
    }

    /**
     * Elasticsearch 설정 로드
     */
    public static function loadElasticConfig($section = 'default')
    {
        $configPath = self::getConfigPath('.ptyElasticConfig.ini');

        if (!file_exists($configPath)) {
            throw new \Exception("Elasticsearch 설정 파일을 찾을 수 없습니다: {$configPath}");
        }

        $config = parse_ini_file($configPath, true);

        if ($config === false) {
            throw new \Exception("Elasticsearch 설정 파일을 파싱할 수 없습니다: {$configPath}");
        }

        if (!isset($config[$section])) {
            throw new \Exception("Elasticsearch 설정에서 [{$section}] 섹션을 찾을 수 없습니다. 사용 가능한 섹션: " . implode(', ', array_keys($config)));
        }

        $sectionConfig = $config[$section];

        // 필수 필드 검증
        if (!isset($sectionConfig['host']) || empty($sectionConfig['host'])) {
            throw new \Exception("Elasticsearch 설정 [{$section}] 섹션에 필수 필드 'host'가 없습니다.");
        }

        return [
            'host' => $sectionConfig['host'],
            'apiKey' => isset($sectionConfig['apiKey']) ? trim($sectionConfig['apiKey'], '"\'') : null,
            'user' => $sectionConfig['user'] ?? null,
            'password' => isset($sectionConfig['password']) ? trim($sectionConfig['password'], '"\'') : null,
        ];
    }
}

error_reporting(E_ALL & ~E_DEPRECATED);

// 메모리 제한 증가 (대용량 데이터 처리를 위해)
ini_set('memory_limit', '2048M');

/**
 * MySQL to Elasticsearch 업로더 클래스
 */
class MySQLToElastic
{
    private $pdo;
    private $elastic;
    private $log;
    private $tableName;
    private $indexName;
    private $batchSize;
    private $primaryKey;
    private $mysqlConfig;
    private $elasticConfig;
    private $sampleDocIds = [];

    public function __construct($tableName, $indexName, $batchSize = 1000, $mysqlConfig = null, $elasticConfig = null)
    {
        $this->tableName = $tableName;
        $this->indexName = $indexName;
        $this->batchSize = $batchSize;
        $this->primaryKey = 'id'; // 기본값, 필요시 변경 가능
        $this->mysqlConfig = $mysqlConfig;
        $this->elasticConfig = $elasticConfig;

        $this->log = new ptyCliLog(prefix: "MySQL→ES", color: ptyCliLog::COLOR_CYAN);

        // MySQL 연결
        $this->initMySQLConnection();

        // Elasticsearch 연결
        $this->initElasticConnection();

        $this->log->info("초기화 완료");
        $this->log->info("테이블: {$this->tableName}");
        $this->log->info("인덱스: {$this->indexName}");
        $this->log->info("배치 크기: {$this->batchSize}");
        $this->log->info("_id 생성 규칙: serviceName, serviceId 컬럼 있으면 {serviceName}_{serviceId}_{primaryKey}, 없으면 {primaryKey}");
    }

    /**
     * MySQL 연결 초기화
     */
    private function initMySQLConnection()
    {
        try {
            $dsn = "mysql:host={$this->mysqlConfig['host']};dbname={$this->mysqlConfig['database']};charset={$this->mysqlConfig['charset']}";
            $this->pdo = new \PDO(
                $dsn,
                $this->mysqlConfig['username'],
                $this->mysqlConfig['password'],
                [
                    \PDO::ATTR_ERRMODE => \PDO::ERRMODE_EXCEPTION,
                    \PDO::ATTR_DEFAULT_FETCH_MODE => \PDO::FETCH_ASSOC,
                    \PDO::ATTR_EMULATE_PREPARES => false,
                    \PDO::MYSQL_ATTR_USE_BUFFERED_QUERY => false, // 메모리 절약을 위해 버퍼링 비활성화
                ]
            );
            $this->log->success("MySQL 연결 성공: {$this->mysqlConfig['host']}/{$this->mysqlConfig['database']}");
        } catch(\PDOException $e) {
            $this->log->error("MySQL 연결 실패: " . $e->getMessage());
            exit(1);
        }
    }

    /**
     * Elasticsearch 연결 초기화
     */
    private function initElasticConnection()
    {
        $this->elastic = new Elastic(
            $this->elasticConfig['apiKey'],
            $this->elasticConfig['host'],
            null,
            $this->elasticConfig['user'],
            $this->elasticConfig['password']
        );
        $authMethod = !empty($this->elasticConfig['apiKey']) ? 'apiKey' : 'user/password';
        $this->log->success("Elasticsearch 연결 성공: {$this->elasticConfig['host']} ({$authMethod})");
    }

    /**
     * 테이블 구조를 읽어서 Elasticsearch 매핑 생성
     */
    public function getTableStructure()
    {
        $this->log->info("테이블 구조 조회 중...");

        $stmt = $this->pdo->query("DESCRIBE `{$this->tableName}`");
        $columns = $stmt->fetchAll();

        $mapping = [];

        foreach ($columns as $column) {
            $field = $column['Field'];
            $type = $column['Type'];
            $key = $column['Key'];

            // Primary Key 저장
            if ($key === 'PRI') {
                $this->primaryKey = $field;
            }

            // MySQL 타입을 Elasticsearch 타입으로 변환
            $esType = $this->convertMySQLTypeToElastic($type);

            $mapping[$field] = ['type' => $esType];

            // text 타입인 경우 keyword 필드도 추가 (정렬, 집계용)
            if ($esType === 'text') {
                $mapping[$field]['fields'] = [
                    'keyword' => [
                        'type' => 'keyword',
                        'ignore_above' => 256
                    ]
                ];
            }
        }

        $this->log->success("테이블 구조 조회 완료 (컬럼 수: " . count($columns) . ")");

        return $mapping;
    }

    /**
     * MySQL 타입을 Elasticsearch 타입으로 변환
     */
    private function convertMySQLTypeToElastic($mysqlType)
    {
        $mysqlType = strtolower($mysqlType);

        // 정수형
        if (preg_match('/^(tinyint|smallint|mediumint|int|bigint)/', $mysqlType)) {
            if (preg_match('/bigint/', $mysqlType)) {
                return 'long';
            }
            return 'integer';
        }

        // 실수형
        if (preg_match('/^(float|double|decimal)/', $mysqlType)) {
            return 'float';
        }

        // 날짜/시간
        if (preg_match('/^(datetime|timestamp)/', $mysqlType)) {
            return 'date';
        }

        if (preg_match('/^date/', $mysqlType)) {
            return 'date';
        }

        // Boolean
        if (preg_match('/^tinyint\(1\)/', $mysqlType)) {
            return 'boolean';
        }

        // TEXT, LONGTEXT 등
        if (preg_match('/text/', $mysqlType)) {
            return 'text';
        }

        // VARCHAR, CHAR 등 - 길이에 따라 keyword 또는 text
        if (preg_match('/^(varchar|char)\((\d+)\)/', $mysqlType, $matches)) {
            $length = (int)$matches[2];
            // 짧은 문자열은 keyword, 긴 문자열은 text
            return $length <= 255 ? 'keyword' : 'text';
        }

        // ENUM
        if (preg_match('/^enum/', $mysqlType)) {
            return 'keyword';
        }

        // JSON
        if (preg_match('/^json/', $mysqlType)) {
            return 'object';
        }

        // 기본값은 keyword
        return 'keyword';
    }

    /**
     * Elasticsearch 인덱스 생성
     */
    public function createIndex($recreate = false)
    {
        $this->log->info("인덱스 생성 중: {$this->indexName}");

        // 인덱스가 이미 존재하는지 확인
        $exists = $this->elastic->indexExists($this->indexName);

        if ($exists) {
            if ($recreate) {
                $this->log->warning("기존 인덱스 삭제 중...");
                $this->elastic->deleteIndex($this->indexName);
                $this->log->success("기존 인덱스 삭제 완료");
            } else {
                $this->log->warning("인덱스가 이미 존재합니다. 매핑 업데이트를 시도합니다.");
                return;
            }
        }

        // 테이블 구조 조회
        $mapping = $this->getTableStructure();

        // 인덱스 생성 요청 데이터
        $indexData = [
            'settings' => [
                'number_of_shards' => 1,
                'number_of_replicas' => 1,
                'analysis' => [
                    'analyzer' => [
                        'korean' => [
                            'type' => 'custom',
                            'tokenizer' => 'standard',
                            'filter' => ['lowercase']
                        ]
                    ]
                ]
            ],
            'mappings' => [
                'properties' => $mapping
            ]
        ];

        try {
            $response = $this->elastic->put($this->indexName, $indexData);
            $this->log->success("인덱스 생성 완료");
        } catch (\Exception $e) {
            $this->log->error("인덱스 생성 실패: " . $e->getMessage());
            throw $e;
        }
    }

    /**
     * 테이블의 전체 레코드 수 조회
     */
    public function getTotalCount()
    {
        $stmt = $this->pdo->query("SELECT COUNT(*) as cnt FROM `{$this->tableName}`");
        $result = $stmt->fetch();
        return (int)$result['cnt'];
    }

    /**
     * 데이터 업로드
     */
    public function uploadData($whereClause = null)
    {
        $this->log->info("=== 데이터 업로드 시작 ===");

        // 전체 레코드 수 조회
        $totalCount = $this->getTotalCount();
        $this->log->info("전체 레코드 수: " . number_format($totalCount));

        if ($totalCount === 0) {
            $this->log->warning("업로드할 데이터가 없습니다.");
            return;
        }

        $offset = 0;
        $successCount = 0;
        $errorCount = 0;

        $startTime = microtime(true);

        while ($offset < $totalCount) {
            $sql = "SELECT * FROM `{$this->tableName}`";

            if ($whereClause) {
                $sql .= " WHERE {$whereClause}";
            }

            $sql .= " LIMIT {$this->batchSize} OFFSET {$offset}";

            $stmt = $this->pdo->query($sql);
            $rows = $stmt->fetchAll();

            if (empty($rows)) {
                break;
            }

            $batchCount = count($rows);
            $progress = min(100, round(($offset + $batchCount) / $totalCount * 100, 2));

            $this->log->info("배치 처리 중... [{$offset}-" . ($offset + $batchCount) . "] / {$totalCount} ({$progress}%)");

            // Bulk API를 위한 데이터 준비
            try {
                $batchDocIds = $this->bulkInsert($rows);
                $successCount += $batchCount;

                // 배치 성공 시 첫/마지막 URL 출력
                if (!empty($batchDocIds)) {
                    $baseUrl = rtrim($this->elasticConfig['host'], '/');
                    $firstId = reset($batchDocIds);
                    $lastId = end($batchDocIds);
                    $this->log->success("  처음: {$baseUrl}/{$this->indexName}/_doc/{$firstId}");
                    $this->log->success("  마지막: {$baseUrl}/{$this->indexName}/_doc/{$lastId}");
                }
            } catch (\Exception $e) {
                $this->log->error("배치 업로드 실패: " . $e->getMessage());
                $errorCount += $batchCount;
            }

            $offset += $batchCount;

            // 메모리 해제 및 가비지 컬렉션
            unset($rows);
            unset($stmt);
            gc_collect_cycles();

            // 진행 상황 요약 (메모리 사용량 포함)
            $elapsed = round(microtime(true) - $startTime, 2);
            $speed = $elapsed > 0 ? round($successCount / $elapsed, 2) : 0;
            $memoryUsage = round(memory_get_usage() / 1024 / 1024, 2);
            $memoryPeak = round(memory_get_peak_usage() / 1024 / 1024, 2);
            $this->log->info("진행: {$successCount}건 성공, {$errorCount}건 실패, 속도: {$speed}건/초, 메모리: {$memoryUsage}MB / Peak: {$memoryPeak}MB");
        }

        $totalTime = round(microtime(true) - $startTime, 2);

        $this->log->success("=== 데이터 업로드 완료 ===");
        $this->log->success("총 {$successCount}건 성공, {$errorCount}건 실패");
        $this->log->success("소요 시간: {$totalTime}초");

        // 예시 URL 출력
        if (!empty($this->sampleDocIds)) {
            $this->log->info("");
            $this->log->info("예시 URL:");
            $baseUrl = rtrim($this->elasticConfig['host'], '/');
            foreach ($this->sampleDocIds as $docId) {
                $this->log->info("  {$baseUrl}/{$this->indexName}/_doc/{$docId}");
            }
        }
    }

    /**
     * Bulk API를 사용한 배치 삽입
     */
    private function bulkInsert($rows)
    {
        $bulkData = '';
        $batchDocIds = []; // 이 배치의 문서 ID들

        foreach ($rows as $row) {
            // 문서 ID 생성
            // serviceName과 serviceId 필드가 있으면 조합하여 ID 생성
            $docId = null;

            if (isset($row['serviceName']) && isset($row['serviceId'])) {
                // serviceName_serviceId_primaryKey 형태로 생성
                $serviceName = $row['serviceName'];
                $serviceId = $row['serviceId'];
                $primaryValue = $row[$this->primaryKey] ?? '';
                $docId = "{$serviceName}_{$serviceId}_{$primaryValue}";
            } else {
                // 기본: primary key 사용
                $docId = $row[$this->primaryKey] ?? null;
            }

            // 배치 문서 ID 수집
            if ($docId !== null) {
                $batchDocIds[] = $docId;
            }

            // 샘플 문서 ID 수집 (최대 5개)
            if (count($this->sampleDocIds) < 5 && $docId !== null) {
                $this->sampleDocIds[] = $docId;
            }

            // 날짜 형식 변환 (MySQL datetime -> Elasticsearch date)
            foreach ($row as $key => $value) {
                if ($value !== null && preg_match('/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$/', $value)) {
                    // MySQL datetime을 ISO 8601 형식으로 변환
                    $row[$key] = str_replace(' ', 'T', $value);
                }
            }

            // index 액션
            $action = [
                'index' => [
                    '_index' => $this->indexName,
                    '_id' => $docId
                ]
            ];

            $bulkData .= json_encode($action, JSON_UNESCAPED_UNICODE) . "\n";
            $bulkData .= json_encode($row, JSON_UNESCAPED_UNICODE) . "\n";

            // 메모리 절약을 위해 사용한 변수 즉시 해제
            unset($action, $row);
        }

        // Bulk API 호출
        $ch = curl_init();

        $url = rtrim($this->elasticConfig['host'], '/') . '/_bulk';

        // 인증 헤더 설정
        $headers = [
            'Content-Type: application/x-ndjson',
        ];

        if (!empty($this->elasticConfig['apiKey'])) {
            $headers[] = 'Authorization: ApiKey ' . $this->elasticConfig['apiKey'];
        } elseif (!empty($this->elasticConfig['user']) && !empty($this->elasticConfig['password'])) {
            $headers[] = 'Authorization: Basic ' . base64_encode($this->elasticConfig['user'] . ':' . $this->elasticConfig['password']);
        }

        curl_setopt_array($ch, [
            CURLOPT_URL => $url,
            CURLOPT_RETURNTRANSFER => true,
            CURLOPT_CUSTOMREQUEST => 'POST',
            CURLOPT_HTTPHEADER => $headers,
            CURLOPT_POSTFIELDS => $bulkData,
            CURLOPT_SSL_VERIFYPEER => false,
            CURLOPT_SSL_VERIFYHOST => false
        ]);

        $response = curl_exec($ch);
        $httpCode = curl_getinfo($ch, CURLINFO_HTTP_CODE);
        curl_close($ch);

        if ($httpCode >= 400) {
            $decodedResponse = json_decode($response, true);

            // 상세 에러 메시지 생성
            $errorMessage = "HTTP {$httpCode}";
            if ($httpCode == 413) {
                $errorMessage = "HTTP 413 Request Entity Too Large - 배치 크기가 너무 큽니다. --batch 값을 줄여보세요.";
            } elseif (isset($decodedResponse['error']['reason'])) {
                $errorMessage = "HTTP {$httpCode}: " . $decodedResponse['error']['reason'];
            } elseif (isset($decodedResponse['error']['type'])) {
                $errorMessage = "HTTP {$httpCode}: " . $decodedResponse['error']['type'];
            }

            // 실패한 배치의 처음과 끝 URL 출력
            $baseUrl = rtrim($this->elasticConfig['host'], '/');
            if (!empty($batchDocIds)) {
                $firstId = reset($batchDocIds);
                $lastId = end($batchDocIds);
                $this->log->error("  실패 범위:");
                $this->log->error("    처음: {$baseUrl}/{$this->indexName}/_doc/{$firstId}");
                $this->log->error("    마지막: {$baseUrl}/{$this->indexName}/_doc/{$lastId}");
            }

            throw new \Exception($errorMessage);
        }

        $result = json_decode($response, true);

        // 메모리 해제
        unset($response, $bulkData);

        // 오류 체크
        if (isset($result['errors']) && $result['errors'] === true) {
            $errorItems = array_filter($result['items'], function($item) {
                return isset($item['index']['error']);
            });

            if (!empty($errorItems)) {
                $this->log->warning("일부 문서 업로드 실패: " . count($errorItems) . "건");
            }

            unset($errorItems);
        }

        unset($result);

        return $batchDocIds;
    }

    /**
     * Primary Key 설정
     */
    public function setPrimaryKey($key)
    {
        $this->primaryKey = $key;
        $this->log->info("Primary Key 설정: {$key}");
    }
}

// ==========================================
// 메인 실행 코드
// ==========================================

$log = new ptyCliLog(prefix: "메인", color: ptyCliLog::COLOR_GREEN);

// 커맨드 라인 인자 파싱 (위치 인자와 옵션 분리)
$positionalArgs = [];
$optionArgs = [];

foreach ($argv as $i => $arg) {
    if ($i === 0) continue; // 스크립트 이름 제외
    if (strpos($arg, '--') === 0) {
        $optionArgs[] = $arg;
    } else {
        $positionalArgs[] = $arg;
    }
}

// 옵션 파싱
$options = [];
foreach ($optionArgs as $arg) {
    if (preg_match('/^--([^=]+)=(.*)$/', $arg, $matches)) {
        $options[$matches[1]] = $matches[2];
    } elseif (preg_match('/^--([^=]+)$/', $arg, $matches)) {
        $options[$matches[1]] = true;
    }
}

$tableName = $positionalArgs[0] ?? null;
$indexName = $positionalArgs[1] ?? null;

if (!$tableName || !$indexName || isset($options['help'])) {
    echo "\n";
    echo "사용법: ./ptyElasticUploadFromMysql <테이블명> <인덱스명> [옵션]\n";
    echo "\n";
    echo "설정 파일:\n";
    echo "  MySQL:         ~/.ptyMysqlConfig.ini\n";
    echo "  Elasticsearch: ~/.ptyElasticConfig.ini\n";
    echo "\n";
    echo "필수 인자:\n";
    echo "  테이블명                   MySQL 테이블 이름 (첫번째 인자)\n";
    echo "  인덱스명                   Elasticsearch 인덱스 이름 (두번째 인자)\n";
    echo "\n";
    echo "선택 인자:\n";
    echo "  --mysql=섹션명            MySQL INI 섹션 (기본값: default)\n";
    echo "  --elastic=섹션명          Elasticsearch INI 섹션 (기본값: default)\n";
    echo "  --batch=N                 배치 크기 (기본값: 100)\n";
    echo "  --recreate                기존 인덱스 삭제 후 재생성\n";
    echo "  --primary=필드명           Primary Key 필드명 (기본값: id)\n";
    echo "  --where='조건'             WHERE 절 추가 (예: --where='enabled=1')\n";
    echo "  --help                    도움말 출력\n";
    echo "\n";
    echo "INI 파일 형식:\n";
    echo "  ~/.ptyMysqlConfig.ini:\n";
    echo "    [default]\n";
    echo "    host=localhost\n";
    echo "    username=root\n";
    echo "    password=\"your_password\"\n";
    echo "    database=your_db\n";
    echo "    charset=utf8mb4\n";
    echo "\n";
    echo "  ~/.ptyElasticConfig.ini:\n";
    echo "    [default]\n";
    echo "    host=https://localhost:9200\n";
    echo "    apiKey=your_api_key          # apiKey 또는 user/password 중 하나 사용\n";
    echo "    # user=elastic\n";
    echo "    # password=\"your_password\"\n";
    echo "\n";
    echo "예시:\n";
    echo "  ./ptyElasticUploadFromMysql new_law_items law_items\n";
    echo "  ./ptyElasticUploadFromMysql users users --mysql=production --elastic=production\n";
    echo "  ./ptyElasticUploadFromMysql new_law_items law_items --batch=500 --recreate\n";
    echo "\n";
    exit(isset($options['help']) ? 0 : 1);
}
$batchSize = isset($options['batch']) ? (int)$options['batch'] : 100;
$recreate = isset($options['recreate']);
$primaryKey = $options['primary'] ?? 'id';
$whereClause = $options['where'] ?? null;
$mysqlSection = $options['mysql'] ?? 'default';
$elasticSection = $options['elastic'] ?? 'default';

$log->info("=== MySQL to Elasticsearch 업로드 시작 ===");

// 설정 로드
try {
    $log->info("MySQL 설정 로드 중... (섹션: {$mysqlSection})");
    $mysqlConfig = ConfigLoader::loadMysqlConfig($mysqlSection);
    $log->success("MySQL 설정 로드 완료: {$mysqlConfig['host']}/{$mysqlConfig['database']}");

    $log->info("Elasticsearch 설정 로드 중... (섹션: {$elasticSection})");
    $elasticConfig = ConfigLoader::loadElasticConfig($elasticSection);
    $log->success("Elasticsearch 설정 로드 완료: {$elasticConfig['host']}");
} catch (\Exception $e) {
    $log->error($e->getMessage());
    exit(1);
}

$log->info("MySQL 테이블: {$tableName}");
$log->info("Elasticsearch 인덱스: {$indexName}");

// 업로더 생성
$uploader = new MySQLToElastic($tableName, $indexName, $batchSize, $mysqlConfig, $elasticConfig);

// Primary Key 설정
if ($primaryKey !== 'id') {
    $uploader->setPrimaryKey($primaryKey);
}

// 인덱스 생성
$uploader->createIndex($recreate);

// 데이터 업로드
$uploader->uploadData($whereClause);

$log->success("=== 모든 작업 완료 ===");

?>
