"""每日把飞书 Wiki 文档表格转换为截图任务并写入数据库。"""

from __future__ import annotations
import re
import json
import os
import sys
import tempfile
import uuid
from datetime import datetime
import time
import redis
import requests
from loguru import logger
from openpyxl import Workbook
from openpyxl.styles import (
    Alignment,
    Border,
    Font,
    PatternFill,
    Side,
)
from openpyxl.utils import get_column_letter

from aidso_geo.clients.tos_client import tos_client
from aidso_geo.models import spider_save_tos
POLL_SECONDS = 120
BASE_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.append(BASE_DIR)

from aidso_geo.utils import bh_utils, tos_utils

APP_ID = "cli_aaeb0c48e9f85be9"
APP_SECRET = "pBC2ty0QMo71YWgkyYOjjhk0v7Jjyg14"
FEISHU_CHAT_ID = "oc_41bc8e496a0d9bb93498a6a6cf6f30c1"

WIKI_TOKEN = "RndAwboqZiuCAFkTGkScaWHPnmg"
TABLE_NAME = "douchacha_data.geo_feishu_snipaste_v3"
LOG_TABLE_NAME = "douchacha_data.geo_feishu_snipaste_log_v3"
RESULT_TABLE_NAME = "douchacha_data.geo_feishu_snipaste_result_v3"
ROUND_COUNT = 30
MAX_ROUND_COUNT = 30
REDIS_QUEUE_KEY = "mt:snipaste_v3:only_content"
REDIS_QUEUE_KEY_SHARE = "mt:snipaste_v3:with_share"
SYNC_HOUR = 0
SYNC_MINUTE = 1
PLATFORM_CONFIGS = {
    "DB": {
        "name": "豆包网页版",
        "sheet_name": "豆包网页版",
        "tos_directory": "doubao",
        "content_redis_key": REDIS_QUEUE_KEY,
        "share_redis_key": REDIS_QUEUE_KEY_SHARE,
    },
    "DP": {
        "name": "DeepSeek网页版",
        "sheet_name": "deepseek网页版",
        "tos_directory": "DP",
        "content_redis_key": "mt:DP:snipaste_v3:only_content",
        "share_redis_key": "mt:DP:snipaste_v3:with_share",
    },
}
THIRTY_ROUND_EXPORT_WORKERS = 20
EXCEL_CELL_MAX_LENGTH = 32767
ILLEGAL_EXCEL_CHARACTERS = re.compile(
    r"[\x00-\x08\x0B-\x0C\x0E-\x1F]"
)

PROMPT_HEADERS = {"问题", "prompt", "问题词", "问题内容"}


class ResultWriteError(RuntimeError):
    pass


class TaskResultUnavailableError(RuntimeError):
    pass


def normalize_platform(platform: str = "DB") -> str:
    normalized_platform = str(platform or "").strip().upper()
    if normalized_platform not in PLATFORM_CONFIGS:
        raise ValueError(
            f"不支持的平台: {platform!r}，"
            f"可选值={list(PLATFORM_CONFIGS)}"
        )
    return normalized_platform


def normalize_pt(pt=None) -> str:
    """
    规范化运行批次。

    不传时使用当天 YYYYMMDD；允许在日期后追加批次号，
    例如 2026072702。
    """
    if pt is None:
        return datetime.now().strftime("%Y%m%d")

    normalized_pt = str(pt).strip()
    if len(normalized_pt) < 8 or not normalized_pt.isdigit():
        raise ValueError(
            "pt 必须是至少 8 位数字，例如 20260727 或 2026072702"
        )
    try:
        datetime.strptime(normalized_pt[:8], "%Y%m%d")
    except ValueError as exc:
        raise ValueError(
            f"pt 的前 8 位不是有效日期: {normalized_pt}"
        ) from exc
    return normalized_pt


def format_pt_display(pt: str, for_filename: bool = False) -> str:
    """把运行批次格式化为日报日期或 Excel 文件名前缀。"""
    normalized_pt = normalize_pt(pt)
    date_format = "%Y年%m月%d日" if for_filename else "%Y-%m-%d"
    date_text = datetime.strptime(
        normalized_pt[:8],
        "%Y%m%d",
    ).strftime(date_format)
    batch_suffix = normalized_pt[8:]
    if not batch_suffix:
        return date_text
    if for_filename:
        return f"{date_text}_批次{batch_suffix}"
    return f"{date_text}（批次{batch_suffix}）"


def init_redis():
    try:
        return redis.Redis(
            host="172.16.0.24",
            port=6379,
            db=4,
            password="aiyingli@@123",
            socket_timeout=5,
            decode_responses=True,
        )
    except Exception:
        logger.exception("Redis 初始化失败")
        return None


def validate_config() -> None:
    if not APP_ID:
        raise RuntimeError("缺少 APP_ID")
    if not APP_SECRET:
        raise RuntimeError("缺少 APP_SECRET")
    if not WIKI_TOKEN:
        raise RuntimeError("缺少 WIKI_TOKEN")


def request_json(method: str, url: str, **kwargs) -> dict:
    response = requests.request(method, url, timeout=30, **kwargs)
    try:
        data = response.json()
    except ValueError as exc:
        raise RuntimeError(
            f"飞书接口未返回 JSON: status={response.status_code}, body={response.text[:500]}"
        ) from exc

    if response.status_code >= 400 or data.get("code", 0) != 0:
        raise RuntimeError(
            f"飞书接口调用失败: status={response.status_code}, response={data}"
        )
    return data


def get_tenant_access_token() -> str:
    data = request_json(
        "POST",
        "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal",
        json={"app_id": APP_ID, "app_secret": APP_SECRET},
    )
    token = data.get("tenant_access_token")
    if not token:
        raise RuntimeError(f"飞书返回中没有 tenant_access_token: {data}")
    return token


def auth_headers(access_token: str) -> dict:
    return {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json; charset=utf-8",
    }


def get_wiki_node(access_token: str) -> dict:
    data = request_json(
        "GET",
        "https://open.feishu.cn/open-apis/wiki/v2/spaces/get_node",
        headers=auth_headers(access_token),
        params={"token": WIKI_TOKEN},
    )
    node = data.get("data", {}).get("node")
    if not node:
        raise RuntimeError(f"飞书返回中没有 Wiki 节点信息: {data}")
    return node


def get_sheet_by_name(
    spreadsheet_token: str,
    access_token: str,
    sheet_name: str,
) -> dict:
    data = request_json(
        "GET",
        (
            "https://open.feishu.cn/open-apis/sheets/v3/spreadsheets/"
            f"{spreadsheet_token}/sheets/query"
        ),
        headers=auth_headers(access_token),
    )
    sheets = data.get("data", {}).get("sheets") or []
    if not sheets:
        raise RuntimeError(f"飞书电子表格中没有工作表: {data}")

    normalized_sheet_name = str(sheet_name or "").strip().lower()
    sheet = next(
        (
            item
            for item in sheets
            if str(item.get("title") or "").strip().lower()
            == normalized_sheet_name
        ),
        None,
    )
    if sheet is None:
        available_sheet_names = [
            str(item.get("title") or "")
            for item in sheets
        ]
        raise RuntimeError(
            f"飞书电子表格中没有名为 {sheet_name!r} 的工作表，"
            f"现有工作表={available_sheet_names}"
        )
    if not sheet.get("sheet_id"):
        raise RuntimeError(f"飞书返回中没有 sheet_id: {sheet}")
    return sheet


def column_name(column_count: int) -> str:
    if column_count <= 0:
        return "A"

    result = []
    while column_count:
        column_count, remainder = divmod(column_count - 1, 26)
        result.append(chr(ord("A") + remainder))
    return "".join(reversed(result))


def get_sheet_rows(
    spreadsheet_token: str,
    access_token: str,
    sheet_name: str,
) -> list[list]:
    sheet = get_sheet_by_name(
        spreadsheet_token,
        access_token,
        sheet_name,
    )
    sheet_id = sheet["sheet_id"]
    grid_properties = sheet.get("grid_properties") or {}
    row_count = int(grid_properties.get("row_count") or 1000)
    column_count = int(grid_properties.get("column_count") or 26)
    last_column = column_name(column_count)
    batch_size = 100
    all_rows = []

    for start_row in range(1, row_count + 1, batch_size):
        end_row = min(start_row + batch_size - 1, row_count)
        cell_range = f"{sheet_id}!A{start_row}:{last_column}{end_row}"
        data = request_json(
            "GET",
            (
                "https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/"
                f"{spreadsheet_token}/values_batch_get"
            ),
            headers=auth_headers(access_token),
            params=[("ranges", cell_range)],
        )
        value_ranges = data.get("data", {}).get("valueRanges") or []
        if value_ranges:
            all_rows.extend(value_ranges[0].get("values") or [])

    if not all_rows:
        raise RuntimeError("飞书电子表格中没有读取到数据")
    return all_rows


def cell_text(value) -> str:
    if value is None:
        return ""
    if isinstance(value, str):
        return value.strip()
    if isinstance(value, (dict, list)):
        return json.dumps(value, ensure_ascii=False)
    return str(value).strip()


def build_tasks(
    rows: list[list],
    pt: str,
    platform: str,
) -> list[dict]:
    platform = normalize_platform(platform)
    business_line_column_index = None
    for row in rows:
        for column_index, cell in enumerate(row):
            if cell_text(cell) == "业务线":
                business_line_column_index = column_index
                break
        if business_line_column_index is not None:
            break

    tasks = []
    for row in rows:
        if not row:
            continue

        prompt = cell_text(row[0])
        if not prompt or prompt.lower() in PROMPT_HEADERS:
            continue

        keywords = []
        for column_index, cell in enumerate(
            row[1:],
            start=1,
        ):
            if column_index == business_line_column_index:
                continue
            keyword = cell_text(cell)
            if keyword:
                keywords.append(keyword)
        business_line = ""
        if (
            business_line_column_index is not None
            and business_line_column_index < len(row)
        ):
            business_line = cell_text(
                row[business_line_column_index]
            )
        tasks.append(
            {
                "taskId": str(uuid.uuid4()),
                "prompt": prompt,
                "keywords": json.dumps(keywords, ensure_ascii=False),
                "pt": pt,
                "sheet_order": len(tasks) + 1,
                "platform": platform,
                "business_line": business_line,
            }
        )
    return tasks


def get_tasks(pt: str, platform: str = "DB") -> list[dict]:
    platform = normalize_platform(platform)
    rows = bh_utils.query_data(
        (
            f"SELECT taskId, prompt, keywords, pt, sheet_order, "
            f"business_line, "
            f"ifNull(platform, 'DB') AS platform "
            f"FROM {TABLE_NAME} WHERE pt = %s "
            f"AND ifNull(platform, 'DB') = %s"
        ),
        (pt, platform),
    )
    if rows is None:
        raise RuntimeError(f"查询 {TABLE_NAME} 失败")
    return rows


