import json
import traceback

import aidso_geo.utils.bh_utils as bh_utils
import time
import os, sys
import ast
import requests
from copy import deepcopy
from datetime import datetime
from enum import Enum

import json
from concurrent.futures import ThreadPoolExecutor, as_completed

from aidso_geo.utils.tos_utils import put_string_to_tos, check_file_in_tos, get_tos_file_size
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(BASE_DIR)

# BASE_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# sys.path.append(BASE_DIR)
from aidso_geo.config.base_config import init_redis
from aidso_geo.utils.ai_utils import ai_get_brand_sentiment_and_mentions, \
    ai_get_brand_sentiment_and_mentions_by_articles, ai_get_product_list, ai_get_product_sentiment_and_mentions, \
    ai_get_product_list_search
from aidso_geo.models import doubao_data_process, deepseek_data_process, yuanbao_data_process, qianwen_data_process, \
    kimi_data_process, wenxin_data_process, baiduai_data_process, doubao_android_data_process, \
    deepseek_android_data_process, douyinai_data_process, qianwen_android_data_process, yuanbao_android_data_process, \
    xiaohongshu_android_data_process
from aidso_geo.utils import ai_utils, spider_interface, ai_interface, url_utils, robot_utils
from aidso_geo.utils import tos_utils
from loguru import logger

redis_client = init_redis()


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


def commit_task(data, process):
    try:
        data["insertime"] = int(time.time())
        data["status"] = process
        bh_utils.insert_data('geo_commit_task', [data])
        # bh_utils.insert_data('geo_commit_task_log', [data])
        return True
    except Exception as e:
        return False


def get_keyword_ranks(text, keywords):
    if isinstance(keywords, str):
        keywords = [keywords]

    matches = []

    for key in keywords:
        start = 0
        while True:
            pos = text.find(key, start)
            if pos == -1:
                break

            matches.append({
                "word": key,
                "pos": pos
            })

            # 允许继续查找后面的相同关键词
            start = pos + 1

    # 按出现位置排序；同位置时短词优先
    matches.sort(key=lambda x: (x["pos"], len(x["word"])))

    result = []
    for key in keywords:
        rank_list = [
            i + 1
            for i, item in enumerate(matches)
            if item["word"] == key
        ]
        result.append({
            "word": key,
            "rank_list": rank_list
        })

    return result


def calculate_rank_min_and_count_sum(data_list):
    positive_ranks = [item['rank'] for item in data_list if item['rank'] > 0]

    counts = [item['count'] for item in data_list]

    min_rank = min(positive_ranks) if positive_ranks else 0

    count_sum = sum(counts)

    return (min_rank, count_sum)


def process_com_map(data):
    if data.get("comWordsMap") is not None:
        com_words_map = data.get("comWordsMap")

        # 如果 comWordsMap 是字符串，解析成 list
        if isinstance(com_words_map, str):
            try:
                com_words_map = json.loads(com_words_map)
            except Exception:
                com_words_map = ast.literal_eval(com_words_map)

        # 兜底：保证是 list
        if not isinstance(com_words_map, list):
            com_words_map = []

        # 再兜底：过滤掉脏数据，避免里面混入 str
        com_words_map = [x for x in com_words_map if isinstance(x, dict)]

        # 写回 data，后面所有逻辑都用标准结构
        data["comWordsMap"] = com_words_map

    return data


def keyword_map_brand(keyword_list):
    def sql_quote(s: str) -> str:
        # 最基础的单引号转义
        return "'" + s.replace("\\", "\\\\").replace("'", "\\'") + "'"

    brand_map = []
    in_list = ",".join(sql_quote(x) for x in keyword_list)
    query_result = []
    if in_list:
        query_result = bh_utils.query_data(
            f"select brand,keyword from geo_keyword_brand_map where keyword in ({in_list}) group by keyword") or []

    found_keywords = {row.get("keyword") for row in (query_result or []) if row.get("keyword")}

    missing_keywords = [kw for kw in keyword_list if kw not in found_keywords]
    if missing_keywords:
        ai_brand = ai_utils.ai_get_brand_name(missing_keywords) or []
        if ai_brand:
            bh_utils.insert_data("geo_keyword_brand_map", ai_brand)
            brand_map.extend(ai_brand)

    brand_map.extend(query_result)

    return brand_map


def extract_keywords(data):
    if not data:
        return []

    result = []

    for item in data:
        if isinstance(item, str):
            if item.strip():
                result.append(item.strip())
        elif isinstance(item, dict):
            keywords = item.get("keywords", [])
            if isinstance(keywords, list):
                for kw in keywords:
                    if isinstance(kw, str) and kw.strip():
                        result.append(kw.strip())
            elif isinstance(keywords, str) and keywords.strip():
                result.append(keywords.strip())
    return result


def convert_rank_data(data):
    result = []

    for item in data:
        rank_list = item.get("rank_list", []) or []

        result.append({
            "word": item.get("word"),
            "rank": min(rank_list) if rank_list else 0,
            "count": len(rank_list)
        })

    return result


def cache_get_ai_brand_list(taskId, platform, response_content, prompt):
    result = bh_utils.query_data(
        f"select * from geo_ai_rank_list where taskId = '{taskId}' and platform = '{platform}' limit 1"
    )

    if result:
        db_row = result[0]
        ai_brand_list = db_row.get("ai_brand_list")

        if ai_brand_list:
            if isinstance(ai_brand_list, str):
                try:
                    return json.loads(ai_brand_list)
                except Exception:
                    return ai_brand_list
            return ai_brand_list

    ai_result = ai_utils.ai_get_brand_list(response_content, prompt)

    bh_utils.insert_data("geo_ai_rank_list", [{"taskId": taskId, "platform": platform, "ai_brand_list": ai_result}])

    return ai_result


def cache_get_ai_product_list(taskId, platform, response_content, prompt):
    result = bh_utils.query_data(
        f"select * from geo_ai_product_list where taskId = '{taskId}' and platform = '{platform}' limit 1"
    )

    if result:
        db_row = result[0]
        ai_brand_list = db_row.get("ai_brand_list")

        if ai_brand_list:
            if isinstance(ai_brand_list, str):
                try:
                    return json.loads(ai_brand_list)
                except Exception:
                    return ai_brand_list
            return ai_brand_list

    ai_result_search = ai_utils.ai_get_product_list(response_content, prompt)
    ai_result = ai_get_product_list_search(ai_result_search)
    bh_utils.insert_data("geo_ai_product_list", [{"taskId": taskId, "platform": platform, "ai_brand_list": ai_result}])

    return ai_result


def split_word_rank(ai_word_list, brand_words, com_word_list, all_word_rank_list):
    word_rank_map = {}
    for item in all_word_rank_list:
        word = item.get("word")
        if not word:
            continue
        word_rank_map[word] = {
            "rank": item.get("rank", 0),
            "count": item.get("count", 0),
            "brand": item.get("brand", '没有品牌'),
        }

    def build_result(word_list):
        result = []
        if word_list:
            for word in word_list:
                rank_info = word_rank_map.get(word, {})
                result.append({
                    "word": word,
                    "rank": rank_info.get("rank", 0),
                    "count": rank_info.get("count", 0),
                    "brand": rank_info.get("brand", '没有品牌')
                })
        return result

    ai_word_list_with_rank = build_result(ai_word_list)
    brand_words_rank = build_result(brand_words)
    com_word_list_rank = build_result(com_word_list)

    return ai_word_list_with_rank, brand_words_rank, com_word_list_rank


