huaweicloud/huaweicloud-skills · GitHub

huawei-cloud-ges-graph

120회 설치

어떤 스킬인가요?

Provides access guide for Huawei Cloud Graph Database GES service. Covers Cypher queries, GQL queries, schema/label management, summary info queries, graph data editing and more. Use this skill when users want to operate Huawei Cloud graph database GES service via terminal.

설치 방법

프로젝트 폴더에서 아래 명령어를 실행하고, 설치할 에이전트를 선택하세요.

npx skills add huaweicloud/huaweicloud-skills --skill huawei-cloud-ges-graph

설치 명령을 직접 실행해야 적용됩니다. 지원 에이전트와 필요한 권한·라이선스는 제작자의 안내를 확인하세요.

스킬 원문 파일

`references/ges_env.csv.example`

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

`scripts/ges_graph_skill.js`

#!/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}`);
        }
    })();
}

`scripts/ges_graph_skill.py`

#!/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()

`scripts/package.json`

{
  "name": "ges-graph-skill",
  "version": "1.0.0",
  "description": "华为云GES图引擎SDK for Node.js",
  "main": "ges_graph_skill.js",
  "scripts": {
    "test": "echo \"Error: no test specified\" && exit 1"
  },
  "keywords": [
    "huawei",
    "GES",
    "graph",
    "database",
    "sdk"
  ],
  "author": "",
  "license": "MIT"
}

`SKILL.md`

---
name: huawei-cloud-ges-graph
description: |
  Provides access guide for Huawei Cloud Graph Database GES service.
  Covers Cypher queries, GQL queries, schema/label management,
  summary info queries, graph data editing and more. Use this skill when users
  want to operate Huawei Cloud graph database GES service via terminal.
Trigger: Graph Database, 图数据库, Cypher查询, GES图, 图引擎, 使用Cypher查一下图数据库, 华为云图引擎, 操作图数据库,
  用Cypher查询图, GES图数据库操作, graph database, "query graph with Cypher", 查询图数据库, 在图数据库中发送语句,
  "Cypher查询", "图数据库GES", "GES图查询"
tags: [huawei-cloud, ges, graph, python, nodejs]
---

> **⚠️ Execution Method (Must Read): This skill executes queries via local Python or Node.js scripts under `scripts/`. Using direct API calls is prohibited.**
>
> - Query scripts are located under the skill directory `scripts/` (e.g., `scripts/ges_graph_skill.py`)
> - All scripts and environment check scripts are inside the skill package. **You must use `skill action=exec` to execute them; do not run them directly in the shell**
> - **Prefer inline execution** (`python -c` or `node -e`) over creating temporary script files. See "Inline Execution (No Temp Files)" section below.
> - **All paths are relative to the skill directory, which is the directory where this SKILL.md resides**

# GES Graph Access Guidance

## Overview

Huawei Cloud Graph Engine Service (GES) persistent edition atomic capability,
providing access guide for graph database operations.
Covers Cypher queries, GQL queries, schema/label management, summary info queries, graph data editing and other core capabilities.

## Directory Structure

The directory conventions are as follows (all paths are relative to the skill directory):

1. `scripts/` - Contains the Python and Node.js execution scripts
   - `ges_graph_skill.py` - Python SDK for GES graph operations
   - `ges_graph_skill.js` - Node.js SDK for GES graph operations
2. `references/` - Contains documentation and configuration examples
   - `ges_env.csv.example` - Environment configuration template

## Prerequisites

Before using this skill, ensure the following conditions are met:

### 1. Runtime Environment
Supports Python or Node.js runtime.
Requires Python 3.8+ or Node.js 14+.

### 2. Graph Instance Configuration

This skill supports both environment variables and configuration files. Environment variables take precedence over configuration files.

**Environment Variables and Parameters**

| Environment Variable | Description | Required |
|---------------------|-------------|----------|
| `GES_GRAPH_IP` | GES service IP | Yes |
| `GES_GRAPH_PORT` | GES service port | Yes |
| `GES_PROJECT_ID` | Project ID | Yes |
| `GES_GRAPH_NAME` | Graph name | Yes |
| `GES_IAM_URL` | IAM service URL | Yes |
| `GES_REGION` | Region | Yes |
| `HUAWEI_CLOUD_AK` | Access Key | Yes* |
| `HUAWEI_CLOUD_SK` | Secret Key | Yes* |
| `GES_USERNAME` | Username | Conditional** |
| `GES_PASSWORD` | Password | Conditional** |
| `GES_DOMAIN_NAME` | Domain name | Conditional** |
| `GES_TOKEN` | Token (highest priority) | Optional |

