package org.fanli.elastic;

import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import org.apache.http.HttpHost;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.support.ActiveShardCount;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.get.GetResult;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptType;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;

public class UpdateUtils {

	private static String clusterName = "my-application";
	private static String host = "192.168.1.200";
	private static Integer port = 9200;

	// 相当于数据库名称
	public static String indexName = "shose";

	// 初始化api客户端
	public static RestHighLevelClient client = new RestHighLevelClient(
			RestClient.builder(new HttpHost(host, port, "http")));

	public static void update1() {
		// Index 、 Document id
		UpdateRequest request = new UpdateRequest("posts", "1");

		// 1、Script 更新
		Map<String, Object> parameters = new HashMap<String, Object>();
		parameters.put("count", 4);
		Script inline = new Script(ScriptType.INLINE, "painless", "ctx._source.field += params.count", parameters);
		request.script(inline);

		// 2、Script 更新
		Script stored = new Script(ScriptType.STORED, null, "increment-field", parameters);
		request.script(stored);

		// 1、部分文档源String以JSON格式提供
		String jsonString = "{" + "\"updated\":\"2017-01-01\"," + "\"reason\":\"daily update\"" + "}";
		request.doc(jsonString, XContentType.JSON);

		// Map
		Map<String, Object> jsonMap = new HashMap<>();
		jsonMap.put("updated", new Date());
		jsonMap.put("reason", "daily update");
		UpdateRequest request2 = new UpdateRequest("posts", "1").doc(jsonMap);

		//
		try {
			XContentBuilder builder = XContentFactory.jsonBuilder();
			builder.startObject();
			{
				builder.timeField("updated", new Date());
				builder.field("reason", "daily update");
			}
			builder.endObject();
			UpdateRequest request3 = new UpdateRequest("posts", "1").doc(builder);
		} catch (Exception e) {
			e.printStackTrace();
		}

		UpdateRequest request4 = new UpdateRequest("posts", "1").doc("updated", new Date(), "reason", "daily update");

		String jsonString2 = "{\"created\":\"2017-01-01\"}";
		request.upsert(jsonString2, XContentType.JSON);

		// 为特定字段配置源包含
		String[] includes = new String[] { "updated", "r*" };
		String[] excludes = Strings.EMPTY_ARRAY;
		request.fetchSource(new FetchSourceContext(true, includes, excludes));

		// 为特定字段配置源排除
		String[] includes2 = Strings.EMPTY_ARRAY;
		String[] excludes2 = new String[] { "updated" };
		request.fetchSource(new FetchSourceContext(true, includes2, excludes2));

		request.setIfSeqNo(2L);
		request.setIfPrimaryTerm(1L);

		// 禁用noop检测
		request.detectNoop(false);

		// 指示脚本必须运行，而不管文档是否存在，即，脚本将负责创建尚不存在的文档。
		request.scriptedUpsert(true);

		// 指示部分文档（如果尚不存在）必须用作upsert文档。
		request.docAsUpsert(true);

		// 设置在继续更新操作之前必须处于活动状态的分片副本数
		request.waitForActiveShards(2);
		// 作为提供碎片拷贝数ActiveShardCount：可以是ActiveShardCount.ALL，
		// ActiveShardCount.ONE或ActiveShardCount.DEFAULT（默认）
		request.waitForActiveShards(ActiveShardCount.ALL);
	}