def keyword_with_brand(all_keyword_with_rank_sentiment, all_keyword_with_brand):
    keyword_brand_map = {
        item.get("keyword"): item.get("brand", "没有品牌")
        for item in all_keyword_with_brand
        if item.get("keyword")
    }

    for item in all_keyword_with_rank_sentiment:
        item["brand"] = keyword_brand_map.get(item.get("word"), "没有品牌")

    return all_keyword_with_rank_sentiment


def build_brand_summary(com_word_list_rank, com_brand):
    # 先构造 word -> 明细 的映射
    word_detail_map = {
        item.get("word"): item
        for item in com_word_list_rank
        if item.get("word") is not None
    }

    result = []

    for brand_item in com_brand:
        brand_name = brand_item.get("brand", "")
        keywords = brand_item.get("keywords") or []

        # 收集该品牌下的所有关键词明细
        vos = []
        for kw in keywords:
            detail = word_detail_map.get(kw)
            if detail:
                vos.append(detail.copy())
            else:
                # 如果关键词在 com_word_list_rank 里不存在，补一个默认结构
                vos.append({
                    "word": kw,
                    "rank": 0,
                    "count": 0
                })

        # 计算外层 rank
        non_zero_ranks = [item.get("rank", 0) for item in vos if item.get("rank", 0) != 0]
        final_rank = min(non_zero_ranks) if non_zero_ranks else 0

        # 计算外层 count
        final_count = sum(item.get("count", 0) or 0 for item in vos)

        result.append({
            "word": brand_name,
            "count": final_count,
            "rank": final_rank,
            "vos": vos
        })

    return result


def pick_brand_from_vos(vos):
    if not vos:
        return "没有品牌"

    # 1. 先找 rank 非 0 的，取 rank 最大的那条
    ranked_items = [item for item in vos if item.get("rank", 0) != 0]
    if ranked_items:
        best = max(ranked_items, key=lambda x: x.get("rank", 0))
        return best.get("brand", "没有品牌") or "没有品牌"

    # 2. 如果全是 0，取第一个不是“没有品牌”的
    for item in vos:
        brand = item.get("brand")
        if brand and brand != "没有品牌":
            return brand

    # 3. 如果全都是“没有品牌”
    return "没有品牌"


def fill_outer_brand(data):
    result = []

    for item in data:
        new_item = item.copy()
        vos = new_item.get("vos", [])
        new_item["brand"] = pick_brand_from_vos(vos)
        result.append(new_item)

    return result


def filter_ai_word_list(ai_word_list_with_rank, brand_words_rank, com_word_list_rank):
    # 收集 brand_words_rank 里的 brand
    def normalize_brand(brand):
        if not brand:
            return ""
        return str(brand).strip().lower()

    brand_words_brand_set = {
        normalize_brand(item.get("brand"))
        for item in brand_words_rank
        if item.get("brand")
    }

    com_words_brand_set = {
        normalize_brand(item.get("brand"))
        for item in com_word_list_rank
        if item.get("brand")
    }

    exclude_brand_set = brand_words_brand_set | com_words_brand_set

    result = []
    for item in ai_word_list_with_rank:
        brand = item.get("brand")

        if brand == "没有品牌":
            result.append(item)
            continue

        if normalize_brand(brand) not in exclude_brand_set:
            result.append(item)

    return result


def merge_brand_rank_by_subset(brand_words_rank, ai_word_list_with_rank):
    """
    规则：
    1. 第一轮：word 完全相等（忽略大小写）时，直接匹配
    2. 第二轮：即使 brand_item 已经有 rank/count，也继续参与匹配：
       - word 双向子串匹配（忽略大小写）
    3. 匹配成功后：
       - 第一轮：仍然保持原逻辑，直接用 ai 的 rank / count 覆盖
       - 第二轮：
         - rank 取更小的非 0 值
         - count 取更大的值
       - 删除该 ai 元素（返回剩余 ai 列表）
    4. 一个 ai 元素只能匹配一次
    """

    def to_int(v):
        try:
            return int(v or 0)
        except Exception:
            return 0

    def merge_rank(old_rank, old_count, new_rank, new_count):
        old_rank = to_int(old_rank)
        old_count = to_int(old_count)
        new_rank = to_int(new_rank)
        new_count = to_int(new_count)

        # rank：取更小的非0值，0 视为无效
        if old_rank > 0 and new_rank > 0:
            rank = min(old_rank, new_rank)
        elif old_rank > 0:
            rank = old_rank
        elif new_rank > 0:
            rank = new_rank
        else:
            rank = 0

        # count：取更大值
        count = max(old_count, new_count)

        return rank, count

    brand_list = deepcopy(brand_words_rank)
    ai_list = deepcopy(ai_word_list_with_rank)
    used_ai_index = set()
    first_round_matched_brand_index = set()

    # ========== 第一轮：word 完全相等，忽略 brand ==========
    for brand_index, brand_item in enumerate(brand_list):
        brand_word = str(brand_item.get("word", "") or "").strip()
        brand_word_lower = brand_word.lower()

        if not brand_word_lower:
            continue

        for ai_index, ai_item in enumerate(ai_list):
            if ai_index in used_ai_index:
                continue

            ai_word = str(ai_item.get("word", "") or "").strip()
            ai_word_lower = ai_word.lower()

            if not ai_word_lower:
                continue

            if brand_word_lower == ai_word_lower:
                # 第一轮保持原逻辑：直接覆盖
                brand_item["rank"] = ai_item.get("rank", 0)
                brand_item["count"] = ai_item.get("count", 0)
                used_ai_index.add(ai_index)
                first_round_matched_brand_index.add(brand_index)
                break

    # ========== 第二轮：word 双向子串匹配，忽略 brand ==========
    for brand_index, brand_item in enumerate(brand_list):
        # 这里不再跳过第一轮命中的 brand_item
        brand_word = str(brand_item.get("word", "") or "").strip()
        brand_word_lower = brand_word.lower()

        if not brand_word_lower:
            continue

        for ai_index, ai_item in enumerate(ai_list):
            if ai_index in used_ai_index:
                continue

            ai_word = str(ai_item.get("word", "") or "").strip()
            ai_word_lower = ai_word.lower()

            if not ai_word_lower:
                continue

            if brand_word_lower in ai_word_lower or ai_word_lower in brand_word_lower:
                merged_rank, merged_count = merge_rank(
                    brand_item.get("rank", 0),
                    brand_item.get("count", 0),
                    ai_item.get("rank", 0),
                    ai_item.get("count", 0),
                )
                brand_item["rank"] = merged_rank
                brand_item["count"] = merged_count
                used_ai_index.add(ai_index)
                # 这里不要 break，继续把后面还能匹配的 ai 一并吃掉

    remain_ai_list = [
        item for idx, item in enumerate(ai_list)
        if idx not in used_ai_index
    ]

    return brand_list, remain_ai_list



def to_int(v):
    try:
        return int(v or 0)
    except Exception:
        return 0


def merge_rank(old_rank, old_count, new_rank, new_count):
    old_rank = to_int(old_rank)
    old_count = to_int(old_count)
    new_rank = to_int(new_rank)
    new_count = to_int(new_count)

    # rank：取更小的非 0 值
    if old_rank > 0 and new_rank > 0:
        rank = min(old_rank, new_rank)
    elif old_rank > 0:
        rank = old_rank
    elif new_rank > 0:
        rank = new_rank
    else:
        rank = 0

    # count：取更大值
    count = max(old_count, new_count)

    return rank, count


