Huawei Cloud Ges Graph

key,value
# GES服务配置
graph_ip,100.95.xxx.xxx
graph_port,80
project_id,your_project_id
graph_name,your_graph_name
graph_id,your_graph_id

# IAM认证配置
iam_url,https://iam.cn-north-4.myhuaweicloud.com/v3/auth/tokens
region,cn-north-7
# AKSK/用户名密码认证 二选一
# AKSK认证
access_key,your_access_key
secret_key,your_secret_key
# 用户名密码认证
username,your_username
password,your_password
domain_name,your_domain_name

#!/usr/bin/env node
/**
 * GES Graph Skill - Node.js Version
 * 华为云图引擎持久化版 (GES) SDK for Node.js
 * 支持Cypher查询、GQL查询、节点/边操作、导入导出等
 */

const https = require('https');
const http = require('http');
const fs = require('fs');
const path = require('path');
const crypto = require('crypto');
const { URL } = require('url');

// ==================== 环境配置 ====================

class EnvConfig {
    static ENV_DIR = path.join(__dirname, '..', '.env');

    static ENV_VAR_MAPPING = {
        'GES_GRAPH_IP': 'graph_ip',
        'GES_GRAPH_PORT': 'graph_port',
        'GES_PROJECT_ID': 'project_id',
        'GES_GRAPH_NAME': 'graph_name',
        'GES_IAM_URL': 'iam_url',
        'GES_USERNAME': 'username',
        'GES_PASSWORD': 'password',
        'GES_DOMAIN_NAME': 'domain_name',
        'GES_REGION': 'region',
        'HUAWEI_CLOUD_AK': 'access_key',
        'HUAWEI_CLOUD_SK': 'secret_key',
    };

    static loadGesEnv() {
        const config = {};
        const envFile = path.join(EnvConfig.ENV_DIR, 'ges_env.csv');
        const fileConfig = EnvConfig._loadCsv(envFile);

        for (const [envVar, configKey] of Object.entries(EnvConfig.ENV_VAR_MAPPING)) {
            const envValue = process.env[envVar];
            if (envValue) {
                config[configKey] = envValue;
            } else if (fileConfig[configKey]) {
                config[configKey] = fileConfig[configKey];
            }
        }

        for (const [key, value] of Object.entries(fileConfig)) {
            if (!config[key]) {
                config[key] = value;
            }
        }

        const requiredFields = ['graph_ip', 'project_id', 'graph_name', 'iam_url', 'username', 'password', 'domain_name', 'region'];
        const missingFields = requiredFields.filter(field => !config[field]);

        if (missingFields.length > 0) {
            console.error('==============================================');
            console.error('⚠️  配置缺失或不完整');
            console.error('==============================================');
            console.error('\n请通过以下方式配置（环境变量优先）：');
            console.error('  - 环境变量: GES_GRAPH_IP, GES_PROJECT_ID, GES_GRAPH_NAME, GES_IAM_URL, GES_REGION');
            console.error('  - AKSK: HUAWEI_CLOUD_AK, HUAWEI_CLOUD_SK');
            console.error('  - 或用户名密码: GES_USERNAME, GES_PASSWORD, GES_DOMAIN_NAME');
            console.error('  - 或配置文件: .env/ges_env.csv');
            console.error('\n缺失的必需配置项:');
            missingFields.forEach(field => console.error(`  - ${field}`));
            console.error('==============================================\n');
        }

        return config;
    }

    static _loadCsv(filepath) {
        const config = {};
        if (fs.existsSync(filepath)) {
            const content = fs.readFileSync(filepath, 'utf-8');
            const lines = content.trim().split('\n');
            for (let i = 1; i < lines.length; i++) {
                const parts = lines[i].split(',');
                if (parts.length >= 2) {
                    config[parts[0].trim()] = parts[1].trim();
                }
            }
        }
        return config;
    }
}

// ==================== Token管理 ====================

class TokenManager {
    constructor() {
        this.token = null;
        this.tokenExpiry = 0;
        this.config = EnvConfig.loadGesEnv();
    }

    async getToken() {
        const currentTime = Date.now();
        if (this.token && currentTime < this.tokenExpiry - 300000) { // 提前5分钟过期
            return this.token;
        }

        const envToken = process.env.GES_TOKEN;
        if (envToken) {
            this.token = envToken;
            this.tokenExpiry = Date.now() + 3600 * 23 * 1000;
            return this.token;
        }

        const accessKey = this.config.access_key;
        const secretKey = this.config.secret_key;

        if (accessKey && secretKey) {
            try {
                this.token = await this._fetchTokenByAksk();
                return this.token;
            } catch (e) {
                console.log(`AKSK方式获取Token失败，尝试密码方式: ${e.message}`);
            }
        }

        const username = this.config.username;
        const password = this.config.password;
        const domainName = this.config.domain_name;

        if (username && password) {
            try {
                this.token = await this._fetchTokenByPassword(username, password, domainName);
                return this.token;
            } catch (e) {
                throw new Error(`密码方式获取Token也失败了: ${e.message}`);
            }
        }

        throw new Error("无法获取Token，请配置AKSK或用户名密码");
    }

    async _fetchTokenByPassword(username, password, domainName) {
        const url = this.config.iam_url;
        const projectId = this.config.project_id;

        if (!url) {
            throw new Error("ges_env.csv中缺少iam_url配置");
        }

        let data;
        if (domainName) {
            data = JSON.stringify({
                auth: {
                    identity: {
                        methods: ['password'],
                        password: {
                            user: {
                                name: username,
                                password: password,
                                domain: { name: domainName }
                            }
                        }
                    },
                    scope: { project: { id: projectId } }
                }
            });
        } else {
            data = JSON.stringify({
                auth: {
                    identity: {
                        methods: ['password'],
                        password: {
                            user: {
                                name: username,
                                password: password
                            }
                        }
                    },
                    scope: { project: { id: projectId } }
                }
            });
        }

        const resp = await this._makeRequest(url, 'POST', data, {
            'Content-Type': 'application/json'
        });

        const token = resp.headers['x-subject-token'];
        if (!token) {
            throw new Error(`密码获取Token失败，未获取到token`);
        }

        this.tokenExpiry = Date.now() + 3600 * 23 * 1000;
        return token;
    }

    async _fetchTokenByAksk() {
        const accessKey = this.config.access_key;
        const secretKey = this.config.secret_key;
        const projectId = this.config.project_id;
        const url = this.config.iam_url;
        const region = this.config.region || 'cn-north-7';

        if (!url) {
            throw new Error("ges_env.csv中缺少iam_url配置");
        }

        const Service = 'iam';
        const ContentType = 'application/json;charset=UTF-8';

        const Body = JSON.stringify({
            auth: {
                identity: {
                    methods: ['hw_access_key'],
                    hw_access_key: {
                        access: { key: accessKey }
                    }
                },
                scope: { project: { id: projectId } }
            }
        });

        // 签名计算
        const BasicDateFormat = "%Y%m%dT%H%M%SZ";
        const ScopeDateFormat = "%Y%m%d";
        const Algorithm = "HWS-HMAC-SHA256";

        const sign = (key, msg) => {
            return crypto.createHmac('sha256', key).update(msg).digest();
        };

        const getSignatureKey = (key, dateStamp, region, service) => {
            const kDate = sign(Buffer.from('HWS' + key), dateStamp);
            const kRegion = sign(kDate, region);
            const kService = sign(kRegion, service);
            return sign(kService, 'hws_request');
        };

        const now = new Date();
        const hwsDate = now.toISOString().replace(/[-:]/g, '').replace(/\.\d{3}/, '');
        const dateStamp = now.toISOString().slice(0, 10).replace(/-/g, '');

        const canonicalUri = '/v3/auth/tokens/';
        const canonicalHeaders = `accept:${'application/json'}\ncontent-type:${ContentType}\nx-hws-date:${hwsDate}\n`;
        const signedHeaders = 'accept;content-type;x-hws-date';
        const payloadHash = crypto.createHash('sha256').update(Body).digest('hex');

        const canonicalRequest = [
            'POST', canonicalUri, '', canonicalHeaders, signedHeaders, payloadHash
        ].join('\n');

        const credentialScope = [dateStamp, region, Service, 'hws_request'].join('/');
        const stringToSign = [
            Algorithm, hwsDate, credentialScope,
            crypto.createHash('sha256').update(canonicalRequest).digest('hex')
        ].join('\n');

        const signingKey = getSignatureKey(secretKey, dateStamp, region, Service);
        const signature = crypto.createHmac('sha256', signingKey).update(stringToSign).digest('hex');

        const authorizationHeader = `${Algorithm} Credential=${accessKey}/${credentialScope}, SignedHeaders=${signedHeaders}, Signature=${signature}`;

        const headers = {
            'Accept': 'application/json',
            'Content-Type': ContentType,
            'X-Hws-Date': hwsDate,
            'X-Identity-Sign': authorizationHeader
        };

        const resp = await this._makeRequest(url, 'POST', Body, headers);

        const token = resp.headers['x-subject-token'];
        if (!token) {
            throw new Error(`AKSK获取Token失败，未获取到token`);
        }

        this.tokenExpiry = Date.now() + 3600 * 23 * 1000;
        return token;
    }

