import redis
from flask import Flask, request, jsonify
from datetime import datetime, timedelta
import json
import os,sys
import io
import gzip
base_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(base_path)
import kafka_utils

app = Flask(__name__)

# 初始化Redis连接
r = redis.Redis(host='172.16.0.24', port=6379, db=0, password="aiyingli@@123",decode_responses=True)

DY_TASK_SET_KEY = "dso_monitor_keyword_rank_third"
DY_TASK_CHANNEL_COUNT_PREFIX = "dy:task:channel:count:"  # 取任务记录
DY_TASK_CHANNEL_COUNT_PREFIX_FAILED = "dy:task:channel:count:failed"   # 失败记录


def get_today_date_str():
    """获取今天的日期字符串，用于统计"""
    return datetime.now().strftime("%Y%m%d")



@app.route('/api/dy/task/get', methods=['GET'])
def get_dy_task():
    channel = request.args.get('channel')
    count_str = request.args.get('count', '1')


    if not channel:
        return jsonify({"status": "error", "message": "渠道参数不能为空"}), 400

    try:
        count = int(count_str)
        if count <= 0:
            raise ValueError("count必须为正整数")
    except ValueError:
        return jsonify({"status": "error", "message": "count参数必须是正整数"}), 400

    tasks = r.spop(DY_TASK_SET_KEY, count)
    if not tasks:
        return jsonify({"status": "empty", "message": "没有可用任务"}), 200

    for task in tasks:
        task_channel_key = f"{DY_TASK_CHANNEL_COUNT_PREFIX}{get_today_date_str()}"
        r.hincrby(task_channel_key, channel, 1)
        r.expire(task_channel_key, 60 * 60 * 24 * 365)

    actual_count = len(tasks)
    return jsonify({
        "status": "success",
        "tasks": tasks,
        "actual_count": actual_count,
        "requested_count": count
    }), 200


@app.route('/api/dy/task/submit', methods=['POST'])
def submit_dy_data():
    try:
        # 判断请求是否为gzip压缩
        if request.headers.get('Content-Encoding') == 'gzip':
            # 读取压缩的请求体并解压缩
            gzip_buffer = io.BytesIO(request.get_data())
            with gzip.GzipFile(fileobj=gzip_buffer, mode='rb') as f:
                decompressed_data = f.read()
            # 将解压缩后的字节数据解析为JSON
            data = json.loads(decompressed_data)
        else:
            # 非压缩请求，正常解析JSON
            data = request.get_json()
    except gzip.BadGzipFile:
        # 处理无效的gzip压缩数据
        return jsonify({
            'success': False,
            'message': '无效的gzip压缩数据'
        }), 400
    except Exception as e:
        # 处理JSON解析失败（包括压缩/非压缩）
        return jsonify({
            'success': False,
            'message': f'JSON解析失败: {str(e)}'
        }), 400

    # 验证请求数据是否为数组
    if not isinstance(data, list):
        return jsonify({
            'success': False,
            'message': '请求参数必须是数组'
        }), 400

    # 定义必须包含的字段
    required_fields = [
        'key_word',
        'rank',
        'video_id',
        'update_pt',
        'create_pt',
        'channel'
    ]

    # 验证每个对象是否包含所有必填字段
    for index, item in enumerate(data):
        if not isinstance(item, dict):
            return jsonify({
                'success': False,
                'message': f'数组中第 {index + 1} 个元素必须是对象'
            }), 400

        missing_fields = [field for field in required_fields if field not in item]
        if missing_fields:
            return jsonify({
                'success': False,
                'message': f'数组中第 {index + 1} 个元素缺少必填字段: {", ".join(missing_fields)}'
            }), 400
    topic = 'dy_aidso_monitor'
    kafka_utils.producer_message(topic,3,data)
    return jsonify({
        'success': True,
        'message': '数据提交成功',
        'received_count': len(data)
    }), 200


@app.route('/api/dy/task/retry', methods=['POST'])
def retry_dy_failed_tasks():
    try:
        try:
            data = request.get_json()
        except json.JSONDecodeError:
            return jsonify({
                "success": False,
                "message": "请求必须是有效的JSON格式"
            }), 400

        if not data:
            return jsonify({
                "success": False,
                "message": "请求体不能为空"
            }), 400

        required_fields = ["keywords", "channel"]
        missing_fields = [field for field in required_fields if field not in data]
        if missing_fields:
            return jsonify({
                "success": False,
                "message": f"缺少必要字段: {', '.join(missing_fields)}",
                "required_fields": required_fields
            }), 400

        # 4. 验证keywords是否为列表类型
        if not isinstance(data["keywords"], list):
            return jsonify({
                "success": False,
                "message": "keywords必须是列表类型（例如: [1, 2, 3]）"
            }), 400

        items = data.get("keywords")
        channel = data.get("channel")

        r.sadd(DY_TASK_SET_KEY, *items)
        item_count = len(items)

        for task in items:
            task_channel_key = f"{DY_TASK_CHANNEL_COUNT_PREFIX_FAILED}{get_today_date_str()}"
            r.hincrby(task_channel_key, channel, 1)
            r.expire(task_channel_key, 60 * 60 * 24 * 365)

        return jsonify({
            "success": True,
            "message": f"成功接收字符串列表，共{item_count}个元素",
        }), 200

    except Exception as e:
        return jsonify({
            "success": False,
            "message": f"服务器处理错误: {str(e)}"
        }), 500

