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

from __future__ import annotations

import json
import os
import sys
import traceback
import uuid
from datetime import datetime
import time
import redis
import requests
from loguru import logger
from openpyxl import Workbook

from aidso_geo.models import spider_save_tos
from aidso_geo.utils.tos_utils import get_string_from_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"
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

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


class ResultWriteError(RuntimeError):
    pass


class TaskResultUnavailableError(RuntimeError):
    pass


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_first_sheet(spreadsheet_token: str, access_token: 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}")

    sheet = sheets[0]
    if not sheet.get("sheet_id"):
        raise RuntimeError(f"飞书返回中没有 sheet_id: {sheets[0]}")
    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) -> list[list]:
    sheet = get_first_sheet(spreadsheet_token, access_token)
    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) -> list[dict]:
    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 cell in row[1:]:
            keyword = cell_text(cell)
            if keyword:
                keywords.append(keyword)
        tasks.append(
            {
                "taskId": str(uuid.uuid4()),
                "prompt": prompt,
                "keywords": json.dumps(keywords, ensure_ascii=False),
                "pt": pt,
                "sheet_order": len(tasks) + 1,
            }
        )
    return tasks


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


def create_round_logs(tasks: list[dict], pt: str) -> int:
    existing_rows = bh_utils.query_data(
        (
            f"SELECT taskId, `count` FROM {LOG_TABLE_NAME} "
            f"WHERE pt = %s"
        ),
        (pt,),
    )
    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,
                }
            )

    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,
) -> int:
    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 toInt32(`count`) BETWEEN %s AND %s "
            f"{task_filter_sql}"
            f"ORDER BY toInt32(`count`), taskId"
        ),
        tuple(query_params),
    )
    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) -> None:
    validate_config()
    pt = normalize_pt(pt)
    tasks = get_tasks(pt)
    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)

        tasks = build_tasks(table_rows, pt)
        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)

    logger.success(
        f"[第一阶段 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")
    scheduler.add_job(
        run_daily_pipeline_safely,
        trigger="cron",
        hour=SYNC_HOUR,
        minute=SYNC_MINUTE,
        second=0,
        id="feishu_snipaste_v3_daily_pipeline",
        replace_existing=True,
        max_instances=1,
        coalesce=True,
        misfire_grace_time=3600,
    )
    logger.success(
        f"[调度 count=1-99] 七阶段每日任务已注册: "
        f"{SYNC_HOUR:02d}:{SYNC_MINUTE:02d}"
    )
    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
    response_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 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 get_task_content_with_retries(
    row: dict,
    file: str,
    redis_key: str,
    max_retries: int = 3,
):
    req_id = row.get("reqId")
    redis_pt = normalize_pt(row.get("pt"))[:8]
    last_error = None
    for retry_index in range(max_retries + 1):
        try:
            raw = tos_utils.get_string_from_tos(file)
        except Exception as exc:
            last_error = exc
            raw = None
        if raw:
            try:
                content = json.loads(raw).get("content")
            except (TypeError, json.JSONDecodeError) as exc:
                last_error = exc
                content = None
            if content:
                return raw, content

        if retry_index >= max_retries:
            break

        retry_value = json.dumps(
            {
                "reqId": req_id,
                "prompt": row.get("prompt"),
                "pt": redis_pt,
            },
            ensure_ascii=False,
        )
        redis_client = init_redis()
        if redis_client is None:
            raise TaskResultUnavailableError("Redis 初始化失败")
        try:
            redis_client.rpush(redis_key, retry_value)
        except Exception as exc:
            raise TaskResultUnavailableError(
                f"TOS 结果缺失任务重发 Redis 失败: "
                f"key={redis_key}, reqId={req_id}"
            ) from exc

        logger.warning(
            f"[pt={row.get('pt')} count={row.get('count')} "
            f"reqId={req_id}] TOS 结果缺失或正文为空，"
            f"开始第 {retry_index + 1}/{max_retries} 次重试"
        )
        try:
            wait_mt_tasks_done(
                redis_key,
                pt=str(row.get("pt") or ""),
                count_label=str(row.get("count") or ""),
            )
        except Exception as exc:
            raise TaskResultUnavailableError(
                f"TOS 结果缺失任务等待 Redis 失败: "
                f"key={redis_key}, reqId={req_id}"
            ) from exc
        time.sleep(60)

    error = TaskResultUnavailableError(
        f"TOS 结果重试 {max_retries} 次后仍不可用: "
        f"reqId={req_id}, file={file}"
    )
    if last_error is not None:
        raise error from last_error
    raise error


