import time
from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk
import json
import redis
import requests
from itertools import islice
import datetime

es = Elasticsearch(
    hosts='elasticsearch-o-0080ceq4wig5.escloud.ivolces.com',
    http_auth=('admin', 'Aiyingli123'),

    timeout=30,
    max_retries=3,
    retry_on_timeout=True  # 超时后是否重试
)

redis_client = redis.Redis(host='172.16.0.24', port=6379, password='aiyingli@@123', socket_timeout=10,
                           decode_responses=True, db=0)


def bulk_update(request_list):
    if not request_list or len(request_list) == 0:
        return
    bulk_actions = []
    for update_request in request_list:
        print(update_request)
        action = {
            "_op_type": "update",
            "_index": update_request["index"],
            "_id": update_request["id"],
            "_type": update_request["type"],
            "doc": update_request.get("doc", {}),
            "doc_as_upsert": True
        }
        if "script" in update_request:
            action["script"] = update_request["script"]
            del action["doc"]
        bulk_actions.append(action)

    try:
        success, failed = bulk(
            client=es,
            actions=bulk_actions,
            chunk_size=1000,
            raise_on_error=False
        )
        print(f" 批量更新完成 | 成功: {success} | 失败: {len(failed)}")
        if failed:
            print(f"失败详情: {json.dumps(failed[:5], ensure_ascii=False)}")
    except Exception as e:
        import traceback
        traceback.print_exc()


def get_timestamp(timestamp):
    dt = datetime.datetime.fromtimestamp(int(timestamp) / 1000).strftime("%Y%m")
    return dt


def split_by_20_itertools(data):
    iterator = iter(data)
    result = []
    while True:
        chunk = list(islice(iterator, 20))
        if not chunk:
            break
        result.append(chunk)
    return result


def check_open(user_ids):
    url = f"http://172.16.1.37:8873/getOpenLive?user_ids={user_ids}"

    payload = {}
    headers = {}

    response = requests.request("GET", url, headers=headers, data=payload)
    try:
        json_data = response.json().get('data')
        return json.loads(json_data).get('data')[0].get('user_live')
    except Exception as e:
        print(e)


def get_open_live():
    grouped = split_by_20_itertools(redis_client.smembers('third_hx_user_id'))

    user_ids_list = []
    for i, group in enumerate(grouped):
        user_ids_list.append(','.join(str(num) for num in group))
    result_list = []
    for user_ids in user_ids_list:
        result_list.append(check_open(user_ids))
    live_list = []
    for result in result_list:
        if len(result) > 0:
            for re in result:
                live_list.append(f"{re.get('room_id')}:::{re.get('user_id')}")
    print(f"扫描到{len(live_list)}场直播")
    redis_client.sadd('HX:THIRD:LIVES', *live_list)


def get_live_info(live_id, user_id):
    url = f"http://172.16.1.37:8873/getLiveInfo?live_id={live_id}&user_id={user_id}"

    payload = {}
    headers = {}

    response = requests.request("GET", url, headers=headers, data=payload)
    json_data = response.json()
    return json_data.get('data')


def search_live_base():
    result_data_list = []

    live_set = redis_client.smembers('HX:THIRD:LIVES')
    if not live_set:
        return
    finish_live = []
    for i in live_set:
        print(f"直播信息{i}获取中")
        live_id = i.split(':::')[0]
        user_id = i.split(':::')[1]
        result_data = get_live_info(live_id, user_id)
        if result_data.get('is_finish') == True:
            finish_live.append(i)
            result_data_list.append({
                "index": f"tiktok_user_live_base_data_kgj_{get_timestamp(result_data.get('create_time'))}",
                "type": "es_user_live_base_data",
                "id": result_data.get('live_id'),
                "doc": {
                    "liveId": result_data.get('live_id'),
                    "cover": result_data.get('cover'),
                    "title": result_data.get('title'),
                    "totalUser": result_data.get('total_user'),
                    "totalUser2": result_data.get('total_user'),
                    "createTime": int(result_data.get('create_time')),
                    "finishTime": int(result_data.get('finish_time')),
                    "duringTime": int(result_data.get('finish_time')) - int(result_data.get('create_time')),
                    "shareUrl": result_data.get('share_url'),
                    "userId": result_data.get('user_id'),
                    "nickname": result_data.get('nickname'),
                    "avatarLarger": result_data.get('avatar_larger'),
                    "isFinish": result_data.get('is_finish'),
                    "followerCount": int(result_data.get('follower_count')),
                },
                "doc_as_upsert": True
            })
        elif result_data.get('is_finish') ==False and result_data.get('create_time') !='0':
            result_data_list.append({
                "index": f"tiktok_user_live_base_data_kgj_{get_timestamp(result_data.get('create_time'))}",
                "type": "es_user_live_base_data",
                "id": result_data.get('live_id'),
                "doc": {
                    "liveId": result_data.get('live_id'),
                    "cover": result_data.get('cover'),
                    "title": result_data.get('title'),
                    "totalUser": result_data.get('total_user'),
                    "totalUser2": result_data.get('total_user'),
                    "maxUserCount": int(result_data.get('max_user_count')),
                    "createTime": int(result_data.get('create_time')),
                    "finishTime": int(result_data.get('finish_time')),
                    "duringTime": int(result_data.get('finish_time')) - int(result_data.get('create_time')),
                    "shareUrl": result_data.get('share_url'),
                    "userId": result_data.get('user_id'),
                    "nickname": result_data.get('nickname'),
                    "avatarLarger": result_data.get('avatar_larger'),
                    "isFinish": result_data.get('is_finish'),
                    "followerCount": int(result_data.get('follower_count')),
                },
                "doc_as_upsert": True
            })
        else:
            finish_live.append(i)
    redis_client.srem('HX:THIRD:LIVES', *finish_live)
    bulk_update(result_data_list)


if __name__ == '__main__':

    while True:
        try:
            get_open_live()
            search_live_base()
            time.sleep(300)
        except Exception as e:
            print(e)