def merge_com_and_ai_by_subset(brand_words_rank, ai_word_list_with_rank):
    """
    规则：
    1. 第一轮：vos 中的 word 和 ai.word 完全相等（忽略大小写）时，直接匹配
    2. 第二轮：即使 vos 已经有 rank/count，也继续参与匹配：
       - word 双向子串匹配（忽略大小写）
    3. 匹配成功后：
       - 第一轮：直接用 ai 的 rank / count 覆盖 vos 的 rank / count
       - 第二轮：
         - rank 取更小的非 0 值
         - count 取更大的值
       - 删除该 ai 元素（返回剩余 ai 列表）
    4. 一个 ai 元素只能匹配一次
    5. 外层 brand_item 的 rank / count 不变，只更新 vos
    """

    brand_list = deepcopy(brand_words_rank)
    ai_list = deepcopy(ai_word_list_with_rank)

    used_ai_index = set()

    # ========== 第一轮：vos.word 完全相等，忽略 brand ==========
    for brand_item in brand_list:
        vos = brand_item.get("vos", [])
        if not isinstance(vos, list):
            continue

        for vo in vos:
            vo_word = str(vo.get("word", "") or "").strip()
            vo_word_lower = vo_word.lower()

            if not vo_word_lower:
                continue

            for ai_index, ai_item in enumerate(ai_list):
                if ai_index in used_ai_index:
                    continue

                ai_word = str(ai_item.get("word", "") or "").strip()
                ai_word_lower = ai_word.lower()

                if not ai_word_lower:
                    continue

                if vo_word_lower == ai_word_lower:
                    vo["rank"] = ai_item.get("rank", 0)
                    vo["count"] = ai_item.get("count", 0)
                    used_ai_index.add(ai_index)
                    break

    # ========== 第二轮：vos.word 双向子串匹配，忽略 brand ==========
    for brand_item in brand_list:
        vos = brand_item.get("vos", [])
        if not isinstance(vos, list):
            continue

        for vo in vos:
            vo_word = str(vo.get("word", "") or "").strip()
            vo_word_lower = vo_word.lower()

            if not vo_word_lower:
                continue

            for ai_index, ai_item in enumerate(ai_list):
                if ai_index in used_ai_index:
                    continue

                ai_word = str(ai_item.get("word", "") or "").strip()
                ai_word_lower = ai_word.lower()

                if not ai_word_lower:
                    continue

                if vo_word_lower in ai_word_lower or ai_word_lower in vo_word_lower:
                    merged_rank, merged_count = merge_rank(
                        vo.get("rank", 0),
                        vo.get("count", 0),
                        ai_item.get("rank", 0),
                        ai_item.get("count", 0),
                    )
                    vo["rank"] = merged_rank
                    vo["count"] = merged_count
                    used_ai_index.add(ai_index)
                    # 第二轮不 break，继续吃后面还能匹配的 ai

    remain_ai_list = [
        item for idx, item in enumerate(ai_list)
        if idx not in used_ai_index
    ]

    return brand_list, remain_ai_list

def to_int_trans(v):
    try:
        return int(v or 0)
    except Exception:
        return 0

def rebuild_count_and_rank(data_list):
    """
    重新计算外层 count 和 rank
    规则：
    1. 外层 count = vos 中所有 count 之和
    2. 外层 rank = vos 中所有大于 0 的 rank 的最小值
    3. 如果 vos 中没有大于 0 的 rank，则外层 rank = 0
    """
    result = deepcopy(data_list)

    for item in result:
        vos = item.get("vos", [])
        if not isinstance(vos, list):
            item["count"] = 0
            item["rank"] = 0
            continue

        total_count = 0
        rank_list = []

        for vo in vos:
            if not isinstance(vo, dict):
                continue

            count = to_int_trans(vo.get("count", 0))
            rank = to_int_trans(vo.get("rank", 0))

            total_count += count

            if rank > 0:
                rank_list.append(rank)

        item["count"] = total_count
        item["rank"] = min(rank_list) if rank_list else 0

    return result


def get_all_brands(*data_lists):
    brands = []
    seen = set()

    for data_list in data_lists:
        if not data_list:
            continue

        for item in data_list:
            brand = item.get("brand")
            if not brand or brand == "没有品牌":
                brand = item.get("word")

            if brand and brand not in seen:
                seen.add(brand)
                brands.append(brand)

    return brands


def get_all_words(*data_lists):
    brands = []
    seen = set()

    for data_list in data_lists:
        if not data_list:
            continue

        for item in data_list:
            brand = item.get("word")
            seen.add(brand)
            brands.append(brand)

    return brands


def parse_mentions(value):
    if not value:
        return ""
    if isinstance(value, list):
        return ", ".join(str(x) for x in value if x)
    if isinstance(value, str):
        try:
            arr = json.loads(value)
            if isinstance(arr, list):
                return ", ".join(str(x) for x in arr if x)
        except Exception:
            return value
    return ""


def cache_ai_get_brand_sentiment_and_mentions(prompt, brands, response_content, task_id, platform, req_id):
    if not brands:
        return []

    seen = set()
    unique_brand_items = []

    for item in brands:
        if not isinstance(item, dict):
            continue

        brand = item.get("brand")
        if brand and brand not in seen:
            seen.add(brand)
            unique_brand_items.append({
                "brand": brand,
                "keyword": item.get("keyword", ""),
                "type": item.get("type", "")
            })

    if not unique_brand_items:
        return []

    unique_brands = [item["brand"] for item in unique_brand_items]

    # brand -> keyword/type 映射，插入 geo_brand_mention_list
    brand_info_map = {
        item["brand"]: {
            "keyword": item.get("keyword", ""),
            "type": item.get("type", "")
        }
        for item in unique_brand_items
    }

    # SQL 单引号转义
    def _escape_sql(s):
        return str(s).replace("'", "''")

    in_str = ",".join(f"'{_escape_sql(brand)}'" for brand in unique_brands)

    sql = f"""
        select *
        from geo_ai_brand_sentiment_and_mentions
        where taskId = '{_escape_sql(task_id)}'
          and platform = '{_escape_sql(platform)}'
          and brand_name in ({in_str})
    """


    db_result = bh_utils.query_data(sql) or []
    brand_mention_list = []
    for item in db_result:
        brand_n = item.get("brand_name")
        info = brand_info_map.get(brand_n, {})
        brand_mention_list .append(
            {
                "req_id": req_id,
                "taskId": task_id,
                "prompt": prompt,
                "platform": platform,
                "keyword": info.get("keyword", ""),
                "type": info.get("type", ""),
                "brand_name": item.get("brand_name"),
                "positive_mentions": parse_mentions(item.get("positive_mentions")),
                "negative_mentions": parse_mentions(item.get("negative_mentions"))
            }
        )

    bh_utils.insert_data("geo_brand_mention_list", brand_mention_list)
    # 数据库中已有的 brand
    db_brand_set = {
        item.get("brand_name")
        for item in db_result
        if item.get("brand_name")
    }

    # 找出缺失的 brand
    missing_brands = [
        brand
        for brand in unique_brands
        if brand not in db_brand_set
    ]

    # 如果数据库里已经有全部 brands，直接返回
    if not missing_brands:
        db_map = {
            item.get("brand_name"): item
            for item in db_result
            if item.get("brand_name")
        }

        return [
            db_map[brand]
            for brand in unique_brands
            if brand in db_map
        ]

    # 只对缺失的 brands 调 AI
    ai_result = ai_get_brand_sentiment_and_mentions(missing_brands, response_content) or []

    # 只补充原来缓存表需要的字段，不加 keyword/type
    for ar in ai_result:
        ar["taskId"] = task_id
        ar["platform"] = platform

    # 插入 AI 缓存表：
    if ai_result:
        bh_utils.insert_data("geo_ai_brand_sentiment_and_mentions", ai_result)

    all_result = db_result + ai_result

    # 插入 geo_brand_mention_list：
    result = []

    for item in all_result:
        brand_name = item.get("brand_name")
        info = brand_info_map.get(brand_name, {})

        result.append({
            "req_id": req_id,
            "prompt": prompt,
            "taskId": task_id,
            "platform": platform,
            "brand_name": brand_name,
            "keyword": info.get("keyword", ""),
            "type": info.get("type", ""),
            "positive_mentions": parse_mentions(item.get("positive_mentions")),
            "negative_mentions": parse_mentions(item.get("negative_mentions"))
        })

    if result:
        bh_utils.insert_data("geo_brand_mention_list", result)

    return all_result


