# -*- coding: utf-8 -*-
import secrets
import string
import time
import uuid
import threading
import redis
import openpyxl
from concurrent.futures import ThreadPoolExecutor, as_completed
import json
import os
from datetime import datetime, timedelta
from openpyxl.workbook import Workbook
import os ,sys
from loguru import logger


from aidso_geo.models import spider_save_tos
from aidso_geo.utils import bh_utils, tos_utils


# =========================
# 配置区
# =========================

import requests
import json

def get_token():
    url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"

    payload = json.dumps({
        "app_id": "cli_a903ede61ff99bcc",
        "app_secret": "yiJkjDnHcbYhIWD0Hb26OfuxUXAu13ra"
    })
    headers = {
        'Content-Type': 'application/json'
    }

    response = requests.request("POST", url, headers=headers, data=payload)

    return response.json().get('tenant_access_token')

container_id = "oc_41bc8e496a0d9bb93498a6a6cf6f30c1"

# 每天 23:00 生成第二天任务
SCHEDULE_HOUR = 23
SCHEDULE_MINUTE = 0
SCHEDULE_SECOND = 0

# 等待任务跑完（Redis mt_third_task 为空）轮询间隔
POLL_SECONDS = 180

# 到第二天 00:00 后稍微缓一会儿再开跑，避免边界抖动
AFTER_MIDNIGHT_DELAY_SECONDS = 60

# Excel 存放目录
# EXCEL_DIR = "/opt/file"
EXCEL_DIR = "/Users/yaowentong/Desktop"

# Redis 去重锁过期时间（秒）
INIT_TASK_LOCK_EXPIRE = 3 * 24 * 3600   # 3天：防止 key 堆积
RUNNER_LOCK_EXPIRE = 3 * 24 * 3600      # 3天：防止 key 堆积


# =========================
# 飞书上传/发消息
# =========================

def upload_file(file_path, file_name):
    url = "https://open.feishu.cn/open-apis/im/v1/files"
    headers = {"Authorization": f"Bearer {get_token()}"}

    files = {
        "file": (
            file_name,
            open(file_path, "rb"),
            "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
        )
    }
    data = {
        "file_type": "xlsx",
        "file_name": file_name
    }

    try:
        resp = requests.post(url=url, headers=headers, files=files, data=data, timeout=30)
        rj = resp.json()
        return (rj.get("data") or {}).get("file_key")
    except FileNotFoundError as e:
        logger.success(f"错误：文件不存在 - {e}")
    except Exception as e:
        logger.success(f"请求失败：{e}")
    finally:
        try:
            f = files.get("file")[1]
            if f:
                f.close()
        except Exception:
            pass

    return None


def chat_message(file_key):
    url = "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id"
    payload = json.dumps({
        "content": f'{{"file_key":"{file_key}"}}',
        "msg_type": "file",
        "receive_id": container_id,
        "uuid": str(uuid.uuid4())
    })
    headers = {
        "Authorization": f"Bearer {get_token()}",
        "Content-Type": "application/json"
    }

    resp = requests.request("POST", url, headers=headers, data=payload, timeout=30)
    logger.success(resp.text)


def send_message(message):
    url = "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id"

    payload = {
        "receive_id": container_id,
        "msg_type": "text",
        "content": json.dumps({
            "text": message
        }, ensure_ascii=False)
    }

    headers = {
       'Authorization': f'Bearer {get_token()}',
       'Content-Type': 'application/json'
    }

    response = requests.request("POST", url, headers=headers, data=json.dumps(payload))

    logger.success(response.text)





def get_task():
    all_items = []
    page_token = None

    while True:
        url = f"https://open.feishu.cn/open-apis/docx/v1/chats/{container_id}/announcement/blocks?page_size=500&revision_id=-1"
        headers = {
            "Authorization": f"Bearer {get_token()}"
        }

        params = {}
        if page_token:
            params["page_token"] = page_token

        try:
            resp = requests.get(url, headers=headers, params=params, timeout=30)
            resp.raise_for_status()
            data = resp.json()


            page_data = data.get("data") or {}
            items = page_data.get("items") or []
            has_more = page_data.get("has_more", False)
            page_token = page_data.get("page_token")

            all_items.extend(items)

            if not has_more:
                break

        except Exception as e:
            logger.success(f"请求失败：{e}")
            break

    return all_items


# =========================
# Redis
# =========================