    _makeRequest(urlStr, method, data, headers) {
        return new Promise((resolve, reject) => {
            const url = new URL(urlStr);
            const isHttps = url.protocol === 'https:';
            const lib = isHttps ? https : http;

            const options = {
                hostname: url.hostname,
                port: url.port || (isHttps ? 443 : 80),
                path: url.pathname,
                method: method,
                headers: headers,
                timeout: 30000
            };

            const req = lib.request(options, (res) => {
                let body = '';
                res.on('data', chunk => body += chunk);
                res.on('end', () => {
                    resolve({
                        statusCode: res.statusCode,
                        headers: res.headers,
                        body: body
                    });
                });
            });

            req.on('error', reject);
            req.on('timeout', () => reject(new Error('Request timeout')));

            if (data) {
                req.write(data);
            }
            req.end();
        });
    }
}

// ==================== OBS客户端 ====================

/**
 * OBS客户端 - 使用虚拟主机方式访问
 *
 * 虚拟主机URL格式: https://bucketname.obs.region.domain/objectkey
 *
 * 签名算法: Authorization: OBS AccessKeyID:Signature
 * 其中: Signature = Base64(HMAC-SHA1(Your_SK, UTF8(StringToSign)))
 *
 * StringToSign格式:
 * - 无Content-Type: METHOD\n\n\nDate\n/bucket/key
 * - 有Content-Type: METHOD\n\nContent-Type\nDate\n/bucket/key
 */
class OBSClient {
    constructor(accessKey, secretKey, region = "cn-north-7") {
        this.accessKey = accessKey;
        this.secretKey = secretKey;
        this.region = region;
        // 使用虚拟主机方式: bucket.obs.endpoint
        this.endpoint = `obs.${region}.ulanqab.huawei.com`;
        this.available = Boolean(accessKey && secretKey);
    }

    isAvailable() {
        return this.available;
    }

    /**
     * 生成OBS签名
     * @param {string} stringToSign - 待签名字符串
     */
    _sign(stringToSign) {
        const crypto = require('crypto');
        const hmac = crypto.createHmac('sha1', this.secretKey);
        hmac.update(stringToSign, 'utf8');
        return hmac.digest('base64');
    }

    /**
     * 发起OBS请求
     * @param {string} method - HTTP方法
     * @param {string} bucket - 桶名
     * @param {string} objectKey - 对象键
     * @param {object} options - 选项
     */
    async _request(method, bucket, objectKey = '', options = {}) {
        const { query = {}, headers = {}, body = null } = options;

        // 构造查询字符串 (实际请求路径)
        const queryString = Object.entries(query)
            .map(([k, v]) => `${k}=${encodeURIComponent(v)}`)
            .join('&');
        const actualPath = objectKey ? `/${objectKey}` : '/';
        const fullPath = queryString ? `${actualPath}?${queryString}` : actualPath;

        // 生成签名 - OBS签名格式
        const date = new Date().toUTCString();
        const contentType = headers['Content-Type'] || '';

        let stringToSign;
        const signPath = objectKey ? `/${bucket}/${objectKey}` : `/${bucket}/`;

        if (contentType) {
            // 有Content-Type时的签名格式: METHOD\n\nContent-Type\nDate\n/path
            stringToSign = [
                method,
                '\n',
                '\n',
                contentType,
                '\n',
                date,
                '\n',
                signPath
            ].join('');
        } else {
            // 无Content-Type时的签名格式: METHOD\n\n\nDate\n/path (3个换行)
            stringToSign = [
                method,
                '\n',
                '\n',
                '\n',
                date,
                '\n',
                signPath
            ].join('');
        }

        const signature = this._sign(stringToSign);

        // 构建请求头
        const requestHeaders = {
            'Host': `${bucket}.${this.endpoint}`,
            'Date': date,
            'Authorization': `OBS ${this.accessKey}:${signature}`,
            ...headers
        };

        return new Promise((resolve, reject) => {
            const req = https.request({
                hostname: `${bucket}.${this.endpoint}`,
                port: 443,
                path: fullPath,
                method: method,
                headers: requestHeaders
            }, (res) => {
                const chunks = [];
                res.on('data', chunk => chunks.push(chunk));
                res.on('end', () => {
                    const body = Buffer.concat(chunks).toString();
                    if (res.statusCode >= 400) {
                        reject(new Error(`OBS请求失败 [${res.statusCode}]: ${body}`));
                        return;
                    }
                    try {
                        resolve({ statusCode: res.statusCode, body: JSON.parse(body) });
                    } catch {
                        resolve({ statusCode: res.statusCode, body: body });
                    }
                });
            });

            req.on('error', reject);
            if (body) req.write(body);
            req.end();
        });
    }

    /**
     * 列出桶中对象
     * @param {string} bucket - 桶名
     * @param {string} prefix - 前缀过滤
     */
    async listObjects(bucket, prefix = "") {
        if (!this.available) {
            throw new Error("OBS客户端不可用，请配置access_key和secret_key");
        }

        const result = await this._request('GET', bucket, '', {
            query: { 'list-type': '2', prefix: prefix }
        });

        // 解析XML响应
        const contents = [];
        const regex = /<Contents><Key>([^<]+)<\/Key><LastModified>([^<]+)<\/LastModified><ETag>([^<]+)<\/ETag><Size>(\d+)<\/Size>/g;
        let match;
        while ((match = regex.exec(result.body)) !== null) {
            contents.push({
                Key: match[1],
                LastModified: match[2],
                ETag: match[3],
                Size: parseInt(match[4])
            });
        }
        return contents;
    }

    /**
     * 上传文件到OBS
     * @param {string} localFile - 本地文件路径
     * @param {string} bucket - 桶名
     * @param {string} objectKey - OBS中的对象键
     */
    async uploadFile(localFile, bucket, objectKey) {
        if (!this.available) {
            throw new Error("OBS客户端不可用");
        }

        const fs = require('fs');
        if (!fs.existsSync(localFile)) {
            throw new Error(`文件不存在: ${localFile}`);
        }

        const fileContent = fs.readFileSync(localFile);

        await this._request('PUT', bucket, objectKey, {
            headers: {
                'Content-Type': 'application/octet-stream',
                'Content-Length': fileContent.length
            },
            body: fileContent
        });

        return true;
    }

    /**
     * 从OBS下载文件
     * @param {string} bucket - 桶名
     * @param {string} objectKey - OBS中的对象键
     * @param {string} localFile - 本地保存路径
     */
    async downloadFile(bucket, objectKey, localFile) {
        if (!this.available) {
            throw new Error("OBS客户端不可用");
        }

        const fs = require('fs');

        const result = await this._request('GET', bucket, objectKey);
        fs.writeFileSync(localFile, result.body);
        return true;
    }

    /**
     * 删除OBS中的对象
     * @param {string} bucket - 桶名
     * @param {string} objectKey - OBS中的对象键
     */
    async deleteObject(bucket, objectKey) {
        if (!this.available) {
            throw new Error("OBS客户端不可用");
        }

        await this._request('DELETE', bucket, objectKey);
        return true;
    }

    /**
     * 获取对象URL
     * @param {string} bucket - 桶名
     * @param {string} objectKey - OBS中的对象键
     */
    getObjectUrl(bucket, objectKey) {
        return `https://${bucket}.${this.endpoint}/${objectKey}`;
    }
}
// ==================== GES客户端 ====================

class GESClient {
    constructor() {
        this.tokenMgr = new TokenManager();
        this.config = EnvConfig.loadGesEnv();
        this.graphIp = this.config.graph_ip || '';
        this.graphPort = this.config.graph_port || '80';
        this.projectId = this.config.project_id || '';
        this.graphName = this.config.graph_name || '';
        this.baseUrl = `http://${this.graphIp}:${this.graphPort}/ges/v1.0/${this.projectId}/graphs/${this.graphName}`;

        // 初始化OBS客户端
        const accessKey = this.config.access_key || '';
        const secretKey = this.config.secret_key || '';
        const region = this.config.region || 'cn-north-7';
        if (accessKey && secretKey) {
            this.obsClient = new OBSClient(accessKey, secretKey, region);
        } else {
            this.obsClient = null;
        }
    }

    getObsClient() {
        return this.obsClient;
    }

    async _getHeaders() {
        return {
            'X-Auth-Token': await this.tokenMgr.getToken(),
            'Content-Type': 'application/json'
        };
    }

