""" 爬虫持续采集入口(深度采集模式)。 流程:搜索笑话站点 → 翻页深度采集 → AI 提取 → 入库。 每批采集 20 条后休息 2-10 分钟,出错休息 3-11 分钟后继续。 按 Ctrl+C 中断。 用法: python crawler/main.py # 默认,无头浏览器 python crawler/main.py --no-headless # 显示浏览器窗口(方便测试) """ import argparse import asyncio import os import sys if sys.platform == "win32": import shutil os.environ["PYTHONIOENCODING"] = "utf-8" os.environ["TERM"] = "dumb" try: shutil.get_terminal_size() except Exception: pass sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from crawler.processor import Processor def parse_args(): parser = argparse.ArgumentParser(description="笑话爬虫 - 深度采集模式") parser.add_argument( "--no-headless", action="store_true", help="显示浏览器窗口(默认无头模式)", ) return parser.parse_args() def main(): args = parse_args() headless = not args.no_headless api_base = os.getenv("API_BASE", "http://localhost:8001") username = os.getenv("CRAWL_USERNAME", "admin") password = os.getenv("CRAWL_PASSWORD", "admin123") keyword_str = os.getenv("CRAWL_KEYWORD", "笑话大全,搞笑段子,冷笑话,幽默笑话,爆笑笑话,笑话集锦") keywords = [k.strip() for k in keyword_str.split(",") if k.strip()] if not keywords: print("错误: 未设置关键词") return print(f"=" * 50) print(f"笑话爬虫 - 深度采集模式") print(f"搜索词: {keywords}") print(f"API 地址: {api_base}") print(f"浏览器: {'显示窗口' if not headless else '无头模式'}") print(f"每批 20 条后休息 2-10 分钟") print(f"出错后休息 3-11 分钟后重试") print(f"按 Ctrl+C 终止") print(f"=" * 50) try: import crawl4ai print(f"crawl4ai 版本: {crawl4ai.__version__}") except ImportError: print("错误: crawl4ai 未安装,请先运行: pip install crawl4ai") return processor = Processor( api_base=api_base, username=username, password=password, headless=headless, ) try: asyncio.run(processor.run_continuous(keywords=keywords, max_pages=3, batch_size=20)) except KeyboardInterrupt: print("\n用户中断,退出") if __name__ == "__main__": main()