def init_redis4():
    try:
        redis_client = redis.Redis(
            host="172.16.0.24",
            port=6379,
            db=4,
            password="aiyingli@@123",
            socket_timeout=5,
            decode_responses=True
        )
        return redis_client
    except Exception as e:
        logger.success(f"redis 初始化失败: {e}")
        return None


def wait_mt_tasks_done(redis_client):
    """
    等待 redis set mt_third_task 为空：
    - 为空 -> 认为“采集任务都跑完了”
    - 不为空 -> 不发飞书，继续等
    """
    while True:
        try:
            size = redis_client.scard("mt_third_task")
            logger.success(f"size:{size}")
        except Exception as e:
            logger.success(f"redis 读取 mt_third_task 失败: {e}")
            size = -1

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


# =========================
# 任务初始化：每天 23:00 抓公告，写第二天 pt 的任务
# =========================

def init_task(pt_for_tomorrow: str):
    """
    生成第二天任务：
    - 从飞书群公告 blocks 里取每行文本作为 prompt
    - 生成 3 轮（cn=0/1/2）写入 geo_third_task_log
    """
    task_list = get_task()

    promp = []
    for i in task_list:
        try:
            if i.get("text"):
                elements = (i.get("text") or {}).get("elements") or []
                if not elements:
                    continue
                content = (((elements[0] or {}).get("text_run") or {}).get("content")) or ""
                content = content.strip()
                if content:
                    promp.append(content)
        except Exception:
            continue

    if not promp:
        logger.success(f"[init_task] 公告里没有取到 prompt，pt={pt_for_tomorrow}")
        return

    data_list = []
    for cn in range(3):
        for index,p in enumerate(promp):
            req_id = str(uuid.uuid4())
            data_list.append({
                "reqId": req_id,
                "prompt": p,
                "platform": "DB",
                "cn": cn,
                "pt": pt_for_tomorrow,
                "rank": index
            })

    bh_utils.insert_data("geo_third_task_log", data_list)
    logger.success(f"[init_task] 已写入任务: pt={pt_for_tomorrow}, prompts={len(promp)}, rows={len(data_list)}")


def init_task_once(pt_for_tomorrow: str):
    """
    只对某个 pt 生成一次任务（避免 scheduler 线程重复触发或重启重复写）
    """
    r = init_redis4()
    if not r:
        # redis 不可用时，退化为直接执行（可能重复写入，至少不影响流程）
        init_task(pt_for_tomorrow)
        return

    lock_key = f"mt:init_task_done:{pt_for_tomorrow}"
    ok = r.set(lock_key, "1", nx=True, ex=INIT_TASK_LOCK_EXPIRE)
    if ok:
        init_task(pt_for_tomorrow)
    else:
        logger.success(f"[init_task_once] 已生成过 pt={pt_for_tomorrow}，跳过")





# =========================
# 解析豆包 SSE 原始数据（保留）
# =========================

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")[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(
                                    "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)
    return (file_path, search_keyword, url_list, think_content, response_content, suggestions)



def timestamp_to_datetime(timestamp, fmt="%Y-%m-%d %H:%M:%S"):
    timestamp = int(timestamp)
    local_time = datetime.fromtimestamp(timestamp)
    return local_time.strftime(fmt)


# =========================
# ExcelWriter
# =========================