def cache_ai_get_product_sentiment_and_mentions(products, response_content, task_id, platform, req_id):
    """
    逻辑：
    1. 先查数据库中 task_id + platform + brands 的已有结果
    2. 如果数据库中已包含所有 brands，直接返回数据库结果
    3. 如果只缺部分 brands，则只对缺失 brands 调用 AI
    4. 将缺失 brands 的 AI 结果插入数据库
    5. 返回完整结果
    """

    if not products:
        return []

    # 去重并保持顺序
    seen = set()
    unique_brands = []
    for brand in products:
        if brand and brand not in seen:
            seen.add(brand)
            unique_brands.append(brand)

    # SQL 单引号转义
    def _escape_sql(s):
        return str(s).replace("'", "''")

    in_str = ",".join(f"'{_escape_sql(brand)}'" for brand in unique_brands)

    sql = f"""
        select *
        from geo_ai_product_sentiment_and_mentions
        where taskId = '{_escape_sql(task_id)}'
          and platform = '{_escape_sql(platform)}'
          and brand_name in ({in_str})
    """
    db_result = bh_utils.query_data(sql) or []

    # 数据库中已有的 brand
    db_brand_set = {
        item.get("brand_name")
        for item in db_result
        if item.get("brand_name")
    }

    # 找出缺失的 brand
    missing_brands = [brand for brand in unique_brands if brand not in db_brand_set]

    # 如果数据库里已经有全部 brands，直接返回
    if not missing_brands:
        # 按传入 brands 顺序返回
        db_map = {item.get("brand_name"): item for item in db_result if item.get("brand_name")}

        return [db_map[brand] for brand in unique_brands if brand in db_map]

    # 只对缺失的 brands 调 AI
    ai_result = ai_get_product_sentiment_and_mentions(missing_brands, response_content) or []

    # 补充公共字段
    for ar in ai_result:
        ar["taskId"] = task_id
        ar["platform"] = platform

    # 插入数据库
    if ai_result:
        bh_utils.insert_data("geo_ai_product_sentiment_and_mentions", ai_result)

    all_result = db_result + ai_result

    result = [
        {
            "req_id": req_id,
            "brand_name": item.get("brand_name"),
            "positive_mentions": parse_mentions(item.get("positive_mentions")),
            "negative_mentions": parse_mentions(item.get("negative_mentions"))
        }
        for item in all_result
    ]

    bh_utils.insert_data("geo_product_mention_list", result)

    return all_result


def replace_word_by_brand(data_list):
    if not data_list:
        return []

    result = []
    for item in data_list:
        if not isinstance(item, dict):
            continue

        new_item = item.copy()
        brand = new_item.get("brand")

        if brand and brand != "没有品牌":
            new_item["word"] = brand

        result.append(new_item)

    return result


def keep_min_rank_by_brand(data_list):
    """
    相同 brand 只保留一条，保留 rank 最小的那条
    """
    brand_map = {}

    for item in data_list or []:
        brand = item.get("brand")
        rank = item.get("rank", 0)

        if brand not in brand_map:
            brand_map[brand] = item
        else:
            old_rank = brand_map[brand].get("rank", 0)
            if rank < old_rank:
                brand_map[brand] = item

    return list(brand_map.values())


sentiment_map = {
    "正向": "POSITIVE",
    "负向": "NEGATIVE",
    "中性": "NEUTRAL"
}


def merge_sentiment_to_brand_words(brand_words_list_with_rank, all_sentiment_and_mentions):
    # 构建 brand_name -> sentiment 映射
    word_sentiment_maps = {}
    for item in all_sentiment_and_mentions:
        brand_name = item.get("brand_name")
        sentiment = item.get("sentiment", "中性")
        if brand_name:
            word_sentiment_maps[brand_name] = sentiment
    result = []
    for item in brand_words_list_with_rank:
        new_item = dict(item)
        word = new_item.get("word")
        count = new_item.get("count", 0)
        if count > 0:
            new_item["sentiment"] = sentiment_map[word_sentiment_maps.get(word, "中性")]

        result.append(new_item)

    return result


def build_mentions_list(brand_words_list_with_rank, all_sentiment_and_mentions):
    valid_words = set()

    for item in brand_words_list_with_rank or []:
        word = str(item.get("word") or "").strip()
        try:
            count = int(item.get("count") or 0)
        except Exception:
            count = 0

        if word and count > 0:
            valid_words.add(word)

    all_positive = []
    all_negative = []

    for item in all_sentiment_and_mentions or []:
        brand_name = str(item.get("brand_name") or "").strip()
        if brand_name not in valid_words:
            continue

        positive_mentions = item.get("positive_mentions") or "[]"
        negative_mentions = item.get("negative_mentions") or "[]"

        if isinstance(positive_mentions, str):
            try:
                positive_mentions = json.loads(positive_mentions)
            except Exception:
                positive_mentions = []
        elif not isinstance(positive_mentions, list):
            positive_mentions = []

        if isinstance(negative_mentions, str):
            try:
                negative_mentions = json.loads(negative_mentions)
            except Exception:
                negative_mentions = []
        elif not isinstance(negative_mentions, list):
            negative_mentions = []

        all_positive.extend([str(x).strip() for x in positive_mentions if str(x).strip()])
        all_negative.extend([str(x).strip() for x in negative_mentions if str(x).strip()])

    return [
        {
            "negativeMentions": list(dict.fromkeys(all_negative)),
            "positiveMentions": list(dict.fromkeys(all_positive))
        }
    ]


def get_first_sentiment(data):
    for item in data or []:
        sentiment = item.get("sentiment")
        if sentiment:
            return sentiment


def get_first_score(data):
    for item in data or []:
        score = item.get("sentimentScore")
        if score:
            return score


def mark_source_mentioned(source_list, word_list):
    result = []

    # 去重并去掉空值
    words = [w.strip() for w in word_list if w and str(w).strip()]

    for item in source_list:
        snippet = item.get("snippet", "") or ""
        mentioned = 1 if any(word in snippet for word in words) else 0
        new_item = {}
        new_item["sourceId"] = item.get("sourceId")
        new_item["hasBrand"] = mentioned
        result.append(new_item)

    return result

def norm_source_id(v):
    if v is None:
        return ""
    return str(v).strip()

def merge_mentioned(snippet_mention, quotes):
    mentioned_map = {item["sourceId"]: item["hasBrand"] for item in snippet_mention}
    for item in quotes:
        item["hasBrand"] = mentioned_map.get(item["sourceId"], 0)
    return quotes