def create_round_logs(
    tasks: list[dict],
    pt: str,
    platform: str = "DB",
) -> int:
    platform = normalize_platform(platform)
    existing_rows = bh_utils.query_data(
        (
            f"SELECT taskId, `count` FROM {LOG_TABLE_NAME} "
            f"WHERE pt = %s "
            f"AND ifNull(platform, 'DB') = %s"
        ),
        (pt, platform),
    )
    if existing_rows is None:
        raise RuntimeError(f"查询 {LOG_TABLE_NAME} 失败")

    existing_keys = {
        (str(row.get("taskId") or ""), str(row.get("count") or ""))
        for row in existing_rows
    }
    log_items = []
    for task in tasks:
        task_id = str(task["taskId"])
        for round_number in range(1, ROUND_COUNT + 1):
            count = str(round_number)
            if (task_id, count) in existing_keys:
                continue
            log_items.append(
                {
                    "reqId": str(uuid.uuid4()),
                    "taskId": task_id,
                    "prompt": task["prompt"],
                    "keywords": task.get("keywords"),
                    "`count`": count,
                    "pt": pt,
                    "platform": platform,
                }
            )

    if not log_items:
        logger.success(
            f"[第一阶段 pt={pt} count=1-30] "
            f"当天 30 轮日志已全部存在"
        )
        return 0
    if not bh_utils.insert_data(LOG_TABLE_NAME, log_items):
        raise RuntimeError(f"写入日志表失败: {LOG_TABLE_NAME}")
    return len(log_items)


def push_rounds_to_redis(
    start_count: int,
    end_count: int,
    redis_key: str,
    pt: str = None,
    task_ids: list = None,
    platform: str = "DB",
) -> int:
    platform = normalize_platform(platform)
    is_share_round = (
        start_count == end_count
        and start_count in (0, 99)
    )
    if (
        start_count < 0
        or (
            end_count > MAX_ROUND_COUNT
            and not is_share_round
        )
        or start_count > end_count
    ):
        raise ValueError(
            f"轮次范围必须满足 0 <= start_count <= end_count "
            f"<= {MAX_ROUND_COUNT}，或使用分享轮次 0、99"
        )
    if not redis_key or not redis_key.strip():
        raise ValueError("redis_key 不能为空")
    redis_key = redis_key.strip()
    pt = normalize_pt(pt)
    redis_pt = pt[:8]

    task_filter_sql = ""
    query_params = [pt, start_count, end_count]
    if task_ids is not None:
        task_ids = [
            str(task_id)
            for task_id in task_ids
            if task_id
        ]
        if not task_ids:
            logger.warning(
                f"[pt={pt} count={start_count}-{end_count}] "
                f"task_ids 为空，没有可发送到 Redis 的任务"
            )
            return 0
        task_id_placeholders = ", ".join(
            ["%s"] * len(task_ids)
        )
        task_filter_sql = (
            f"AND taskId IN ({task_id_placeholders}) "
        )
        query_params.extend(task_ids)

    rows = bh_utils.query_data(
        (
            f"SELECT reqId, prompt, pt FROM {LOG_TABLE_NAME} "
            f"WHERE pt = %s "
            f"AND ifNull(platform, 'DB') = %s "
            f"AND toInt32(`count`) BETWEEN %s AND %s "
            f"{task_filter_sql}"
            f"ORDER BY toInt32(`count`), taskId"
        ),
        tuple([query_params[0], platform, *query_params[1:]]),
    )
    if rows is None:
        raise RuntimeError(
            f"查询第 {start_count}-{end_count} 轮任务失败: {LOG_TABLE_NAME}"
        )
    if not rows:
        logger.warning(
            f"[pt={pt} count={start_count}-{end_count}] "
            f"没有可发送到 Redis 的任务"
        )
        return 0

    values = [
        json.dumps(
            {
                "reqId": row.get("reqId"),
                "prompt": row.get("prompt"),
                "pt": redis_pt,
            },
            ensure_ascii=False,
        )
        for row in rows
    ]

    redis_client = init_redis()
    if redis_client is None:
        raise RuntimeError("Redis 初始化失败")

    try:
        with redis_client.pipeline(transaction=True) as pipeline:
            pipeline.delete(redis_key)
            pipeline.rpush(redis_key, *values)
            pipeline.execute()
    except Exception as exc:
        raise RuntimeError(
            f"发送 Redis 队列失败: key={redis_key}"
        ) from exc
    logger.success(
        f"[pt={pt} count={start_count}-{end_count}] "
        f"Redis 队列发送完成: key={redis_key}, rows={len(values)}"
    )
    return len(values)


def sync_feishu_to_database(
    pt: str = None,
    platform: str = "DB",
) -> None:
    validate_config()
    platform = normalize_platform(platform)
    platform_config = PLATFORM_CONFIGS[platform]
    pt = normalize_pt(pt)
    tasks = get_tasks(pt, platform=platform)
    title = ""

    if not tasks:
        access_token = get_tenant_access_token()
        node = get_wiki_node(access_token)
        title = node.get("title") or ""
        obj_type = node.get("obj_type")
        if obj_type != "sheet":
            raise RuntimeError(
                f"当前 Wiki 节点类型是 {obj_type!r}，仅支持 sheet"
            )
        table_rows = get_sheet_rows(
            node["obj_token"],
            access_token,
            platform_config["sheet_name"],
        )

        tasks = build_tasks(table_rows, pt, platform)
        if not tasks:
            raise RuntimeError("飞书文档中没有解析到可写入的表格任务")
        if not bh_utils.insert_data(TABLE_NAME, tasks):
            raise RuntimeError(f"写入任务表失败: {TABLE_NAME}")
    else:
        logger.success(
            f"[第一阶段 pt={pt} count=1-30] "
            f"当前批次主任务已存在，直接补充轮次日志: "
            f"rows={len(tasks)}"
        )

    inserted_log_count = create_round_logs(
        tasks,
        pt,
        platform=platform,
    )

    logger.success(
        f"[第一阶段 platform={platform} pt={pt} count=1-30] "
        f"飞书数据库同步完成: title={title}, "
        f"tasks={len(tasks)}, inserted_logs={inserted_log_count}"
    )


def start_scheduler() -> None:
    from apscheduler.schedulers.blocking import BlockingScheduler

    scheduler = BlockingScheduler(timezone="Asia/Shanghai")
    for platform in PLATFORM_CONFIGS:
        scheduler.add_job(
            run_daily_pipeline_safely,
            trigger="cron",
            hour=SYNC_HOUR,
            minute=SYNC_MINUTE,
            second=0,
            kwargs={"platform": platform},
            id=f"feishu_snipaste_v3_daily_pipeline_{platform}",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
            misfire_grace_time=3600,
        )
    logger.success(
        f"[调度 platforms={list(PLATFORM_CONFIGS)} count=1-99] "
        f"七阶段每日任务已注册: "
        f"{SYNC_HOUR:02d}:{SYNC_MINUTE:02d}, "
        f"jobs={len(PLATFORM_CONFIGS)}"
    )
    scheduler.start()

def doubao_process_original_data(file_path, original_content):
    url_list = ""
    think_content = ""
    response_content = ""
    search_keyword = []
    suggestions = []
    is_think = False
    rich_media_block = []
    think_bool = False
    file_path_result = os.path.dirname(file_path)
    content_list = original_content.split("\n")

    for i in content_list:

        if i != "":

            if not i.startswith("data:"):
                continue

            payload = i[len("data:"):].lstrip()
            if not payload:
                continue

            try:
                json_content = json.loads(payload)

            except (IndexError, json.JSONDecodeError):
                continue
            if json_content.get('query_list'):
                extra = json_content.get('ack_client_meta').get('conversation_info').get('extra')
                if json_content.get('ack_client_meta').get('conversation_info').get('extra'):
                    extra_json = json.loads(extra)
                    inner_user_ip = extra_json.get('inner_user_ip')
                    inner_pc_version = extra_json.get('inner_pc_version')
                    if inner_user_ip:
                        tos_utils.put_string_to_tos(f"{file_path_result}/inner_user_ip.txt", inner_user_ip)
                    if inner_pc_version:
                        tos_utils.put_string_to_tos(f"{file_path_result}/inner_pc_version.txt", inner_pc_version)
            if json_content.get('event_type') == 2001:

                even_data = json.loads(json_content.get('event_data'))

                message_data = even_data.get('message')
                if even_data.get('tts_content') is not None:
                    response_content = even_data.get('tts_content')
                if message_data.get('content_type') == 2007 :
                    for i in json.loads(message_data.get('content')).get("search_result").get("video_card").get("card_list"):
                        rich_media_block.append(i)
                if message_data.get('content_type') == 10040 and message_data.get('is_finish') is None:
                    think_bool = True

                    continue
                if message_data.get('content_type') == 10040 and message_data.get('is_finish') == True:
                    think_bool = False
                    continue
                if think_bool:
                    if json.loads(message_data.get('content')).get('text') is not None:
                        think_content += json.loads(message_data.get('content')).get('text')
                    content_json = json.loads(message_data.get('content'))

                    if message_data.get('content_type') == 10025 and content_json.get('results') is not None:
                        think_content += "\n\n"
                        think_content += "**搜索"
                        think_content += str(len(content_json.get('queries')))
                        think_content += "个关键词，参考"
                        think_content += str(len(content_json.get('results')))
                        think_content += "篇文章**"
                        think_content += "\n\n"

                if message_data.get('content_type') == 10025:

                    content_json = json.loads(message_data.get('content'))
                    if content_json.get('queries') is not None and content_json.get('results') is not None:
                        search_keyword = search_keyword + json.loads(message_data.get('content')).get('queries')

                    if content_json.get('scene') == 2:
                        url_list = content_json.get('results')

                if message_data.get('content_type') == 2002:
                    suggestions = suggestions + json.loads(message_data.get('content')).get('suggestions')
            else:

                if json_content.get("patch_op"):
                    if json_content.get('patch_op')[0].get("patch_object") == 111:
                        if json_content.get('patch_op')[0].get("patch_value").get("tts_content"):
                            response_content += json_content.get('patch_op')[0].get("patch_value").get("tts_content")
                    if json_content.get('patch_op')[0].get("patch_object") == 1:
                        if json_content.get('patch_op')[0].get("patch_value",{}).get("content_block"):
                            if json_content.get('patch_op')[0].get("patch_value",{}).get("content_block")[0].get(
                                    "content",{}).get("search_query_result_block",{}):
                                search_keyword = json_content.get('patch_op')[0].get("patch_value").get("content_block")[
                                    0].get("content").get("search_query_result_block").get("queries")
                                url_list = json_content.get('patch_op')[0].get("patch_value").get("content_block")[0].get(
                                    "content").get("search_query_result_block").get("results")
                                if is_think == True and json_content.get('patch_op')[0].get("patch_value").get("content_block")[0].get(
                                    "content").get("search_query_result_block").get("results") is not None:
                                    think_content += "\n\n"
                                    think_content += "**"
                                    think_content+=json_content.get('patch_op')[0].get("patch_value").get("content_block")[0].get(
                                    "content").get("search_query_result_block").get("summary")
                                    think_content += "**"
                                    think_content += "\n\n"

                            if json_content.get('patch_op')[0].get("patch_value").get("content_block")[0].get(
                                    "block_type") == 10000 and json_content.get('patch_op')[0].get("patch_value").get("content_block")[0].get(
                                    "meta_info"):

                                meta_info = json_content.get('patch_op')[0].get("patch_value").get("content_block")[0].get(
                                    "meta_info")
                                for meta in  meta_info:
                                    target_title = json.loads(meta.get('info')).get('title')
                                    index = next(
                                        (
                                            item["text_card"]["index"]
                                            for item in url_list
                                            if item.get("text_card", {}).get("title") == target_title
                                        ),
                                        None
                                    )
                                    if index:
                                        response_content+=f"[reference:{index}]"

                            if json_content.get('patch_op')[0].get("patch_value").get("content_block")[0].get(
                                    "block_type") ==10000 and json_content.get('patch_op')[0].get("patch_value").get("content_block")[0].get(
                                    "parent_id") and len(json_content.get('patch_op'))>1:

                                is_think = True
                                if json_content.get('patch_op')[0].get("patch_value").get("content_block")[0].get("content").get('text_block').get("text"):
                                    think_content+=json_content.get('patch_op')[0].get("patch_value").get("content_block")[0].get("content").get("text_block").get("text")
                                    continue
                            if json_content.get('patch_op')[0].get("patch_value").get("content_block")[0].get(
                                    "block_type") == 10040:
                                is_think = False
                                continue
                    if json_content.get('patch_op')[0].get("patch_object") == 50:
                        for sug in json.loads(
                                json_content.get('patch_op')[0].get("patch_value").get("ext").get("sp_v2")):
                            suggestions.append(sug.get("content"))
                if is_think:
                    if json_content.get("text"):
                        think_content += json_content.get("text")
                else:
                    if json_content.get("content"):
                        content_block = json_content.get("content").get('content_block')
                        if content_block:
                            if len(content_block)>0:
                                if content_block[0].get('block_type') ==10050:
                                    for i in content_block[0].get('content').get('rich_media_block').get('creations'):
                                        rich_media_block.append(i.get('video'))
                                if content_block[0].get('block_type') ==10000:

                                    if content_block[0].get("content").get("text_block").get("text"):
                                        response_content = content_block[0].get("content").get("text_block").get("text")

    suggestions = list(set(suggestions))
    spider_save_tos.process_and_save_files(file_path, search_keyword, url_list, think_content, response_content,
                                           suggestions,rich_media_block)
    return (file_path, search_keyword, url_list, think_content, response_content, suggestions)