	public static void update2() {

		// 添加列
		UpdateRequest request = new UpdateRequest("posts", "1");
		String jsonString = "{" + "\"updated\":\"2017-01-01\"," + "\"reason\":\"daily update\"" + "}";
		request.doc(jsonString, XContentType.JSON);
		try {
			UpdateResponse updateResponse = client.update(request, RequestOptions.DEFAULT);

//			
//			// 返回的值UpdateResponse允许检索有关已执行操作的信息
//			String index = updateResponse.getIndex();
//			String id = updateResponse.getId();
//			long version = updateResponse.getVersion();
//			if (updateResponse.getResult() == DocWriteResponse.Result.CREATED) {
//				// 处理首次创建文档的情况（upsert）
//			} else if (updateResponse.getResult() == DocWriteResponse.Result.UPDATED) {
//				// 处理文档更新的情况
//			} else if (updateResponse.getResult() == DocWriteResponse.Result.DELETED) {
//				// 处理文件被删除的情况
//			} else if (updateResponse.getResult() == DocWriteResponse.Result.NOOP) {
//				// 处理文档不受更新影响的情况，即未对文档执行任何操作（空转）
//			}
//
//			// UpdateRequest 通过fetchSource方法启用了源检索后，响应将包含更新文档的源：
//			GetResult result = updateResponse.getGetResult(); // 检索更新的文档为 GetResult
//			if (result.isExists()) {
//				// 检索更新文档的来源为 String
//				String sourceAsString = result.sourceAsString();
//				// 检索更新文档的来源为 Map<String, Object>
//				Map<String, Object> sourceAsMap = result.sourceAsMap();
//				// 检索更新文档的来源为 byte[]
//				byte[] sourceAsBytes = result.source();
//			} else {
//				// 处理响应中不存在文档源的情况（默认情况下就是这种情况）
//			}
//
//			// 可以检查分片故障：
//			ReplicationResponse.ShardInfo shardInfo = updateResponse.getShardInfo();
//			if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
//				// 处理成功分片数量少于总分片数量的情况
//			}
//			if (shardInfo.getFailed() > 0) {
//				for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
//					String reason = failure.reason();
//				}
//				// 处理潜在的故障
//			}
//
//			// 当UpdateRequest抵靠不存在一个文件执行时，响应具有404状态码，ElasticsearchException获取引发这就需要如下进行处理：
//			UpdateRequest request2 = new UpdateRequest("posts", "does_not_exist").doc("field", "value");
//			try {
//				UpdateResponse updateResponse2 = client.update(request2, RequestOptions.DEFAULT);
//			} catch (ElasticsearchException e) {
//				if (e.status() == RestStatus.NOT_FOUND) {
//					// 处理由于文档不存在而引发的异常
//				}
//			}
//
//			// 如果存在版本冲突，ElasticsearchException将引发：
//			UpdateRequest request3 = new UpdateRequest("posts", "1").doc("field", "value").setIfSeqNo(101L)
//					.setIfPrimaryTerm(200L);
//			try {
//				UpdateResponse updateResponse3 = client.update(request3, RequestOptions.DEFAULT);
//			} catch (ElasticsearchException e) {
//				if (e.status() == RestStatus.CONFLICT) {
//					// 	引发的异常表示返回了版本冲突错误。
//				}
//			}

		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public static void update3() {
		UpdateRequest request = new UpdateRequest("posts", "1");
		String jsonString = "{" + "\"updated\":\"2019-09-09\"," + "\"reason\":\"daily update9\"" + "}";
		request.doc(jsonString, XContentType.JSON);
		try {
			UpdateResponse updateResponse = client.update(request, RequestOptions.DEFAULT);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	
	public static void Multupdate1() {
		// 批量API仅支持以JSON或SMILE编码的文档。提供任何其他格式的文档将导致错误。
		BulkRequest request = new BulkRequest();
		request.add(new IndexRequest("posts").id("1").source(XContentType.JSON, "field", "foo"));
		request.add(new IndexRequest("posts").id("2").source(XContentType.JSON, "field", "bar"));
		request.add(new IndexRequest("posts").id("3").source(XContentType.JSON, "field", "baz"));

		BulkRequest request2 = new BulkRequest();
		request2.add(new DeleteRequest("posts", "3"));
		request2.add(new UpdateRequest("posts", "2").doc(XContentType.JSON, "other", "test"));
		request2.add(new IndexRequest("posts").id("4").source(XContentType.JSON, "field", "baz"));

		// 可选参数
		request.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
		request.setRefreshPolicy("wait_for");

		request.pipeline("pipelineId");
		request.routing("routingId");
		BulkRequest defaulted = new BulkRequest("posts");

		// 开始执行
		try {
			BulkResponse bulkResponse = client.bulk(request, RequestOptions.DEFAULT);

			for (BulkItemResponse bulkItemResponse : bulkResponse) {
				DocWriteResponse itemResponse = bulkItemResponse.getResponse();

				switch (bulkItemResponse.getOpType()) {
				case INDEX:
					// 检索操作的响应（成功与否），可以是 IndexResponse，UpdateResponse或DeleteResponse可全部被视为
					// DocWriteResponse实例
				case CREATE:
					// 处理更新操作的响应
					IndexResponse indexResponse = (IndexResponse) itemResponse;
					break;
				case UPDATE:
					UpdateResponse updateResponse = (UpdateResponse) itemResponse;
					// 处理更新操作的响应
					break;
				case DELETE:
					// 处理删除操作的响应
					DeleteResponse deleteResponse = (DeleteResponse) itemResponse;
				}
			}

			if (bulkResponse.hasFailures()) {
				// 批量响应提供了一种方法来快速检查一个或多个操作是否失败：
			}

			// 这种情况下，有必要遍历所有操作结果，以检查操作是否失败，如果失败，则检索相应的失败：
			for (BulkItemResponse bulkItemResponse : bulkResponse) {
				if (bulkItemResponse.isFailed()) {
					BulkItemResponse.Failure failure = bulkItemResponse.getFailure();
				}
			}

			BulkProcessor.Listener listener = new BulkProcessor.Listener() {
				@Override
				public void beforeBulk(long executionId, BulkRequest request) {

				}

				@Override
				public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {

				}

				@Override
				public void afterBulk(long executionId, BulkRequest request, Throwable failure) {

				}
			};

		} catch (IOException e) {
			e.printStackTrace();
		}
	}


	public static void main(String[] args) {
		update3();
		System.out.println("------------ 测试结束 -------------------------");
	}

}