main.py

import asyncio
import argparse
import time
from typing import Dict, List, Optional, Any, Coroutine
from DrissionPage import Chromium, ChromiumOptions
from autobrowser import wait_for_new_topics, process_topic

EDGE_BROWSER_PATH = r'C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe'
EDGE_DEBUG_PORT = 9223

async def browse_with_browser(browser_choice: str, mode: str, browser_paths: Dict[str, str], num_topics: int = 10) -> None:
    """使用指定浏览器进行浏览"""
    co: ChromiumOptions
    if browser_choice == "":
        co = ChromiumOptions()
    elif browser_choice == "edge":
        co = ChromiumOptions().set_browser_path(EDGE_BROWSER_PATH).set_local_port(EDGE_DEBUG_PORT)
    elif browser_choice in browser_paths:
        path: str = browser_paths[browser_choice]
        co = ChromiumOptions().set_browser_path(path)
    else:
        print("无效的选择，将使用系统默认路径。")
        co = ChromiumOptions()

    browser: Chromium = Chromium(co)
    tab = browser.new_tab('https://linux.do/')
    topic_list_body = tab('.topic-list-body')
    
    topic_list: List[Any] = []
    if mode == 'short' or mode == 'long':
        # 循环加载主题，直到满足条件或没有更多主题
        while True:
            # 获取当前已加载的主题元素
            elements = topic_list_body.eles('t=tr')
            current_count = len(elements)
            
            # 如果是 short 模式且已加载足够数量的主题，则停止
            if mode == 'short' and current_count >= num_topics:
                topic_list = list(elements)[:num_topics]
                break
                
            # 滚动到底部以加载更多主题
            tab.scroll.to_bottom()
            
            # 等待一段时间并轮询，检查是否有新主题加载
            start_wait_time = time.time()
            while time.time() - start_wait_time < 5: # 最多等待5秒
                await asyncio.sleep(0.5) # 每0.5秒检查一次
                elements = topic_list_body.eles('t=tr')
                if len(elements) > current_count:
                    # 有新主题加载，跳出内层等待循环
                    break
            else:
                # 等待超时，没有新主题加载，认为已加载完毕
                topic_list = list(elements)
                print('没有更多主题加载，结束程序')
                break
                
    semaphore: asyncio.Semaphore = asyncio.Semaphore(3)
    tasks: List[Coroutine[Any, Any, None]] = [process_topic(topic, n, semaphore) for n, topic in enumerate(topic_list)]
    await asyncio.gather(*tasks)
    
    # 关闭浏览器
    browser.quit()

async def main() -> None:
    parser = argparse.ArgumentParser(description='浏览 linux.do 主题')
    parser.add_argument('-m', '--mode', choices=['short', 'long'], default='short',
                       help='浏览模式: short(浏览10个主题) 或 long(浏览所有主题)')
    parser.add_argument('-b', '--browser', choices=['single', 'all'], default='single',
                       help='浏览器模式: single(单个浏览器) 或 all(所有浏览器)')
    parser.add_argument('-n', '--num', type=int, default=10, help='指定在short模式下浏览的主题数量')
    args = parser.parse_args()

    browser_paths: Dict[str, str] = {
        'chrome': r'',
        '115': r'',
        'doubao': r'', 
    }
    browser_choices: List[str] = ['edge', *browser_paths.keys()]
    
    if args.browser == 'single':
        print("请选择要使用的浏览器:")
        for key in browser_choices:
            print(f"- {key}")
        choice: str = input("请输入你的选择: ").lower().strip()
        await browse_with_browser(choice, args.mode, browser_paths, args.num)
    else:
        for browser_name in browser_choices:
            print(f"\n正在使用 {browser_name} 浏览...")
            await browse_with_browser(browser_name, args.mode, browser_paths, args.num)

if __name__ == "__main__":
    asyncio.run(main())