def deepseek_process_original_data(file_path, original_content):

    url_list = ""
    think_content = ""
    response_content = ""
    search_keyword = []
    is_think = False
    think_bool = False
    suggestions = []
    response_bool = False
    url_id_map = {}
    # 按空行分割内容，过滤空字符串
    content_list = [item for item in original_content.split("\n\n") if item]


    for item in content_list:
        if item.startswith("event: "):
            # 可根据需要补充event数据处理逻辑
            continue

        # 处理data类型数据
        if item.startswith("data: "):
            try:
                # 提取并解析JSON数据
                data_str = item.split("data: ")[1]
                json_data = json.loads(data_str)

            except (IndexError, json.JSONDecodeError):
                continue  # 跳过格式错误的数据

            if isinstance(json_data.get('v'),dict):
                if  json_data.get('v').get('response').get('thinking_enabled') == True:

                    is_think=True
                fragments = json_data.get('v').get('response').get('fragments')
                if fragments:
                    if isinstance(fragments, list):
                        if fragments[0].get('type') == 'RESPONSE':
                            response_content += fragments[0].get('content')
                        if fragments[0].get('type') == 'THINK':
                            think_content += fragments[0].get('content')
                            think_bool = True
                        if fragments[0].get('type') == 'SEARCH':
                            if isinstance(fragments[0].get('queries'), list):
                                for sea in fragments[0].get('queries'):
                                    search_keyword.append(sea.get('query'))
                # if json_data.get('v').get('response').get('search_enabled') == True:
                #     if len(json_data.get('v').get('response').get('fragments'))>0:
                #         query_list = json_data.get('v').get('response').get('fragments')[0].get('queries')
                #         if query_list:
                #             result = [item.get('query', '') for item in query_list]
                #             search_keyword.extend(result)

            if is_think:

                if json_data.get('p') == 'response/fragments' and json_data.get('v')[0].get('type') == 'SEARCH':
                    search_keyword.append(json_data.get('v')[0].get('queries')[0].get('query'))

                if json_data.get('p') == 'response/fragments/0/results' or json_data.get(
                        'p') == 'response/fragments/-1/results':
                    url_list = json_data.get('v')

                if json_data.get('p') =='response/fragments' and json_data.get('v')[0].get('type') == 'THINK':
                    think_bool = True
                    think_content+=json_data.get('v')[0].get('content')
                    continue
                if json_data.get('p') == 'response' :
                    if isinstance(json_data.get('v')[0].get('v'),list):
                        if json_data.get('v')[0].get('v')[0].get('type') == 'TOOL_SEARCH':
                            for q in json_data.get('v')[0].get('v')[0].get('queries'):
                                search_keyword.append(q.get('query'))

                    if json_data.get('v')[1].get('p') == 'fragments' or json_data.get('v')[1].get('p') == 'response/fragments':
                        if  json_data.get('v')[1].get('v')[0].get('type') == 'THINK':
                            think_content+=json_data.get('v')[1].get('v')[0].get('content')
                            think_bool = True
                            continue
                    if json_data.get('v')[0].get('p') == 'fragments' or json_data.get('v')[0].get('p') == 'response/fragments':
                        if  json_data.get('v')[0].get('v')[0].get('type') == 'THINK':
                            think_content+=json_data.get('v')[0].get('v')[0].get('content')
                            think_bool = True
                            continue
                    if json_data.get('v')[0].get('v') == 'FINISHED':
                        response_bool = False
                if json_data.get('p') == 'response/fragments' and json_data.get('v')[0].get('type') == 'RESPONSE':
                    think_bool = False
                    response_bool = True
                    response_content += json_data.get('v')[0].get('content')
                    continue
                if json_data.get('p') == 'response/fragments/1/elapsed_secs':
                    think_bool = False
                if isinstance(json_data.get('v'),list):
                    v_json_data = json_data.get('v')
                    has_tool_search = False
                    tool_search_id = None

                    for item in v_json_data:
                        if item.get('p') == 'references':
                            references = item.get('v') or []

                            if isinstance(references, list):
                                for reference in references:
                                    if (
                                            isinstance(reference, dict)
                                            and reference.get('type') == 'TOOL_OPEN'
                                    ):
                                        has_tool_search = True
                                        tool_search_id = reference.get('id')
                                        break
                    for v in v_json_data:
                        if v.get('p') == 'content' and has_tool_search:
                            p_content = v.get('v') or ''

                            p_content = re.sub(
                                r'\[reference:\d+\]',
                                f'[reference:{tool_search_id}]',
                                p_content
                            )
                            response_content+=p_content

                        if v.get('p') == 'fragments':
                            if v.get('p') == 'fragments':
                                fragments = v.get('v') or []

                                if isinstance(fragments, list):
                                    for fragment in fragments:
                                        if (
                                                isinstance(fragment, dict)
                                                and fragment.get('type') == 'TOOL_OPEN'
                                                and fragment.get('result') is not None
                                        ):
                                            url_id_map[fragment.get('result').get('url')] = fragment.get('id')

                if json_data.get('p') == 'response/fragments/-1/elapsed_secs':
                    think_bool = False
                if json_data.get('p') == 'response/fragments':
                    response_bool = False
                if response_bool:
                    if isinstance(json_data.get('v'), str):
                        response_content += json_data.get('v')
                if think_bool:
                    if isinstance(json_data.get('v'), str):
                        think_content+=json_data.get('v')
            else:
                if json_data.get('p') == 'response/fragments/-1/content':
                    response_bool = True
                if json_data.get('p') == 'response/fragments' and json_data.get('v')[0].get('type') == 'SEARCH':
                    search_keyword.append(json_data.get('v')[0].get('queries')[0].get('query'))
                if json_data.get('p') == 'response/fragments/0/results' or json_data.get('p') == 'response/fragments/-1/results':
                    url_list = json_data.get('v')
                if json_data.get('p') == 'response/fragments' and json_data.get('v')[0].get('type') == 'RESPONSE':
                    response_content+=json_data.get('v')[0].get('content')
                    response_bool = True
                    continue
                if json_data.get('p') == 'response':
                    if json_data.get('v')[1].get('p') == 'fragments':
                        if  json_data.get('v')[1].get('v')[0].get('type') == 'RESPONSE':
                            response_content+=json_data.get('v')[1].get('v')[0].get('content')
                            response_bool = True
                            continue
                    if json_data.get('v')[0].get('p') == 'fragments':
                        if  json_data.get('v')[0].get('v')[0].get('type') == 'RESPONSE':
                            response_content+=json_data.get('v')[0].get('v')[0].get('content')
                            response_bool = True
                            continue
                    if json_data.get('v')[0].get('v') == 'FINISHED':
                        response_bool = False
                if  json_data.get('p') == 'response/fragments' and json_data.get('v')[0].get('type') == 'TIP':
                    response_bool= False
                    continue
                if json_data.get('v') == 'FINISHED':
                    response_bool = False
                if response_bool:
                    if isinstance(json_data.get('v'),str):
                        response_content+=json_data.get('v')
    used_ids = set(url_id_map.values())
    next_id = 1

    for item in url_list:
        url = item.get('url')

        if url in url_id_map:
            item['cite_index'] = url_id_map[url]

        elif item.get('cite_index') is None:
            while next_id in used_ids:
                next_id += 1

            item['cite_index'] = next_id
            used_ids.add(next_id)
            next_id += 1
    think_content = think_content.replace('DEEP_SEARCH','')
    spider_save_tos.process_and_save_files(file_path, search_keyword, url_list, think_content, response_content,
                                           suggestions)
    return (file_path, search_keyword, url_list, think_content, response_content, suggestions)


PLATFORM_PROCESS_MAP = {
    "DB": doubao_process_original_data,
    "DP": deepseek_process_original_data,
}


def get_platform_tos_directory(platform: str) -> str:
    platform = normalize_platform(platform)
    return PLATFORM_CONFIGS[platform]["tos_directory"]


def wait_mt_tasks_done(
    redis_key: str,
    pt: str = None,
    count_label: str = None,
    timeout_seconds: int = None,
    max_consecutive_failures: int = 3,
):
    redis_client = init_redis()
    """
    等待 Redis List 队列为空：
    - 为空 -> 认为“采集任务都跑完了”
    - 不为空 -> 不发飞书，继续等
    - 默认不限制等待时间
    """
    if not redis_key or not redis_key.strip():
        raise ValueError("redis_key 不能为空")
    redis_key = redis_key.strip()
    if redis_client is None:
        raise RuntimeError("Redis 初始化失败")

    log_prefix = (
        f"[pt={pt or '-'} count={count_label or '-'}]"
    )
    started_at = time.monotonic()
    consecutive_failures = 0

    while True:
        if (
            timeout_seconds is not None
            and time.monotonic() - started_at >= timeout_seconds
        ):
            raise TimeoutError(
                f"{log_prefix} 等待 Redis 队列超时: key={redis_key}, "
                f"timeout_seconds={timeout_seconds}"
            )
        try:
            size = redis_client.llen(redis_key)
            consecutive_failures = 0
            logger.success(
                f"{log_prefix} Redis 队列剩余任务: "
                f"key={redis_key}, size={size}"
            )
        except Exception as e:
            consecutive_failures += 1
            logger.warning(
                f"{log_prefix} Redis 队列长度读取失败: "
                f"key={redis_key}, error={e}"
            )
            if consecutive_failures >= max_consecutive_failures:
                raise RuntimeError(
                    f"{log_prefix} Redis 队列连续读取失败: "
                    f"key={redis_key}, "
                    f"failures={consecutive_failures}"
                ) from e
            time.sleep(POLL_SECONDS)
            continue

        if size == 0:
            return
        time.sleep(POLL_SECONDS)