@app.route('/api/dy/result/search', methods=['GET'])
def get_dy_channel_stats():
    channel = request.args.get('channel')
    all_task = r.scard(DY_TASK_SET_KEY)

    today_date = get_today_date_str()

    failed_key = f"{DY_TASK_CHANNEL_COUNT_PREFIX_FAILED}{today_date}"
    success_key = f"{DY_TASK_CHANNEL_COUNT_PREFIX}{today_date}"


    if channel is None:
        success_data = r.hgetall(success_key)
        failed_data = r.hgetall(failed_key)
    else:
        success_data = r.hget(success_key, channel)
        failed_data = r.hget(failed_key, channel)

    return jsonify({
        "status": "success",
        "data": {
            "success": success_data,
            "failed": failed_data,
            "all_task": all_task
        }
    }), 200


# Redis键名定义
XHS_TASK_SET_KEY = "xhs:task:all"
XHS_TASK_CHANNEL_COUNT_PREFIX = "xhs:task:channel:count:"  # 取任务记录
XHS_TASK_CHANNEL_COUNT_PREFIX_FAILED = "xhs:task:channel:count:failed"   # 失败记录

@app.route('/api/xhs/task/get', methods=['GET'])
def get_xhs_task():
    channel = request.args.get('channel')
    count_str = request.args.get('count', '1')

    if not channel:
        return jsonify({"status": "error", "message": "渠道参数不能为空"}), 400

    try:
        count = int(count_str)
        if count <= 0:
            raise ValueError("count必须为正整数")
    except ValueError:
        return jsonify({"status": "error", "message": "count参数必须是正整数"}), 400

    tasks = r.spop(XHS_TASK_SET_KEY, count)
    if not tasks:
        return jsonify({"status": "empty", "message": "没有可用任务"}), 200

    for task in tasks:
        task_channel_key = f"{XHS_TASK_CHANNEL_COUNT_PREFIX}{get_today_date_str()}"
        r.hincrby(task_channel_key, channel, 1)
        r.expire(task_channel_key, 60 * 60 * 24 * 365)

    actual_count = len(tasks)
    return jsonify({
        "status": "success",
        "tasks": tasks,
        "actual_count": actual_count,
        "requested_count": count
    }), 200


@app.route('/api/xhs/task/submit', methods=['POST'])
def submit_xhs_data():
    # 获取请求数据
    data = request.get_json()

    # 验证请求数据是否为数组
    if not isinstance(data, list):
        return jsonify({
            'success': False,
            'message': '请求参数必须是数组'
        }), 400

    # 定义必须包含的字段
    required_fields = [
        'key_word',
        'rank',
        'video_id',
        'update_pt',
        'create_pt',
        'channel'
    ]

    # 验证每个对象是否包含所有必填字段
    for index, item in enumerate(data):
        if not isinstance(item, dict):
            return jsonify({
                'success': False,
                'message': f'数组中第 {index + 1} 个元素必须是对象'
            }), 400

        missing_fields = [field for field in required_fields if field not in item]
        if missing_fields:
            return jsonify({
                'success': False,
                'message': f'数组中第 {index + 1} 个元素缺少必填字段: {", ".join(missing_fields)}'
            }), 400
    topic = 'xhs_aidso_monitor'
    kafka_utils.producer_message(topic,3,data)
    return jsonify({
        'success': True,
        'message': '数据提交成功',
        'received_count': len(data)
    }), 200


@app.route('/api/xhs/task/retry', methods=['POST'])
def retry_xhs_failed_tasks():
    try:
        try:
            data = request.get_json()
        except json.JSONDecodeError:
            return jsonify({
                "success": False,
                "message": "请求必须是有效的JSON格式"
            }), 400

        if not data:
            return jsonify({
                "success": False,
                "message": "请求体不能为空"
            }), 400

        required_fields = ["keywords", "channel"]
        missing_fields = [field for field in required_fields if field not in data]
        if missing_fields:
            return jsonify({
                "success": False,
                "message": f"缺少必要字段: {', '.join(missing_fields)}",
                "required_fields": required_fields
            }), 400

        # 4. 验证keywords是否为列表类型
        if not isinstance(data["keywords"], list):
            return jsonify({
                "success": False,
                "message": "keywords必须是列表类型（例如: [1, 2, 3]）"
            }), 400

        items = data.get("keywords")
        channel = data.get("channel")

        r.sadd(DY_TASK_SET_KEY, *items)
        item_count = len(items)

        for task in items:
            task_channel_key = f"{XHS_TASK_CHANNEL_COUNT_PREFIX_FAILED}{get_today_date_str()}"
            r.hincrby(task_channel_key, channel, 1)
            r.expire(task_channel_key, 60 * 60 * 24 * 365)

        return jsonify({
            "success": True,
            "message": f"成功接收字符串列表，共{item_count}个元素",
        }), 200

    except Exception as e:
        return jsonify({
            "success": False,
            "message": f"服务器处理错误: {str(e)}"
        }), 500

@app.route('/api/xhs/result/search', methods=['GET'])
def get_xhs_channel_stats():
    channel = request.args.get('channel')
    all_task = r.scard(DY_TASK_SET_KEY)

    today_date = get_today_date_str()

    failed_key = f"{XHS_TASK_CHANNEL_COUNT_PREFIX_FAILED}{today_date}"
    success_key = f"{XHS_TASK_CHANNEL_COUNT_PREFIX}{today_date}"


    if channel is None:
        success_data = r.hgetall(success_key)
        failed_data = r.hgetall(failed_key)
    else:
        success_data = r.hget(success_key, channel)
        failed_data = r.hget(failed_key, channel)

    return jsonify({
        "status": "success",
        "data": {
            "success": success_data,
            "failed": failed_data,
            "all_task": all_task
        }
    }), 200



if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8085,debug=True)
