douyin-live/test/test.py

import asyncio
import os
import queue
import sys
import threading
import time

sys.path.append(os.path.dirname(os.path.dirname(__file__)))
from conf.settings import WORK_DIR,CONF_DIR,OUTPUT,logger,Detection_Interval

Thread_que = queue.Queue()
# from detection import Detection
# def analyze():
#     detection = Detection()
#     detection.analyze_link('123')

async def main_task():
    stop_event = asyncio.Event()
    task_auto_detect = asyncio.create_task(auto_detect_task(stop_event))
    while True:
        logger.info("m")
        await asyncio.sleep(1)

def run_task():
    asyncio.run(main_task())
    
async def auto_detect_task(stop_event:asyncio.Event=None):
    loop = asyncio.get_event_loop()
    while 1:
        # await asyncio.sleep(Detection_Interval)
        await asyncio.sleep(1)
        logger.debug(f"thread_scan_lives finish:{stop_event.is_set()}")
        if stop_event.is_set():
            logger.debug("stop event")
            break

def imitate_server():
    thread = threading.Thread(target=run_task)
    thread.start()
    while True:
        try:
            indat = input('input:')
            
        except KeyboardInterrupt:
            logger.info("KeyboardInterrupt")
            break
    # Thread_que.put_nowait("stop")
    logger.info('put stop msg')
    thread.join()
    logger.info('exit')
    # sys.exit(0)
    
if __name__ == "__main__":
    # imitate_server()
    live_page_user_name = {}
    ret = {"live_url": "live_url"}
    ret.update(live_page_user_name)
    print(ret.update(live_page_user_name))
    print(ret)