def process_rounds_to_result(
    pt: str,
    start_count: int,
    end_count: int,
    redis_key: str,
    task_ids: list = None,
    platform: str = "DB",
) -> int:
    """
    发送指定轮次，等待并补发未完成任务，
    最后解析 TOS 正文、统计关键词并写入结果表。
    """
    platform = normalize_platform(platform)
    pt = normalize_pt(pt)
    redis_pt = pt[:8]
    tos_directory = get_platform_tos_directory(platform)
    process_original_data = PLATFORM_PROCESS_MAP[platform]
    normalized_task_ids = None
    task_filter_sql = ""
    task_filter_params = []
    if task_ids is not None:
        normalized_task_ids = [
            str(task_id)
            for task_id in task_ids
            if task_id
        ]
        if not normalized_task_ids:
            logger.warning(
                f"[pt={pt} count={start_count}-{end_count}] "
                f"task_ids 为空，跳过轮次处理"
            )
            return 0
        task_id_placeholders = ", ".join(
            ["%s"] * len(normalized_task_ids)
        )
        task_filter_sql = (
            f"AND taskId IN ({task_id_placeholders})"
        )
        task_filter_params = normalized_task_ids

    push_rounds_to_redis(
        start_count,
        end_count,
        redis_key,
        pt=pt,
        task_ids=normalized_task_ids,
        platform=platform,
    )
    wait_mt_tasks_done(
        redis_key,
        pt=pt,
        count_label=f"{start_count}-{end_count}",
    )
    time.sleep(60)

    for retry_index in range(3):
        diff_result = bh_utils.query_data(
            f"""
            SELECT a1.reqId, a1.prompt, a1.pt
            FROM (
                SELECT reqId, prompt, pt
                FROM {LOG_TABLE_NAME}
                WHERE pt = %s
                  AND ifNull(platform, 'DB') = %s
                  AND toInt32(`count`) BETWEEN %s AND %s
                  {task_filter_sql}
            ) AS a1
            LEFT JOIN geo_third_task AS a2 ON a1.reqId = a2.reqId
            WHERE a2.reqId IS NULL
            """,
            (
                pt,
                platform,
                start_count,
                end_count,
                *task_filter_params,
            ),
        )
        if diff_result is None:
            raise RuntimeError(
                f"查询未完成任务失败: pt={pt}, "
                f"count={start_count}-{end_count}"
            )

        values = [
            json.dumps(
                {
                    "reqId": row.get("reqId"),
                    "prompt": row.get("prompt"),
                    "pt": redis_pt,
                },
                ensure_ascii=False,
            )
            for row in diff_result
        ]
        if not values:
            logger.success(
                f"[pt={pt} count={start_count}-{end_count}] "
                f"没有需要补发的任务"
            )
            break

        redis_client = init_redis()
        if redis_client is None:
            raise RuntimeError("Redis 初始化失败")

        try:
            redis_client.rpush(redis_key, *values)
        except Exception as exc:
            raise RuntimeError(
                f"发送 Redis 队列失败: key={redis_key}"
            ) from exc

        logger.success(
            f"[pt={pt} count={start_count}-{end_count}] "
            f"第 {retry_index + 1} 次补发完成: key={redis_key}, "
            f"rows={len(values)}"
        )
        wait_mt_tasks_done(
            redis_key,
            pt=pt,
            count_label=f"{start_count}-{end_count}",
        )
        time.sleep(60)

    logger.success(
        f"[pt={pt} count={start_count}-{end_count}] "
        f"{redis_key} 已为空，开始解析结果"
    )

    result = bh_utils.query_data(
        f"""
        SELECT *
        FROM {LOG_TABLE_NAME}
        WHERE pt = %s
          AND ifNull(platform, 'DB') = %s
          AND toInt32(`count`) BETWEEN %s AND %s
          {task_filter_sql}
        """,
        (
            pt,
            platform,
            start_count,
            end_count,
            *task_filter_params,
        ),
    )
    if result is None:
        raise RuntimeError(
            f"查询待解析任务失败: pt={pt}, "
            f"count={start_count}-{end_count}"
        )

    from concurrent.futures import ThreadPoolExecutor, as_completed

    def parse_single_result(row):
        req_id = row.get("reqId")
        if not req_id:
            return None

        file = (
            f"geo_snipaste/{redis_pt}/{tos_directory}/"
            f"{req_id}/text.json"
        )
        try:
            keywords_value = row.get("keywords")
            if isinstance(keywords_value, str):
                try:
                    keywords = json.loads(keywords_value)
                except json.JSONDecodeError:
                    keywords = [keywords_value]
            elif isinstance(keywords_value, list):
                keywords = keywords_value
            else:
                keywords = []

            try:
                raw = tos_utils.get_string_from_tos(file)
            except Exception as exc:
                raise TaskResultUnavailableError(
                    f"TOS 结果读取失败: reqId={req_id}, file={file}"
                ) from exc
            if not raw:
                raise TaskResultUnavailableError(
                    f"TOS 结果不存在: reqId={req_id}, file={file}"
                )
            try:
                content = json.loads(raw).get("content")
            except (TypeError, json.JSONDecodeError) as exc:
                raise TaskResultUnavailableError(
                    f"TOS 结果格式异常: reqId={req_id}, file={file}"
                ) from exc
            if not content:
                raise TaskResultUnavailableError(
                    f"TOS 正文为空: reqId={req_id}, file={file}"
                )

            (
                _,
                _,
                _,
                _,
                response_content,
                _,
            ) = process_original_data(
                file,
                content,
            )

            keyword_counts = {}
            for keyword in keywords:
                keyword = str(keyword).strip()
                if not keyword:
                    continue
                keyword_counts[keyword] = (
                    response_content.count(keyword)
                )

            keywords_count = [
                {
                    "keyword": keyword,
                    "word_count": word_count,
                }
                for keyword, word_count
                in keyword_counts.items()
            ]
            has_keyword = int(
                any(
                    item["word_count"] > 0
                    for item in keywords_count
                )
            )

            return {
                "reqId": req_id,
                "taskId": row.get("taskId"),
                "prompt": row.get("prompt"),
                "keywords": row.get("keywords"),
                "`count`": int(row.get("count")),
                "pt": int(row.get("pt")),
                "has_keyword": has_keyword,
                "keywords_count": keywords_count,
                "platform": platform,
            }
        except Exception as exc:
            logger.warning(
                f"[pt={pt} count={start_count}-{end_count}] "
                f"处理 {file} 失败: {exc}"
            )
            if isinstance(exc, TaskResultUnavailableError):
                raise
            if isinstance(exc, ResultWriteError):
                raise
            return None

    def parse_result_batch(rows, batch_label):
        if not rows:
            return [], []
        worker_count = min(20, len(rows))
        logger.info(
            f"[pt={pt} count={start_count}-{end_count}] "
            f"开始并发解析结果: batch={batch_label}, "
            f"workers={worker_count}, tasks={len(rows)}"
        )
        parsed_results = []
        missing_rows = []
        completed_count = 0
        with ThreadPoolExecutor(max_workers=worker_count) as executor:
            future_to_row = {
                executor.submit(parse_single_result, row): row
                for row in rows
            }
            for future in as_completed(future_to_row):
                row = future_to_row[future]
                try:
                    result_item = future.result()
                except TaskResultUnavailableError:
                    missing_rows.append(row)
                    result_item = None
                except Exception as exc:
                    logger.error(
                        f"[pt={row.get('pt')} "
                        f"count={row.get('count')} "
                        f"reqId={row.get('reqId')}] "
                        f"并发任务处理失败: {exc}"
                    )
                    raise
                if result_item is not None:
                    parsed_results.append((row, result_item))
                completed_count += 1
                if (
                    completed_count % 500 == 0
                    or completed_count == len(rows)
                ):
                    logger.info(
                        f"[pt={pt} count={start_count}-{end_count}] "
                        f"并发解析进度: batch={batch_label}, "
                        f"completed={completed_count}/{len(rows)}, "
                        f"success={len(parsed_results)}, "
                        f"missing={len(missing_rows)}"
                    )
        return parsed_results, missing_rows

    def parse_rows_with_tos_retries(rows, batch_label):
        parsed_results, missing_rows = parse_result_batch(
            rows,
            batch_label,
        )
        all_parsed_results = list(parsed_results)

        for tos_retry_round in range(1, 4):
            if not missing_rows:
                break

            retry_values = [
                json.dumps(
                    {
                        "reqId": row.get("reqId"),
                        "prompt": row.get("prompt"),
                        "pt": redis_pt,
                    },
                    ensure_ascii=False,
                )
                for row in missing_rows
            ]
            redis_client = init_redis()
            if redis_client is None:
                raise TaskResultUnavailableError("Redis 初始化失败")
            try:
                redis_client.rpush(redis_key, *retry_values)
            except Exception as exc:
                raise TaskResultUnavailableError(
                    f"TOS 缺失任务批量重发 Redis 失败: "
                    f"key={redis_key}, "
                    f"retry_round={tos_retry_round}"
                ) from exc

            logger.warning(
                f"[pt={pt} count={start_count}-{end_count}] "
                f"TOS 结果缺失，开始第 {tos_retry_round}/3 "
                f"轮批量重试: rows={len(retry_values)}"
            )
            wait_mt_tasks_done(
                redis_key,
                pt=pt,
                count_label=(
                    f"{start_count}-{end_count} "
                    f"tos_retry={tos_retry_round}/3"
                ),
            )
            time.sleep(60)

            retry_parsed_results, missing_rows = parse_result_batch(
                missing_rows,
                (
                    f"{batch_label} "
                    f"tos_retry={tos_retry_round}/3"
                ),
            )
            all_parsed_results.extend(retry_parsed_results)

        if missing_rows:
            logger.error(
                f"[pt={pt} count={start_count}-{end_count}] "
                f"TOS 结果批量重试 3 次后仍缺失: "
                f"rows={len(missing_rows)}, "
                f"reqIds={[row.get('reqId') for row in missing_rows]}"
            )

        return all_parsed_results

    def insert_result_items(result_items, batch_size=500):
        if not result_items:
            return 0

        total_count = len(result_items)
        total_batches = (
            total_count + batch_size - 1
        ) // batch_size
        inserted_count = 0

        for batch_index, batch_start in enumerate(
            range(0, total_count, batch_size),
            start=1,
        ):
            batch_items = result_items[
                batch_start:batch_start + batch_size
            ]
            last_error = None

            for attempt in range(1, 4):
                try:
                    success = bh_utils.insert_data(
                        RESULT_TABLE_NAME,
                        batch_items,
                        retry_times=0,
                    )
                except Exception as exc:
                    last_error = exc
                    success = False

                if success:
                    inserted_count += len(batch_items)
                    logger.success(
                        f"[pt={pt} "
                        f"count={start_count}-{end_count}] "
                        f"结果批量写入完成: "
                        f"batch={batch_index}/{total_batches}, "
                        f"rows={len(batch_items)}, "
                        f"progress={inserted_count}/{total_count}, "
                        f"attempt={attempt}/3"
                    )
                    break

                logger.warning(
                    f"[pt={pt} count={start_count}-{end_count}] "
                    f"结果批量写入失败: "
                    f"batch={batch_index}/{total_batches}, "
                    f"rows={len(batch_items)}, "
                    f"attempt={attempt}/3, "
                    f"error={last_error or 'insert_data 返回 False'}"
                )
                if attempt < 3:
                    time.sleep(2)
            else:
                raise ResultWriteError(
                    f"关键词统计结果批量写入连续失败 3 次: "
                    f"pt={pt}, "
                    f"count={start_count}-{end_count}, "
                    f"batch={batch_index}/{total_batches}, "
                    f"rows={len(batch_items)}"
                ) from last_error

        return inserted_count

    inserted_count = 0
    parsed_results = parse_rows_with_tos_retries(
        result,
        "initial",
    )
    is_count_zero = start_count == 0 and end_count == 0

    if not is_count_zero:
        inserted_count = insert_result_items(
            [
                result_item
                for _, result_item in parsed_results
            ]
        )
    else:
        pending_results = []
        final_result_items = []
        for row, result_item in parsed_results:
            if int(result_item.get("has_keyword") or 0) == 1:
                final_result_items.append(result_item)
            else:
                pending_results.append((row, result_item))

        for share_retry_round in range(1, 6):
            if not pending_results:
                break

            retry_values = [
                json.dumps(
                    {
                        "reqId": row.get("reqId"),
                        "prompt": row.get("prompt"),
                        "pt": redis_pt,
                    },
                    ensure_ascii=False,
                )
                for row, _ in pending_results
            ]
            redis_client = init_redis()
            if redis_client is None:
                raise TaskResultUnavailableError("Redis 初始化失败")
            try:
                redis_client.rpush(redis_key, *retry_values)
            except Exception as exc:
                raise TaskResultUnavailableError(
                    f"count=0 未命中任务批量重发 Redis 失败: "
                    f"key={redis_key}, "
                    f"retry_round={share_retry_round}"
                ) from exc

            logger.warning(
                f"[pt={pt} count=0] 分享回答未命中关键词，"
                f"开始第 {share_retry_round}/5 轮批量重试: "
                f"rows={len(retry_values)}"
            )
            wait_mt_tasks_done(
                redis_key,
                pt=pt,
                count_label=(
                    f"0 batch_retry={share_retry_round}/5"
                ),
            )
            time.sleep(60)

            retry_rows = [
                row
                for row, _ in pending_results
            ]
            retry_parsed_results = parse_rows_with_tos_retries(
                retry_rows,
                f"count=0 retry={share_retry_round}/5",
            )
            retry_result_by_req_id = {
                str(result_item.get("reqId")): result_item
                for _, result_item in retry_parsed_results
            }

            next_pending_results = []
            for row, previous_result_item in pending_results:
                req_id = str(row.get("reqId") or "")
                result_item = retry_result_by_req_id.get(
                    req_id,
                    previous_result_item,
                )
                if int(result_item.get("has_keyword") or 0) == 1:
                    final_result_items.append(result_item)
                else:
                    next_pending_results.append((row, result_item))
            pending_results = next_pending_results

        for _, result_item in pending_results:
            final_result_items.append(result_item)

        inserted_count = insert_result_items(final_result_items)

    logger.success(
        f"[pt={pt} count={start_count}-{end_count}] "
        f"指定轮次处理完成: "
        f"results={len(result)}, inserted={inserted_count}"
    )
    return inserted_count