class ExcelWriter:
    """
    Excel写入工具类，专门用于写入taskID和response两列数据
    """

    def __init__(self, file_path):
        """
        初始化Excel写入工具
        :param file_path: Excel文件保存路径（如: './output.xlsx'）
        """
        self.file_path = file_path
        self.workbook = None
        self.worksheet = None
        # 初始化工作簿和工作表
        self._init_workbook()

    def _init_workbook(self):
        """初始化工作簿和工作表，若文件已存在则打开，不存在则新建"""
        # 检查文件是否存在
        if os.path.exists(self.file_path):
            self.workbook = openpyxl.load_workbook(self.file_path)
            # 取第一个工作表
            self.worksheet = self.workbook.active
            # 检查表头是否存在，不存在则添加
            if self.worksheet.cell(row=1, column=1).value != 'taskID' or \
                    self.worksheet.cell(row=1, column=2).value != 'response':
                # 在第一行插入表头
                self.worksheet.insert_rows(1)
                self.worksheet.cell(row=1, column=1, value='taskID')
                self.worksheet.cell(row=1, column=2, value='response')
        else:
            # 新建工作簿
            self.workbook = Workbook()
            self.worksheet = self.workbook.active
            # 设置表头
            self.worksheet.cell(row=1, column=1, value='问题')
            self.worksheet.cell(row=1, column=2, value='平台')
            self.worksheet.cell(row=1, column=3, value='回答')
            self.worksheet.cell(row=1, column=4, value='思考过程')
            self.worksheet.cell(row=1, column=5, value='引用来源')
            self.worksheet.cell(row=1, column=6, value='分享链接')
            self.worksheet.cell(row=1, column=7, value='截图')
            self.worksheet.cell(row=1, column=8, value='查询时间')
            self.worksheet.cell(row=1, column=9, value='ocr结果')
            self.worksheet.cell(row=1, column=10, value='排名')
            self.worksheet.cell(row=1, column=11, value='ip')
            self.worksheet.cell(row=1, column=12, value='版本号')




    def write_batch_rows(self, data_list):
        if not isinstance(data_list, list) or len(data_list) == 0:
            raise ValueError("data_list必须是非空的列表")

        next_row = self.worksheet.max_row + 1

        # 批量写入数据
        for idx, (prompt, platform, content, think, quote,share, snipaste, insertime,ocr,rank,inner_user_ip,inner_pc_version) in enumerate(data_list):
            self.worksheet.cell(row=next_row + idx, column=1, value=prompt)
            self.worksheet.cell(row=next_row + idx, column=2, value=platform)
            self.worksheet.cell(row=next_row + idx, column=3, value=content)
            self.worksheet.cell(row=next_row + idx, column=4, value=think)
            self.worksheet.cell(row=next_row + idx, column=5, value=quote)
            self.worksheet.cell(row=next_row + idx, column=6, value=share)
            self.worksheet.cell(row=next_row + idx, column=7, value=snipaste)
            self.worksheet.cell(row=next_row + idx, column=8, value=insertime)
            self.worksheet.cell(row=next_row + idx, column=9, value=ocr)
            self.worksheet.cell(row=next_row + idx, column=10, value=rank)
            self.worksheet.cell(row=next_row + idx, column=11, value=inner_user_ip)
            self.worksheet.cell(row=next_row + idx, column=12, value=inner_pc_version)

        # 保存文件
        self.workbook.save(self.file_path)

    def close(self):
        """关闭工作簿，释放资源"""
        if self.workbook:
            self.workbook.close()

# =========================
# 导出 Excel
# =========================
def get_today_date_string():
    import datetime as d
    today = d.date.today()
    return today.strftime("%Y%m%d")
def rand_16():
    chars = string.digits + string.ascii_lowercase  # 0-9a-z
    return ''.join(secrets.choice(chars) for _ in range(16))




def get_context(image_url):
    url = "https://ark.cn-beijing.volces.com/api/v3/responses"

    payload = json.dumps({
        "model": "doubao-seed-2-0-lite-260428",
        "input": [
            {
                "role": "user",
                "content": [
                    {
                        "type": "input_image",
                        "image_url": image_url
                    },
                    {
                        "type": "input_text",
                        "text": "提取图中文字"
                    }
                ]
            }
        ]
    })
    headers = {
        'Authorization': 'Bearer fcc424e5-58af-494d-9683-5787413a26c9',
        'Content-Type': 'application/json'
    }
    try:
        response = requests.request("POST", url, headers=headers, data=payload)

        data = response.json()

        return next(
            (
                content.get("text", "")
                for item in data.get("output", [])
                if item.get("type") == "message"
                for content in item.get("content", [])
                if content.get("type") == "output_text"
            ),
            ""
        )
    except Exception as e:
        return ""