def merge_sentiment(snippet_mention, quotes):
    mentioned_map = {norm_source_id(item.get("sourceId")): item.get("sentiment", "NEUTRAL")for item in snippet_mention}
    for item in quotes:
        sid = norm_source_id(item.get("sourceId"))
        item["sentiment"] = mentioned_map.get(sid, "NEUTRAL")

    return quotes


def format_mentions(data_list):
    for item in data_list:
        positive_mentions = item.get("positive_mentions", "[]")
        negative_mentions = item.get("negative_mentions", "[]")

        try:
            item["positive_mentions"] = json.loads(positive_mentions) if isinstance(positive_mentions,
                                                                                    str) else positive_mentions
        except Exception:
            item["positive_mentions"] = []

        try:
            item["negative_mentions"] = json.loads(negative_mentions) if isinstance(negative_mentions,
                                                                                    str) else negative_mentions
        except Exception:
            item["negative_mentions"] = []

    return data_list


def process_brand_sentiment_by_snippets(prompt, platform, taskId, reqId, brand_words, snippet_list):
    """
    根据 brand_words 和 snippet_list 调用 ai_get_brand_sentiment_and_mentions

    参数:
        brand_words: list[str]
        snippet_list: list[dict]，每项至少包含 snippet 字段

    返回:
        list[dict]
    """
    results = []

    if not brand_words or not snippet_list:
        return results

    db_result = bh_utils.query_data(f"select * from geo_quote_mention_list where reqId = '{reqId}'")
    if db_result:
        return format_mentions(db_result)

    # 全量兜底 brand_str
    fallback_brand_str = [",".join(dict.fromkeys(brand_words))]
    articles_result = ai_get_brand_sentiment_and_mentions_by_articles(fallback_brand_str, snippet_list)
    today = datetime.now().strftime("%Y%m%d")
    for articles in articles_result:
        de = {
            "reqId": reqId,
            "taskId": taskId,
            "platform": platform,
            "prompt": prompt,
            "sourceId": articles.get("sourceId"),
            "sentiment": sentiment_map[articles.get('sentiment', '中性')],
            "positive_mentions": articles.get('positive_mentions'),
            "negative_mentions": articles.get('negative_mentions'),
            "pt": today
        }
        results.append(de)

    bh_utils.insert_data("geo_quote_mention_list", results)

    return results


def calc_brand_favorability_simple(data_list):
    result = []

    for item in data_list:
        pos_count = len(item.get("positive_mentions", []) or [])
        neg_count = len(item.get("negative_mentions", []) or [])
        total = pos_count + neg_count
        favorability = int((pos_count / total) * 100) if total > 0 else 50

        result.append({
            "brand_name": item.get("brand_name"),
            "favorability_score": favorability
        })

    return result


def attach_favorability_score(score_list, word_list):
    # 先构建品牌分数字典
    score_map = {
        item.get("brand_name"): item.get("favorability_score", 0)
        for item in score_list
    }

    result = []
    for item in word_list:
        new_item = dict(item)
        new_item["sentimentScore"] = score_map.get(item.get("brand"), 0)
        result.append(new_item)

    return result


def attach_favorability_brand_score(score_list, word_list):
    # 先构建品牌分数字典

    score_map = {
        item.get("brand_name"): item.get("favorability_score", 0)
        for item in score_list
    }

    result = []
    for item in word_list:
        new_item = dict(item)
        new_item["sentimentScore"] = score_map.get(item.get("word"), 0)
        result.append(new_item)

    return result


def replace_no_brand_inplace(data_list):
    for item in data_list:
        if item.get("brand") == "没有品牌":
            item["brand"] = item.get("keyword")
    return data_list


def add_keyword_type(ai_word_list, brand_words, com_word_list, all_keyword_with_brand):
    """
    根据 keyword 属于哪个 list，给 all_keyword_with_brand 每一项增加 type

    type:
    - brand: 自有品牌词
    - com: 竞品词
    - ai: AI 识别词
    - unknown: 未匹配
    """

    ai_set = set(ai_word_list or [])
    brand_set = set(brand_words or [])
    com_set = set(com_word_list or [])

    result = []

    for item in all_keyword_with_brand:
        brand = item.get("brand")
        keyword = item.get("keyword")

        if keyword in brand_set:
            word_type = "brand"
        elif keyword in com_set:
            word_type = "com"
        elif keyword in ai_set:
            word_type = "ai"
        else:
            word_type = "unknown"

        result.append({
            "brand": brand,
            "keyword": keyword,
            "type": word_type
        })

    return result


