notion/main.py

# from notion.block import CalloutBlock, BulletedListBlock, TextBlock
# from auth import client
from notion_client import AsyncClient
from pprint import pprint
import asyncio
import platform
from dateutil.parser import parse
from dateutil.tz import gettz
import datetime
import requests
import json
from loguru import logger

# def update_callout():
#     page_callout_extract = client.get_block(
#         "https://www.notion.so/facat/76583e3369764976868d11bc761abc93"
#     )
#     page_callout_update = client.get_block(
#         "https://www.notion.so/facat/8f4f0ed22f9f4afb9b8faee15b33d8ed"
#     )
#     page_callout_extract.locked = True
#     page_callout_update.locked = True
#     print("The old title is:", page_callout_extract.title)
#     callout_list = []
#     for child in page_callout_extract.children:
#         if type(child) == CalloutBlock:
#             callout_list.append(child.title)
#             print(f"找到Callout{child.title}")
#     # 清除重点的内容
#     for child in page_callout_update.children:
#         child.remove()
#     # 更新重点的内容
#     for callout_list in callout_list:
#         page_callout_update.children.add_new(BulletedListBlock, title=callout_list)
#     page_callout_update.children.add_new(TextBlock, title="End!")
#     page_callout_extract.locked = False
#
#
# async def sdk_callout():
#     async with AsyncClient(
#         auth="secret_WoGb0MDPDdwiEMQr8ZU6i50u7fT6WD6RzXkHu6RAF4R"
#     ) as notion:
#         page = await notion.pages.retrieve("8f4f0ed22f9f4afb9b8faee15b33d8ed")
#         children_info = await notion.blocks.children.list(page["id"])
#         children = children_info["results"]
#         for child in children:
#             # text=await notion.blocks.children.list(child['id'])
#             pprint(child)
#
#
# async def extract_callout():
#     async with AsyncClient(
#         auth="secret_WoGb0MDPDdwiEMQr8ZU6i50u7fT6WD6RzXkHu6RAF4R"
#     ) as notion:
#         extract_page_block_info = await notion.blocks.children.list('76583e3369764976868d11bc761abc93')
#         extract_blocks = extract_page_block_info["results"]
#         for blocks in extract_blocks:
#             if blocks['type']=='synced_block':
#                 callouts=await notion.blocks.children.list(blocks['id'])
#                 pprint(callouts)


def test_if_mention(block):
    if "paragraph" in block:
        paragraph = block["paragraph"]
        if "text" in paragraph:
            texts = paragraph["text"]
            for text in texts:
                if "mention" in text:
                    return text["mention"]
    return None


def send_msg(text):
    r = requests.post(
        "https://mmssgg.happymap.xyz/sendmessage",
        headers={"Content-Type": "application/json"},
        data=json.dumps({"message": text}),
    )


async def check_un_do():
    notion = AsyncClient(auth="secret_WoGb0MDPDdwiEMQr8ZU6i50u7fT6WD6RzXkHu6RAF4R")
    todo_database_id = "31ba234684e54e57a10e7e155fe7767c"
    uncompleted_todos_query = await notion.databases.query(
        **{
            "database_id": todo_database_id,
            "filter": {"property": "状态", "select": {"equals": "未完成"}},
        }
    )
    uncompleted_todos = uncompleted_todos_query["results"]
    for uncompleted_todo in uncompleted_todos:
        # pprint(uncompleted_todo)
        task_name = uncompleted_todo["properties"]["Name"]["title"][0]["plain_text"]
        task_url = uncompleted_todo["url"]
        page_id = uncompleted_todo["id"]
        page_blocks_query = await notion.blocks.children.list(page_id)
        page_blocks = page_blocks_query["results"]
        for block in page_blocks:
            mention = test_if_mention(block)
            # pprint(block)
            if mention:
                mention_date = mention["date"]["start"]
                time_zone = gettz("Asia/Chongqing")
                mention_datetime = parse(mention_date)
                now = datetime.datetime.now(time_zone)
                duration = now - mention_datetime
                duration_seconds = duration.total_seconds()
                if -5 * 60 < duration_seconds < 0:  # 提前5分钟通知
                    send_msg(f"任务 {task_name} 要开始了。地址:{task_url}")
                    logger.info(f"提示了任务 {task_name} ")
                    pass
                # pprint(mention)


async def loop():
    while True:
        logger.info("检查任务。")
        await check_un_do()
        await asyncio.sleep(3 * 60)  # 3分钟检查一次


if __name__ == "__main__":
    # update_callout()
    if platform.system() == "Windows":
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    asyncio.run(loop())
    print("PyCharm")