def create_and_process_share_rounds(
    pt: str,
    redis_key: str,
    hit_start_count: int = 1,
    hit_end_count: int = 30,
    hit_threshold: int = 24,
    platform: str = "DB",
) -> int:
    """
    第 1-30 轮中关键词命中轮数达到阈值的 taskId，
    生成 count=0 的分享任务并完成 Redis、TOS 和结果入库流程。
    """
    platform = normalize_platform(platform)
    qualified_rows = bh_utils.query_data(
        f"""
        SELECT taskId
        FROM {RESULT_TABLE_NAME}
        WHERE pt = %s
          AND ifNull(platform, 'DB') = %s
          AND `count` BETWEEN %s AND %s
          AND has_keyword = 1
        GROUP BY taskId
        HAVING count(DISTINCT `count`) >= %s
        """,
        (
            int(pt),
            platform,
            hit_start_count,
            hit_end_count,
            hit_threshold,
        ),
    )
    if qualified_rows is None:
        raise RuntimeError("第三阶段查询满足分享条件的 taskId 失败")

    task_ids = [
        str(row.get("taskId"))
        for row in qualified_rows
        if row.get("taskId")
    ]
    if not task_ids:
        logger.success(
            f"[第三阶段 pt={pt} count=0] "
            f"没有 taskId 达到分享阈值: threshold={hit_threshold}"
        )
        return 0

    task_id_placeholders = ", ".join(["%s"] * len(task_ids))
    tasks = bh_utils.query_data(
        f"""
        SELECT taskId, prompt, keywords, pt
        FROM {TABLE_NAME}
        WHERE pt = %s
          AND ifNull(platform, 'DB') = %s
          AND taskId IN ({task_id_placeholders})
        """,
        (
            pt,
            platform,
            *task_ids,
        ),
    )
    if tasks is None:
        raise RuntimeError("第三阶段从主表查询分享任务失败")

    existing_rows = bh_utils.query_data(
        f"""
        SELECT taskId
        FROM {LOG_TABLE_NAME}
        WHERE pt = %s
          AND ifNull(platform, 'DB') = %s
          AND toInt32(`count`) = 0
          AND taskId IN ({task_id_placeholders})
        """,
        (
            pt,
            platform,
            *task_ids,
        ),
    )
    if existing_rows is None:
        raise RuntimeError("第三阶段查询已生成分享任务失败")

    existing_task_ids = {
        str(row.get("taskId"))
        for row in existing_rows
        if row.get("taskId")
    }
    log_items = []
    for task in tasks:
        task_id = str(task.get("taskId"))
        if task_id in existing_task_ids:
            continue
        log_items.append(
            {
                "reqId": str(uuid.uuid4()),
                "taskId": task_id,
                "prompt": task.get("prompt"),
                "keywords": task.get("keywords"),
                "`count`": "0",
                "pt": str(task.get("pt") or pt),
                "platform": platform,
            }
        )

    inserted_log_count = len(log_items)
    if log_items:
        if not bh_utils.insert_data(LOG_TABLE_NAME, log_items):
            raise RuntimeError("第三阶段写入分享任务失败")
        logger.success(
            f"[第三阶段 pt={pt} count=0] "
            f"分享任务生成完成: "
            f"task_ids={len(task_ids)}, rows={inserted_log_count}"
        )
    else:
        logger.success(
            f"[第三阶段 pt={pt} count=0] 分享任务均已存在"
        )

    process_rounds_to_result(
        pt=pt,
        start_count=0,
        end_count=0,
        redis_key=redis_key,
        platform=platform,
    )
    return inserted_log_count


def create_and_process_unqualified_share_rounds(
    pt: str,
    redis_key: str,
    hit_start_count: int = 1,
    hit_end_count: int = 30,
    hit_threshold: int = 24,
    platform: str = "DB",
) -> int:
    """
    第 1-30 轮中关键词命中轮数不足阈值的 taskId，
    生成 count=99 的分享任务并完成 Redis、TOS 和结果入库流程。
    """
    platform = normalize_platform(platform)
    qualified_rows = bh_utils.query_data(
        f"""
        SELECT taskId
        FROM {RESULT_TABLE_NAME}
        WHERE pt = %s
          AND ifNull(platform, 'DB') = %s
          AND `count` BETWEEN %s AND %s
          AND has_keyword = 1
        GROUP BY taskId
        HAVING count(DISTINCT `count`) >= %s
        """,
        (
            int(pt),
            platform,
            hit_start_count,
            hit_end_count,
            hit_threshold,
        ),
    )
    if qualified_rows is None:
        raise RuntimeError("第五阶段查询已达标 taskId 失败")

    qualified_task_ids = {
        str(row.get("taskId"))
        for row in qualified_rows
        if row.get("taskId")
    }
    tasks = bh_utils.query_data(
        f"""
        SELECT taskId, prompt, keywords, pt
        FROM {TABLE_NAME}
        WHERE pt = %s
          AND ifNull(platform, 'DB') = %s
        """,
        (pt, platform),
    )
    if tasks is None:
        raise RuntimeError("第五阶段从主表查询未达标任务失败")

    unqualified_tasks = [
        task
        for task in tasks
        if task.get("taskId")
        and str(task.get("taskId")) not in qualified_task_ids
    ]
    if not unqualified_tasks:
        logger.success(
            f"[第五阶段 pt={pt} count=99] "
            f"没有需要生成的未达标任务"
        )
        return 0

    existing_rows = bh_utils.query_data(
        f"""
        SELECT taskId
        FROM {LOG_TABLE_NAME}
        WHERE pt = %s
          AND ifNull(platform, 'DB') = %s
          AND toInt32(`count`) = 99
        """,
        (pt, platform),
    )
    if existing_rows is None:
        raise RuntimeError("第五阶段查询已生成 count=99 任务失败")

    existing_task_ids = {
        str(row.get("taskId"))
        for row in existing_rows
        if row.get("taskId")
    }
    log_items = []
    for task in unqualified_tasks:
        task_id = str(task.get("taskId"))
        if task_id in existing_task_ids:
            continue
        log_items.append(
            {
                "reqId": str(uuid.uuid4()),
                "taskId": task_id,
                "prompt": task.get("prompt"),
                "keywords": task.get("keywords"),
                "`count`": "99",
                "pt": str(task.get("pt") or pt),
                "platform": platform,
            }
        )

    inserted_log_count = len(log_items)
    if log_items:
        if not bh_utils.insert_data(LOG_TABLE_NAME, log_items):
            raise RuntimeError("第五阶段写入 count=99 分享任务失败")
        logger.success(
            f"[第五阶段 pt={pt} count=99] "
            f"未达标分享任务生成完成: "
            f"task_ids={len(unqualified_tasks)}, "
            f"rows={inserted_log_count}"
        )
    else:
        logger.success(
            f"[第五阶段 pt={pt} count=99] "
            f"未达标分享任务均已存在"
        )

    process_rounds_to_result(
        pt=pt,
        start_count=99,
        end_count=99,
        redis_key=redis_key,
        platform=platform,
    )
    return inserted_log_count