def result_v2(response_content, data):
    data = process_com_map(data)
    reqId = data.get('reqId')
    type = data.get('type')
    keywords = data.get('keywords')
    product_map = data.get('productWordsMap')
    prompt = data.get('prompt')
    platform = data.get('platform')
    taskId = data.get('taskId')
    brand_words = data.get('brandWords', [])
    comWords = data.get('comWords', [])
    if data.get('comWordsMap'):
        comWords = data.get('comWordsMap')

    com_word_list = extract_keywords(comWords)
    # 获取ai所有的词
    # -------------------------
    # 获取ai提及词
    ai_word_list = cache_get_ai_brand_list(taskId, platform, response_content, prompt)
    # 获取所有词
    all_word_list = []
    if isinstance(ai_word_list, list):
        all_word_list += ai_word_list
    if isinstance(brand_words, list):
        all_word_list += brand_words
    if isinstance(com_word_list, list):
        all_word_list += com_word_list


    brand_words_list_with_rank = []
    com_word_list_with_rank = []
    ai_word_list_with_rank = []
    keywords_list_with_rank = []
    product_map_list_with_rank = []
    product_map_list_with_rank_result = []
    all_product_list_with_rank = []

    all_word_set_list = list(set(all_word_list))
    ## 去除空字符串 死循环
    all_word_set_list = [str(x).strip() for x in all_word_set_list if str(x).strip()]
    # 获取所有词的排名 word + rank_list
    all_word_rank_list = get_keyword_ranks(response_content, all_word_set_list)
    # 获取所有词的排名 word + rank + count

    all_keyword_with_rank = convert_rank_data(all_word_rank_list)

    # 获取所有词的品牌
    all_keyword_with_brand = keyword_map_brand(all_word_list)

    # 追加所有词的品牌
    all_keyword_with_rank_sentiment_brand = keyword_with_brand(all_keyword_with_rank, all_keyword_with_brand)

    # 拆分为 all_词 品牌词 竞品词
    ai_word_list_with_rank, brand_words_list_with_rank, com_word_list_with_rank = split_word_rank(ai_word_list,
                                                                                                  brand_words,
                                                                                                  com_word_list,
                                                                                                  all_keyword_with_rank_sentiment_brand)

    # ---------------------------------调整排名-----------------------------

    brand_words_list_with_rank, ai_word_list_with_rank = merge_brand_rank_by_subset(brand_words_list_with_rank,
                                                                                    ai_word_list_with_rank)

    # 兼容map场景
    if data.get('comWordsMap'):
        # 计算出vos与外层数据
        com_word_list_with_rank = build_brand_summary(com_word_list_with_rank, comWords)
        # 外层品牌
        com_word_list_with_rank = fill_outer_brand(com_word_list_with_rank)
        # 去除竞品词在ai中的
        com_word_list_with_rank, ai_word_list_with_rank = merge_com_and_ai_by_subset(com_word_list_with_rank,
                                                                                     ai_word_list_with_rank)
        # 重新计算外层rank count
        com_word_list_with_rank = rebuild_count_and_rank(com_word_list_with_rank)


    else:
        # list 调整竞争品牌词排名 去除竞争品牌词中ai的排名
        com_word_list_with_rank, ai_word_list_with_rank = merge_brand_rank_by_subset(com_word_list_with_rank,
                                                                                     ai_word_list_with_rank)
    # ---------------------------------调整排名-----------------------------

    # 去除all
    ai_word_list_with_rank = filter_ai_word_list(ai_word_list_with_rank, brand_words_list_with_rank,
                                                 com_word_list_with_rank)

    # ---------------------------------处理品牌-----------------------------

    ai_word_list_with_rank = replace_word_by_brand(ai_word_list_with_rank)
    # 同一品牌只返回一个
    ai_word_list_with_rank = keep_min_rank_by_brand(ai_word_list_with_rank)
    # ---------------------------------处理品牌-----------------------------

    # ---------------------------------情感倾向-----------------------------
    # 获取所有品牌 或者词
    brands = get_all_brands(ai_word_list_with_rank) + get_all_words(brand_words_list_with_rank, com_word_list_with_rank)

    # 获取所有词情感倾向 正面词 负面词
    all_keyword_with_brand_replace = replace_no_brand_inplace(all_keyword_with_brand)
    brands_with_type = add_keyword_type(ai_word_list, brand_words, com_word_list, all_keyword_with_brand_replace)

    all_sentiment_and_mentions = cache_ai_get_brand_sentiment_and_mentions(prompt, brands_with_type, response_content,
                                                                           taskId, platform, reqId)
    all_sentiment_and_mentions_list = format_mentions(all_sentiment_and_mentions)
    all_sentiment_and_mentions_score = calc_brand_favorability_simple(all_sentiment_and_mentions_list)

    #
    brand_words_list_with_rank = merge_sentiment_to_brand_words(brand_words_list_with_rank, all_sentiment_and_mentions)
    com_word_list_with_rank = merge_sentiment_to_brand_words(com_word_list_with_rank, all_sentiment_and_mentions)
    ai_word_list_with_rank = merge_sentiment_to_brand_words(ai_word_list_with_rank, all_sentiment_and_mentions)
    # 计算分
    brand_words_list_with_rank = attach_favorability_brand_score(all_sentiment_and_mentions_score,
                                                                 brand_words_list_with_rank)

    com_word_list_with_rank = attach_favorability_brand_score(all_sentiment_and_mentions_score, com_word_list_with_rank)
    ai_word_list_with_rank = attach_favorability_score(all_sentiment_and_mentions_score, ai_word_list_with_rank)

    mentionsList = build_mentions_list(brand_words_list_with_rank, all_sentiment_and_mentions)
    # ---------------------------------情感倾向-----------------------------

    # ---------------------------------最小排名最小次数-----------------------------
    min_com_rank, com_count_sum = calculate_rank_min_and_count_sum(com_word_list_with_rank)
    min_brand_rank, brand_count_sum = calculate_rank_min_and_count_sum(brand_words_list_with_rank)
    # ---------------------------------最小排名最小次数-----------------------------

    # ---------------------------------关键词-----------------------------
    if keywords:
        keywords = [str(x).strip() for x in keywords if str(x).strip()]

        keywords_rank_list = get_keyword_ranks(response_content, keywords)
        keywords_list_with_rank = convert_rank_data(keywords_rank_list)

    # ---------------------------------产品词-----------------------------
    if product_map:
        # 获取所有产品词
        product_list = extract_keywords(product_map)

        # 获取ai提取产品词
        ai_product_list = cache_get_ai_product_list(taskId, platform, response_content, prompt)

        if isinstance(ai_product_list, list):
            product_list += ai_product_list

        product_list = list(set(product_list))

        product_list = [str(x).strip() for x in product_list if str(x).strip()]


        # 获取所有产品排名
        product_rank_list = get_keyword_ranks(response_content, product_list)


        all_product_list_with_rank = convert_rank_data(product_rank_list)


        # 所有产品
        product_all = get_all_words(all_product_list_with_rank)
        # 获取产品的情感倾向以及正面次负面词
        all_product_sentiment_and_mentions = cache_ai_get_product_sentiment_and_mentions(product_all, response_content,
                                                                                         taskId,
                                                                                         platform, reqId)
        # 计算所有的分
        all_product_sentiment_and_mentions = format_mentions(all_product_sentiment_and_mentions)
        all_product_sentiment_and_mentions_score = calc_brand_favorability_simple(all_product_sentiment_and_mentions)

        # 获取产品词的排名
        product_map_list_with_rank = build_brand_summary(all_product_list_with_rank, product_map)


        # 获取所有词的情感倾向
        product_map_list_with_rank = merge_sentiment_to_brand_words(product_map_list_with_rank,
                                                                    all_product_sentiment_and_mentions)

        # 删除ai中出现的品牌词

        product_map_list_with_rank, all_product_list_with_rank = merge_com_and_ai_by_subset(product_map_list_with_rank,
                                                                                            all_product_list_with_rank)

        # 获取产品词的分

        product_map_list_with_rank = attach_favorability_brand_score(all_product_sentiment_and_mentions_score,
                                                                     product_map_list_with_rank)

        product_map_list_with_rank_result = rebuild_count_and_rank(product_map_list_with_rank)

        # 获取所有品牌的分
        all_product_list_with_rank = attach_favorability_brand_score(all_product_sentiment_and_mentions_score,
                                                                     all_product_list_with_rank)

        all_product_list_with_rank = merge_sentiment_to_brand_words(all_product_list_with_rank,
                                                                    all_product_sentiment_and_mentions)

    # ---------------------------------引用来源-----------------------------
    path = f'geo/{taskId}/{platform}/quote.txt'
    quto = tos_utils.get_string_from_tos(path)
    snippet_list = []
    quotes = []
    if quto:
        json_quto = json.loads(quto)
        for q in json_quto:
            t = {
                "img": q.get("site_icon"),
                "platform": q.get("platform"),
                "sourceId": q.get("quto_id"),
                "title": q.get("title"),
                "url": q.get("url"),
                "site_name": q.get("site_name")
            }
            quotes.append(t)
        for q in json_quto:
            k = {
                "snippet": q.get("snippet"),
                "sourceId": q.get("quto_id"),
            }
            snippet_list.append(k)
    if quotes:
        prompt = data["prompt"]
        for item in quotes:
            item["prompt"] = prompt

    if snippet_list and brand_words and type != 'stream':
        # if snippet_list and brand_words :
        snippet_mention = mark_source_mentioned(snippet_list, brand_words)
        quotes = merge_mentioned(snippet_mention, quotes)

        sentiment_list = process_brand_sentiment_by_snippets(prompt, platform, taskId, reqId, brand_words, snippet_list)
        quotes = merge_sentiment(sentiment_list, quotes)

    # ---------------------------------引用来源-----------------------------

    # ---------------------------------结果整合-----------------------------
    result = {
        "brandVos": brand_words_list_with_rank,
        "competitorVos": com_word_list_with_rank,
        "keywordsVos": keywords_list_with_rank,
        "productVos": product_map_list_with_rank_result,
        "productAllVos": all_product_list_with_rank,
        "allVos": ai_word_list_with_rank,
        "fullVos": [],
        "minBrandRank": min_brand_rank,
        "minCompetitorRank": min_com_rank,
        "platform": platform,
        "prompt": prompt,
        "reqId": reqId,
        "sources": quotes,
        "taskId": taskId,
        "totalBrandCount": brand_count_sum,
        "totalCompetitorCount": com_count_sum,
        "mentionsList": mentionsList
    }
    # 修复里外不一致
    result["productVos"] = rebuild_count_and_rank(result.get("productVos") or [])
    # 情感分
    pos_count = len(mentionsList[0].get("positiveMentions", []) or [])
    neg_count = len(mentionsList[0].get("negativeMentions", []) or [])
    total = pos_count + neg_count
    sentimentScore = int((pos_count / total) * 100) if total > 0 else 50
    result['sentimentScore'] = sentimentScore

    # 品牌情感倾向
    if brand_words_list_with_rank:
        sentiment_result = get_first_sentiment(brand_words_list_with_rank)
        if sentiment_result:
            result['sentiment'] = sentiment_result


    # ---------------------------------结果整合-----------------------------

    # ---------------------------------上传结果-----------------------------
    tos_utils.put_string_to_tos(f'geo/{taskId}/{platform}/result.json', result)
    commit_task(data, 'SUCCESS')
    resp_cache_key = f"geo:task_check:resp:{reqId}"
    resp = {
        "code": 200,
        "msg": "success",
        "data": {
            "status": 'ING',
            "result": {}
        }
    }
    _cache_set_json(resp_cache_key, resp, 1)
    logger.success(f"{data['reqId']}--{data['platform']}--{data['prompt']}--ALL:SUCCESS")
    # ---------------------------------上传结果-----------------------------
    return True


