title_with_video/new_history.py

"""
@author: luojunhui
"""
import asyncio
import time
import traceback

from tasks.new_history_task import NewHistoryTask
from applications.db import AsyncMySQLClient
from applications.log import logging
from applications.feishu import bot


async def main():
    """
    main job
    :return:
    """
    async_mysql_pool = AsyncMySQLClient()
    try:
        await async_mysql_pool.init_pool()
        logging(
            code="newHistory0001",
            info="Init MySQL pool successfully",
            alg="historyContentIdTask",
            function="main"
        )
    except Exception as e:
        logging(
            code="newHistory0002",
            info="Init MySQL pool failed",
            alg="historyContentIdTask",
            function="main",
            data={
                "error": str(e),
                "traceback": traceback.format_exc()
            }
        )
        bot(
            title="newHistoryContentIdTask INIT MYSQL FAILS",
            detail={
                "error": str(e),
                "traceback": traceback.format_exc()
            }
        )
    try:
        history_content_id_task = NewHistoryTask(async_mysql_pool)
    except Exception as e:
        logging(
            code="newHistory0003",
            info="Init newHistory0003 failed",
            alg="historyContentIdTask",
            function="main",
            data={
                "error": str(e),
                "traceback": traceback.format_exc()
            }
        )
        bot(
            title="newHistoryContentIdTask INIT TASK FAILS",
            detail={
                "error": str(e),
                "traceback": traceback.format_exc()
            }
        )
        return
    await history_content_id_task.deal_new()

if __name__ == '__main__':
    while True:
        asyncio.run(main())
        time.sleep(5)