> * **AK/SK required** for AKSK authentication.
> ** **Username/password conditionally required** when AKSK is not available.

**Configuration File (`.env/ges_env.csv`)**

Config file path: `.env/ges_env.csv`

| Config Item | Description | Required | Example Value |
|-------------|-------------|----------|---------------|
| graph_ip | GES service IP | Yes | 100.95.xxx.xxx |
| graph_port | GES service port | Yes | 80 |
| project_id | Project ID | Yes | your_project_id |
| graph_name | Graph name | Yes | your_graph_name |
| iam_url | IAM service URL | Yes | (see region table below) |
| access_key | Access Key | Yes* | your_access_key |
| secret_key | Secret Key | Yes* | your_secret_key |
| username | Username | Conditional** | your_username |
| password | Password | Conditional** | your_password |
| domain_name | Domain name | Conditional** | your_domain_name |
| region | Region | Yes | cn-north-4 |

> * **access_key/secret_key required** for AKSK authentication.
> ** **username/password/domain_name conditionally required** when AKSK is not available.

### IAM URLs by Region

| Region     | URL                                                 | protocol |
|------------|-----------------------------------------------------|----------|
| cn-north-4 | iam.cn-north-4.myhuaweicloud.com/v3/auth/tokens     | HTTPS |
| ap-southeast-1 | iam.ap-southeast-1.myhuaweicloud.com/v3/auth/tokens | HTTPS |

Three methods are supported (priority from high to low):
1. Environment variable `GES_TOKEN`
2. AKSK method (`HUAWEI_CLOUD_AK` + `HUAWEI_CLOUD_SK` or `access_key` + `secret_key`)
3. Password method (username + password + domain_name)

## ⛔ Prohibited Operations (Safety Guardrail)

> **This skill prohibits the following operations:**

| Prohibited Operation | Description | Reason |
|----------------------|-------------|--------|
| ❌ Printing sensitive credentials | Printing AK/SK,  password, token, or other sensitive information | Risk of sensitive information leakage |

> **The following high-risk operations require explicit agent confirmation before execution:**

| High-Risk Operation | Description | Confirmation Prompt |
|---------------------|-------------|---------------------|
| ⚠️ Clearing all graph data | `clear_graph()` or similar clear operations | "Are you sure you want to clear all data in the graph? This operation is irreversible." |
| ⚠️ Batch deleting nodes/edges | Unconditional bulk deletion of all nodes or edges | "Are you sure you want to batch delete all nodes/edges? This operation is irreversible." |

> **If a user requests any of the above high-risk operations, explicit confirmation must be obtained:**
> "This is a high-risk operation and requires your explicit confirmation. Please reply with 'confirm' to proceed."


## Cypher and GQL Query Languages

GES supports two query languages: Cypher (Neo4j-compatible) and GQL (international standard graph query language).

### Cypher Usage

**Python:**
```python
# Execute a Cypher query
result = skill.execute_query("MATCH (n) RETURN n LIMIT 10")

# Create a node (_ID_ is used only during creation to set a custom ID)
result = skill.execute_query(
    "CREATE (n:Person {_ID_: 'p001', name: '张三'}) RETURN n"
)

# Match query
result = skill.execute_query(
    "MATCH (n:Person)-[:KNOWS]->(m) WHERE n.name = '张三' RETURN m"
)

# Update a node (match via id() function)
result = skill.execute_query(
    "MATCH (n) WHERE id(n) = 123 SET n.name = '李四' RETURN n"
)

# Delete a node
result = skill.execute_query(
    "MATCH (n) WHERE id(n) = 123 DETACH DELETE n"
)

# Aggregate query
result = skill.execute_query(
    "MATCH (n:Person) RETURN n.city, count(*) as cnt ORDER BY cnt DESC"
)
```