def to_excel2(pt, cn):

    query_list_log = bh_utils.query_data(
        f"SELECT DISTINCT reqId FROM  geo_third_task_log  where cn = {cn} and pt = {pt}"
    )
    uuid_list = []

    for i in query_list_log:
        uuid_list.append(i.get('reqId'))
    uuid_str = ",".join([f"'{uuid}'" for uuid in uuid_list])
    query_list = bh_utils.query_data(
        f"SELECT DISTINCT reqId,prompt, platform,insertime,rank  FROM  geo_third_task  WHERE reqId IN ({uuid_str}) order by  rank asc"
    )

    task_list = []
    for i in query_list:
        task_list.append(
            {
                "reqId": i.get("reqId"),
                "prompt": i.get("prompt"),
                "platform": i.get("platform"),
                "insertime": i.get("insertime"),
                "rank": i.get("rank")
            }
        )
    plat_form_map = {
        "DP": "deepseek网页版",
        "DB": "豆包网页版",
        "TXYB": "腾讯元宝",
        "TYQW": "通义千问",
        "KIMI": "kimi",
        "WXYY": "文心一言",
        "BDAI": "百度ai",
        "DYAI": "抖音ai",
        "DOUBA": "豆包安卓版",
        "DPA": "deepseek安卓版",
    }

    def build_row(i):
        reqId = i.get("reqId")
        prompt = i.get("prompt")
        platform = i.get("platform")
        insertime = i.get("insertime")
        rank = i.get("rank")

        file = f"geo_snipaste/{pt}/doubao/{reqId}/text.json"
        context_path = f"geo_snipaste/{pt}/doubao/{reqId}/context.txt"
        quote_path = f"geo_snipaste/{pt}/doubao/{reqId}/quote.txt"
        think = f"geo_snipaste/{pt}/doubao/{reqId}/think.txt"
        inner_user_ip= f"geo_snipaste/{pt}/doubao/{reqId}/inner_user_ip.txt"
        inner_pc_version= f"geo_snipaste/{pt}/doubao/{reqId}/inner_pc_version.txt"

        png_url = f'https://tcdn.aidso.com/geo_snipaste/{pt}/doubao/{reqId}/png.png'
        text_json_str = tos_utils.get_string_from_tos(file) or "{}"

        return (
            prompt,
            plat_form_map.get(platform),
            tos_utils.get_string_from_tos(context_path),
            tos_utils.get_string_from_tos(think),
            tos_utils.get_string_from_tos(quote_path),
            json.loads(text_json_str).get('share_url', ''),
            png_url,
            timestamp_to_datetime(insertime),
            get_context(png_url),
            rank,
            tos_utils.get_string_from_tos(inner_user_ip),
            tos_utils.get_string_from_tos(inner_pc_version)

        )

    batch_data = [None] * len(task_list)

    with ThreadPoolExecutor(max_workers=50) as executor:
        future_map = {
            executor.submit(build_row, i): index
            for index, i in enumerate(task_list)
        }

        for future in as_completed(future_map):
            index = future_map[future]

            try:
                batch_data[index] = future.result()
            except Exception as e:
                logger.exception(f"导出数据失败 index={index}, task={task_list[index]}, err={e}")
                batch_data[index] = None

    # 去掉失败的数据
    batch_data = [row for row in batch_data if row is not None]

    logger.success(f"时间{pt}--{cn}---导出数据{len(batch_data)}条")

    out_path = os.path.join(EXCEL_DIR, f"{pt}美团{cn}.xlsx")

    excel_writer = ExcelWriter(out_path)
    excel_writer.write_batch_rows(batch_data)
    excel_writer.close()

    return out_path






# =========================
# 时间等待工具
# =========================

def wait_until_time(hour: int, minute: int, second: int = 0):
    now = datetime.now()
    target = now.replace(hour=hour, minute=minute, second=second, microsecond=0)
    if target <= now:
        target = target + timedelta(days=1)
    time.sleep(max(0, (target - now).total_seconds()))


def wait_until_date(date_obj):
    while True:
        if datetime.now().date() >= date_obj:
            return
        time.sleep(60)


# =========================
# 处理 cn：必须等 mt_third_task 为空才发消息
# =========================

