from flask import jsonify, request,Blueprint
import os, sys
import json
import queue
from concurrent.futures import ThreadPoolExecutor
from loguru import logger
BASE_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

sys.path.append(BASE_DIR)
from aidso_geo.models.process import process_call_back, commit_task, main_process
from aidso_geo.config.base_config import init_redis, init_redis8
from aidso_geo.utils import bh_utils, tos_utils, url_utils
line_app = Blueprint("line_app", __name__)
redis_client = init_redis()
redis_client8 = init_redis8()
MAX_WORKERS = 30
MAX_PENDING_TASKS = 200

executor = ThreadPoolExecutor(max_workers=MAX_WORKERS)
pending_queue = queue.Queue(maxsize=MAX_PENDING_TASKS)


def submit_background_task(func, *args, **kwargs) -> bool:
    try:
        pending_queue.put_nowait(1)
    except queue.Full:
        return False

    def _run():
        try:
            func(*args, **kwargs)
        finally:
            pending_queue.get()
            pending_queue.task_done()

    executor.submit(_run)
    return True


def _cache_get_json(key: str):
    v = redis_client.get(key)
    if not v:
        return None
    try:
        return json.loads(v)
    except Exception:
        # 缓存脏数据直接清掉
        redis_client.delete(key)
        return None


def _cache_set_json(key: str, obj: dict, ttl: int):
    redis_client.set(key, json.dumps(obj, ensure_ascii=False), ex=ttl)


@line_app.route('/api/geo/task_check', methods=['GET'])
def task_check():
    req_id = request.args.get('reqId')
    if not req_id:
        return jsonify({
            "code": 400,
            "status": "参数错误：reqId 不能为空",
            "data": {}
        })

    resp_cache_key = f"geo:task_check:resp:{req_id}"
    cached_resp = _cache_get_json(resp_cache_key)
    if cached_resp:
        return jsonify(cached_resp)

    try:
        data = bh_utils.query_data(f"select * from geo_commit_task where reqId ='{req_id}'")
    except Exception as e:
        return jsonify({
            "code": 500,
            "status": f"查询失败：{str(e)}",
            "data": {}
        })

    if not data:
        return jsonify({
            "code": 400,
            "msg": "reqId 未提交",
            "data": {}
        })
    task_info = data[0]
    status = task_info['status']
    logger.success(f"{task_info['reqId']}--{task_info['platform']}--{task_info['prompt']}---{status}")
    if status in ('ING', 'PROCESSING','QUEUE','PROCESS_FAIL','FAIL'):
        resp = {
            "code": 200,
            "msg": "success",
            "data": {
                "status": 'ING',
                "result": {}
            }
        }
        _cache_set_json(resp_cache_key, resp, 10)
        return jsonify(resp)

    elif status == 'SUCCESS':
        try:
            result_path = f'geo/{task_info["taskId"]}/{task_info["platform"]}/result.json'
            result_content = tos_utils.get_string_from_tos(result_path)
            result_data = json.loads(result_content)
            return jsonify({
                "code": 200,
                "msg": "success",
                "data": {
                    "status": 'success',
                    "result": result_data
                }
            })
        except Exception as e:
            return jsonify({
                "code": 500,
                "status": f"获取结果失败：{str(e)}",
                "data": {}
            })
    else:
        resp = {
            "code": 200,
            "msg": "success",
            "data": {
                "status": 'ING',
                "result": {}
            }
        }
        _cache_set_json(resp_cache_key, resp, 10)
        return jsonify(resp)


@line_app.route('/api/geo/task_commit', methods=['POST'])
def task_commit():
    req_id = ""

    try:
        data = request.get_json()
        if not data:
            return jsonify({
                "code": 400,
                "msg": "请求数据为空，请检查请求体",
                "reqId": req_id
            })

        req_id = data.get("reqId", "")
        type = data.get('type',"")
        platform = data.get('platform',"")
        prompt = data.get('prompt',"")
        required_fields = ["prompt", "taskId", "reqId", "platform", "type"]
        missing_fields = [field for field in required_fields if field not in data]
        if missing_fields:
            return jsonify({
                "code": 400,
                "status": f"缺少必要字段：{', '.join(missing_fields)}",
                "reqId": req_id
            })
        if type == 'stream':
            insert_ok = commit_task(data, 'ING')
            ok = submit_background_task(main_process, data)
            if insert_ok and ok:
                return jsonify({
                    "code": 200,
                    "msg": 'success',
                    "reqId": req_id
                })
        else:
            logger.success(f"{data['reqId']}--{platform}--{prompt}--任务提交--{type}")
            ret = redis_client.lpush("geo:task_commit:list",json.dumps(data, ensure_ascii=False))
            if ret and ret > 0:
                resp_cache_key = f"geo:task_check:resp:{req_id}"
                resp = {
                    "code": 200,
                    "msg": "success",
                    "data": {
                        "status": 'ING',
                        "result": {}
                    }
                }
                if type == 'stream_batch':
                    _cache_set_json(resp_cache_key, resp, 600)
                else:
                    _cache_set_json(resp_cache_key, resp, 18000)
                return jsonify({
                    "code": 200,
                    "msg": "任务已提交",
                    "reqId": req_id
                })

        return jsonify({
            "code": 503,
            "msg": "服务器繁忙，请稍后重试",
            "reqId": req_id
        })

    except Exception as e:
        return jsonify({
            "code": 500,
            "msg": f"服务器内部错误：{str(e)}",
            "reqId": req_id
        })