class QueueType(Enum):
    STREAM = (1, "stream")
    STREAM_BATCH = (2, "stream_batch")
    BATCH = (3, "batch")
    SUCCESS = (4, "success")

    def __init__(self, priority, queue_name):
        self.priority = priority
        self.queue_name = queue_name

    @classmethod
    def get_by_name(cls, name):
        """根据队列名称获取枚举实例"""
        for queue in cls:
            if queue.queue_name == name:
                return queue
        raise ValueError(f"不存在名为 {name} 的队列")

    @classmethod
    def get_next_priority_queue(cls, current_queue):
        """根据当前队列获取下一个优先级的队列（最后一个返回自身）"""
        # 获取所有队列并按优先级排序（确保顺序正确）
        sorted_queues = sorted(cls, key=lambda q: q.priority)
        current_index = sorted_queues.index(current_queue)

        # 下一个索引：如果是最后一个则保持当前索引，否则+1
        next_index = current_index + 1 if current_index < len(sorted_queues) - 1 else current_index
        return sorted_queues[next_index]


def handle_task_failure(task_type: str) -> str:
    """
    接收队列类型名称，返回下一个优先级的队列名称
    :param task_type: 原始队列名称（如 "stream"、"batch" 等）
    :return: 下一个优先级的队列名称
    """
    try:
        # 根据名称获取原始队列枚举实例
        original_queue = QueueType.get_by_name(task_type)
    except ValueError:
        raise ValueError(f"不支持的任务类型: {task_type}")

    # 获取下一个优先级的队列
    next_queue = QueueType.get_next_priority_queue(original_queue)
    return next_queue.queue_name


def platform_process(data):
    PLATFORM_PROCESS_MAP = {
        'DB': doubao_data_process.doubao_process_original_data,
        'DP': deepseek_data_process.deepseek_process_original_data,
        'TXYB': yuanbao_data_process.yuanbao_process_original_data,
        'KIMI': kimi_data_process.kimi_process_original_data,
        'WXYY': wenxin_data_process.wenxin_process_original_data,
        'TYQW': qianwen_data_process.qianwen_process_original_data,
        'BDAI': baiduai_data_process.baiduai_process_original_data,
        'DPA': deepseek_android_data_process.deepseek_android_process_original_data,
        'DOUBA': doubao_android_data_process.doubao_mobile_process_original_data,
        'DYAI': douyinai_data_process.douyin_ai_process_original_data,
        'TYQWA': qianwen_android_data_process.qianwen_android_process_original_data,
        'TXYBA': yuanbao_android_data_process.yuanbao_android_process_original_data,
        'XHSA': xiaohongshu_android_data_process.xiaohongshu_android_process_original_data,
    }

    try:
        commit_task(data, 'PROCESSING')
        platform = data["platform"]
        original_path = f'geo/{data["taskId"]}/{platform}/original.text'
        context_path = f'geo/{data["taskId"]}/{platform}/context.txt'
        response_content = None


        #——————————————###################
        # has_original = check_file_in_tos(original_path)
        # has_context = check_file_in_tos(context_path)
        # # 重新跑逻辑 当俩个都有走下面
        # if has_original and has_context:
        #     process_func = PLATFORM_PROCESS_MAP.get(platform)
        #     if process_func:
        #         _, _, _, _, response_content, _ = process_func(data)
        # # 2. 只有 context：直接读 context
        # elif has_context:
        #     response_content = tos_utils.get_string_from_tos(context_path)
        #
        # # 3. 只有 original：跑平台解析逻辑
        # elif has_original:
        #     process_func = PLATFORM_PROCESS_MAP.get(platform)
        #     if process_func:
        #         _, _, _, _, response_content, _ = process_func(data)
        #——————————————###################


        #-------------
        if check_file_in_tos(context_path):
            response_content = tos_utils.get_string_from_tos(context_path)
        else:
            process_func = PLATFORM_PROCESS_MAP.get(platform)
            if process_func:
                _, _, _, _, response_content, _ = process_func(data)
        #-------------
        if response_content:
            result_v2(response_content, data)
        else:
            scheduler(data)

        return True
    except Exception as e:
        logger.exception(
            f"{data['reqId']}--{data['platform']}--{data['prompt']}--PROCESS_FAIL"
        )
        logger.error(f"{data['reqId']}--{data['platform']}--{data['prompt']}--{e}--PROCESS_FAIL", exc_info=True)
        robot_utils.feishu_tobot(f"{data['reqId']}--{data['platform']}--{data['prompt']}--{e}--PROCESS_FAIL")
        commit_task(data, 'PROCESS_FAIL')
        scheduler(data)
        return False


def main_process(data):
    task_id = data.get('taskId')
    platform = data.get('platform')
    type = data.get('type')
    req_id = data.get('reqId')
    prompt = data.get('prompt')
    data["prompt"] = prompt.strip()
    data["search_enabled"] = data.get('searchEnabled', '1')
    if data.get('thinkingEnabled'):
        data["thinking_enabled"] = data.get('thinkingEnabled')
    check = bh_utils.query_data(
        f"select status from geo_commit_task where taskId ='{task_id}' and platform = '{platform}'")

    if any(item.get("status") == "SUCCESS" for item in (check or [])):
        platform_process(data)
    else:
        if type == 'stream':
            logger.success(f"{req_id}--{platform}--{prompt}--{type}")
            bool_result = spider_interface.get_platform_response(data)
            if bool_result:
                size = get_tos_file_size(f"geo/{task_id}/{platform}/original.text")
                if size is None or size < 1024:
                    scheduler(data)
                else:
                    logger.success(f"{req_id}--{platform}--{prompt}--ING_IN_TOS")
                    platform_process(data)
            else:
                logger.error(f"{req_id}--{platform}--{prompt}--400")
                scheduler(data)
        elif type in ('stream_batch', 'batch') and not check:
            task_send_queue(data, type)


