title_with_video/historyTask.py

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

from concurrent.futures import ThreadPoolExecutor, as_completed

from tasks.history_task import historyContentIdTask
from applications.db import AsyncMySQLClient
from applications.log import logging
from applications.feishu import bot
from applications.config.const import HistoryContentIdTaskConst


async def main(publish_flag):
    """
    main job
    :return:
    """
    async_mysql_pool = AsyncMySQLClient()
    try:
        await async_mysql_pool.init_pool()
        logging(
            code="history0001",
            info="Init MySQL pool successfully",
            alg="historyContentIdTask",
            function="main"
        )
    except Exception as e:
        logging(
            code="history0002",
            info="Init MySQL pool failed",
            alg="historyContentIdTask",
            function="main",
            data={
                "error": str(e),
                "traceback": traceback.format_exc()
            }
        )
        bot(
            title="historyContentIdTask INIT MYSQL FAILS",
            detail={
                "error": str(e),
                "traceback": traceback.format_exc()
            }
        )
    try:
        history_content_id_task = historyContentIdTask(async_mysql_pool)
    except Exception as e:
        logging(
            code="history0003",
            info="Init historyContentIdTask failed",
            alg="historyContentIdTask",
            function="main",
            data={
                "error": str(e),
                "traceback": traceback.format_exc()
            }
        )
        return
    await history_content_id_task.deal(publish_flag)


def run_thread(publish_flag, sleep_seconds):
    """
    执行进程
    """
    while True:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        loop.run_until_complete(main(publish_flag=publish_flag))
        loop.close()
        now_str = datetime.datetime.now().__str__()
        logging(
            code="history0004",
            info="History task finished",
            alg="run_thread2"
        )
        print("{}    请求执行完成， 等待{}s".format(now_str, sleep_seconds))
        time.sleep(sleep_seconds)


if __name__ == '__main__':
    const = HistoryContentIdTaskConst()

    with ThreadPoolExecutor(max_workers=2) as executor:
        futures = [
            executor.submit(run_thread, const.NEED_PUBLISH, const.NEED_PUBLISH_WAIT_TIME),
            executor.submit(run_thread, const.DO_NOT_NEED_PUBLISH, const.DO_NOT_NEED_PUBLISH_WAIT_TIME)
        ]

        try:
            for future in as_completed(futures):
                future.result()
        except KeyboardInterrupt:
            print("Stopping all threads...")