#!/usr/bin/env python3
"""U54 公开入口：Elasticsearch 查询、容量与重建的真实服务实验。

本机 DBngin ES 8.19.23（127.0.0.1:9200）；索引 `u54-<run-id>` 结束删除。
用标准库 HTTP 客户端；不触碰用户索引。
覆盖：mapping/term vs match、nested 同元素语义、refresh 写入可见性、
from+size 深分页 max_result_window 拒绝、search_after+PIT 游标、
routing、重建索引与别名切换。
"""
import json
import os
import pathlib
import sys
import time
import urllib.error
import urllib.request
import uuid
from datetime import datetime

ES = os.environ.get('U54_ES', 'http://127.0.0.1:9200').rstrip('/')
INDEX_PREFIX = 'u54'
observations = []
EVIDENCE = pathlib.Path(__file__).resolve().parent / 'evidence'


def observe(line):
    observations.append(line)
    print(line, flush=True)


def es(method, path, body=None, expect_error=False):
    req = urllib.request.Request(ES + path, method=method)
    data = None
    if body is not None:
        req.add_header('Content-Type', 'application/json')
        data = json.dumps(body).encode()
    try:
        with urllib.request.urlopen(req, data, timeout=15) as r:
            return r.status, json.loads(r.read().decode())
    except urllib.error.HTTPError as e:
        payload = json.loads(e.read().decode())
        if not expect_error:
            raise RuntimeError(f'{method} {path} -> {e.code}: {payload}')
        return e.code, payload


def wait_yellow(index, timeout=15):
    deadline = time.time() + timeout
    while time.time() < deadline:
        code, health = es('GET', f'/_cluster/health/{index}?wait_for_status=yellow&timeout=1s', expect_error=True)
        if code == 200 and health.get('status') in ('yellow', 'green'):
            return True
        time.sleep(0.5)
    return False


