美团截图30遍

aidso_geo/core/feishu_snipaste_v3.py

+"""每日把飞书 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")