AutoScraperX/core/utils/helpers.py

import asyncio

from typing import List, Dict

from datetime import datetime
from core.utils.feishu_data_async import FeishuDataAsync
from core.utils.gpt4o_mini_help import GPT4oMini


async def get_title_filter_word() -> List[str]:
    """
    获取飞书表配置的标题过滤词
    https://w42nne6hzg.feishu.cn/sheets/KsoMsyP2ghleM9tzBfmcEEXBnXg?sheet=BS9uyu
    :return:
    """
    spreadsheet_token = "KsoMsyP2ghleM9tzBfmcEEXBnXg"
    sheet_id = "BS9uyu"
    async with FeishuDataAsync() as feishu:
        feishu_data = await feishu.get_values(spreadsheet_token=spreadsheet_token, sheet_id=sheet_id)
        return feishu_data[1]

async def generate_titles(sheet_id: str,video_obj: Dict):
    title_list = await get_title_filter_word()
    title = video_obj.get("title")
    if not title:
        return
    contains_keyword = any(keyword in title for keyword in title_list)
    if contains_keyword:
        new_title = await GPT4oMini.get_ai_mini_title(title)
        current_time = datetime.now()
        formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
        values = [
                video_obj["v_url"],
                video_obj["url"],
                title,
                new_title,
                formatted_time,
        ]
        await insert_feishu_data(sheet_id, values)

async def insert_feishu_data(sheet_id: str,values: List):
    spreadsheet_token = "KsoMsyP2ghleM9tzBfmcEEXBnXg"
    async with FeishuDataAsync() as feishu:
        await feishu.insert_values(spreadsheet_token=spreadsheet_token, sheet_id=sheet_id,ranges="A2:Z2",values=values)




if __name__ == '__main__':
     filter_word = asyncio.run(insert_feishu_data())
     print(filter_word)