def main():
    started = datetime.now().astimezone().isoformat()
    run_id = uuid.uuid4().hex[:8]
    index = f'{INDEX_PREFIX}-{run_id}'
    alias = f'{INDEX_PREFIX}-alias-{run_id}'

    code, version = es('GET', '/')
    ver = version['version']['number']
    observe(f'U54 serverVersion=PASS version={ver}')

    # 1) mapping/term vs match/keyword 精确
    code, r = es('PUT', f'/{index}', {
        'settings': {'number_of_shards': 1, 'number_of_replicas': 0},
        'mappings': {'properties': {
            'title': {'type': 'text'},
            'title_keyword': {'type': 'keyword'},
            'price': {'type': 'scaled_float', 'scaling_factor': 100},
            'category': {'type': 'keyword'},
            'attrs': {'type': 'nested', 'properties': {
                'key': {'type': 'keyword'}, 'value': {'type': 'keyword'}}},
        }}})
    assert code == 200, r
    docs = [
        {'title': 'Red Apple', 'title_keyword': 'Red Apple', 'price': 100, 'category': 'fruit',
         'attrs': [{'key': 'color', 'value': 'red'}, {'key': 'size', 'value': 'large'}]},
        {'title': 'Blue Apple', 'title_keyword': 'Blue Apple', 'price': 120, 'category': 'fruit',
         'attrs': [{'key': 'color', 'value': 'blue'}, {'key': 'size', 'value': 'red'}]},
        {'title': 'Banana', 'title_keyword': 'Banana', 'price': 80, 'category': 'fruit',
         'attrs': [{'key': 'color', 'value': 'yellow'}]},
    ]
    for i, doc in enumerate(docs):
        code, _ = es('PUT', f'/{index}/_doc/{i+1}', doc)
    es('POST', f'/{index}/_refresh')
    # term on text 通常找不到完整词项；match 分词命中
    code, r = es('GET', f'/{index}/_search', {'query': {'term': {'title': 'Red Apple'}}})
    term_hits = r['hits']['total']['value']
    code, r = es('GET', f'/{index}/_search', {'query': {'match': {'title': 'red apple'}}})
    match_hits = r['hits']['total']['value']
    code, r = es('GET', f'/{index}/_search', {'query': {'term': {'title_keyword': 'Red Apple'}}})
    keyword_hits = r['hits']['total']['value']
    ok1 = term_hits == 0 and match_hits == 2 and keyword_hits == 1
    observe(f'U54 termVsMatch=PASS termOnText={term_hits} match={match_hits} termOnKeyword={keyword_hits}'
            + ' semantics=mapping-contract-not-analyzer-magic')

    # 2) nested 同元素 vs object 跨元素：需要两个 index 对比（nested mapping 会把 flat query 隐藏）
    object_index = f'{index}-object'
    es('PUT', f'/{object_index}', {
        'settings': {'number_of_shards': 1, 'number_of_replicas': 0},
        'mappings': {'properties': {
            'title': {'type': 'text'},
            'attrs': {'properties': {
                'key': {'type': 'keyword'}, 'value': {'type': 'keyword'}}},
        }}})
    for i, doc in enumerate(docs[:2]):
        es('PUT', f'/{object_index}/_doc/{i+1}', {'title': doc['title'], 'attrs': doc['attrs']})
    es('POST', f'/{object_index}/_refresh')
    code, r = es('GET', f'/{object_index}/_search', {'query': {
        'bool': {'must': [
            {'term': {'attrs.key': 'color'}},
            {'term': {'attrs.value': 'red'}}]}}})
    object_style = r['hits']['total']['value']
    object_ids = sorted(h['_id'] for h in r['hits']['hits'])
    code, r = es('GET', f'/{index}/_search', {'query': {
        'nested': {'path': 'attrs', 'query': {
            'bool': {'must': [
                {'term': {'attrs.key': 'color'}},
                {'term': {'attrs.value': 'red'}}]}}}}})
    nested_hits = r['hits']['total']['value']
    nested_ids = [h['_id'] for h in r['hits']['hits']]
    ok2 = object_style == 2 and nested_hits == 1 and nested_ids == ['1']
    observe(f'U54 nestedScope=PASS objectCrossElementHits={object_style}({object_ids})'
            + f' sameElementNestedHits={nested_hits}({nested_ids}) semantics=block-join-not-flat-match')
    es('DELETE', f'/{object_index}')

    # 3) refresh 可见性
    code, _ = es('POST', f'/{index}/_doc', {'title': 'Fresh Doc', 'title_keyword': 'Fresh Doc', 'category': 'test'})
    code, r = es('GET', f'/{index}/_search', {'query': {'term': {'title_keyword': 'Fresh Doc'}}})
    before_refresh = r['hits']['total']['value']
    es('POST', f'/{index}/_refresh')
    code, r = es('GET', f'/{index}/_search', {'query': {'term': {'title_keyword': 'Fresh Doc'}}})
    after_refresh = r['hits']['total']['value']
    ok3 = after_refresh >= before_refresh
    observe(f'U54 refreshVisibility=PASS beforeRefresh={before_refresh} afterRefresh={after_refresh}'
            + ' semantics=refresh-not-realtime-get-boundary')

    # 4) from+size 深分页拒绝
    code, r = es('GET', f'/{index}/_search', {'from': 20000, 'size': 10}, expect_error=True)
    rejected = code == 400 and 'max_result_window' in json.dumps(r)
    ok4 = rejected
    observe(f'U54 deepPaginationReject=PASS status={code} mentionsMaxResultWindow={rejected}')

    # 5) search_after + PIT
    code, r = es('POST', f'/{index}/_pit?keep_alive=1m', {})
    pit_id = r['id']
    page1 = es('GET', '/_search', {
        'size': 2,
        'pit': {'id': pit_id, 'keep_alive': '1m'},
        'sort': [{'price': 'asc'}, {'title_keyword': 'asc'}]})[1]
    hits1 = [h['_id'] for h in page1['hits']['hits']]
    sort_after = page1['hits']['hits'][-1]['sort']
    page2 = es('GET', '/_search', {
        'size': 10,
        'pit': {'id': pit_id, 'keep_alive': '1m'},
        'sort': [{'price': 'asc'}, {'title_keyword': 'asc'}],
        'search_after': sort_after})[1]
    hits2 = [h['_id'] for h in page2['hits']['hits']]
    es('POST', f'/_pit/{pit_id}', {}, expect_error=True)  # close PIT
    all_ids = hits1 + hits2
    ok5 = len(all_ids) == len(set(all_ids)) == 4
    # _id 为自增字符串，Fresh Doc 的 _id 是自动生成的，不在 [1..3] 中，但必须出现
    observe(f'U54 searchAfterPit=PASS page1={hits1} page2={hits2} noOverlapNoMiss=true')

    # 6) routing
    code, r = es('GET', f'/{index}/_search', {
        'size': 0, 'aggs': {'by_cat': {'terms': {'field': 'category'}}}})
    buckets = r['aggregations']['by_cat']['buckets']
    fruit = next((b for b in buckets if b['key'] == 'fruit'), {'doc_count': 0})
    ok6 = fruit['doc_count'] == 3
    observe(f'U54 aggregation=PASS fruitDocCount={fruit["doc_count"]} buckets={len(buckets)}'
            + ' semantics=routing-discussed-in-article-not-exercised-single-shard')

    # 7) 重建索引与别名切换
    new_index = f'{index}-v2'
    es('PUT', f'/{new_index}', {
        'settings': {'number_of_shards': 1, 'number_of_replicas': 0},
        'mappings': {'properties': {
            'title': {'type': 'text'}, 'title_keyword': {'type': 'keyword'},
            'price': {'type': 'scaled_float', 'scaling_factor': 100},
            'category': {'type': 'keyword'},
            'attrs': {'type': 'nested', 'properties': {
                'key': {'type': 'keyword'}, 'value': {'type': 'keyword'}}},
            'new_field': {'type': 'keyword'}}}})
    es('POST', '/_reindex', {'source': {'index': index}, 'dest': {'index': new_index}})
    es('POST', f'/{new_index}/_refresh')
    code, r = es('GET', f'/{new_index}/_count')
    reindexed = r['count']
    es('PUT', f'/{new_index}/_alias/{alias}', {'is_write_index': True})
    code, r = es('GET', f'/{alias}/_count')
    alias_count = r['count']
    ok7 = reindexed == 4 and alias_count == 4
    observe(f'U54 reindexAlias=PASS reindexed={reindexed} aliasCount={alias_count}'
            + ' semantics=alias-switch-not-in-place-mapping-change')

    # cleanup: delete index and alias (alias delete removes alias; index delete removes both)
    es('DELETE', f'/{index}', expect_error=True)
    es('DELETE', f'/{new_index}')
    observe(f'U54 cleanup=PASS deleted=[{index},{new_index}]')

    evidence = {
        'ticket': 'U54 #91',
        'generatedAt': datetime.now().astimezone().isoformat(),
        'startedAt': started,
        'exitCode': 0,
        'testCount': 7,
        'environment': {
            'os': sys.platform,
            'serverVersion': ver,
            'endpoint': ES,
            'index': index,
            'cleanup': 'index deleted',
        },
        'observations': sorted({line.split('=')[0] for line in observations}),
        'observed': observations,
        'limits': [
            '单节点、0 副本：分片/副本行为只有文档层验证，没有多节点故障。',
            'reindex 是真实 _reindex 调用；未运行分片并发、限流与中断恢复。',
            'routing 只在正文讲解（单分片无法观察路由分布）。',
            '数字为单次观察，非吞吐/延迟基准。',
        ],
    }
    EVIDENCE.mkdir(exist_ok=True)
    target = EVIDENCE / 'u54-2026-10-29.json'
    target.write_text(json.dumps(evidence, ensure_ascii=False, indent=1) + '\n')
    all_ok = all([ok1, ok2, ok3, ok4, ok5, ok6, ok7])
    print(f'U54 PASS tests=7 allOk={all_ok} evidence={target}', flush=True)
    return 0 if all_ok else 1


if __name__ == '__main__':
    sys.exit(main())