@line_app.route('/api/geo/call_back', methods=['POST'])
def data_call_back():
    req_id = ""
    try:
        data = request.get_json()
        task_data = data.get("task_data")
        result = data.get("result")
        req_id = task_data.get('reqId')
        platform = task_data.get('platform')
        logger.success(f"{req_id}--{platform}-------CALL_BACK")
        # ret = redis_client8.lpush("geo:call_back:list",json.dumps(data, ensure_ascii=False))
        #
        # if ret and ret > 0:
        #     return jsonify({
        #         "code": 200,
        #         "msg": "任务已提交",
        #         "reqId": req_id
        #     })
        # else:
        #     return jsonify({
        #         "code": 400,
        #         "msg": "提交失败",
        #         "reqId": req_id
        #     })
        #
        ok = submit_background_task(process_call_back, task_data, result)
        if not ok:
            return jsonify({
                "code": 503,
                "msg": "服务器繁忙，请稍后重试",
                "reqId": req_id
            })
        return jsonify({
            "code": 200,
            "msg": "任务已提交",
            "reqId": req_id
        })
    except Exception as e:
        return jsonify({
            "code": 500,
            "msg": f"服务器内部错误：{str(e)}",
            "reqId": req_id
        })

@line_app.route('/api/geo/check_quto', methods=['GET'])
def check_quto():
    req_id = ""
    try:
        req_id = request.args.get('reqId')
        if not req_id:
            return jsonify({
                "code": 400,
                "msg": "参数错误：reqId 不能为空",
                "data": {}
            })
        try:
            data = bh_utils.query_data(f"select * from geo_commit_task where reqId ='{req_id}'")
        except Exception as e:
            return jsonify({
                "code": 500,
                "msg": f"查询失败：{str(e)}",
                "data": {}
            })

        if not data:
            return jsonify({
                "code": 400,
                "msg": "reqId 未提交",
                "data": {}
            })
        task_info = data[0]
        taskId = task_info["taskId"]
        platform = task_info["platform"]
        quote_path = f'geo/{task_info["taskId"]}/{task_info["platform"]}/quote.txt'
        result_path = f'geo/{task_info["taskId"]}/{task_info["platform"]}/result.json'

        result_content = tos_utils.get_string_from_tos(result_path)
        result_data = json.loads(result_content)

        if result_data.get("sources"):
            quote_content = tos_utils.get_string_from_tos(quote_path)
            quote_data = json.loads(quote_content)

            urls = []

            for c in quote_data:
                u = c.get('url')
                if isinstance(u, str) and u.startswith('http'):
                    urls.append(u)
                else:
                    urls.append(c.get('title'))

            url_ids_map = url_utils.generate_numeric_url_id_v2(urls)

            for c in quote_data:
                u = c.get('url')
                if isinstance(u, str) and u.startswith('http'):
                    key = u
                else:
                    key = c.get('title')
                c['quto_id'] = url_ids_map.get(key)

            bh_utils.insert_data("geo_quote_result_v2", quote_data)

            tos_utils.put_string_to_tos(quote_path, quote_data)

            quotes = bh_utils.query_data(
                f"select site_icon as img ,platform ,quto_id as sourceId,title,url,site_name as site_name from geo_quote_result_v2 where task_id ='{taskId}' and platform =  '{platform}'")

            result_data['sources'] = quotes
            tos_utils.put_string_to_tos(result_path,result_data)
        logger.success(f"{req_id}--check_quto-----success")
        return jsonify({
            "code": 200,
            "msg": "success",
            "data": {
                "status": 'success',
                "result": result_data
            }
        })
    except Exception as e:
        return jsonify({
            "code": 500,
            "msg": f"服务器内部错误：{str(e)}",
            "reqId": req_id
        })


@line_app.route('/api/geo/get_mention_count', methods=['POST'])
def get_mention_count():
    try:
        data = request.get_json(force=True) or {}
        req_ids = data.get('reqIds', [])

        in_sql = ",".join([f"'{i}'" for i in list(dict.fromkeys(req_ids))])

        result = bh_utils.query_data(f"""
    
        WITH t AS (
        SELECT
            req_id,
            arrayFilter(x -> x != '',
                arrayMap(x -> replaceRegexpAll(x, '(^\\s+)|(\\s+$)', ''),
                    splitByChar(',', ifNull(positive_mentions, '')))
            ) AS pos_arr,
            arrayFilter(x -> x != '',
                arrayMap(x -> replaceRegexpAll(x, '(^\\s+)|(\\s+$)', ''),
                    splitByChar(',', ifNull(negative_mentions, '')))
            ) AS neg_arr
        FROM geo_brand_mention_list
        WHERE req_id IN ({in_sql}
        )
    ),
    pos AS (
        SELECT
            req_id,
            uniqExact(word) AS positive_distinct,
            count() AS positive_total
        FROM (
            SELECT req_id, arrayJoin(pos_arr) AS word
            FROM t
        )
        GROUP BY req_id
    ),
    neg AS (
        SELECT
            req_id,
            uniqExact(word) AS negative_distinct,
            count() AS negative_total
        FROM (
            SELECT req_id, arrayJoin(neg_arr) AS word
            FROM t
        )
        GROUP BY req_id
    )
    SELECT
        a.req_id,
        ifNull(p.positive_distinct, 0) AS positive_word_count,
        ifNull(p.positive_total, 0) AS positive_mention_count,
        ifNull(n.negative_distinct, 0) AS negative_word_count,
        ifNull(n.negative_total, 0) AS negative_mention_count
    FROM
        (SELECT DISTINCT req_id FROM t) a
        LEFT JOIN pos p ON a.req_id = p.req_id
        LEFT JOIN neg n ON a.req_id = n.req_id
    ORDER BY a.req_id;
    
    
        """)
        return jsonify({
            "code": 200,
            "msg": "success",
            "data": result or []
        })
    except Exception as e:
        return jsonify({
            "code": 400,
            "msg": "err",
            "data": []
        })