def process_cn_after_tasks_done(pt: str, cn: int):
    redis_client = init_redis4()

    logger.success(f"[pt={pt} cn={cn}] 等待 mt_third_task 跑空...")
    wait_mt_tasks_done(redis_client)
    logger.success(f"[pt={pt} cn={cn}] mt_third_task 已为空，初步完成")

    for i in range(3):
        diff_result = bh_utils.query_data(
            f"select a1.reqId,a1.prompt,a1.`rank` from (select reqId,prompt,`rank` from geo_third_task_log where pt = {pt} and cn = {cn}) as a1 left join geo_third_task as a2 on a1.reqId = a2.reqId where a2.reqId is null")
        if not diff_result:
            logger.success(f"[pt={pt} cn={cn}] 第{i + 1}轮检查已无缺失")
            break
        for row in diff_result:
            row["thinking_enabled"] = "1"
        values = [json.dumps(row, ensure_ascii=False) for row in diff_result]

        redis_client.sadd("mt_third_task", *values)
        logger.success(f"[pt={pt} cn={cn}] 缺失数据开始补充")
        wait_mt_tasks_done(redis_client)
        logger.success(f"[pt={pt} cn={cn}] 缺失数据补充完成")

    time.sleep(120)
    logger.success(f"[pt={pt} cn={cn}] mt_third_task 已为空，开始汇总导出/发送")
    # 1) 解析原始 json
    result = bh_utils.query_data(f"select * from geo_third_task_log where cn = {cn} and pt = {pt}") or []

    for i in result:
        req_id = i.get("reqId")
        if not req_id:
            continue
        file = f"geo_snipaste/{pt}/doubao/{req_id}/text.json"
        try:
            raw = tos_utils.get_string_from_tos(file)
            if not raw:
                continue
            content = json.loads(raw).get("content")
            if content:
                doubao_process_original_data(file, content)
        except Exception as e:
            logger.success(f"[pt={pt} cn={cn}] 处理 {file} 失败: {e}")

    # 2) 导出 Excel
    out_path = to_excel2(pt, cn)

    # 3) 上传 + 发飞书
    file_name = os.path.basename(out_path)
    file_key = upload_file(out_path, file_name)
    if file_key:
        chat_message(file_key)
        logger.success(f"[pt={pt} cn={cn}] 已发送飞书文件: {file_name}")
    else:
        logger.success(f"[pt={pt} cn={cn}] 上传失败，未发送飞书消息")


# =========================

# =========================

def scheduler_thread():
    """
    永远保证 23:00 生成“第二天任务”（不会因为 runner 跑太久被阻塞）
    """
    while True:
        wait_until_time(SCHEDULE_HOUR, SCHEDULE_MINUTE, SCHEDULE_SECOND)

        tomorrow = datetime.now().date() + timedelta(days=1)
        pt_tomorrow = tomorrow.strftime("%Y%m%d")

        try:
            init_task_once(pt_tomorrow)
        except Exception as e:
            logger.success(f"[scheduler] init_task_once 失败: {e}")

        # 防止同一分钟内重复触发（例如 sleep 被中断）
        time.sleep(2)

def runner_thread():
    """
    - 程序启动后先跑一次“当天 pt”的 cn0->cn1->cn2
      * 同一天多次启动：每个 cn 只会跑一次（靠 redis lock_key: pt+cn）
    - 之后进入跨天循环：每天 00:00 + delay 后再跑一次（daily 不去重）
    """
    r = init_redis4()


    def run_once_for_pt(pt_today: str):
        for cn in (0, 1, 2):
            try:
                rows = bh_utils.query_data(
                    f"select reqId,prompt,rank,pt from geo_third_task_log where cn = {cn} and pt = {pt_today}"
                ) or []
                for row in rows:
                    row["thinking_enabled"] = "1"
                values = [json.dumps(row, ensure_ascii=False) for row in rows]

                if values:
                    r.sadd("mt_third_task", *values)
                process_cn_after_tasks_done(pt_today, cn)

            except Exception as e:
                logger.success(f"[runner] pt={pt_today} cn={cn} 执行失败: {e}")
                continue

    # ========= 1) 启动立即跑一次（按 cn 去重） =========
    try:
        pt_today = datetime.now().date().strftime("%Y%m%d")

        for cn in (0, 1, 2):
            # 关键：锁 key 加上 cn
            lock_key = f"mt:runner_startup_ran:{pt_today}:{cn}"

            # redis 不可用：就直接跑（会重复，至少不中断流程）
            ok = (r.set(lock_key, "1", nx=True, ex=RUNNER_LOCK_EXPIRE) if r else True)

            if ok:
                logger.success(f"[runner] startup run for pt={pt_today} cn={cn}")

                # startup 只跑一个 cn：把 run_once_for_pt 拆开跑
                try:
                    rows = bh_utils.query_data(
                        f"select reqId,prompt,rank from geo_third_task_log where cn = {cn} and pt = {pt_today}"
                    ) or []

                    values = [json.dumps(row, ensure_ascii=False) for row in rows]
                    if values:
                        r.sadd("mt_third_task", *values)

                    process_cn_after_tasks_done(pt_today, cn)
                except Exception as e:
                    logger.success(f"[runner] startup pt={pt_today} cn={cn} 执行失败: {e}")

            else:
                logger.success(f"[runner] startup already done for pt={pt_today} cn={cn}, skip")
        logger.success(f"[runner] pt={pt_today} all finish")
    except Exception as e:
        logger.success(f"[runner] 启动跑一次失败: {e}")

    # ========= 2) 之后按原逻辑：跨天再跑（每天跑一次，不去重） =========
    while True:
        today_date = datetime.now().date()
        tomorrow_date = today_date + timedelta(days=1)

        wait_until_date(tomorrow_date)
        time.sleep(AFTER_MIDNIGHT_DELAY_SECONDS)

        pt_today = datetime.now().date().strftime("%Y%m%d")
        logger.success(f"[runner] daily run for pt={pt_today}")

        run_once_for_pt(pt_today)

        time.sleep(60)

        logger.success(f"[runner] pt={pt_today} all finish")