**Node.js:**
```javascript
const { GESGraphSkill } = require('./ges_graph_skill.js');
const skill = new GESGraphSkill();

// Execute a Cypher query
const result = await skill.executeQuery("MATCH (n) RETURN n LIMIT 10");

// Create a node (_ID_ is used only during creation to set a custom ID)
const result = await skill.executeQuery(
    "CREATE (n:Person {_ID_: 'p001', name: '张三'}) RETURN n"
);

// Match query
const result = await skill.executeQuery(
    "MATCH (n:Person)-[:KNOWS]->(m) WHERE n.name = '张三' RETURN m"
);

// Update a node
const result = await skill.executeQuery(
    "MATCH (n) WHERE id(n) = 123 SET n.name = '李四' RETURN n"
);

// Delete a node
const result = await skill.executeQuery(
    "MATCH (n) WHERE id(n) = 123 DETACH DELETE n"
);

// Aggregate query
const result = await skill.executeQuery(
    "MATCH (n:Person) RETURN n.city, count(*) as cnt ORDER BY cnt DESC"
);

// Path query
const result = await skill.executeQuery(
    "MATCH p=(n:Person)-[*1..3]->(m) WHERE n.name = '张三' RETURN p"
);
```

### Inline Execution (No Temp Files)

Execute Cypher queries directly via `skill action=exec` without creating any script files.

> **Note:** When running via `skill action=exec`, the working directory is the project root. The installed skill is located under `.agents/skills/huawei-cloud-ges-graph`, **not** under `skills/ai/ges/`.

**Python:**
```bash
skill exec py -c "
import sys, os; cwd=os.getcwd()
skill_dir = os.path.join(cwd, '.agents', 'skills', 'huawei-cloud-ges-graph')
sys.path.insert(0, os.path.join(skill_dir, 'scripts'))
from ges_graph_skill import get_skill
import json
r = get_skill().execute_query('MATCH (n) RETURN n LIMIT 5')
print(json.dumps(r, ensure_ascii=False, indent=2))
"
```

**Node.js:**
```bash
skill exec node -e "
const path = require('path');
const cwd = process.cwd();
const scriptPath = path.join(cwd, '.agents', 'skills', 'huawei-cloud-ges-graph', 'scripts', 'ges_graph_skill.js');
const { GESGraphSkill } = require(scriptPath);
(async () => {
    const r = await new GESGraphSkill().executeQuery('MATCH (n) RETURN n LIMIT 5');
    console.log(JSON.stringify(r, null, 2));
})();
"
```

#### Common Cypher Statements

| Category | Statement | Description |
|----------|-----------|-------------|
| Schema | `call db.schema()` | Get graph schema information |
| Indexes | `call db.indexes()` | View all indexes |
| Kill query | `call dbms.killQuery('queryId')` | Terminate a running query |
| Running queries | `call dbms.listQueries()` | View current queries |
| System parameters | `call dbms.parameter('needNodeIndex', false)` | Remove index constraint (large graph scenarios) |

### GQL Usage (Supported by this Skill)

GQL is the ISO/IEC 39075 standardized graph query language. GES invokes it via `action_id=execute-gql-query`.

```python
# GQL requires the underlying _request method
client = GESClient()
result = client._request('POST', '/action?action_id=execute-gql-query', json={
    "statements": [{
        "statement": "INSERT (n:Person{_ID_:'p001', firstName:'Eywa'}) RETURN n",
        "parameters": {},
        "resultDataContents": ["row"]
    }]
})
```

#### Common GQL Statements

| Category | Statement | Description |
|----------|-----------|-------------|
| Insert | `INSERT (n:Person{_ID_:'p001', firstName:'Eywa'}) RETURN n` | Insert node |
| Match | `MATCH (n:Person WHERE element_id(n)='p001') RETURN n` | Conditional match |
| Update | `MATCH (n:Person WHERE element_id(n)='p001') SET n.lastName='Higgo' RETURN n` | Update properties |
| Remove property | `MATCH (n:Person WHERE element_id(n)='p001') REMOVE n.lastName RETURN n` | Remove property |
| Delete node | `MATCH (n:Person WHERE element_id(n)='p001') DELETE n` | Delete node |
| Filter | `MATCH (n:Person)-[:KNOWS]->(m) FILTER element_id(n)='7933' AND m.gender='male' RETURN m` | Filter results |
| FOR loop | `FOR a IN [1,2,3] RETURN a` | Loop statement |
| LET variable | `LET a = 1, b = 2 RETURN a, b` | Variable definition |
| UNION | `... UNION ALL ...` | Merge result sets |

#### Cypher vs GQL Key Differences

| Feature | Cypher | GQL |
|---------|--------|-----|
| Internal ID | `id(n)` | `element_id(n)` |
| Custom ID (insert only) | `_ID_` property | `_ID_` property |
| Node matching | `MATCH (n)` | `MATCH (n WHERE ...)` |
| SET statement | `SET n.prop = value` | `SET n.prop = value` |
| Remove property | `REMOVE n.prop` | `REMOVE n.prop` |
| Loops | Not supported | `FOR x IN [...]` |
| Variable definition | Not supported | `LET x = value` |## Core Interfaces (Core Commands)