    async _request(method, path, data = null, params = null) {
        let url = `${this.baseUrl}${path}`;
        
        if (params) {
            const queryString = Object.entries(params)
                .map(([k, v]) => `${encodeURIComponent(k)}=${encodeURIComponent(v)}`)
                .join('&');
            url += (url.includes('?') ? '&' : '?') + queryString;
        }

        const headers = await this._getHeaders();
        
        // 使用原生http模块发送请求
        const urlObj = new URL(url);
        const isHttps = urlObj.protocol === 'https:';
        const lib = isHttps ? https : http;

        return new Promise((resolve, reject) => {
            const options = {
                hostname: urlObj.hostname,
                port: urlObj.port || (isHttps ? 443 : 80),
                path: urlObj.pathname + urlObj.search,
                method: method,
                headers: headers,
                timeout: 60000,
                rejectUnauthorized: false
            };

            const req = lib.request(options, (res) => {
                let body = '';
                res.on('data', chunk => body += chunk);
                res.on('end', () => {
                    if (res.statusCode >= 400) {
                        reject(new Error(`API请求失败 [${res.statusCode}]: ${body}`));
                        return;
                    }
                    try {
                        resolve(JSON.parse(body));
                    } catch {
                        resolve({ raw: body });
                    }
                });
            });

            req.on('error', reject);
            req.on('timeout', () => reject(new Error('Request timeout')));

            if (data) {
                req.write(JSON.stringify(data));
            }
            req.end();
        });
    }

    // ==================== Cypher查询相关API ====================

    async executeCypher(statement, parameters = null, executionMode = "sync", 
                       resultDataContents = "row", limit = 1000) {
        const data = {
            statements: [{
                statement: statement,
                parameters: parameters || {},
                executionMode: executionMode,
                resultDataContents: [resultDataContents],
                limit: limit
            }]
        };

        return this._request('POST', `/action?action_id=execute-cypher-query`, data);
    }

    async executeCypherAsync(statement, parameters = null, resultDataContents = "row") {
        const result = await this.executeCypher(statement, parameters, "async", resultDataContents);
        // jobId可能在顶层或results[0]中
        return result.jobId || (result.results && result.results[0] && result.results[0].jobId) || '';
    }

    async getJobStatus(jobId) {
        return this._request('GET', `/jobs/${jobId}/status`);
    }

    // ==================== GQL查询相关API ====================

    async executeGql(statement, parameters = null, executionMode = "sync",
                    resultDataContents = "row", limit = 1000, includeStats = true) {
        const data = {
            statements: [{
                statement: statement,
                parameters: parameters || {},
                executionMode: executionMode,
                resultDataContents: [resultDataContents],
                limit: limit,
                includeStats: includeStats
            }]
        };

        return this._request('POST', `/action?action_id=execute-gql-query`, data);
    }

    async executeGqlAsync(statement, parameters = null, resultDataContents = "row") {
        const result = await this.executeGql(statement, parameters, "async", resultDataContents);
        // jobId可能在顶层或results[0]中
        return result.jobId || (result.results && result.results[0] && result.results[0].jobId) || '';
    }

    // ==================== Schema相关API ====================

    async getSchema() {
        return this.executeCypher("call db.schema()", null, "sync", "graph");
    }

    async createOrUpdateLabel(label, properties, labelType = "vertex") {
        const data = {
            name: label,
            type: labelType,
            properties: []
        };

        for (const prop of properties) {
            data.properties.push({
                property: {
                    name: prop.name,
                    dataType: prop.dataType || "string",
                    cardinality: prop.cardinality || "single"
                }
            });
        }

        return this._request('POST', '/schema/labels', data);
    }

    async getLabelSchema(label) {
        return this._request('GET', `/schema?label=${label}`);
    }

    async createEdgeType(edgeType, properties = null) {
        const data = {
            type: "edge",
            properties: []
        };

        if (properties) {
            for (const prop of properties) {
                data.properties.push({
                    property: {
                        name: prop.name,
                        dataType: prop.dataType || "string",
                        cardinality: prop.cardinality || "single"
                    }
                });
            }
        }

        return this._request('POST', `/schema?label=${edgeType}`, data);
    }

    async getGraphSummary(labelDetails = false) {
        return this._request('GET', `/summary?label_details=${labelDetails}`);
    }

    // ==================== 节点操作API ====================

    async addNode(nodeId, labels = null, properties = null) {
        labels = labels || [];
        properties = properties || {};

        const labelStr = labels.length > 0 ? ':' + labels.join(':') : '';
        
        const propsStr = Object.keys(properties).map(k => `${k}: $${k}`).join(', ');
        const propsClause = propsStr ? `, ${propsStr}` : '';

        const statement = `CREATE (n${labelStr} {_ID_: $id${propsClause}}) RETURN n`;
        const params = { id: nodeId, ...properties };

        return this.executeCypher(statement, params);
    }

    async addNodesBatch(nodes) {
        const statements = [];

        for (let i = 0; i < nodes.length; i++) {
            const node = nodes[i];
            const nodeId = node.id;
            const labels = node.labels || [];
            const properties = node.properties || {};

            if (!nodeId) continue;

            const labelStr = labels.length > 0 ? ':' + labels.join(':') : '';
            const propsStr = Object.keys(properties).map(k => `${k}: $${k}_${i}`).join(', ');
            const propsClause = propsStr ? `, ${propsStr}` : '';

            const statement = `CREATE (n${labelStr} {_ID_: $id_${i}${propsClause}})`;
            
            const params = { [`id_${i}`]: nodeId };
            for (const [k, v] of Object.entries(properties)) {
                params[`${k}_${i}`] = v;
            }

            statements.push({
                statement: statement,
                parameters: params,
                resultDataContents: ["row"]
            });
        }

        const data = { statements: statements };
        return this._request('POST', '/action?action_id=execute-cypher-query', data);
    }

    async deleteNode(nodeId) {
        const statement = "MATCH (n) WHERE id(n) = $id DETACH DELETE n";
        return this.executeCypher(statement, { id: nodeId });
    }

    async updateNode(nodeId, properties) {
        const propsStr = Object.keys(properties).map(k => `n.${k} = $${k}`).join(', ');
        const statement = `MATCH (n) WHERE id(n) = $id SET ${propsStr} RETURN n`;
        const params = { id: nodeId, ...properties };
        return this.executeCypher(statement, params);
    }

    async getNode(nodeId) {
        const statement = "MATCH (n) WHERE id(n) = $id RETURN n";
        return this.executeCypher(statement, { id: nodeId });
    }

    // ==================== 边操作API ====================

    async addEdge(startNodeId, endNodeId, edgeType, properties = null) {
        properties = properties || {};

        let propsClause = '';
        if (Object.keys(properties).length > 0) {
            const propsStr = Object.keys(properties).map(k => `${k}: $${k}`).join(', ');
            propsClause = ` {${propsStr}}`;
        }

        const statement = `MATCH (a), (b) WHERE id(a) = $start AND id(b) = $end CREATE (a)-[r:${edgeType}${propsClause}]->(b) RETURN r`;
        const params = { start: startNodeId, end: endNodeId, ...properties };

        return this.executeCypher(statement, params);
    }

    async deleteEdge(startNodeId, endNodeId, edgeType = null) {
        let statement;
        if (edgeType) {
            statement = `MATCH (a)-[r:${edgeType}]->(b) WHERE id(a) = $start AND id(b) = $end DELETE r`;
        } else {
            statement = "MATCH (a)-[r]->(b) WHERE id(a) = $start AND id(b) = $end DELETE r";
        }

        return this.executeCypher(statement, { start: startNodeId, end: endNodeId });
    }

    async getEdges(nodeId, direction = "both") {
        let statement;
        if (direction === "out") {
            statement = "MATCH (n)-[r]->(m) WHERE id(n) = $id RETURN r, m";
        } else if (direction === "in") {
            statement = "MATCH (n)<-[r]-(m) WHERE id(n) = $id RETURN r, m";
        } else {
            statement = "MATCH (n)-[r]-(m) WHERE id(n) = $id RETURN r, m";
        }

        return this.executeCypher(statement, { id: nodeId });
    }

    // ==================== Label操作API ====================

    async addLabelToNode(nodeId, label) {
        const statement = "MATCH (n) WHERE id(n) = $id SET n:`$label` RETURN n";
        return this.executeCypher(statement, { id: nodeId, label: label });
    }

    async removeLabelFromNode(nodeId, label) {
        // GES不支持直接移除label
        const statement = "MATCH (n) WHERE id(n) = $id RETURN n";
        return this.executeCypher(statement, { id: nodeId });
    }

    async getNodesByLabel(label, limit = 100) {
        const statement = `MATCH (n:\`${label}\`) RETURN n LIMIT ${limit}`;
        return this.executeCypher(statement);
    }

    // ==================== 导入导出API ====================

    async exportGraph(exportPath, vertexSetName = "set_vertex", 
                     edgeSetName = "set_edge", schemaName = "schema.xml", 
                     obsParameters = null) {
        const data = {
            graphExportPath: exportPath,
            vertexSetName: vertexSetName,
            edgeSetName: edgeSetName,
            schemaName: schemaName
        };

        if (obsParameters) {
            data.obsParameters = obsParameters;
        }

        const resp = await this._request('POST', '/action?action_id=export-graph', data);
        return resp.jobId || '';
    }