def run_once_for_pt(pt_today: str):
    r = init_redis4()
    cn = 0
    try:
        rows = bh_utils.query_data(
            f"select reqId,prompt,rank from geo_third_task_log where cn = {cn} and pt = {pt_today}"
        ) or []

        values = [json.dumps(row, ensure_ascii=False) for row in rows]
        if values:
            r.sadd("mt_third_task", *values)
        process_cn_after_tasks_done(pt_today, cn)

    except Exception as e:
        logger.success(f"[runner] pt={pt_today} cn={cn} 执行失败: {e}")

def send_task(promp_list,pt,count):

    data_list = []
    for cn in range(count):
        for index,p in enumerate(promp_list):
            req_id = str(uuid.uuid4())
            data_list.append({
                "reqId": req_id,
                "prompt": p,
                "platform": "DB",
                "cn": cn,
                "pt": pt,
                "rank": index
            })
    logger.success(f"[init_task] 已写入任务: pt={pt}, prompts={len(promp_list)}, rows={len(data_list)}")
    bh_utils.insert_data("geo_third_task_log", data_list)

def webhook_snipaste(promp_list):
    count = 1

    pt = datetime.now().strftime("%Y%m%d%H%M")
    tos_pt = datetime.now().strftime("%Y%m%d")
    send_task(promp_list, pt, count)
    r = init_redis4()
    for cn in range(count):
        rows = bh_utils.query_data(
            f"select reqId,prompt,rank from geo_third_task_log where cn = {cn} and pt = {pt}"
        ) or []
        for row in rows:
            row["thinking_enabled"] = "1"
        values = [json.dumps(row, ensure_ascii=False) for row in rows]

        if values:
            r.sadd("mt_third_task", *values)

        logger.success(f"[pt={pt} cn={cn}] 等待 mt_third_task 跑空...")
        wait_mt_tasks_done(r)
        logger.success(f"[pt={pt} cn={cn}] mt_third_task 已为空，初步完成")

        for i in range(3):
            diff_result = bh_utils.query_data(
                f"select a1.reqId,a1.prompt,a1.`rank` from (select reqId,prompt,`rank` from geo_third_task_log where pt = {pt} and cn = {cn}) as a1 left join geo_third_task as a2 on a1.reqId = a2.reqId where a2.reqId is null")
            for row in diff_result:
                row["thinking_enabled"] = "1"
            values = [json.dumps(row, ensure_ascii=False) for row in diff_result]
            if not values:
                logger.success(f"[pt={pt} cn={cn}] 第{i + 1}轮检查已无缺失")
                break
            r.sadd("mt_third_task", *values)
            logger.success(f"[pt={pt} cn={cn}] 缺失数据开始补充")
            wait_mt_tasks_done(r)
            logger.success(f"[pt={pt} cn={cn}] 缺失数据补充完成")

        time.sleep(180)
        logger.success(f"[pt={pt} cn={cn}] mt_third_task 已为空，开始汇总导出/发送")

        result = bh_utils.query_data(f"select * from geo_third_task_log where cn = {cn} and pt = {pt}") or []
        for i in result:
            req_id = i.get("reqId")
            if not req_id:
                continue
            file = f"geo_snipaste/{tos_pt}/doubao/{req_id}/text.json"
            try:
                raw = tos_utils.get_string_from_tos(file)
                if not raw:
                    continue
                content = json.loads(raw).get("content")
                if content:
                    doubao_process_original_data(file, content)
            except Exception as e:
                logger.success(f"[pt={pt} cn={cn}] 处理 {file} 失败: {e}")

        query_list_log = bh_utils.query_data(
            f"SELECT DISTINCT reqId FROM  geo_third_task_log  where cn = {cn} and pt = {pt}"
        )
        uuid_list = []

        for i in query_list_log:
            uuid_list.append(i.get('reqId'))

        uuid_str = ",".join([f"'{uuid}'" for uuid in uuid_list])
        query_list = bh_utils.query_data(
            f"SELECT DISTINCT reqId,prompt, platform,insertime,rank  FROM  geo_third_task  WHERE reqId IN ({uuid_str}) order by  rank asc"
        )

        task_list = []
        for i in query_list:
            task_list.append(
                {
                    "reqId": i.get("reqId"),
                    "prompt": i.get("prompt"),
                    "platform": i.get("platform"),
                    "insertime": i.get("insertime"),
                    "rank": i.get("rank")
                }
            )
        plat_form_map = {
            "DP": "deepseek网页版",
            "DB": "豆包网页版",
            "TXYB": "腾讯元宝",
            "TYQW": "通义千问",
            "KIMI": "kimi",
            "WXYY": "文心一言",
            "BDAI": "百度ai",
            "DYAI": "抖音ai",
            "DOUBA": "豆包安卓版",
            "DPA": "deepseek安卓版",
        }

        def build_row(i):
            reqId = i.get("reqId")
            prompt = i.get("prompt")
            platform = i.get("platform")
            insertime = i.get("insertime")
            rank = i.get("rank")

            file = f"geo_snipaste/{tos_pt}/doubao/{reqId}/text.json"
            context_path = f"geo_snipaste/{tos_pt}/doubao/{reqId}/context.txt"
            quote_path = f"geo_snipaste/{tos_pt}/doubao/{reqId}/quote.txt"
            think = f"geo_snipaste/{tos_pt}/doubao/{reqId}/think.txt"
            inner_user_ip = f"geo_snipaste/{pt}/doubao/{reqId}/inner_user_ip.txt"
            inner_pc_version = f"geo_snipaste/{pt}/doubao/{reqId}/inner_pc_version.txt"

            png_url = f'https://tcdn.aidso.com/geo_snipaste/{tos_pt}/doubao/{reqId}/png.png'
            text_json_str = tos_utils.get_string_from_tos(file) or "{}"

            return (
                prompt,
                plat_form_map.get(platform),
                tos_utils.get_string_from_tos(context_path),
                tos_utils.get_string_from_tos(think),
                tos_utils.get_string_from_tos(quote_path),
                json.loads(text_json_str).get('share_url', ''),
                png_url,
                timestamp_to_datetime(insertime),
                get_context(png_url),
                rank,
                inner_user_ip,
                inner_pc_version
            )

        batch_data = [None] * len(task_list)

        with ThreadPoolExecutor(max_workers=50) as executor:
            future_map = {
                executor.submit(build_row, i): index
                for index, i in enumerate(task_list)
            }

            for future in as_completed(future_map):
                index = future_map[future]

                try:
                    batch_data[index] = future.result()
                except Exception as e:
                    logger.exception(f"导出数据失败 index={index}, task={task_list[index]}, err={e}")
                    batch_data[index] = None

        # 去掉失败的数据
        batch_data = [row for row in batch_data if row is not None]

        logger.success(f"时间{pt}--{cn}---导出数据{len(batch_data)}条")
        out_path = os.path.join(EXCEL_DIR, f"{pt}美团{cn}.xlsx")
        excel_writer = ExcelWriter(out_path)
        excel_writer.write_batch_rows(batch_data)
        excel_writer.close()
        file_name = os.path.basename(out_path)
        file_key = upload_file(out_path, file_name)
        if file_key:
            chat_message(file_key)
            logger.success(f"[pt={pt} cn={cn}] 已发送飞书文件: {file_name}")
        else:
            logger.success(f"[pt={pt} cn={cn}] 上传失败，未发送飞书消息")


# =========================
# 主入口
# =========================

if __name__ == "__main__":

    t1 = threading.Thread(target=scheduler_thread, daemon=True)
    t2 = threading.Thread(target=runner_thread, daemon=True)
    t1.start()
    t2.start()
    t1.join()
    t2.join()