def export_share_rounds_to_excel(
    pt: str,
    output_path: str = None,
    share_count: int = 0,
    platform: str = "DB",
) -> str:
    """
    导出当天指定分享轮次的任务及其 TOS 结果到 Excel。
    """
    if share_count not in (0, 99):
        raise ValueError("Excel 导出仅支持分享轮次 count=0 或 count=99")
    platform = normalize_platform(platform)
    platform_config = PLATFORM_CONFIGS[platform]
    pt = normalize_pt(pt)
    tos_pt = pt[:8]
    tos_directory = get_platform_tos_directory(platform)
    stage_name = "第四阶段" if share_count == 0 else "第六阶段"
    file_label = "达标数据" if share_count == 0 else "未达标"
    qualification_filter_sql = (
        "AND ifNull(a3.hit_round_count, 0) >= 24"
        if share_count == 0
        else "AND ifNull(a3.hit_round_count, 0) < 24"
    )

    rows = bh_utils.query_data(
        f"""
        SELECT DISTINCT
            a1.reqId,
            a1.taskId,
            a1.prompt,
            a2.platform,
            a2.insertime,
            a5.sheet_order,
            a3.processed_round_count,
            a3.hit_round_count,
            a4.has_keyword AS current_has_keyword
        FROM {LOG_TABLE_NAME} AS a1
        LEFT JOIN geo_third_task AS a2 ON a1.reqId = a2.reqId
        LEFT JOIN {TABLE_NAME} AS a5
          ON a1.taskId = a5.taskId
         AND a1.pt = a5.pt
         AND ifNull(a1.platform, 'DB')
             = ifNull(a5.platform, 'DB')
        LEFT JOIN (
            SELECT
                taskId,
                count(*) AS processed_round_count,
                sum(round_has_keyword) AS hit_round_count
            FROM (
                SELECT
                    taskId,
                    `count`,
                    max(has_keyword) AS round_has_keyword
                FROM {RESULT_TABLE_NAME}
                WHERE pt = %s
                  AND ifNull(platform, 'DB') = %s
                  AND `count` BETWEEN 1 AND 30
                GROUP BY taskId, `count`
            ) AS round_result
            GROUP BY taskId
        ) AS a3 ON a1.taskId = a3.taskId
        LEFT JOIN {RESULT_TABLE_NAME} AS a4
          ON a1.reqId = a4.reqId
         AND toInt32(a1.`count`) = a4.`count`
         AND ifNull(a1.platform, 'DB')
             = ifNull(a4.platform, 'DB')
        WHERE a1.pt = %s
          AND ifNull(a1.platform, 'DB') = %s
          AND toInt32(a1.`count`) = %s
          {qualification_filter_sql}
        ORDER BY a5.sheet_order, a1.reqId
        """,
        (
            int(pt),
            platform,
            pt,
            platform,
            share_count,
        ),
    )
    if rows is None:
        raise RuntimeError(
            f"{stage_name}查询 count={share_count} 导出任务失败"
        )

    task_keyword_rows = bh_utils.query_data(
        f"""
        SELECT taskId, keywords
        FROM {TABLE_NAME}
        WHERE pt = %s
          AND ifNull(platform, 'DB') = %s
        ORDER BY sheet_order, taskId
        """,
        (pt, platform),
    )
    if task_keyword_rows is None:
        raise RuntimeError(
            f"{stage_name}查询当天任务关键词失败"
        )

    round_keyword_rows = bh_utils.query_data(
        f"""
        SELECT taskId, `count`, keywords_count
        FROM {RESULT_TABLE_NAME}
        WHERE pt = %s
          AND ifNull(platform, 'DB') = %s
          AND `count` BETWEEN 1 AND 30
        """,
        (int(pt), platform),
    )
    if round_keyword_rows is None:
        raise RuntimeError(
            f"{stage_name}查询 1-30 轮关键词次数失败"
        )

    all_keywords = []
    seen_keywords = set()
    for task in task_keyword_rows:
        keywords_value = task.get("keywords")
        if isinstance(keywords_value, str):
            try:
                keywords = json.loads(keywords_value)
            except json.JSONDecodeError:
                keywords = [keywords_value]
        elif isinstance(keywords_value, list):
            keywords = keywords_value
        else:
            keywords = []

        for keyword in keywords:
            keyword = str(keyword).strip()
            if not keyword or keyword in seen_keywords:
                continue
            seen_keywords.add(keyword)
            all_keywords.append(keyword)

    keyword_totals_by_task = {}
    recommended_rounds_by_task = {}
    seen_task_rounds = set()
    for round_row in round_keyword_rows:
        task_id = str(round_row.get("taskId") or "")
        count = int(round_row.get("count") or 0)
        if not task_id or not 1 <= count <= 30:
            continue

        task_round_key = (task_id, count)
        if task_round_key in seen_task_rounds:
            continue
        seen_task_rounds.add(task_round_key)

        keywords_count_value = round_row.get("keywords_count")
        if isinstance(keywords_count_value, str):
            try:
                keywords_count = json.loads(keywords_count_value)
            except json.JSONDecodeError:
                keywords_count = []
        elif isinstance(keywords_count_value, list):
            keywords_count = keywords_count_value
        else:
            keywords_count = []

        task_keyword_totals = keyword_totals_by_task.setdefault(
            task_id,
            {},
        )
        has_recommended_keyword = False
        for keyword_item in keywords_count:
            if not isinstance(keyword_item, dict):
                continue
            keyword = str(
                keyword_item.get("keyword") or ""
            ).strip()
            if not keyword:
                continue
            try:
                word_count = int(
                    keyword_item.get("word_count") or 0
                )
            except (TypeError, ValueError):
                word_count = 0
            task_keyword_totals[keyword] = (
                task_keyword_totals.get(keyword, 0)
                + int(word_count > 0)
            )
            if keyword != "美团" and word_count > 0:
                has_recommended_keyword = True

        if has_recommended_keyword:
            recommended_rounds_by_task.setdefault(
                task_id,
                set(),
            ).add(count)

    if output_path is None:
        formatted_date = format_pt_display(
            pt,
            for_filename=True,
        )
        output_path = os.path.join(
            "/Users/yaowentong/Desktop",
            (
                f"{formatted_date}_"
                f"{platform_config['name']}_{file_label}.xlsx"
            ),
        )

    output_dir = os.path.dirname(os.path.abspath(output_path))
    os.makedirs(output_dir, exist_ok=True)

    workbook = Workbook()
    worksheet = workbook.active
    worksheet.title = (
        "达标分享结果"
        if share_count == 0
        else "未达标分享结果"
    )
    worksheet.append(
        [
            "问题",
            "是否达标",
            "达标轮次",
            "提及率",
            "推荐率",
            "本次是否达标",
            *all_keywords,
            "平台",
            "回答",
            "思考过程",
            "引用来源",
            "分享链接",
            "截图",
            "查询时间",
            "IP",
            "版本号",
            "文档顺序",
        ]
    )

    def timestamp_to_datetime(timestamp):
        if timestamp in (None, ""):
            return ""
        return datetime.fromtimestamp(
            int(timestamp)
        ).strftime("%Y-%m-%d %H:%M:%S")

    for row in rows:
        req_id = row.get("reqId")
        if not req_id:
            continue

        task_id = str(row.get("taskId") or "")
        prompt = row.get("prompt")
        row_platform = str(row.get("platform") or platform)
        insertime = row.get("insertime")
        sheet_order = row.get("sheet_order")
        processed_round_count = row.get("processed_round_count")
        hit_round_count = row.get("hit_round_count")
        hit_round_count_value = int(hit_round_count or 0)
        hit_rate = hit_round_count_value / 30
        recommendation_rate = (
            len(recommended_rounds_by_task.get(task_id, set()))
            / 30
        )
        if not processed_round_count:
            qualified_text = "没处理"
        elif hit_round_count_value >= 24:
            qualified_text = "是"
        else:
            qualified_text = "否"

        current_has_keyword = row.get("current_has_keyword")
        if current_has_keyword is None:
            current_qualified_text = "没处理"
        elif int(current_has_keyword) == 1:
            current_qualified_text = "是"
        else:
            current_qualified_text = "否"

        task_keyword_totals = keyword_totals_by_task.get(
            task_id,
            {},
        )
        keyword_total_values = [
            task_keyword_totals.get(keyword, 0)
            for keyword in all_keywords
        ]

        base_path = (
            f"geo_snipaste/{tos_pt}/"
            f"{tos_directory}/{req_id}"
        )
        text_json_path = f"{base_path}/text.json"
        context_path = f"{base_path}/context.txt"
        think_path = f"{base_path}/think.txt"
        quote_path = f"{base_path}/quote.txt"
        inner_user_ip_path = f"{base_path}/inner_user_ip.txt"
        inner_pc_version_path = (
            f"{base_path}/inner_pc_version.txt"
        )
        png_url = (
            f"https://tcdn.aidso.com/"
            f"geo_snipaste/{tos_pt}/"
            f"{tos_directory}/{req_id}/png.png"
        )

        text_json_str = (
            tos_utils.get_string_from_tos(text_json_path) or "{}"
        )
        try:
            share_url = json.loads(text_json_str).get(
                "share_url",
                "",
            )
        except json.JSONDecodeError:
            share_url = ""

        worksheet.append(
            [
                prompt,
                qualified_text,
                hit_round_count_value,
                hit_rate,
                recommendation_rate,
                current_qualified_text,
                *keyword_total_values,
                PLATFORM_CONFIGS.get(
                    row_platform,
                    platform_config,
                ).get("name", row_platform),
                tos_utils.get_string_from_tos(context_path),
                tos_utils.get_string_from_tos(think_path),
                tos_utils.get_string_from_tos(quote_path),
                share_url,
                png_url,
                timestamp_to_datetime(insertime),
                tos_utils.get_string_from_tos(
                    inner_user_ip_path
                ),
                tos_utils.get_string_from_tos(
                    inner_pc_version_path
                ),
                sheet_order,
            ]
        )

    for column_name in ("D", "E"):
        for cell in worksheet[column_name][1:]:
            cell.number_format = "0.00%"

    workbook.save(output_path)
    workbook.close()
    logger.success(
        f"{stage_name} Excel 导出完成: pt={pt}, "
        f"count={share_count}, "
        f"rows={len(rows)}, path={output_path}"
    )
    return output_path


def export_unqualified_share_rounds_to_excel(
    pt: str,
    output_path: str = None,
    platform: str = "DB",
) -> str:
    """
    第六阶段导出当天 count=99 的未达标分享结果。
    """
    return export_share_rounds_to_excel(
        pt=pt,
        output_path=output_path,
        share_count=99,
        platform=platform,
    )


def upload_file_to_feishu(file_path: str) -> str:
    if not os.path.isfile(file_path):
        raise FileNotFoundError(f"待上传文件不存在: {file_path}")

    access_token = get_tenant_access_token()
    headers = {
        "Authorization": f"Bearer {access_token}",
    }
    file_name = os.path.basename(file_path)
    with open(file_path, "rb") as file_object:
        data = request_json(
            "POST",
            "https://open.feishu.cn/open-apis/im/v1/files",
            headers=headers,
            data={
                "file_type": "stream",
                "file_name": file_name,
            },
            files={
                "file": (
                    file_name,
                    file_object,
                    (
                        "application/vnd.openxmlformats-"
                        "officedocument.spreadsheetml.sheet"
                    ),
                )
            },
        )

    file_key = data.get("data", {}).get("file_key")
    if not file_key:
        raise RuntimeError(f"飞书上传文件未返回 file_key: {data}")
    return file_key


def send_file_to_feishu_chat(
    file_key: str,
    chat_id: str,
) -> str:
    access_token = get_tenant_access_token()
    data = request_json(
        "POST",
        "https://open.feishu.cn/open-apis/im/v1/messages",
        headers=auth_headers(access_token),
        params={"receive_id_type": "chat_id"},
        json={
            "receive_id": chat_id,
            "msg_type": "file",
            "content": json.dumps(
                {"file_key": file_key},
                ensure_ascii=False,
            ),
        },
    )
    message_id = data.get("data", {}).get("message_id")
    if not message_id:
        raise RuntimeError(f"飞书发送文件未返回 message_id: {data}")
    return message_id


def upload_excel_to_feishu_chat(
    file_path: str,
    chat_id: str,
    pt: str = None,
    count_label: str = None,
) -> str:
    file_key = upload_file_to_feishu(file_path)
    message_id = send_file_to_feishu_chat(
        file_key=file_key,
        chat_id=chat_id,
    )
    logger.success(
        f"[pt={pt or '-'} count={count_label or '-'}] "
        f"Excel 已发送到飞书群: file={file_path}, "
        f"chat_id={chat_id}, message_id={message_id}"
    )
    return message_id