### Cypher Query

```python
# Execute a Cypher query
result = skill.execute_query("MATCH (n) RETURN n LIMIT 10")

# Execute a Cypher query with parameters
result = skill.execute_query(
    "MATCH (n) WHERE n.name = $name RETURN n",
    parameters={"name": "张三"}
)
```

### Vertex Operations

```python
# Add a node
result = skill.client.add_node(
    node_id="mem_001",
    labels=["Memory", "conversation"],
    properties={"content": "User said Hello", "timestamp": 1234567890}
)

# Batch add nodes
result = skill.client.add_nodes_batch([
    {"id": "mem_001", "labels": ["Memory"], "properties": {"content": "test1"}},
    {"id": "mem_002", "labels": ["Memory"], "properties": {"content": "test2"}}
])

# Get a node
result = skill.client.get_node("mem_001")

# Update a node
result = skill.client.update_node("mem_001", {"content": "New content"})

# Delete a node
result = skill.client.delete_node("mem_001")
```

### Edge Operations

```python
# Add an edge
result = skill.client.add_edge(
    start_node_id="mem_001",
    end_node_id="mem_002",
    edge_type="RELATED_TO",
    properties={"weight": 0.8}
)

# Delete an edge
result = skill.client.delete_edge("mem_001", "mem_002", "RELATED_TO")

# Get edges of a node
result = skill.client.get_edges("mem_001", direction="both")
```

### Label Operations

```python
# Add a label to a node
result = skill.client.add_label_to_node("mem_001", "important")

# Query nodes by label
result = skill.client.get_nodes_by_label("Memory", limit=100)
```

### Graph Management

```python
# Get schema information
result = skill.get_schema_info()

# Get graph statistics
result = skill.get_statistics()

# Clear all data in the graph (dangerous operation)
result = skill.clear_all_memories()
```

### Import/Export

```python
# Import graph data
job_id = skill.client.import_graph(
    schema_path="obs://bucket/schema.xml",
    vertex_path="obs://bucket/vertex",
    edge_path="obs://bucket/edge"
)

# Export graph data (access_key/secret_key are read from .env automatically)
job_id = skill.client.export_graph(
    export_path="obs://bucket/export",
    vertex_set_name="set_vertex",
    edge_set_name="set_edge"
)
```

## GES Syntax Guide

### Node ID Handling

GES uses the special `_ID_` property to handle string-type node IDs:

- **Creating nodes** uses the `_ID_` property:
  ```cypher
  CREATE (n:Memory{_ID_: 'mem_001', content: 'test'})
  ```

- **Other operations** use the `id()` function:
  ```cypher
  MATCH (n) WHERE id(n) = 'mem_001' RETURN n
  MATCH (n)-[r]->(m) WHERE id(n) = 'mem_001' RETURN r
  ```

### Schema Requirements

GES requires that schema (Labels and Properties) be defined before corresponding nodes can be created. Schema can be defined through:
1. Creating Labels and properties via the GES management console
2. Importing data with schema through the import interface

## Response Format

All Cypher interfaces return a unified JSON format:

```json
{
  "results": [
    {
      "columns": ["column1", "column2"],
      "data": [
        {"row": ["value1", "value2"], "meta": [null, null]}
      ]
    }
  ],
  "errors": []
}
```

## Error Handling

```python
from ges_graph_skill import get_skill

skill = get_skill()
try:
    result = skill.execute_query("MATCH (n) RETURN n")
except Exception as e:
    print(f"Error: {e}")
```

## Important Notes

1. **Token validity**: Token is valid for 24 hours; the code refreshes automatically
2. **Schema constraint**: Ensure Labels and Properties are defined before creating nodes
3. **Dangerous operations**: `clear_graph()` deletes all data in the graph; use with caution
4. **Asynchronous operations**: For large data import/export, asynchronous mode is recommended

## Reference Documentation

- **Configuration file format**: `references/ges_env.csv.example` — Environment configuration file template and field descriptions
- **Graph database format**: https://support.huaweicloud.com/usermanual-ges/ges_01_0153.html — Detailed GES graph data format documentation
- **How to access the business API**: https://support.huaweicloud.com/api-ges/ges_03_0112.html — GES business API access guide