def process_success(data):
    data['type'] = 'success'
    platform = data.get('platform')

    if platform == 'DB':
        text, quto = ai_interface.get_doubao_message(data.get('prompt'))
        if quto:
            quto_list = []
            for index, item in enumerate(quto):
                raw_data = {
                    "url": item.get('url'),
                    "title": item.get('title'),
                    "snippet": item.get('summary'),
                    "index": index,
                    "published_at": item.get('publish_time'),
                    "site_name": item.get('site_name'),
                    "site_icon": item.get('logo_url'),
                }
                quto_list.append(raw_data)

            urls = []

            for c in quto_list:
                c['task_id'] = data.get('taskId')
                c['platform'] = platform
                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 quto_list:
                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", quto_list)
            put_string_to_tos(f"geo/{data.get('taskId')}/{data.get('platform')}/quote.txt", quto_list)
    # elif platform == 'KIMI':
    #     text = ai_interface.get_kimi_message(data.get('prompt'))
    # elif platform == 'DP':
    #     text = ai_interface.get_deepseek_message(data.get('prompt'))
    elif platform == 'TYQW':
        text = ai_interface.get_qianwewn_message(data.get('prompt'))
    else:
        text, quto = ai_interface.get_doubao_message(data.get('prompt'))
    put_string_to_tos(f"geo/{data.get('taskId')}/{data.get('platform')}/context.txt", text)
    result_v2(text, data)


def process_call_back(task_data, result):
    task_id = task_data.get('taskId')
    platform = task_data.get('platform')
    req_id = task_data.get('reqId')
    spider_data = result.get('data')

    if spider_data == 400:
        scheduler(task_data)
    else:
        logger.success(f"{req_id}--{platform}-------CALL_BACK_START")
        put_string_to_tos(f'geo/{task_id}/{platform}/original.text', spider_data)
        req_check = bh_utils.query_data(f"select status from geo_commit_task where reqId ='{req_id}'")
        req_status = ""
        if req_check and len(req_check) > 0:
            req_status = req_check[0].get("status", "")
        if req_check and req_status != 'SUCCESS':
            size = get_tos_file_size(f"geo/{task_id}/{platform}/original.text")
            if size is None or size < 1024:
                scheduler(task_data)
            else:
                platform_process(task_data)


def scheduler(data):
    commit_task(data, 'ING')
    task_id = data.get('taskId')
    platform = data.get('platform')
    req_id = data.get('reqId')
    fail_queue_type = data.get('type')
    prompt = data.get('prompt')
    degrade_count_key = f"geo:degrade_count:{task_id}:{platform}"
    degrade_count = redis_client.incr(degrade_count_key)
    if degrade_count == 1:
        redis_client.expire(degrade_count_key, 7200)

    if degrade_count > 6:
        logger.success(f"{data['reqId']}--{data['platform']}--{data['prompt']}--降级次数超过6次，直接走success")
        process_success(data)
        return
    if fail_queue_type == 'batch':
        batch_retry_key = f"geo:batch_retry_count:{task_id}:{platform}"

        batch_retry_count = redis_client.incr(batch_retry_key)
        if batch_retry_count == 1:
            redis_client.expire(batch_retry_key, 7200)

        if batch_retry_count <= 1:
            data['type'] = 'batch'
            key = f"{platform}:geo:batch:list"

            logger.success(
                f"{req_id}--{platform}--{prompt}--batch失败，重新投递batch--第{batch_retry_count}次batch重试"
            )

            redis_client.lpush(key, json.dumps(data))
            return

        logger.success(
            f"{req_id}--{platform}--{prompt}--batch已重试{batch_retry_count - 1}次，直接走success"
        )
        process_success(data)
        return

    next_queue_type = handle_task_failure(fail_queue_type)
    logger.success(
        f"{data['reqId']}--{data['platform']}--{data['prompt']}--降级成功-当前{next_queue_type}--第{degrade_count}次降级")

    if next_queue_type == 'success':
        process_success(data)
    else:
        data['type'] = next_queue_type
        key = f"{platform}:geo:{next_queue_type}:list"
        redis_client.lpush(key, json.dumps(data))


def task_send_queue(data, queue):
    logger.success(f"{data['reqId']}--{data['platform']}--{data['prompt']}--任务提交--{queue}")

    if data.get('thinkingEnabled'):
        data["thinking_enabled"] = str(data.get('thinkingEnabled'))
    if data.get('searchEnabled'):
        data["search_enabled"] = str(data.get('searchEnabled'))

    data["search_enabled"] = "1"

    commit_task(data, 'ING')
    redis_client.lpush(f"{data['platform']}:geo:{queue}:list", json.dumps(data))

def handle_item(i):
    if i.get('comWordsMap'):
        i['comWordsMap'] = json.loads(i.get('comWordsMap'))
    if i.get('brandWords'):
        i['brandWords'] = json.loads(i.get('brandWords'))
    if i.get('comWords'):
        i['comWords'] = json.loads(i.get('comWords'))
    if i.get('keywords'):
        i['keywords'] = json.loads(i.get('keywords'))
    if i.get('productWordsMap'):
        i['productWordsMap'] = json.loads(i.get('productWordsMap'))
    type_t = i.get('type')
    # type_t = 'batch'

    # return task_send_queue(i,type_t)
    return platform_process(i)


def run_data(PAGE_SIZE,MAX_WORKERS):
    logger.success(
       "开始执行"
    )
    last_insertime = 1781253622
    while True:
        query_sql = f"""
            SELECT *
            FROM geo_commit_task
            WHERE insertime < {last_insertime}
              AND status = 'SUCCESS'
            ORDER BY insertime DESC
            LIMIT {PAGE_SIZE}
        """

        data_list = bh_utils.query_data(query_sql) or []

        if not data_list:
            logger.success("处理结束")
            break

        # 关键：处理之前先保存本批次最后一条的原始 insertime
        next_last_insertime = int(data_list[-1].get("insertime"))

        logger.success(
            f"本批次查询到 {len(data_list)} 条，"
            f"当前 last_insertime={last_insertime}, "
            f"下一批 next_last_insertime={next_last_insertime}"
        )

        with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
            futures = [executor.submit(handle_item, i) for i in data_list]

            for future in as_completed(futures):
                try:
                    future.result()
                except Exception as e:
                    logger.exception(f"platform_process 执行异常: {e}")

        # 用处理前保存的原始游标推进
        last_insertime = next_last_insertime

        logger.success(f"本批次处理完成，更新 last_insertime={last_insertime}")

if __name__ == '__main__':

    data_list = bh_utils.query_data(f"select * from geo_commit_task where status ='ING' ")
    # data_list = bh_utils.query_data(query_sql)
    # print(data_list)
    # # #
    # # #
    # # # # #
    def handle_item(i):
        if i.get('comWordsMap'):
            i['comWordsMap'] = json.loads(i.get('comWordsMap'))
        if i.get('brandWords'):
            i['brandWords'] = json.loads(i.get('brandWords'))
        if i.get('comWords'):
            i['comWords'] = json.loads(i.get('comWords'))
        if i.get('keywords'):
            i['keywords'] = json.loads(i.get('keywords'))
        if i.get('productWordsMap'):
            i['productWordsMap'] = json.loads(i.get('productWordsMap'))
        type_t = i.get('type')
        # type_t = 'batch'

        # return task_send_queue(i,type_t)
        return platform_process(i)

    if data_list:
        with ThreadPoolExecutor(max_workers=30) as executor:
            futures = [executor.submit(handle_item, i) for i in data_list]

            for future in as_completed(futures):
                try:
                    future.result()
                except Exception as e:
                    logger.exception(f"platform_process 执行异常: {e}")

    # keywords = []
    # get_keyword_ranks()