def _parse_json_list(value) -> list:
    if isinstance(value, list):
        return value
    if not isinstance(value, str) or not value.strip():
        return []
    try:
        parsed = json.loads(value)
    except json.JSONDecodeError:
        return [value]
    return parsed if isinstance(parsed, list) else []


def _parse_task_keywords(value) -> list[str]:
    keywords = []
    seen_keywords = set()
    for item in _parse_json_list(value):
        keyword = str(item or "").strip()
        if not keyword or keyword in seen_keywords:
            continue
        seen_keywords.add(keyword)
        keywords.append(keyword)
    return keywords


def _parse_round_keywords_count(value) -> dict:
    keyword_counts = {}
    for item in _parse_json_list(value):
        if not isinstance(item, dict):
            continue
        keyword = str(item.get("keyword") or "").strip()
        if not keyword:
            continue
        try:
            word_count = int(item.get("word_count") or 0)
        except (TypeError, ValueError):
            word_count = 0
        keyword_counts[keyword] = max(
            keyword_counts.get(keyword, 0),
            word_count,
        )
    return keyword_counts


def _excel_safe_text(value) -> str:
    text = str(value or "")
    text = ILLEGAL_EXCEL_CHARACTERS.sub("", text)
    return text[:EXCEL_CELL_MAX_LENGTH]


def _safe_get_tos_text(path: str) -> str:
    try:
        return tos_utils.get_string_from_tos(path) or ""
    except Exception as exc:
        logger.warning(
            f"30轮明细读取 TOS 失败: path={path}, error={exc}"
        )
        return ""


def _timestamp_to_text(timestamp) -> str:
    if timestamp in (None, ""):
        return ""
    try:
        timestamp_number = int(timestamp)
        if timestamp_number > 10 ** 12:
            timestamp_number //= 1000
        return datetime.fromtimestamp(timestamp_number).strftime(
            "%Y-%m-%d %H:%M:%S"
        )
    except (TypeError, ValueError, OSError, OverflowError):
        return str(timestamp)


def _sheet_order_sort_key(row: dict) -> tuple:
    try:
        sheet_order = int(row.get("sheet_order"))
        return (
            0,
            sheet_order,
            str(row.get("prompt") or ""),
        )
    except (TypeError, ValueError):
        return (
            1,
            0,
            str(row.get("prompt") or ""),
        )


def _chinese_round_number(number: int) -> str:
    digits = "零一二三四五六七八九"
    if number < 10:
        return digits[number]
    if number == 10:
        return "十"
    if number < 20:
        return f"十{digits[number % 10]}"
    if number % 10 == 0:
        return f"{digits[number // 10]}十"
    return f"{digits[number // 10]}十{digits[number % 10]}"


def _safe_file_name_part(value: str) -> str:
    normalized_value = str(value or "").strip() or "未分类"
    normalized_value = re.sub(
        r'[\\/:*?"<>|]',
        "_",
        normalized_value,
    )
    return normalized_value.strip(". ") or "未分类"


def _style_thirty_round_worksheet(
    worksheet,
    total_columns: int,
) -> None:
    worksheet.freeze_panes = "A2"
    worksheet.auto_filter.ref = worksheet.dimensions
    worksheet.sheet_view.showGridLines = False
    worksheet.row_dimensions[1].height = 32

    header_fill = PatternFill("solid", fgColor="6D28D9")
    header_font = Font(color="FFFFFF", bold=True)
    header_alignment = Alignment(
        horizontal="center",
        vertical="center",
        wrap_text=True,
    )
    row_border = Border(
        bottom=Side(style="thin", color="D8DCE6")
    )
    for cell in worksheet[1]:
        cell.fill = header_fill
        cell.font = header_font
        cell.alignment = header_alignment

    for row in worksheet.iter_rows(
        min_row=2,
        max_row=worksheet.max_row,
        max_col=total_columns,
    ):
        for cell in row:
            cell.alignment = Alignment(
                vertical="top",
                wrap_text=True,
            )
            cell.border = row_border

    column_widths = {
        "问题": 42,
        "批次": 16,
        "30次是否达标": 16,
        "本次是否达标": 16,
        "平台": 18,
        "回答": 70,
        "思考过程": 55,
        "引用来源": 55,
        "查询时间": 22,
        "IP": 18,
        "版本号": 20,
    }
    for column_index, cell in enumerate(
        worksheet[1],
        start=1,
    ):
        worksheet.column_dimensions[
            get_column_letter(column_index)
        ].width = column_widths.get(str(cell.value), 16)


def _upload_thirty_round_excel_to_tos(
    file_path: str,
    object_key: str,
    retry_times: int = 3,
) -> str:
    client = tos_client.get_client()
    bucket = tos_client.bucket_name
    with open(file_path, "rb") as file_object:
        file_content = file_object.read()

    last_error = None
    for attempt in range(1, retry_times + 1):
        try:
            client.put_object(
                bucket,
                object_key,
                content=file_content,
                content_type=(
                    "application/vnd.openxmlformats-officedocument."
                    "spreadsheetml.sheet"
                ),
            )
            logger.success(
                f"[第七阶段 count=1-30] 30轮明细已上传 TOS: "
                f"tos://{bucket}/{object_key}, "
                f"attempt={attempt}/{retry_times}"
            )
            return f"tos://{bucket}/{object_key}"
        except Exception as exc:
            last_error = exc
            logger.warning(
                f"[第七阶段 count=1-30] 30轮明细上传失败: "
                f"object_key={object_key}, "
                f"attempt={attempt}/{retry_times}, error={exc}"
            )

    raise RuntimeError(
        f"30轮明细上传 TOS 连续失败 {retry_times} 次: "
        f"object_key={object_key}"
    ) from last_error


def export_thirty_round_details(
    pt: str,
    platform: str = "DB",
) -> list[dict]:
    """
    按业务线导出当前 pt、当前平台的 1-30 轮明细并上传到 TOS。

    返回可直接写入飞书日报的业务线及 CDN 下载地址列表。
    """
    from collections import defaultdict
    from concurrent.futures import ThreadPoolExecutor

    platform = normalize_platform(platform)
    pt = normalize_pt(pt)
    platform_config = PLATFORM_CONFIGS[platform]
    tos_directory = get_platform_tos_directory(platform)

    log_rows = bh_utils.query_data(
        f"""
        SELECT
            a1.reqId,
            a1.taskId,
            a1.prompt,
            a1.pt,
            toInt32(a1.`count`) AS round_count,
            a2.sheet_order,
            a2.business_line,
            a3.insertime
        FROM {LOG_TABLE_NAME} AS a1
        LEFT JOIN {TABLE_NAME} AS a2
          ON a1.taskId = a2.taskId
         AND a1.pt = a2.pt
         AND ifNull(a1.platform, 'DB')
             = ifNull(a2.platform, 'DB')
        LEFT JOIN geo_third_task AS a3
          ON a1.reqId = a3.reqId
        WHERE toString(a1.pt) = %s
          AND ifNull(a1.platform, 'DB') = %s
          AND toInt32(a1.`count`) BETWEEN 1 AND 30
        """,
        (pt, platform),
    )
    if log_rows is None:
        raise RuntimeError("第七阶段查询 1-30 轮日志数据失败")

    result_rows = bh_utils.query_data(
        f"""
        SELECT
            reqId,
            taskId,
            pt,
            toInt32(`count`) AS round_count,
            has_keyword
        FROM {RESULT_TABLE_NAME}
        WHERE toString(pt) = %s
          AND ifNull(platform, 'DB') = %s
          AND toInt32(`count`) BETWEEN 1 AND 30
        """,
        (pt, platform),
    )
    if result_rows is None:
        raise RuntimeError("第七阶段查询 1-30 轮结果数据失败")

    result_by_req_id = {}
    task_round_hits = defaultdict(dict)
    for result_row in result_rows:
        req_id = str(result_row.get("reqId") or "")
        task_id = str(result_row.get("taskId") or "")
        result_pt = str(result_row.get("pt") or "")
        try:
            round_count = int(
                result_row.get("round_count") or 0
            )
            has_keyword = int(
                result_row.get("has_keyword") or 0
            )
        except (TypeError, ValueError):
            continue
        if (
            not req_id
            or not task_id
            or not 1 <= round_count <= 30
        ):
            continue

        current_result = result_by_req_id.setdefault(
            req_id,
            {
                "has_keyword": 0,
            },
        )
        current_result["has_keyword"] = max(
            current_result["has_keyword"],
            has_keyword,
        )
        task_key = (result_pt, task_id)
        task_round_hits[task_key][round_count] = max(
            task_round_hits[task_key].get(round_count, 0),
            has_keyword,
        )

    rows_by_business_line = defaultdict(list)
    for log_row in log_rows:
        business_line = str(
            log_row.get("business_line") or ""
        ).strip() or "未分类"
        rows_by_business_line[business_line].append(log_row)

    downloads = []
    for business_line in sorted(rows_by_business_line):
        business_line_rows = rows_by_business_line[business_line]

        rows_by_round = defaultdict(list)
        for log_row in business_line_rows:
            try:
                round_count = int(
                    log_row.get("round_count") or 0
                )
            except (TypeError, ValueError):
                continue
            if 1 <= round_count <= 30:
                rows_by_round[round_count].append(log_row)

        safe_business_line = _safe_file_name_part(business_line)
        file_name = (
            f"{pt}_{platform_config['name']}_"
            f"{safe_business_line}_30轮明细.xlsx"
        )
        local_path = os.path.join(
            tempfile.gettempdir(),
            file_name,
        )
        headers = [
            "问题",
            "批次",
            "30次是否达标",
            "本次是否达标",
            "平台",
            "回答",
            "思考过程",
            "引用来源",
            "查询时间",
            "IP",
            "版本号",
        ]

        def build_excel_row(log_row: dict) -> list:
            req_id = str(log_row.get("reqId") or "")
            task_id = str(log_row.get("taskId") or "")
            row_pt = str(log_row.get("pt") or "")
            result_item = result_by_req_id.get(req_id)
            round_hits = task_round_hits.get(
                (row_pt, task_id),
                {},
            )
            if not round_hits:
                thirty_round_qualified = "没处理"
            else:
                hit_round_count = sum(
                    1
                    for value in round_hits.values()
                    if value == 1
                )
                thirty_round_qualified = (
                    "是" if hit_round_count >= 24 else "否"
                )

            if result_item is None:
                current_qualified = "没处理"
            else:
                current_qualified = (
                    "是"
                    if int(
                        result_item.get("has_keyword") or 0
                    ) == 1
                    else "否"
                )

            base_path = (
                f"geo_snipaste/{row_pt[:8]}/"
                f"{tos_directory}/{req_id}"
            )
            return [
                _excel_safe_text(log_row.get("prompt")),
                row_pt,
                thirty_round_qualified,
                current_qualified,
                platform_config["name"],
                _excel_safe_text(
                    _safe_get_tos_text(
                        f"{base_path}/context.txt"
                    )
                ),
                _excel_safe_text(
                    _safe_get_tos_text(
                        f"{base_path}/think.txt"
                    )
                ),
                _excel_safe_text(
                    _safe_get_tos_text(
                        f"{base_path}/quote.txt"
                    )
                ),
                _timestamp_to_text(log_row.get("insertime")),
                _excel_safe_text(
                    _safe_get_tos_text(
                        f"{base_path}/inner_user_ip.txt"
                    )
                ),
                _excel_safe_text(
                    _safe_get_tos_text(
                        f"{base_path}/inner_pc_version.txt"
                    )
                ),
            ]

        workbook = Workbook()
        workbook.remove(workbook.active)
        total_rows = 0
        with ThreadPoolExecutor(
            max_workers=THIRTY_ROUND_EXPORT_WORKERS
        ) as executor:
            for round_count in range(1, 31):
                worksheet = workbook.create_sheet(
                    title=(
                        f"第"
                        f"{_chinese_round_number(round_count)}轮"
                    )
                )
                worksheet.append(headers)
                round_rows = [
                    row
                    for row in sorted(
                        rows_by_round.get(round_count, []),
                        key=_sheet_order_sort_key,
                    )
                    if row.get("reqId")
                ]
                for excel_row in executor.map(
                    build_excel_row,
                    round_rows,
                ):
                    worksheet.append(excel_row)
                    total_rows += 1

                _style_thirty_round_worksheet(
                    worksheet,
                    len(headers),
                )
                logger.info(
                    f"[第七阶段 platform={platform} pt={pt} "
                    f"business_line={business_line} "
                    f"count={round_count}] 30轮明细 Sheet 完成: "
                    f"rows={len(round_rows)}"
                )

        workbook.save(local_path)
        workbook.close()
        object_key = (
            f"geo_snipaste/{pt[:8]}/"
            f"{tos_directory}/{file_name}"
        )
        try:
            _upload_thirty_round_excel_to_tos(
                file_path=local_path,
                object_key=object_key,
            )
        finally:
            try:
                os.remove(local_path)
            except OSError:
                pass

        cdn_url = (
            f"https://tcdn.aidso.com/"
            f"geo_snipaste/{pt[:8]}/"
            f"{tos_directory}/{file_name}"
        )
        downloads.append(
            {
                "business_line": business_line,
                "url": cdn_url,
            }
        )
        logger.success(
            f"[第七阶段 platform={platform} pt={pt} "
            f"business_line={business_line} count=1-30] "
            f"30轮明细导出完成: rows={total_rows}, "
            f"url={cdn_url}"
        )

    if not downloads:
        logger.warning(
            f"[第七阶段 platform={platform} pt={pt} count=1-30] "
            f"没有可导出的30轮业务线数据"
        )
    return downloads