    async importGraph(schemaPath, vertexPath = null, edgePath = null, obsParameters = null) {
        const data = { schemaPath: schemaPath };

        if (vertexPath) data.vertexsetPath = vertexPath;
        if (edgePath) data.edgesetPath = edgePath;
        if (obsParameters) data.obsParameters = obsParameters;

        const resp = await this._request('POST', '/action?action_id=import-graph', data);
        return resp.jobId || '';
    }

    // ==================== 图管理API ====================

    async clearGraph(useApi = true) {
        if (useApi) {
            return this._request('POST', '/action?action_id=clear-graph', {});
        } else {
            const statement = "MATCH (n) DETACH DELETE n";
            return this.executeCypher(statement);
        }
    }

    async getGraphStats() {
        return this.getGraphSummary(true);
    }

    // ==================== 索引操作API ====================

    async createVertexIndex(indexName, label = null) {
        const data = {
            indexName: indexName,
            indexType: "GlobalCompositeVertexIndex",
            hasLabel: Boolean(label),
            indexProperty: []
        };

        return this._request('POST', '/indices', data);
    }

    async createEdgeIndex(indexName, label = null) {
        const data = {
            indexName: indexName,
            indexType: "GlobalCompositeEdgeIndex",
            hasLabel: Boolean(label),
            indexProperty: []
        };

        return this._request('POST', '/indices', data);
    }
}

// ==================== GES Graph Skill主类 ====================

class GESGraphSkill {
    constructor() {
        this.client = new GESClient();
    }

    // ==================== 高级操作接口 ====================

    async _ensureLabelProperties(label, newProperties) {
        // 获取已有属性列表
        const schema = await this.client.getLabelSchema(label);
        const existingProps = [];

        // 解析已有属性
        if (schema && schema.results && schema.results[0]) {
            const data = schema.results[0].data;
            if (data && data.length > 0) {
                const properties = data[0].row[0].properties || [];
                for (const prop of properties) {
                    existingProps.push({
                        property: {
                            name: prop.name,
                            dataType: prop.dataType,
                            cardinality: prop.cardinality || "single"
                        }
                    });
                }
            }
        }

        // 合并新属性（去重）
        const existingNames = new Set(existingProps.map(p => p.property.name));
        for (const prop of newProperties) {
            if (!existingNames.has(prop.property.name)) {
                existingProps.push(prop);
            }
        }

        // 如果没有新属性要添加，直接返回
        if (existingProps.length === 0 && newProperties.length === 0) {
            return;
        }

        // 更新Label Schema（追加新属性）
        const data = {
            type: "vertex",
            properties: existingProps
        };

        await this.client._request('POST', `/schema?label=${label}`, data);
    }

    async executeQuery(cypher, parameters = null) {
        return this.client.executeCypher(cypher, parameters);
    }

    async executeGql(gql, parameters = null) {
        return this.client.executeGql(gql, parameters);
    }

    async getSchemaInfo() {
        return this.client.getSchema();
    }

    async getStatistics() {
        return this.client.getGraphStats();
    }

}

// ==================== 便捷函数 ====================

function getClient() {
    return new GESClient();
}

function getSkill() {
    return new GESGraphSkill();
}

// ==================== 导出 ====================

module.exports = {
    GESClient,
    GESGraphSkill,
    TokenManager,
    OBSClient,
    getClient,
    getSkill
};

// ==================== 主程序测试 ====================