def process_rounds_to_result(
    pt: str,
    start_count: int,
    end_count: int,
    redis_key: str,
    task_ids: list = None,
) -> int:
    """
    发送指定轮次，等待并补发未完成任务，
    最后解析 TOS 正文、统计关键词并写入结果表。
    """
    pt = normalize_pt(pt)
    redis_pt = pt[:8]
    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,
    )
    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 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,
                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 toInt32(`count`) BETWEEN %s AND %s
          {task_filter_sql}
        """,
        (
            pt,
            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}/doubao/"
            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 = []

            raw, content = get_task_content_with_retries(
                row=row,
                file=file,
                redis_key=redis_key,
            )

            (
                _,
                _,
                _,
                _,
                response_content,
                _,
            ) = doubao_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,
            }
        except Exception as exc:
            logger.warning(
                f"[pt={pt} count={start_count}-{end_count}] "
                f"处理 {file} 失败: {exc}"
            )
            if isinstance(
                exc,
                (ResultWriteError, TaskResultUnavailableError),
            ):
                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 = []
        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 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))
        return parsed_results

    def insert_result_item(result_item, share_retry_count=0):
        req_id = result_item.get("reqId")
        if not bh_utils.insert_data(
            "douchacha_data.geo_feishu_snipaste_result_v3",
            [result_item],
        ):
            raise ResultWriteError(
                f"关键词统计结果写入失败: reqId={req_id}"
            )
        logger.success(
            f"[pt={result_item.get('pt')} "
            f"count={result_item.get('`count`')} "
            f"reqId={req_id}] 关键词统计完成: "
            f"has_keyword={result_item.get('has_keyword')}, "
            f"keywords_count={result_item.get('keywords_count')}, "
            f"share_retries={share_retry_count}"
        )
        return 1

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

    if not is_count_zero:
        for _, result_item in parsed_results:
            inserted_count += insert_result_item(result_item)
    else:
        pending_results = []
        for row, result_item in parsed_results:
            if int(result_item.get("has_keyword") or 0) == 1:
                inserted_count += insert_result_item(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_result_batch(
                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:
                    inserted_count += insert_result_item(
                        result_item,
                        share_retry_count=share_retry_round,
                    )
                else:
                    next_pending_results.append((row, result_item))
            pending_results = next_pending_results

        for _, result_item in pending_results:
            inserted_count += insert_result_item(
                result_item,
                share_retry_count=5,
            )

    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,
) -> int:
    """
    第 1-30 轮中关键词命中轮数达到阈值的 taskId，
    生成 count=0 的分享任务并完成 Redis、TOS 和结果入库流程。
    """
    qualified_rows = bh_utils.query_data(
        """
        SELECT taskId
        FROM douchacha_data.geo_feishu_snipaste_result_v3
        WHERE pt = %s
          AND `count` BETWEEN %s AND %s
          AND has_keyword = 1
        GROUP BY taskId
        HAVING count(DISTINCT `count`) >= %s
        """,
        (
            int(pt),
            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 taskId IN ({task_id_placeholders})
        """,
        (
            pt,
            *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 toInt32(`count`) = 0
          AND taskId IN ({task_id_placeholders})
        """,
        (
            pt,
            *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),
            }
        )

    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,
    )
    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,
) -> int:
    """
    第 1-30 轮中关键词命中轮数不足阈值的 taskId，
    生成 count=99 的分享任务并完成 Redis、TOS 和结果入库流程。
    """
    qualified_rows = bh_utils.query_data(
        """
        SELECT taskId
        FROM douchacha_data.geo_feishu_snipaste_result_v3
        WHERE pt = %s
          AND `count` BETWEEN %s AND %s
          AND has_keyword = 1
        GROUP BY taskId
        HAVING count(DISTINCT `count`) >= %s
        """,
        (
            int(pt),
            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
        """,
        (pt,),
    )
    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 toInt32(`count`) = 99
        """,
        (pt,),
    )
    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),
            }
        )

    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,
    )
    return inserted_log_count


def export_share_rounds_to_excel(
    pt: str,
    output_path: str = None,
    share_count: int = 0,
) -> str:
    """
    导出当天指定分享轮次的任务及其 TOS 结果到 Excel。
    """
    if share_count not in (0, 99):
        raise ValueError("Excel 导出仅支持分享轮次 count=0 或 count=99")
    pt = normalize_pt(pt)
    tos_pt = pt[:8]
    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
        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 douchacha_data.geo_feishu_snipaste_result_v3
                WHERE pt = %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 douchacha_data.geo_feishu_snipaste_result_v3 AS a4
          ON a1.reqId = a4.reqId
         AND toInt32(a1.`count`) = a4.`count`
        WHERE a1.pt = %s
          AND toInt32(a1.`count`) = %s
          {qualification_filter_sql}
        ORDER BY a5.sheet_order, a1.reqId
        """,
        (int(pt), pt, 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
        ORDER BY sheet_order, taskId
        """,
        (pt,),
    )
    if task_keyword_rows is None:
        raise RuntimeError(
            f"{stage_name}查询当天任务关键词失败"
        )

    round_keyword_rows = bh_utils.query_data(
        """
        SELECT taskId, `count`, keywords_count
        FROM douchacha_data.geo_feishu_snipaste_result_v3
        WHERE pt = %s
          AND `count` BETWEEN 1 AND 30
        """,
        (int(pt),),
    )
    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 = {}
    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,
            {},
        )
        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 output_path is None:
        formatted_date = format_pt_display(
            pt,
            for_filename=True,
        )
        output_path = os.path.join(
            "/Users/yaowentong/Desktop",
            f"{formatted_date}_{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",
            "版本号",
            "文档顺序",
        ]
    )

    plat_form_map = {
        "DB": "豆包网页版",
    }

    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")
        platform = row.get("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")
        if not processed_round_count:
            qualified_text = "没处理"
        elif int(hit_round_count or 0) >= 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}/doubao/{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}/doubao/{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,
                current_qualified_text,
                *keyword_total_values,
                plat_form_map.get(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,
            ]
        )

    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,
) -> str:
    """
    第六阶段导出当天 count=99 的未达标分享结果。
    """
    return export_share_rounds_to_excel(
        pt=pt,
        output_path=output_path,
        share_count=99,
    )


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 build_daily_run_report(pt: str) -> str:
    """
    汇总当天各阶段运行结果，生成 Markdown 报告。
    """
    pt = normalize_pt(pt)
    tasks = bh_utils.query_data(
        f"""
        SELECT taskId, prompt, sheet_order
        FROM {TABLE_NAME}
        WHERE pt = %s
        """,
        (pt,),
    )
    if tasks is None:
        raise RuntimeError("最后阶段查询当天主任务失败")

    result_rows = bh_utils.query_data(
        """
        SELECT taskId, `count`, has_keyword
        FROM douchacha_data.geo_feishu_snipaste_result_v3
        WHERE pt = %s
          AND `count` BETWEEN 1 AND 30
        """,
        (int(pt),),
    )
    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 = [
        "## 美团 GEO 今日任务运行总结",
        "",
        f"**统计日期：** {format_pt_display(pt)}",
        "",
        (
            "**统计口径：** 同一问题按第 1–30 轮去重统计"
            "关键词命中轮次数，命中次数达到 24 次视为达标。"
        ),
        "",
        f"- 今日问题任务总数：{len(task_map)} 个",
        f"- 达标问题：{len(thirty_round_qualified)} 个",
        f"- 未达标问题：{len(thirty_round_failed)} 个",
        "",
        "### 未达标问题 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,
) -> str:
    card_title = "美团 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) -> None:
    """
    按顺序执行飞书截图任务的第一至第七阶段。
    """
    pt = normalize_pt(pt)
    start_count = 1
    end_count = 30

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

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

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

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

    # 第四阶段：导出 count=0 的达标分享结果。
    excel_path = export_share_rounds_to_excel(pt=pt)
    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=REDIS_QUEUE_KEY_SHARE,
        hit_start_count=1,
        hit_end_count=30,
        hit_threshold=24,
    )

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

    # 第七阶段：生成并发送今日运行总结。
    markdown_report = build_daily_run_report(pt=pt)
    send_markdown_report_to_feishu_chat(
        markdown_text=markdown_report,
        chat_id=FEISHU_CHAT_ID,
    )

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


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


if __name__ == "__main__":
    # run_daily_pipeline()
    start_scheduler()
    # run_daily_pipeline(pt="2026072702")