def build_daily_run_report(
    pt: str,
    platform: str = "DB",
    thirty_round_downloads: list = None,
) -> str:
    """
    汇总当天各阶段运行结果，生成 Markdown 报告。
    """
    platform = normalize_platform(platform)
    platform_name = PLATFORM_CONFIGS[platform]["name"]
    pt = normalize_pt(pt)
    tasks = bh_utils.query_data(
        f"""
        SELECT taskId, prompt, sheet_order
        FROM {TABLE_NAME}
        WHERE pt = %s
          AND ifNull(platform, 'DB') = %s
        """,
        (pt, platform),
    )
    if tasks is None:
        raise RuntimeError("最后阶段查询当天主任务失败")

    result_rows = bh_utils.query_data(
        f"""
        SELECT taskId, `count`, has_keyword
        FROM {RESULT_TABLE_NAME}
        WHERE pt = %s
          AND ifNull(platform, 'DB') = %s
          AND `count` BETWEEN 1 AND 30
        """,
        (int(pt), platform),
    )
    if result_rows is None:
        raise RuntimeError("最后阶段查询关键词统计结果失败")

    task_map = {
        str(task.get("taskId")): task
        for task in tasks
        if task.get("taskId")
    }
    hit_rounds_by_task = {
        task_id: set()
        for task_id in task_map
    }
    for row in result_rows:
        task_id = str(row.get("taskId") or "")
        if (
            task_id in hit_rounds_by_task
            and int(row.get("has_keyword") or 0) == 1
        ):
            hit_rounds_by_task[task_id].add(
                int(row.get("count"))
            )

    thirty_round_qualified = {
        task_id
        for task_id, hit_rounds in hit_rounds_by_task.items()
        if len(
            {
                count
                for count in hit_rounds
                if 1 <= count <= 30
            }
        ) >= 24
    }
    thirty_round_failed = set(task_map) - thirty_round_qualified

    markdown_lines = [
        f"## {platform_name} GEO 今日任务运行总结",
        "",
        f"**统计日期：** {format_pt_display(pt)}",
        "",
        (
            "**统计口径：** 同一问题按第 1–30 轮去重统计"
            "关键词命中轮次数，命中次数达到 24 次视为达标。"
        ),
        "",
        f"- 今日问题任务总数：{len(task_map)} 个",
        f"- 达标问题：{len(thirty_round_qualified)} 个",
        f"- 未达标问题：{len(thirty_round_failed)} 个",
    ]
    if thirty_round_downloads:
        markdown_lines.extend(
            [
                "",
                "**30轮数据如下：**",
                "",
            ]
        )
        for download in thirty_round_downloads:
            business_line = str(
                download.get("business_line") or "未分类"
            )
            download_url = str(download.get("url") or "")
            if not download_url:
                continue
            markdown_lines.append(
                f"- {business_line}："
                f"[点击下载30轮明细]({download_url})"
            )
    markdown_lines.extend(
        [
            "",
            "### 未达标问题 Top 10",
            "",
        ]
    )

    def failed_task_sort_key(task):
        task_id = str(task.get("taskId") or "")
        hit_count = len(
            {
                count
                for count in hit_rounds_by_task.get(task_id, set())
                if 1 <= count <= 30
            }
        )
        try:
            sheet_order = int(task.get("sheet_order"))
        except (TypeError, ValueError):
            sheet_order = 10 ** 9
        return (
            hit_count,
            sheet_order,
            str(task.get("prompt") or ""),
        )

    failed_tasks = sorted(
        (
            task_map[task_id]
            for task_id in thirty_round_failed
        ),
        key=failed_task_sort_key,
    )[:10]
    if failed_tasks:
        for index, task in enumerate(failed_tasks, start=1):
            markdown_lines.append(
                f"{index}. {task.get('prompt') or ''}"
            )
    else:
        markdown_lines.append("无")

    return "\n".join(markdown_lines)


def send_markdown_report_to_feishu_chat(
    markdown_text: str,
    chat_id: str,
    platform: str = "DB",
) -> str:
    platform = normalize_platform(platform)
    card_title = (
        f"{PLATFORM_CONFIGS[platform]['name']} "
        f"GEO 今日任务运行总结"
    )
    card_markdown_lines = []
    for line in markdown_text.splitlines():
        stripped_line = line.strip()
        if stripped_line.startswith("#"):
            heading_text = stripped_line.lstrip("#").strip()
            if heading_text == card_title:
                continue
            card_markdown_lines.extend(
                [
                    "",
                    f"**{heading_text}**",
                    "",
                ]
            )
            continue
        card_markdown_lines.append(line)
    card_markdown = "\n".join(card_markdown_lines).strip()

    access_token = get_tenant_access_token()
    card = {
        "config": {
            "wide_screen_mode": True,
        },
        "header": {
            "template": "blue",
            "title": {
                "tag": "plain_text",
                "content": card_title,
            },
        },
        "elements": [
            {
                "tag": "markdown",
                "content": card_markdown,
            }
        ],
    }
    data = request_json(
        "POST",
        "https://open.feishu.cn/open-apis/im/v1/messages",
        headers=auth_headers(access_token),
        params={"receive_id_type": "chat_id"},
        json={
            "receive_id": chat_id,
            "msg_type": "interactive",
            "content": json.dumps(
                card,
                ensure_ascii=False,
            ),
        },
    )
    message_id = data.get("data", {}).get("message_id")
    if not message_id:
        raise RuntimeError(
            f"飞书发送运行报告未返回 message_id: {data}"
        )
    logger.success(
        f"[第七阶段 count=1-30] 运行报告已发送: "
        f"chat_id={chat_id}, message_id={message_id}"
    )
    return message_id


def run_daily_pipeline(
    pt: str = None,
    platform: str = "DB",
) -> None:
    """
    按顺序执行飞书截图任务的第一至第七阶段。
    """
    platform = normalize_platform(platform)
    platform_config = PLATFORM_CONFIGS[platform]
    pt = normalize_pt(pt)
    start_count = 1
    end_count = 30

    logger.success(
        f"[每日任务 platform={platform} pt={pt} count=1-99] "
        f"七阶段任务开始执行"
    )

    # 第一阶段：飞书同步到主表和 30 轮日志表。
    sync_feishu_to_database(pt=pt, platform=platform)

    # 第二阶段：所有任务发送 1-30 轮，完成三轮补发、解析和结果入库。
    process_rounds_to_result(
        pt=pt,
        start_count=start_count,
        end_count=end_count,
        redis_key=platform_config["content_redis_key"],
        platform=platform,
    )

    # 第三阶段：1-30 轮命中不少于 24 轮的任务生成 count=0 分享任务。
    create_and_process_share_rounds(
        pt=pt,
        redis_key=platform_config["share_redis_key"],
        hit_start_count=1,
        hit_end_count=30,
        hit_threshold=24,
        platform=platform,
    )

    # 第四阶段：导出 count=0 的达标分享结果。
    excel_path = export_share_rounds_to_excel(
        pt=pt,
        platform=platform,
    )
    upload_excel_to_feishu_chat(
        file_path=excel_path,
        chat_id=FEISHU_CHAT_ID,
        pt=pt,
        count_label="0",
    )

    # 第五阶段：1-30 轮命中不足 24 轮的任务生成 count=99 分享任务。
    create_and_process_unqualified_share_rounds(
        pt=pt,
        redis_key=platform_config["share_redis_key"],
        hit_start_count=1,
        hit_end_count=30,
        hit_threshold=24,
        platform=platform,
    )

    # 第六阶段：导出 count=99 的未达标分享结果。
    unqualified_excel_path = (
        export_unqualified_share_rounds_to_excel(
            pt=pt,
            platform=platform,
        )
    )
    upload_excel_to_feishu_chat(
        file_path=unqualified_excel_path,
        chat_id=FEISHU_CHAT_ID,
        pt=pt,
        count_label="99",
    )

    # 第七阶段：导出30轮明细，并在今日运行总结中发送下载链接。
    thirty_round_downloads = export_thirty_round_details(
        pt=pt,
        platform=platform,
    )
    markdown_report = build_daily_run_report(
        pt=pt,
        platform=platform,
        thirty_round_downloads=thirty_round_downloads,
    )
    send_markdown_report_to_feishu_chat(
        markdown_text=markdown_report,
        chat_id=FEISHU_CHAT_ID,
        platform=platform,
    )

    logger.success(
        f"[每日任务 platform={platform} pt={pt} count=1-99] "
        f"七阶段任务执行完成"
    )


def run_daily_pipeline_safely(
    pt: str = None,
    platform: str = "DB",
) -> None:
    try:
        run_daily_pipeline(pt=pt, platform=platform)
    except Exception:
        logger.exception(
            f"[每日任务 platform={platform} count=1-99] "
            f"七阶段任务执行失败"
        )


if __name__ == "__main__":
    # pt = '20260803'
    # platform_db = 'DB'
    # platform_dp = 'DP'
    # run_daily_pipeline(pt=pt, platform=platform_db)
    # run_daily_pipeline(pt=pt, platform=platform_dp)
    start_scheduler()