if (require.main === module) {
    (async () => {
        const skill = new GESGraphSkill();

        console.log("=== 测试获取Token ===");
        try {
            const token = await skill.client.tokenMgr.getToken();
            console.log(`Token获取成功: ${token.substring(0, 20)}...`);
        } catch (e) {
            console.log(`Token获取失败: ${e.message}`);
        }

        console.log("\n=== 测试GQL API ===");
        console.log("\n1. 执行GQL查询: MATCH (n) RETURN n LIMIT 3");
        try {
            const result = await skill.executeGql("MATCH (n) RETURN n LIMIT 3");
            console.log(`结果: ${JSON.stringify(result).substring(0, 500)}...`);
        } catch (e) {
            console.log(`错误: ${e.message}`);
        }

        console.log("\n2. 执行GQL统计: MATCH (n) RETURN count(*) as total");
        try {
            const result = await skill.executeGql("MATCH (n) RETURN count(*) as total");
            console.log(`结果: ${JSON.stringify(result).substring(0, 500)}...`);
        } catch (e) {
            console.log(`错误: ${e.message}`);
        }

        console.log("\n=== 测试Cypher API ===");
        console.log("\n1. 执行Cypher查询: MATCH (n) RETURN n LIMIT 3");
        try {
            const result = await skill.executeQuery("MATCH (n) RETURN n LIMIT 3");
            console.log(`结果: ${JSON.stringify(result).substring(0, 500)}...`);
        } catch (e) {
            console.log(`错误: ${e.message}`);
        }
    })();
}

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
GES Memory Skill - 图引擎持久化版原子能力
用于LLM记忆系统的图数据库操作：添加节点、关系、查询、导入导出等
"""

import os
import csv
import json
import time
import requests
from typing import Dict, List, Optional, Any, Union

# 项目根目录
CURRENT_FILE = os.path.abspath(__file__)
SKILL_DIR = os.path.dirname(os.path.dirname(CURRENT_FILE))  # skill根目录（SKILL.md所在目录）
SCRIPTS_DIR = os.path.dirname(CURRENT_FILE)                  # scripts/
SKILLS_DIR = os.path.dirname(SKILL_DIR)     # skills
CODEMATE_DIR = os.path.dirname(SKILLS_DIR)  # .codemate
PROJECT_ROOT = os.path.dirname(CODEMATE_DIR)  # ges_skill (项目根目录)


class EnvConfig:
    """环境配置管理 - 支持环境变量和配置文件读取"""

    ENV_DIR = os.path.join(SKILL_DIR, '.env')

    ENV_VAR_MAPPING = {
        'GES_GRAPH_IP': 'graph_ip',
        'GES_GRAPH_PORT': 'graph_port',
        'GES_PROJECT_ID': 'project_id',
        'GES_GRAPH_NAME': 'graph_name',
        'GES_IAM_URL': 'iam_url',
        'GES_USERNAME': 'username',
        'GES_PASSWORD': 'password',
        'GES_DOMAIN_NAME': 'domain_name',
        'GES_REGION': 'region',
        'HUAWEI_CLOUD_AK': 'access_key',
        'HUAWEI_CLOUD_SK': 'secret_key',
    }

    @staticmethod
    def load_ges_env() -> Dict[str, str]:
        """加载GES环境配置，优先从环境变量读取，备用从.env文件读取"""
        config = {}

        env_file = os.path.join(EnvConfig.ENV_DIR, 'ges_env.csv')
        file_config = EnvConfig._load_csv(env_file)

        for env_var, config_key in EnvConfig.ENV_VAR_MAPPING.items():
            env_value = os.environ.get(env_var)
            if env_value:
                config[config_key] = env_value
            elif file_config.get(config_key):
                config[config_key] = file_config.get(config_key)

        for key, value in file_config.items():
            if key not in config:
                config[key] = value

        required_fields = ['graph_ip', 'project_id', 'graph_name', 'iam_url', 'username', 'password', 'domain_name', 'region']
        missing_fields = [field for field in required_fields if not config.get(field)]

        if missing_fields:
            print('==============================================')
            print('⚠️  配置缺失或不完整')
            print('==============================================')
            print('\n请通过以下方式配置（环境变量优先）：')
            print('  - 环境变量: GES_GRAPH_IP, GES_PROJECT_ID, GES_GRAPH_NAME, GES_IAM_URL, GES_REGION')
            print('  - AKSK: HUAWEI_CLOUD_AK, HUAWEI_CLOUD_SK')
            print('  - 或用户名密码: GES_USERNAME, GES_PASSWORD, GES_DOMAIN_NAME')
            print('  - 或配置文件: .env/ges_env.csv')
            print('\n缺失的必需配置项:')
            for field in missing_fields:
                print(f'  - {field}')
            print('==============================================\n')

        return config

    @staticmethod
    def _load_csv(filepath: str) -> Dict[str, str]:
        """读取CSV配置文件"""
        config = {}
        if os.path.exists(filepath):
            with open(filepath, 'r', encoding='utf-8') as f:
                reader = csv.DictReader(f)
                for row in reader:
                    config[row['key']] = row['value']
        return config


class TokenManager:
    """Token管理 - 支持AKSK方式获取Token"""

    def __init__(self):
        self.token = None
        self.token_expiry = 0
        self.config = EnvConfig.load_ges_env()

    def get_token(self) -> str:
        """获取Token，如果过期则重新获取"""
        current_time = time.time()
        if self.token and current_time < self.token_expiry - 300:
            return self.token

        env_token = os.environ.get('GES_TOKEN')
        if env_token:
            self.token = env_token
            self.token_expiry = time.time() + 3600 * 23
            return self.token

        access_key = self.config.get('access_key')
        secret_key = self.config.get('secret_key')

        if access_key and secret_key:
            try:
                self.token = self._fetch_token_by_aksk()
                return self.token
            except Exception as e:
                print(f"AKSK方式获取Token失败，尝试密码方式: {e}")

        username = self.config.get('username')
        password = self.config.get('password')
        domain_name = self.config.get('domain_name')

        if username and password:
            try:
                self.token = self._fetch_token_by_password(username, password, domain_name)
                return self.token
            except Exception as e:
                raise Exception(f"密码方式获取Token也失败了: {e}")

        raise Exception("无法获取Token，请配置AKSK或用户名密码")

    def _fetch_token_by_password(self, username: str, password: str, domain_name: str = None) -> str:
        """通过密码方式获取Token"""
        url = self.config.get('iam_url')
        project_id = self.config.get('project_id')

        if not url:
            raise Exception("ges_env.csv中缺少iam_url配置")

        headers = {'Content-Type': 'application/json'}

        # 构建请求体
        if domain_name:
            data = {
                'auth': {
                    'identity': {
                        'methods': ['password'],
                        'password': {
                            'user': {
                                'name': username,
                                'password': password,
                                'domain': {'name': domain_name}
                            }
                        }
                    },
                    'scope': {'project': {'id': project_id}}
                }
            }
        else:
            data = {
                'auth': {
                    'identity': {
                        'methods': ['password'],
                        'password': {
                            'user': {
                                'name': username,
                                'password': password
                            }
                        }
                    },
                    'scope': {'project': {'id': project_id}}
                }
            }

        resp = requests.post(url, headers=headers, json=data, timeout=30, verify=False)

        if resp.status_code not in [200, 201]:
            raise Exception(f"密码获取Token失败: {resp.status_code}, {resp.text[:200]}")

        token = resp.headers.get('X-Subject-Token')
        if not token:
            raise Exception(f"密码获取Token失败，未获取到token: {resp.text[:200]}")

        self.token_expiry = time.time() + 3600 * 23
        return token

    def _fetch_token_by_aksk(self) -> str:
        """通过AKSK方式获取Token"""
        import hashlib
        import hmac
        import datetime

        access_key = self.config.get('access_key')
        secret_key = self.config.get('secret_key')
        project_id = self.config.get('project_id')
        url = self.config.get('iam_url')
        region = self.config.get('region', 'cn-north-7')

        if not url:
            raise Exception("ges_env.csv中缺少iam_url配置")

        Service = 'iam'
        ContentType = 'application/json;charset=UTF-8'

        # 构建请求体
        Body = '{"auth": {"identity": {"methods": ["hw_access_key"],"hw_access_key": {"access": {"key": "' + access_key + '"}}},"scope": {"project": {"id": "' + project_id + '"}}}}'

        # 签名计算
        BasicDateFormat = "%Y%m%dT%H%M%SZ"
        ScopeDateFormat = "%Y%m%d"
        Algorithm = "HWS-HMAC-SHA256"

        def sign(key, msg):
            return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()

        def getSignatureKey(key, date_stamp, region, service):
            kDate = sign(('HWS' + key).encode('utf-8'), date_stamp)
            kRegion = sign(kDate, region)
            kService = sign(kRegion, service)
            kSigning = sign(kService, 'hws_request')
            return kSigning

        t = datetime.datetime.utcnow()
        hws_date = t.strftime(BasicDateFormat)
        date_stamp = t.strftime(ScopeDateFormat)
        canonical_uri = '/v3/auth/tokens/'
        canonical_headers = 'accept:' + 'application/json' + '\n' + 'content-type:' + ContentType + '\n' + 'x-hws-date:' + hws_date + '\n'
        signed_headers = 'accept;content-type;x-hws-date'
        payload_hash = hashlib.sha256(Body.encode('utf-8')).hexdigest()
        canonical_request = '\n'.join(
            ['POST', canonical_uri, '', canonical_headers, signed_headers, payload_hash])
        credential_scope = '/'.join([date_stamp, region, Service, 'hws_request'])
        string_to_sign = '\n'.join(
            [Algorithm, hws_date, credential_scope, hashlib.sha256(canonical_request.encode('utf-8')).hexdigest()])
        signing_key = getSignatureKey(secret_key, date_stamp, region, Service)
        signature = hmac.new(signing_key, string_to_sign.encode('utf-8'), hashlib.sha256).hexdigest()
        authorization_header = Algorithm + ' ' + 'Credential=' + access_key + '/' + credential_scope + ', ' + 'SignedHeaders=' + signed_headers + ', ' + 'Signature=' + signature

        headers = {
            'Accept': 'application/json',
            'Content-Type': ContentType,
            'X-Hws-Date': hws_date,
            'X-Identity-Sign': authorization_header
        }

        resp = requests.post(url, data=Body, headers=headers, timeout=30, verify=False)

        if resp.status_code not in [200, 201]:
            raise Exception(f"AKSK获取Token失败: {resp.status_code}, {resp.text[:200]}")

        token = resp.headers.get('X-Subject-Token')
        if not token:
            raise Exception(f"AKSK获取Token失败，未获取到token: {resp.text[:200]}")

        self.token_expiry = time.time() + 3600 * 23  # Token有效期通常是24小时
        return token


class OBSClient:
    """华为云OBS对象存储客户端 (使用HMAC-SHA1签名，不依赖SDK)

    虚拟主机URL格式: https://bucketname.obs.region.domain/objectkey

    签名算法: Authorization: OBS AccessKeyID:Signature
    其中: Signature = Base64(HMAC-SHA1(Your_SK, UTF8(StringToSign)))

    StringToSign格式:
    - 无Content-Type: METHOD\n\n\nDate\n/bucket/key (3个换行)
    - 有Content-Type: METHOD\n\nContent-Type\nDate\n/bucket/key
    """

    def __init__(self, access_key: str, secret_key: str, region: str = "cn-north-7"):
        """初始化OBS客户端

        Args:
            access_key: OBS访问密钥
            secret_key: OBS秘密密钥
            region: OBS区域
        """
        self.access_key = access_key
        self.secret_key = secret_key
        self.region = region
        self.server = f"obs.{region}.ulanqab.huawei.com"
        self._available = bool(access_key and secret_key)

        # 禁用SSL警告
        import urllib3
        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

    def is_available(self) -> bool:
        """检查OBS是否可用"""
        return self._available

    def _get_signature(self, method: str, bucket: str, object_key: str,
                       content_type: str = '', date: str = None) -> tuple:
        """计算OBS签名

        Returns:
            (signature, date)
        """
        import hmac
        import hashlib
        import base64
        from datetime import datetime

        if date is None:
            date = datetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S GMT')

        # 签名的path
        sign_path = f'/{bucket}/{object_key}' if object_key else f'/{bucket}/'

        if content_type:
            # 有Content-Type时的签名格式: METHOD\n\nContent-Type\nDate\n/path
            string_to_sign = f'{method}\n\n{content_type}\n{date}\n{sign_path}'
        else:
            # 无Content-Type时的签名格式: METHOD\n\n\nDate\n/path (3个换行)
            string_to_sign = f'{method}\n\n\n{date}\n{sign_path}'

        # HMAC-SHA1签名
        h = hmac.new(self.secret_key.encode('utf-8'), string_to_sign.encode('utf-8'), hashlib.sha1)
        signature = base64.b64encode(h.digest()).decode('utf-8')

        return signature, date

    def _request(self, method: str, bucket: str, object_key: str = '',
                 query: dict = None, headers: dict = None, body: bytes = None):
        """发起OBS请求"""
        import urllib.parse
        import requests

        # 构造URL
        path = f'/{object_key}' if object_key else '/'

        # 添加查询参数
        if query:
            query_string = '&'.join([f'{k}={urllib.parse.quote(str(v))}' for k, v in query.items()])
            full_path = f'{path}?{query_string}'
        else:
            full_path = path

        url = f'https://{bucket}.{self.server}{full_path}'

        # 构建请求头
        content_type = headers.get('Content-Type', '') if headers else ''
        signature, date = self._get_signature(method, bucket, object_key, content_type)

        request_headers = {
            'Host': f'{bucket}.{self.server}',
            'Date': date,
            'Authorization': f'OBS {self.access_key}:{signature}'
        }

        if headers:
            for k, v in headers.items():
                request_headers[k] = str(v) if k == 'Content-Length' else v

        # 发送请求
        response = requests.request(
            method=method,
            url=url,
            headers=request_headers,
            data=body,
            verify=False
        )

        if response.status_code >= 400:
            raise Exception(f'OBS请求失败 [{response.status_code}]: {response.text}')

        return response

    def list_buckets(self) -> List[str]:
        """列举所有桶"""
        if not self._available:
            raise Exception("OBS客户端不可用")
        # 暂不支持桶列表
        return []

    def list_objects(self, bucket: str, prefix: str = "") -> List[Dict]:
        """列举桶中的对象

        Args:
            bucket: 桶名
            prefix: 对象前缀

        Returns:
            对象列表 (包含Key, Size等信息)
        """
        if not self._available:
            raise Exception("OBS客户端不可用")

        response = self._request(
            method='GET',
            bucket=bucket,
            object_key='',
            query={'list-type': '2', 'prefix': prefix}
        )

        # 解析XML响应
        import re
        contents = []
        for match in re.finditer(r'<Contents><Key>([^<]+)</Key><LastModified>([^<]+)</LastModified><ETag>([^<]+)</ETag><Size>(\d+)</Size>', response.text):
            contents.append({
                'Key': match.group(1),
                'LastModified': match.group(2),
                'ETag': match.group(3),
                'Size': int(match.group(4))
            })
        return contents

    def upload_file(self, local_file: str, bucket: str, object_key: str) -> bool:
        """上传本地文件到OBS

        Args:
            local_file: 本地文件路径
            bucket: 目标桶名
            object_key: OBS中的对象键

        Returns:
            是否成功
        """
        if not self._available:
            raise Exception("OBS客户端不可用")

        import os
        if not os.path.exists(local_file):
            raise Exception(f"文件不存在: {local_file}")

        with open(local_file, 'rb') as f:
            file_content = f.read()

        self._request(
            method='PUT',
            bucket=bucket,
            object_key=object_key,
            headers={
                'Content-Type': 'application/octet-stream',
                'Content-Length': len(file_content)
            },
            body=file_content
        )
        return True

    def download_file(self, bucket: str, object_key: str, local_file: str) -> bool:
        """从OBS下载文件到本地

        Args:
            bucket: 桶名
            object_key: OBS中的对象键
            local_file: 本地保存路径

        Returns:
            是否成功
        """
        if not self._available:
            raise Exception("OBS客户端不可用")

        response = self._request(
            method='GET',
            bucket=bucket,
            object_key=object_key
        )

        with open(local_file, 'wb') as f:
            f.write(response.content)
        return True

    def delete_object(self, bucket: str, object_key: str) -> bool:
        """删除OBS中的对象

        Args:
            bucket: 桶名
            object_key: OBS中的对象键

        Returns:
            是否成功
        """
        if not self._available:
            raise Exception("OBS客户端不可用")

        self._request(
            method='DELETE',
            bucket=bucket,
            object_key=object_key
        )
        return True

    def get_object_url(self, bucket: str, object_key: str) -> str:
        """获取对象URL

        Args:
            bucket: 桶名
            object_key: OBS中的对象键

        Returns:
            对象的公开URL
        """
        return f'https://{bucket}.{self.server}/{object_key}'


class GESClient:
    """GES图数据库API客户端"""

    def __init__(self):
        self.token_mgr = TokenManager()
        self.config = EnvConfig.load_ges_env()
        self.graph_ip = self.config.get('graph_ip', '')
        self.graph_port = self.config.get('graph_port', '80')
        self.project_id = self.config.get('project_id', '')
        self.graph_name = self.config.get('graph_name', '')
        self.base_url = f"http://{self.graph_ip}:{self.graph_port}/ges/v1.0/{self.project_id}/graphs/{self.graph_name}"

        # 初始化OBS客户端
        access_key = self.config.get('access_key', '')
        secret_key = self.config.get('secret_key', '')
        region = self.config.get('region', 'cn-north-7')
        if access_key and secret_key:
            self.obs_client = OBSClient(access_key, secret_key, region)
        else:
            self.obs_client = None

    def get_obs_client(self) -> Optional[OBSClient]:
        """获取OBS客户端"""
        return self.obs_client

    def _get_headers(self) -> Dict[str, str]:
        """获取请求头"""
        return {
            'X-Auth-Token': self.token_mgr.get_token(),
            'Content-Type': 'application/json'
        }

    def _request(self, method: str, path: str, **kwargs) -> Dict:
        """发送API请求"""
        url = f"{self.base_url}{path}"
        headers = self._get_headers()

        resp = requests.request(method, url, headers=headers, **kwargs, verify=False)

        if resp.status_code >= 400:
            raise Exception(f"API请求失败 [{resp.status_code}]: {resp.text}")

        try:
            return resp.json()
        except:
            return {'raw': resp.text}

    # ==================== Cypher查询相关API ====================

    def execute_cypher(self, statement: str, parameters: Dict = None,
                       execution_mode: str = "sync", result_data_contents: str = "row",
                       limit: int = 1000) -> Dict:
        """执行Cypher查询

        Args:
            statement: Cypher查询语句
            parameters: 查询参数（可选）
            execution_mode: 执行模式，sync/async（默认sync）
            result_data_contents: 返回格式，row/graph/raw（默认row）
            limit: 返回结果数量限制（默认1000）

        Returns:
            API响应结果
        """
        data = {
            "statements": [{
                "statement": statement,
                "parameters": parameters or {},
                "executionMode": execution_mode,
                "resultDataContents": [result_data_contents],
                "limit": limit
            }]
        }

        return self._request('POST', '/action?action_id=execute-cypher-query', json=data)

    def execute_gql(self, statement: str, parameters: Dict = None,
                    execution_mode: str = "sync", result_data_contents: str = "row",
                    limit: int = 1000, include_stats: bool = True) -> Dict:
        """执行GQL查询

        Args:
            statement: GQL查询语句
            parameters: 查询参数（可选）
            execution_mode: 执行模式，sync/async（默认sync）
            result_data_contents: 返回格式，row/graph/raw（默认row）
            limit: 返回结果数量限制（默认1000）
            include_stats: 是否包含统计信息（默认True）

        Returns:
            API响应结果
        """
        data = {
            "statements": [{
                "statement": statement,
                "parameters": parameters or {},
                "executionMode": execution_mode,
                "resultDataContents": [result_data_contents],
                "limit": limit,
                "includeStats": include_stats
            }]
        }

        return self._request('POST', '/action?action_id=execute-gql-query', json=data)

    def execute_gql_async(self, statement: str, parameters: Dict = None,
                          result_data_contents: str = "row") -> str:
        """异步执行GQL查询，返回job_id

        Args:
            statement: GQL查询语句
            parameters: 查询参数（可选）
            result_data_contents: 返回格式

        Returns:
            job_id，用于后续查询任务状态
        """
        result = self.execute_gql(statement, parameters, "async", result_data_contents)
        # 从响应中提取job_id
        return result.get('jobId', '')

    def execute_cypher_async(self, statement: str, parameters: Dict = None,
                             result_data_contents: str = "row") -> str:
        """异步执行Cypher查询，返回job_id

        Args:
            statement: Cypher查询语句
            parameters: 查询参数（可选）
            result_data_contents: 返回格式

        Returns:
            job_id，用于后续查询任务状态
        """
        result = self.execute_cypher(statement, parameters, "async", result_data_contents)
        # 从响应中提取job_id
        return result.get('jobId', '')

    def get_job_status(self, job_id: str) -> Dict:
        """查询Job状态

        Args:
            job_id: Job ID

        Returns:
            Job状态信息
        """
        return self._request('GET', f'/jobs/{job_id}/status')

    # ==================== Schema相关API ====================

    def get_schema(self) -> Dict:
        """获取图schema信息

        Returns:
            图的schema信息（点边类型、属性等）
        """
        return self.execute_cypher("call db.schema()", result_data_contents="graph")

    def _ensure_label_properties(self, label: str, new_properties: List[Dict]) -> Dict:
        """确保Label具有所需的属性（获取已有属性后追加新属性）

        Args:
            label: 标签名称
            new_properties: 新增属性列表

        Returns:
            执行结果
        """
        # 获取已有属性列表
        try:
            schema = self.get_label_schema(label)
            existing_props = []

            if schema and "results" in schema and schema["results"]:
                data = schema["results"][0].get("data", [])
                if data and len(data) > 0:
                    properties = data[0].get("row", [{}])[0].get("properties", [])
                    for prop in properties:
                        existing_props.append({
                            "property": {
                                "name": prop.get("name", ""),
                                "dataType": prop.get("dataType", "string"),
                                "cardinality": prop.get("cardinality", "single")
                            }
                        })
        except Exception:
            existing_props = []

        # 合并新属性（去重）
        existing_names = {p["property"]["name"] for p in existing_props}
        for prop in new_properties:
            if prop.get("property", {}).get("name") not in existing_names:
                existing_props.append(prop)

        # 更新Label Schema（追加新属性）
        data = {
            "type": "vertex",
            "properties": existing_props
        }

        return self._request("POST", f"/schema?label={label}", json=data)

    def create_or_update_label(self, label: str, properties: List[Dict], label_type: str = "vertex") -> Dict:
        """创建或更新Label及其属性

        Args:
            label: 标签名称
            properties: 属性列表，每个包含 name, dataType, cardinality
            label_type: 类型，vertex 或 edge

        Returns:
            执行结果

        Example:
            properties = [
                {"name": "content", "dataType": "string", "cardinality": "single"},
                {"name": "timestamp", "dataType": "long", "cardinality": "single"}
            ]
        """
        data = {
            "name": label,
            "type": label_type,
            "properties": []
        }

        for prop in properties:
            data["properties"].append({
                "property": {
                    "name": prop.get("name", ""),
                    "dataType": prop.get("dataType", "string"),
                    "cardinality": prop.get("cardinality", "single")
                }
            })

        return self._request('POST', '/schema/labels', json=data)

    def get_label_schema(self, label: str) -> Dict:
        """获取指定Label的schema信息

        Args:
            label: 标签名称

        Returns:
            Label的schema信息
        """
        return self._request('GET', f'/schema?label={label}')

    def create_edge_type(self, edge_type: str, properties: List[Dict] = None) -> Dict:
        """创建边类型及其属性

        Args:
            edge_type: 边类型名称
            properties: 属性列表（可选）

        Returns:
            执行结果
        """
        data = {
            "type": "edge",
            "properties": []
        }

        if properties:
            for prop in properties:
                data["properties"].append({
                    "property": {
                        "name": prop.get("name", ""),
                        "dataType": prop.get("dataType", "string"),
                        "cardinality": prop.get("cardinality", "single")
                    }
                })

        return self._request('POST', f'/schema?label={edge_type}', json=data)

    def get_graph_summary(self, label_details: bool = False) -> Dict:
        """获取图概要信息

        Args:
            label_details: 是否显示label详情

        Returns:
            图概要信息
        """
        return self._request('GET', f'/summary?label_details={str(label_details).lower()}')

    # ==================== 节点操作API ====================

    def add_node(self, node_id: str, labels: List[str] = None, properties: Dict = None) -> Dict:
        """添加节点

        Args:
            node_id: 节点ID
            labels: 节点标签列表，如 ["user", "person"]
            properties: 节点属性字典

        Returns:
            执行结果
        """
        labels = labels or []
        properties = properties or {}

        # 构建Cypher语句，使用GES特殊的 _ID_ 属性
        label_str = ':'.join(labels) if labels else ''
        if label_str:
            label_str = ':' + label_str

        props_str = ', '.join([f'{k}: ${k}' for k in properties.keys()])
        if props_str:
            props_str = ', ' + props_str

        # GES使用 _ID_ 来设置点的字符串ID
        statement = f"CREATE (n{label_str} {{_ID_: $id{props_str}}}) RETURN n"
        params = {'id': node_id}
        params.update(properties)

        return self.execute_cypher(statement, parameters=params)

    def add_nodes_batch(self, nodes: List[Dict]) -> Dict:
        """批量添加节点

        Args:
            nodes: 节点列表，每个节点包含id, labels, properties

        Returns:
            执行结果
        """
        statements = []
        params_dict = {}

        for i, node in enumerate(nodes):
            node_id = node.get('id', '')
            labels = node.get('labels', [])
            properties = node.get('properties', {})

            if not node_id:
                continue

            label_str = ':'.join(labels) if labels else ''
            if label_str:
                label_str = ':' + label_str

            props_str = ', '.join([f'{k}: ${k}_{i}' for k in properties.keys()])
            if props_str:
                props_str = ', ' + props_str

            # GES使用 _ID_ 来设置点的字符串ID
            statement = f"CREATE (n{label_str} {{_ID_: $id_{i}{props_str}}})"
            statements.append({
                "statement": statement,
                "parameters": {f'id_{i}': node_id},
                "resultDataContents": ["row"]
            })

            # 添加属性参数
            for k, v in properties.items():
                params_dict[f'{k}_{i}'] = v

        data = {"statements": statements}
        return self._request('POST', '/action?action_id=execute-cypher-query', json=data)

    def delete_node(self, node_id: str) -> Dict:
        """删除节点

        Args:
            node_id: 节点ID

        Returns:
            执行结果
        """
        # GES中使用 _ID_ 属性来匹配字符串ID
        statement = "MATCH (n) WHERE id(n) = $id DETACH DELETE n"
        return self.execute_cypher(statement, parameters={'id': node_id})

    def update_node(self, node_id: str, properties: Dict) -> Dict:
        """更新节点属性

        Args:
            node_id: 节点ID
            properties: 要更新的属性

        Returns:
            执行结果
        """
        # GES中使用 _ID_ 属性来匹配字符串ID
        props_str = ', '.join([f'n.{k} = ${k}' for k in properties.keys()])
        statement = f"MATCH (n) WHERE id(n) = $id SET {props_str} RETURN n"
        params = {'id': node_id}
        params.update(properties)

        return self.execute_cypher(statement, parameters=params)

    def get_node(self, node_id: str) -> Dict:
        """根据ID获取节点

        Args:
            node_id: 节点ID

        Returns:
            节点信息
        """
        # GES中使用 _ID_ 属性来匹配字符串ID
        statement = "MATCH (n) WHERE id(n) = $id RETURN n"
        return self.execute_cypher(statement, parameters={'id': node_id})

    # ==================== 边操作API ====================

    def add_edge(self, start_node_id: str, end_node_id: str,
                 edge_type: str, properties: Dict = None) -> Dict:
        """添加边

        Args:
            start_node_id: 起始节点ID
            end_node_id: 终止节点ID
            edge_type: 边类型/标签
            properties: 边属性

        Returns:
            执行结果
        """
        properties = properties or {}

        # 构建属性字符串
        if properties:
            props_parts = []
            for k in properties.keys():
                props_parts.append(f'{k}: ${k}')
            props_str = ', '.join(props_parts)
            props_clause = f' {{{props_str}}}'
        else:
            props_clause = ''

        # GES中使用 id() 函数来匹配节点ID
        statement = f"MATCH (a), (b) WHERE id(a) = $start AND id(b) = $end CREATE (a)-[r:{edge_type}{props_clause}]->(b) RETURN r"
        params = {'start': start_node_id, 'end': end_node_id}
        params.update(properties)

        return self.execute_cypher(statement, parameters=params)

    def delete_edge(self, start_node_id: str, end_node_id: str, edge_type: str = None) -> Dict:
        """删除边

        Args:
            start_node_id: 起始节点ID
            end_node_id: 终止节点ID
            edge_type: 边类型（可选）

        Returns:
            执行结果
        """
        # GES中使用 id() 函数来匹配节点ID
        if edge_type:
            statement = "MATCH (a)-[r:{}]->(b) WHERE id(a) = $start AND id(b) = $end DELETE r".format(edge_type)
        else:
            statement = "MATCH (a)-[r]->(b) WHERE id(a) = $start AND id(b) = $end DELETE r"

        return self.execute_cypher(statement, parameters={'start': start_node_id, 'end': end_node_id})

    def get_edges(self, node_id: str, direction: str = "both") -> Dict:
        """获取节点的边

        Args:
            node_id: 节点ID
            direction: 方向，both/in/out

        Returns:
            边的信息
        """
        if direction == "out":
            statement = "MATCH (n)-[r]->(m) WHERE id(n) = $id RETURN r, m"
        elif direction == "in":
            statement = "MATCH (n)<-[r]-(m) WHERE id(n) = $id RETURN r, m"
        else:
            statement = "MATCH (n)-[r]-(m) WHERE id(n) = $id RETURN r, m"

        return self.execute_cypher(statement, parameters={'id': node_id})

    # ==================== Label操作API ====================

    def add_label_to_node(self, node_id: str, label: str) -> Dict:
        """为节点添加Label

        Args:
            node_id: 节点ID
            label: 要添加的标签

        Returns:
            执行结果
        """
        statement = "MATCH (n) WHERE id(n) = $id SET n:`$label` RETURN n"
        return self.execute_cypher(statement, parameters={'id': node_id, 'label': label})

    def remove_label_from_node(self, node_id: str, label: str) -> Dict:
        """从节点移除Label

        Args:
            node_id: 节点ID
            label: 要移除的标签

        Returns:
            执行结果
        """
        # GES不支持直接移除label，但可以通过重新创建节点来实现
        statement = "MATCH (n) WHERE id(n) = $id RETURN n"
        result = self.execute_cypher(statement, parameters={'id': node_id})

        # 注意：GES实际的label移除可能需要更复杂的操作
        return result

    def get_nodes_by_label(self, label: str, limit: int = 100) -> Dict:
        """根据Label查询节点

        Args:
            label: 节点标签
            limit: 返回数量限制

        Returns:
            节点列表
        """
        statement = f"MATCH (n:`{label}`) RETURN n LIMIT {limit}"
        return self.execute_cypher(statement)

    # ==================== 导入导出API ====================

    def export_graph(self, export_path: str, vertex_set_name: str = "set_vertex",
                     edge_set_name: str = "set_edge", schema_name: str = "schema.xml",
                     obs_parameters: Dict = None) -> str:
        """导出整个图数据到OBS

        Args:
            export_path: OBS导出路径，格式: bucket/path，如 "claude/export_test/01"
            vertex_set_name: 点数据集名称，如 "set_vertex"
            edge_set_name: 边数据集名称，如 "set_edge"
            schema_name: Schema文件名，如 "schema.xml"
            obs_parameters: OBS访问参数，包含accessKey和secretKey

        Returns:
            job_id，用于查询导出任务状态

        Example:
            obs_params = {
                'accessKey': 'your_access_key',
                'secretKey': 'your_secret_key'
            }
            job_id = client.export_graph(
                export_path='claude/export_test/01',
                vertex_set_name='set_vertex',
                edge_set_name='set_edge',
                schema_name='schema.xml',
                obs_parameters=obs_params
            )
        """
        data = {
            "graphExportPath": export_path,
            "vertexSetName": vertex_set_name,
            "edgeSetName": edge_set_name,
            "schemaName": schema_name
        }

        if obs_parameters:
            data["obsParameters"] = obs_parameters

        resp = self._request('POST', '/action?action_id=export-graph', json=data)
        return resp.get('jobId', '')

    def export_query_result(self, cypher: str, export_path: str,
                            obs_parameters: Dict = None, wait_completion: bool = True) -> str:
        """导出Cypher查询结果到OBS

        注意：此API可能不被当前GES版本支持，返回404错误。
        cypher的RETURN中必须指明id或属性，用于标识节点/边

        Args:
            cypher: Cypher查询语句，RETURN中需要包含id或属性
            export_path: OBS导出路径，格式: bucket/path
            obs_parameters: OBS访问参数，包含accessKey和secretKey
            wait_completion: 是否等待查询完成后才返回export jobId

        Returns:
            export_job_id，用于查询导出任务状态

        Example:
            # 查询并导出，注意RETURN中包含id
            obs_params = {'accessKey': 'xxx', 'secretKey': 'yyy'}
            job_id = client.export_query_result(
                cypher='MATCH (n) RETURN id(n) as id, n.name as name',
                export_path='claude/export_result/01',
                obs_parameters=obs_params
            )

        Note:
            当前GES版本可能不支持此API，如返回404请使用其他方式导出数据
        """
        # 先执行异步查询
        data = {
            "statements": [{
                "statement": cypher,
                "parameters": {},
                "executionMode": "async",
                "resultDataContents": ["row"]
            }]
        }
        resp = self._request('POST', '/action?action_id=execute-cypher-query', json=data)

        # 从results中获取jobId
        query_job_id = resp.get('results', [{}])[0].get('jobId', '')
        if not query_job_id:
            raise Exception(f"无法获取查询jobId: {resp}")

        # 如果需要等待查询完成
        if wait_completion:
            # 等待查询执行完成（调用方也可以自行轮询）
            import time
            for _ in range(60):  # 最多等待2分钟
                time.sleep(2)
                status = self.get_job_status(query_job_id)
                if status.get('status') == 'complete':
                    break
                if status.get('status') == 'failed':
                    raise Exception(f"查询执行失败: {status}")

        # 导出查询结果到OBS
        export_data = {
            "exportPath": export_path,
            "fileName": "query_result.csv",
            "obsParameters": obs_parameters
        }
        export_resp = self._request('POST', f'/jobs/{query_job_id}/action?action_id=export-result', json=export_data)

        return export_resp.get('jobId', '')

    def export_job_result(self, job_id: str, export_path: str, file_name: str = "result.csv",
                          obs_parameters: Dict = None) -> Dict:
        """导出指定Job的结果到文件

        Args:
            job_id: Job ID（通常是Cypher异步查询的jobId）
            export_path: OBS导出路径，格式: bucket/path
            file_name: 导出文件名
            obs_parameters: OBS访问参数，包含accessKey和secretKey

        Returns:
            执行结果
        """
        data = {
            "exportPath": export_path,
            "fileName": file_name,
        }

        if obs_parameters:
            data["obsParameters"] = obs_parameters

        return self._request('POST', f'/jobs/{job_id}/action?action_id=export-result', json=data)

    def import_graph(self, schema_path: str, vertex_path: str = None,
                     edge_path: str = None, obs_parameters: Dict = None) -> str:
        """导入图数据

        Args:
            schema_path: OBS上的Schema文件路径，格式: bucket/path/schema.xml
            vertex_path: OBS上的点数据文件夹路径，格式: bucket/path/vertices/
            edge_path: OBS上的边数据文件夹路径，格式: bucket/path/edges/
            obs_parameters: OBS访问参数，包含accessKey和secretKey

        Returns:
            job_id，用于查询导入任务状态

        Note:
            - vertex_path和edge_path必须使用不同的文件夹
            - OBS路径格式: bucket/object-key，不需要obs://前缀
            - 建议目录结构:
                obs://bucket/schema/schema.xml
                obs://bucket/data/vertices/
                obs://bucket/data/edges/
        """
        data = {
            "schemaPath": schema_path
        }

        if vertex_path:
            data["vertexsetPath"] = vertex_path
        if edge_path:
            data["edgesetPath"] = edge_path
        if obs_parameters:
            data["obsParameters"] = obs_parameters

        resp = self._request('POST', '/action?action_id=import-graph', json=data)
        return resp.get('jobId', '')

    # ==================== 图管理API ====================

    def clear_graph(self, use_api: bool = True) -> Dict:
        """清空图中所有数据

        Args:
            use_api: 是否使用专用API方式（默认True）。若为False，则使用Cypher方式。

        Returns:
            执行结果，包含job_id（API方式）或查询结果（Cypher方式）

        Note:
            - API方式 (use_api=True): 使用 action_id=clear-graph，会返回job_id
            - Cypher方式 (use_api=False): 使用 MATCH (n) DETACH DELETE n
            - 推荐使用API方式，更彻底且支持异步执行
        """
        if use_api:
            # 使用专用clear-graph API，更彻底
            return self._request('POST', '/action?action_id=clear-graph', json={})
        else:
            # 备用Cypher方式
            statement = "MATCH (n) DETACH DELETE n"
            return self.execute_cypher(statement)

    def get_graph_stats(self) -> Dict:
        """获取图统计信息

        Returns:
            图的统计信息（节点数、边数等）
        """
        return self.get_graph_summary(label_details=True)

    # ==================== 索引操作API ====================

    def create_vertex_index(self, index_name: str, label: str = None) -> Dict:
        """创建点索引

        Args:
            index_name: 索引名称
            label: 点标签（可选）

        Returns:
            执行结果
        """
        data = {
            "indexName": index_name,
            "indexType": "GlobalCompositeVertexIndex",
            "hasLabel": True if label else False,
            "indexProperty": []
        }

        return self._request('POST', '/indices', json=data)

    def create_edge_index(self, index_name: str, label: str = None) -> Dict:
        """创建边索引

        Args:
            index_name: 索引名称
            label: 边标签（可选）

        Returns:
            执行结果
        """
        data = {
            "indexName": index_name,
            "indexType": "GlobalCompositeEdgeIndex",
            "hasLabel": True if label else False,
            "indexProperty": []
        }

        return self._request('POST', '/indices', json=data)


class GESMemorySkill:
    """GES Memory Skill主类 - 提供LLM记忆系统的原子能力

    注意：GES持久化版只支持以下导出能力：
    - export_graph: 导出整图数据到OBS

    不支持以下功能：
    - export_query_result: 导出Cypher查询结果（job结果导出）
    - export_job_result: 导出Job结果到文件

    这是GES持久化版的设计限制，如需导出查询结果请使用其他方式。
    """

    def __init__(self):
        self.client = GESClient()

    def execute_query(self, cypher: str, parameters: Dict = None) -> Dict:
        """执行自定义Cypher查询

        Args:
            cypher: Cypher查询语句
            parameters: 查询参数

        Returns:
            查询结果
        """
        return self.client.execute_cypher(cypher, parameters=parameters)

    def execute_gql(self, gql: str, parameters: Dict = None) -> Dict:
        """执行自定义GQL查询

        Args:
            gql: GQL查询语句
            parameters: 查询参数

        Returns:
            查询结果
        """
        return self.client.execute_gql(gql, parameters=parameters)

    def get_schema_info(self) -> Dict:
        """获取图谱Schema信息

        Returns:
            Schema信息
        """
        return self.client.get_schema()

    def get_statistics(self) -> Dict:
        """获取图谱统计信息

        Returns:
            统计信息
        """
        return self.client.get_graph_stats()


# 便捷函数
def get_client() -> GESClient:
    """获取GES客户端实例"""
    return GESClient()


def get_skill() -> GESMemorySkill:
    """获取GES Memory Skill实例"""
    return GESMemorySkill()