基于Playwright与FastAPI的智能LinkedIn职位爬虫：自动化求职与AI分析实战

1. 项目概述：一个基于Playwright与FastAPI的智能LinkedIn职位爬虫

如果你正在寻找一份海外，特别是欧洲或北美地区的工作，每天手动刷新LinkedIn的职位列表绝对是一项耗时又低效的苦差事。更别提还要从海量信息中筛选出那些真正符合你技术栈、期望地点和签证政策的机会。今天分享的这个项目，正是为了解决这个痛点而生的。它是一个全自动的LinkedIn职位信息爬取、分析与推送系统，核心由Python驱动，结合了Playwright进行浏览器自动化、FastAPI构建后端服务，并巧妙地集成了AI能力进行职位描述分析。我自己在找工作的那段时间，就深受信息过载和筛选效率低下的困扰，于是动手搭建了这套工具，它不仅能7x24小时不间断地帮我监控目标职位，还能通过智能逻辑表达式进行精准过滤，并通过Telegram Bot实时推送到我手机上，让我几乎不会错过任何潜在机会。

简单来说，这个项目就是一个高度定制化的“职位猎手”。它模拟真人行为浏览LinkedIn，绕过一些基础的防爬机制，抓取公开的职位信息。然后，它并非简单地罗列数据，而是内置了一套强大的分析引擎：利用类似ChatGPT的AI服务解析职位描述，提取关键技能要求，判断公司是否提供签证担保，甚至能翻译非母语的招聘广告。最核心的亮点在于其“嵌套逻辑表达式过滤器”，你可以像编写程序逻辑一样，用and、or、括号来组合你的求职条件，实现极其精细的筛选。最后，所有匹配的职位会通过一个Telegram机器人推送给你，整个过程完全自动化。它适合任何有明确求职目标、希望提升求职效率的开发者，尤其是目标海外市场的技术人才。接下来，我将详细拆解这个项目的设计思路、技术实现细节以及我在搭建和使用过程中积累的实战经验。

2. 核心架构与设计思路解析

2.1 为什么选择Playwright + FastAPI的技术栈？

在项目启动时，我评估过几种常见的爬虫方案。传统的requests+BeautifulSoup组合对于静态页面很高效，但面对像LinkedIn这样大量依赖JavaScript渲染、交互复杂的单页应用（SPA）就显得力不从心。Selenium是经典的选择，但它的执行速度相对较慢，且资源消耗较大。最终我选择了Playwright，主要基于以下几点考量：

1. 对现代Web技术的完美支持：Playwright由微软开发，天生为自动化测试和爬取现代Web应用而生。它支持所有主流浏览器（Chromium, Firefox, WebKit），并且能自动等待页面元素加载、网络请求完成，这对于处理动态内容至关重要。
2. 强大的浏览器上下文与指纹管理：LinkedIn有反爬机制，会检测浏览器指纹。Playwright允许我们创建独立的浏览器上下文（Context），每个上下文可以拥有独立的Cookie、缓存和用户代理（UA），甚至可以模拟不同的设备、时区、语言。这为我们实现“拟人化”爬取、降低被封禁风险提供了底层支持。
3. 卓越的性能与可靠性：Playwright的API设计非常直观，执行速度比Selenium快，且更稳定。其自动等待机制减少了编写大量time.sleep的需要，让代码更健壮。

选择FastAPI作为后端框架，则是看中了它的高性能和开发效率。这个爬虫系统不仅仅是一个脚本，它需要管理代理池、用户过滤器、任务队列，并提供API接口供前端（Telegram Bot）调用。FastAPI的异步特性（基于asyncio）能与Playwright的异步API完美结合，实现高并发爬取。此外，其自动生成的交互式API文档（Swagger UI）也极大方便了开发和调试。整个系统的数据流转通过SQLAlchemy与数据库交互，确保了数据持久化和结构化存储。

2.2 系统工作流程与模块职责

整个系统可以清晰地划分为四个核心模块，它们协同工作，形成一个闭环的数据流水线：

1. 爬虫工作者（Worker）：这是系统的“采集触角”。它是一个或多个独立的进程，负责执行实际的爬取任务。每个工作者实例会：

   • 从任务队列中获取一个搜索关键词（如“Python developer Berlin”）。
   • 从代理池中选取一个可用的HTTP代理。
   • 启动一个Playwright浏览器实例（可配置为无头模式以节省资源），并配置好浏览器指纹（UA、视窗大小等）。
   • 使用代理访问LinkedIn，执行搜索，模拟滚动、点击等操作，抓取列表页和详情页的HTML数据。
   • 将原始HTML数据解析成结构化的职位信息（公司、职位、地点、描述等），并发送到后端API进行存储。

2. 后端API服务（FastAPI App）：这是系统的“大脑和中枢”。它提供了一系列RESTful API端点，负责：

   • 接收并存储爬虫工作者发送的职位数据。
   • 管理用户账户和他们的个性化过滤规则（即那些复杂的逻辑表达式）。
   • 提供接口给Telegram Bot，用于用户注册、设置过滤器、查询信息。
   • 在后台，它会定时或在收到新职位数据时，触发“匹配引擎”。

3. 匹配与AI分析引擎：这是系统的“智能过滤器”。当一个新的职位被存入数据库后，引擎会启动：

   • 逻辑表达式匹配：核心功能。引擎会取出所有活跃用户的过滤器，将职位信息（标题、描述、地点等）作为输入，动态“计算”这些逻辑表达式。例如，对于用户过滤器(python and django) and (berlin or munich)，引擎会检查职位描述中是否同时包含“python”和“django”，并且地点是否包含“berlin”或“munich”。只有完全匹配的职位才会进入下一阶段。
   • AI深度分析：对于通过初步过滤的职位，引擎会调用外部的AI服务（项目中使用的是替代ChatGPT API的第三方服务，以控制成本）进行深度分析。这包括：
     • 硬技能提取：从大段的职位描述中，精准提取出技术要求，如“Python”, “Docker”, “AWS”等，并以列表形式返回。
     • 签证赞助分析：分析描述中是否有“visa sponsorship”, “work permit”, “relocation assistance”等关键词，判断该公司是否为外籍员工提供签证支持。这对于国际求职者至关重要。
     • 翻译：如果职位描述是德语、法语等非用户母语，可调用AI进行翻译，确保用户完全理解职位要求。
   • 将分析结果附加到职位数据中。

4. Telegram Bot推送服务：这是系统的“交付终端”。它是一个独立的服务，与Telegram官方Bot API交互。

   • 用户通过与Bot对话来设置和管理自己的求职过滤器。
   • 当匹配引擎发现一个职位符合某个用户的条件时，后端API会调用Bot服务。
   • Bot服务将格式化后的职位信息（包含公司、职位、地点、关键技能、签证情况、原始链接等）即时推送到用户的Telegram私聊或指定的频道中。

注意：关于代理与合规性的重要考量：项目文档提到了使用代理。在实际部署中，务必使用合法、可靠的代理服务，并严格遵守目标网站（LinkedIn）的robots.txt协议和服务条款。本项目设计使用代理的主要目的是分散请求频率、模拟不同地理位置的访问，以降低单个IP被限制的风险，而非进行恶意爬取。爬取时应仅针对公开的、无需登录即可查看的职位信息，并设置合理的请求间隔（如-p参数可能用于限制只爬取“流行国家”，也是一种控制频率的策略）。任何爬虫项目都应秉持负责任的态度，避免对目标服务器造成过大负担。

3. 关键模块深度剖析与实操要点

3.1 Playwright爬虫工作者的实现细节与避坑指南

爬虫工作者是整个系统中最容易出问题的部分，因为它直接与“变幻莫测”的前端页面交互。以下是我在实现过程中总结的关键步骤和注意事项。

3.1.1 浏览器实例与上下文的初始化

直接使用Playwright的同步API虽然简单，但在高并发（-w 5表示启动5个工作者）场景下，异步API能更好地利用系统资源。每个工作者启动时，应该创建一个异步的浏览器实例和独立的上下文。

# 示例代码结构 (main.py 中工作者部分) import asyncio from playwright.async_api import async_playwright async def create_browser_context(proxy_url=None): async with async_playwright() as p: # 启动浏览器，推荐使用Chromium，平衡兼容性和性能 browser = await p.chromium.launch(headless=True) # 生产环境用--headless # 创建浏览器上下文，这是管理独立会话的关键 context = await browser.new_context( viewport={'width': 1920, 'height': 1080}, user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ...', # 使用常见UA # 如果配置了代理，在此处传入 proxy={'server': proxy_url} if proxy_url else None, # 可以设置语言、时区等，让指纹更真实 locale='en-US', timezone_id='America/New_York', ) # 可以在此处为上下文添加初始Cookie或执行登录（如果需要，但本项目使用公开视图） # await context.add_cookies([...]) return browser, context

实操心得：指纹管理：LinkedIn会通过一系列浏览器属性（如WebGL、Canvas、字体、插件列表等）生成浏览器指纹来识别自动化工具。Playwright的new_context方法可以覆盖大部分基础指纹。但对于高级检测，可能需要更复杂的策略，例如使用playwright-stealth这类插件来隐藏自动化特征。在项目中，如果发现爬取很快被阻断，除了检查代理，首要怀疑对象就是浏览器指纹。

3.1.2 页面导航、等待与数据提取

LinkedIn的页面加载大量依赖AJAX，简单的page.goto后立即查询元素大概率会失败。必须使用Playwright强大的等待机制。

async def scrape_linkedin_job_list(context, keyword): page = await context.new_page() try: # 1. 导航到搜索URL search_url = f"https://www.linkedin.com/jobs/search/?keywords={keyword}" # 使用`wait_until='networkidle'` 等待主要网络活动停止 await page.goto(search_url, wait_until='networkidle') # 2. 等待职位列表容器出现 # 使用`page.wait_for_selector` 并设置超时 list_selector = ".jobs-search__results-list" await page.wait_for_selector(list_selector, timeout=15000) # 3. 模拟滚动加载更多（LinkedIn是滚动加载） for _ in range(3): # 滚动3次，可根据需要调整 await page.evaluate("window.scrollTo(0, document.body.scrollHeight)") await page.wait_for_timeout(2000) # 等待新内容加载 # 更优的做法是等待某个加载指示器出现再消失 # try: # await page.wait_for_selector('.loading-indicator', state='visible', timeout=1000) # await page.wait_for_selector('.loading-indicator', state='hidden', timeout=5000) # except: # break # 4. 提取职位卡片元素 job_cards = await page.query_selector_all(".job-card-container") jobs_data = [] for card in job_cards: # 使用`elementhandle.evaluate`在浏览器环境中提取数据，比多次调用`get_attribute`快 job_info = await card.evaluate(""" el => { const titleElem = el.querySelector('.job-card-list__title'); const companyElem = el.querySelector('.job-card-container__company-name'); const locationElem = el.querySelector('.job-card-container__metadata-item'); const linkElem = el.querySelector('.job-card-list__external-link'); return { title: titleElem?.innerText.trim() || '', company: companyElem?.innerText.trim() || '', location: locationElem?.innerText.trim() || '', link: linkElem?.href || el.querySelector('a')?.href || '', external_id: el.getAttribute('data-job-id') || '' // 使用唯一ID }; } """) if job_info['link']: # 可以在这里进一步点击进入详情页抓取描述 # desc = await scrape_job_detail(page, job_info['link']) # job_info['description'] = desc jobs_data.append(job_info) return jobs_data except Exception as e: print(f"爬取关键词 '{keyword}' 时出错: {e}") return [] finally: await page.close()

避坑指南：选择器与页面结构：LinkedIn的前端类名（如.job-card-container）可能会随时更新。这是所有基于CSS选择器爬虫的最大风险。解决方案：

1. 使用更稳定的选择器：优先选择带有># 假设有一个Proxy模型，通过FastAPI管理 import random import aiohttp from sqlalchemy.ext.asyncio import AsyncSession async def get_random_working_proxy(db_session: AsyncSession): """ 从数据库获取一个随机且最近验证可用的代理。 """ # 查询状态为‘active’且最近检查成功的代理 stmt = select(Proxy).where( Proxy.is_active == True, Proxy.last_checked > (datetime.utcnow() - timedelta(hours=1)) # 一小时内检查过的 ).order_by(func.random()) # 随机排序 result = await db_session.execute(stmt) proxy = result.scalar_one_or_none() if proxy: return f"http://{proxy.ip}:{proxy.port}" else: # 如果没有可用代理，可能需要触发代理采集或使用无代理模式（风险高） return None async def validate_proxy(proxy_url): """ 验证代理是否有效且能访问LinkedIn。 """ try: async with aiohttp.ClientSession() as session: # 使用一个简单的LinkedIn公开页面进行测试，如robots.txt async with session.get('https://www.linkedin.com/robots.txt', proxy=proxy_url, timeout=10) as resp: return resp.status == 200 except: return False

   在工作者启动时，应该先调用get_random_working_proxy获取一个代理，然后在创建浏览器上下文时传入。每次爬取任务结束后，可以根据任务成功与否更新该代理的“健康状态”。对于连续失败的代理，应将其标记为is_active=False，并安排后续重新验证。

   3.2 嵌套逻辑表达式过滤器的安全实现

   这是项目中最具创新性但也最需谨慎处理的功能。允许用户输入类似Python表达式的字符串（如(django or fastapi) and (germany)）并在服务器端执行匹配，存在巨大的安全风险（代码注入）。

   3.2.1 安全的表达式评估策略

   项目文档提到“Isolated Secure System”，这是一个正确的方向。绝不能使用Python内置的eval()函数直接执行用户输入。我的实现方案如下：

   1. 词法分析与语法解析：首先，将用户输入的字符串解析成一系列令牌（Token）和抽象语法树（AST）。我们可以定义一个非常有限的语法，只支持AND,OR,NOT逻辑运算符、括号()以及由字母数字和下划线组成的“关键词”。
   2. 使用AST解释器：自己编写一个简单的解释器来遍历AST。解释器的核心是一个evaluate函数，它接收AST节点和当前职位的文本数据，然后递归地计算逻辑值。
   3. 完全沙盒化：整个解析和评估过程不调用任何Python的动态执行功能。所有操作都在我们自定义的解释器内完成。

   # 一个极度简化的示例，展示思路 import re from enum import Enum from typing import List, Union class TokenType(Enum): KEYWORD = 'KEYWORD' AND = 'AND' OR = 'OR' NOT = 'NOT' LPAREN = '(' RPAREN = ')' EOF = 'EOF' class ASTNode: pass class BinOpNode(ASTNode): def __init__(self, left, op, right): self.left = left self.op = op # TokenType.AND or TokenType.OR self.right = right class NotNode(ASTNode): def __init__(self, operand): self.operand = operand class KeywordNode(ASTNode): def __init__(self, value): self.value = value.lower() # 统一转为小写，实现大小写不敏感匹配 class Parser: # ... 实现词法分析和语法分析，将字符串"(python and django) or java" 转化为AST pass class Evaluator: def evaluate(self, node: ASTNode, text: str) -> bool: """根据AST节点和文本内容，返回布尔值。""" text_lower = text.lower() if isinstance(node, KeywordNode): return node.value in text_lower elif isinstance(node, NotNode): return not self.evaluate(node.operand, text) elif isinstance(node, BinOpNode): left_result = self.evaluate(node.left, text) right_result = self.evaluate(node.right, text) if node.op == TokenType.AND: return left_result and right_result elif node.op == TokenType.OR: return left_result or right_result # 对于括号，在AST中已被处理为改变运算顺序，此处无需特殊处理 return False # 使用示例 user_filter = "(python and django) or (java and spring)" parser = Parser() ast = parser.parse(user_filter) evaluator = Evaluator() job_description = "We are looking for a backend developer skilled in Python and Django framework." is_match = evaluator.evaluate(ast, job_description) # 返回 True job_description2 = "Java Spring Boot developer position." is_match2 = evaluator.evaluate(ast, job_description2) # 返回 True

   3.2.2 增强的安全性措施

   • 输入清洗与验证：在解析前，严格限制输入字符集（如只允许字母、数字、空格、()&|!等）。拒绝任何包含可疑字符（如引号、分号、点号、方括号等）的输入。
   • 关键词白名单（可选）：可以维护一个允许的技术栈、地点等关键词白名单。在解析出KeywordNode后，检查其值是否在白名单内，不在则视为匹配失败或直接拒绝该过滤器。这能极大限制攻击面，但会降低灵活性。
   • 复杂度限制：限制表达式的长度、嵌套深度和运算符数量，防止DoS攻击（过于复杂的表达式消耗大量CPU时间）。
   • 日志与审计：记录所有用户提交的过滤器表达式，便于在出现问题时追溯。

   重要提醒：即使采取了上述措施，在公开服务中提供此类功能仍需极度谨慎。项目作者提到的“使用在线沙盒”或“用ChatGPT API分析安全性”是更高级的防御思路。对于个人使用或小范围团队内部使用，自研的解释器方案通常是足够的。

   3.3 与AI服务集成进行智能分析

   调用外部AI服务（如OpenAI ChatGPT API或兼容API）是提升数据价值的关键。这里需要平衡分析深度、成本和响应速度。

   3.3.1 设计高效的提示词（Prompt）

   AI分析的质量很大程度上取决于提示词。我们需要为不同的分析任务设计结构化、清晰的提示。

   # 硬技能提取的提示词示例 HARD_SKILLS_PROMPT = """ 你是一个专业的IT招聘分析师。请从以下职位描述中，提取出明确要求或强烈暗示的硬技能（编程语言、框架、工具、平台等）。 请只返回一个JSON数组，格式如下：["技能1", "技能2", ...]。不要返回任何其他解释性文字。 职位描述： {job_description} """ # 签证赞助分析的提示词示例 VISA_SPONSOR_PROMPT = """ 分析以下职位描述，判断该公司是否可能为合适的候选人提供工作签证（Visa Sponsorship）或搬迁支持（Relocation Assistance）。 请只返回一个JSON对象，格式如下：{{"provides_visa_sponsorship": true/false, "confidence": "high/medium/low", "relevant_snippets": ["原文片段1", ...]}}。 你的判断应基于描述中是否包含如“visa sponsorship”, “work permit”, “relocation”, “will sponsor”, “must be eligible to work in [Country]”等关键词或类似表述。 职位描述： {job_description} """

   3.3.2 实现容错与降级处理

   AI服务可能不稳定、超时或返回非预期格式。代码必须有健壮的容错机制。

   import aiohttp import json import asyncio from tenacity import retry, stop_after_attempt, wait_exponential class AIServiceClient: def __init__(self, api_key, base_url): self.api_key = api_key self.base_url = base_url self.session = aiohttp.ClientSession(headers={'Authorization': f'Bearer {api_key}'}) @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) async def analyze_skills(self, description): """分析职位技能，重试3次""" if not description or len(description) < 50: return [] # 描述太短，直接返回空 prompt = HARD_SKILLS_PROMPT.format(job_description=description[:3000]) # 限制长度 payload = { "model": "gpt-3.5-turbo", # 或更经济的模型 "messages": [{"role": "user", "content": prompt}], "temperature": 0.1, # 低随机性，确保结果稳定 "max_tokens": 200 } try: async with self.session.post(f"{self.base_url}/chat/completions", json=payload, timeout=30) as resp: if resp.status == 200: result = await resp.json() content = result['choices'][0]['message']['content'].strip() # 尝试解析JSON，如果失败，则尝试清理字符串或返回空 try: skills = json.loads(content) if isinstance(skills, list): return skills except json.JSONDecodeError: # 可能AI返回了非JSON内容，可以尝试用正则提取或用简单关键词匹配降级处理 print(f"AI返回非JSON内容: {content}") return self._fallback_skill_extraction(description) else: raise Exception(f"AI API error: {resp.status}") except (asyncio.TimeoutError, aiohttp.ClientError) as e: print(f"AI服务请求失败: {e}") raise # 触发重试 return [] # 最终失败返回空 def _fallback_skill_extraction(self, description): """降级方案：使用预定义的关键词列表进行简单匹配""" common_skills = ["python", "java", "javascript", "react", "aws", "docker", "kubernetes", "sql", "django", "fastapi"] found = [] desc_lower = description.lower() for skill in common_skills: if skill in desc_lower: found.append(skill) return found

   3.3.3 成本控制策略

   AI API调用是按Token收费的。为了控制成本：

   • 缓存结果：对相同的职位描述（可通过MD5哈希判断）的AI分析结果进行缓存，避免重复分析。
   • 分析前过滤：只对通过逻辑表达式初步筛选的“高潜力”职位进行AI深度分析，而不是对所有爬取的职位都分析。
   • 使用更经济的模型：对于技能提取和签证分析这类相对简单的任务，使用gpt-3.5-turbo而非gpt-4，可以节省大量成本。
   • 设置预算和限额：在代码或监控中设置每日/每周的API调用上限。

   4. 系统部署、运维与问题排查实录

   4.1 基于Docker的部署与配置详解

   项目推荐使用Docker进行部署，这确保了环境的一致性。我们来看一下关键的配置步骤。

   4.1.1 环境变量配置文件 (.env)

   这是系统的核心配置文件，必须妥善保管，不应提交到版本库。

   # .env 文件示例 # 数据库配置 (使用PostgreSQL) DATABASE_URL=postgresql+asyncpg://user:password@postgres_host:5432/linkedin_scraper_db # Telegram Bot配置 TELEGRAM_BOT_TOKEN=YOUR_BOT_TOKEN_FROM_BOTFATHER TELEGRAM_CHANNEL_ID=@your_channel_username # 或具体的频道ID（负数） TELEGRAM_ADMIN_ID=YOUR_PERSONAL_CHAT_ID # 用于接收系统告警 # AI服务配置 (例如使用OpenAI兼容API) AI_API_KEY=your_ai_service_api_key AI_BASE_URL=https://api.openai.com/v1 # 或第三方服务地址 # 安全密钥 (用于JWT令牌签名等，可用`openssl rand -hex 32`生成) SECRET_KEY=your_super_secret_random_string_here # 爬虫相关配置 REQUEST_DELAY_MS=2000 # 请求间延迟，模拟人类操作，单位毫秒 MAX_CONCURRENT_WORKERS=5 # 最大并发工作者数量 HEADLESS_BROWSER=true # 是否使用无头浏览器

   4.1.2 Docker Compose 编排

   一个典型的docker-compose.yml文件会包含数据库、后端API和初始化服务。

   version: '3.8' services: postgres: image: postgres:15-alpine environment: POSTGRES_DB: linkedin_scraper_db POSTGRES_USER: user POSTGRES_PASSWORD: password volumes: - postgres_data:/var/lib/postgresql/data ports: - "5432:5432" # 仅开发暴露，生产环境应使用内部网络 backend: build: . # build: ./backend # 如果后端代码在单独目录 depends_on: - postgres environment: # 通过.env文件注入，在docker-compose中可以使用env_file指令 env_file: - .env ports: - "8000:8000" # 暴露FastAPI服务端口 command: uvicorn main:app --host 0.0.0.0 --port 8000 --reload # 开发用--reload volumes: # 挂载代码目录，便于开发时热重载 - ./app:/app # 可以添加Redis服务用于任务队列（如果未来扩展） # redis: # image: redis:7-alpine # ports: # - "6379:6379" volumes: postgres_data:

   4.1.3 数据库迁移与初始数据加载

   使用Alembic进行数据库版本管理是Python项目的标准实践。在项目启动时，需要创建数据库表结构并可能加载一些初始数据（如预定义的关键词列表）。

   # 1. 初始化Alembic（如果尚未） alembic init migrations # 2. 修改alembic.ini中的数据库连接字符串指向你的环境（如使用.env变量） # 3. 生成迁移脚本（当模型发生变化时） alembic revision --autogenerate -m "Initial tables" # 4. 应用迁移 alembic upgrade head # 5. 加载初始数据（如keywords.json） # 假设FastAPI应用有一个端点或脚本可以加载数据 python -m scripts.load_initial_data

   4.2 爬虫工作者的启动与管理

   根据文档，Playwright部分建议在宿主机而非Docker容器内运行，这可能是由于浏览器依赖或GPU加速（如果需要）的考虑。使用并发工作者（-w 5）可以显著提升爬取效率。

   # 在宿主机上运行工作者 # 首先确保安装了Playwright的浏览器：playwright install chromium cd /path/to/worker_directory # 启动5个并发工作者，使用无头模式，只爬取流行国家（假设-p参数为此意） python main.py -w 5 --headless -p # 可以使用进程管理工具如PM2、Supervisor来管理工作者进程，确保崩溃后自动重启 # 使用PM2示例 (需先安装Node.js和pm2) pm2 start main.py --name linkedin-worker --interpreter python3 -- -w 3 --headless pm2 save pm2 startup # 设置开机自启

   工作者脚本（main.py）的核心循环逻辑：

   # main.py 简化结构 import asyncio import argparse from concurrent.futures import ProcessPoolExecutor import signal import sys from worker import run_single_worker def main(): parser = argparse.ArgumentParser() parser.add_argument('-w', '--workers', type=int, default=1, help='Number of concurrent workers') parser.add_argument('--headless', action='store_true', help='Run browser in headless mode') parser.add_argument('-p', '--popular-only', action='store_true', help='Only scrape jobs from popular countries') args = parser.parse_args() # 优雅退出处理 stop_event = asyncio.Event() def signal_handler(sig, frame): print("\nShutdown signal received...") stop_event.set() signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) async def run_all_workers(): tasks = [] for i in range(args.workers): # 每个worker是一个独立的异步任务 task = asyncio.create_task( run_single_worker(worker_id=i, headless=args.headless, popular_only=args.popular_only, stop_event=stop_event) ) tasks.append(task) # 等待所有任务完成（或被停止信号中断） await asyncio.gather(*tasks, return_exceptions=True) try: asyncio.run(run_all_workers()) except KeyboardInterrupt: pass finally: print("All workers stopped.") if __name__ == '__main__': main()

   4.3 常见问题排查与解决方案速查表

   在长期运行中，你几乎一定会遇到以下问题。这里是我的排查笔记。

   问题现象	可能原因	排查步骤与解决方案
   爬取不到任何数据	1. LinkedIn页面结构已更新。 2. IP被封锁。 3. 浏览器指纹被检测。 4. 代理全部失效。	1.手动检查：用相同的关键词和地点在真实浏览器中访问LinkedIn，查看元素选择器是否还能用。 2.检查日志：查看工作者日志是否有TimeoutError或ElementNotFound错误。 3.切换代理/IP：尝试不使用代理或更换一个全新的代理IP测试。 4.调整指纹：在browser.new_context中更换user_agent、viewport尺寸，或启用playwright-stealth。 5.增加延迟：在关键操作（如点击、滚动）后增加page.wait_for_timeout(3000)，模拟真人操作速度。
   Telegram Bot无响应	1. Bot Token错误或失效。 2.TELEGRAM_CHANNEL_ID格式错误。 3. Bot未加入频道或没有发送消息权限。 4. 网络问题导致API请求失败。	1.验证Token：用curl或浏览器访问https://api.telegram.org/bot<YOUR_TOKEN>/getMe，看是否返回正确信息。 2.检查ID格式：个人聊天ID是数字，频道ID以@开头或是负数。确保Bot是频道管理员并有post_messages权限。 3.查看Bot日志：检查发送消息的API调用是否返回403（无权限）或400（参数错误）。 4.测试简单消息：写一个简单的脚本，只发送一条“Hello”消息，隔离问题。
   AI分析服务返回错误或超时	1. API Key过期或额度不足。 2. 网络连接问题。 3. 请求频率超限。 4. 提示词格式导致AI返回非JSON。	1.检查账户余额/用量：登录AI服务提供商的控制台查看。 2.测试连通性：用curl或ping测试到API端点的网络。 3.实现退避重试：使用tenacity库为AI请求添加指数退避重试机制。 4.简化/标准化提示词：确保提示词明确要求返回JSON，并做好解析失败的异常处理（降级到关键词匹配）。 5.添加请求限流：在代码中控制调用频率，例如每秒不超过1-2次请求。
   数据库连接失败	1..env中的DATABASE_URL配置错误。 2. 数据库服务未启动。 3. 网络或防火墙阻止连接。 4. 数据库用户权限不足。	1.验证连接字符串：使用psql或数据库客户端工具直接连接测试。 2.检查容器/服务状态：docker-compose ps确认PostgreSQL容器正在运行。 3.检查日志：查看后端应用和数据库容器的日志，寻找连接错误信息。 4.确认网络：在应用容器内使用nc -zv postgres_host 5432测试端口连通性。
   逻辑表达式匹配结果不符合预期	1. 表达式解析有bug。 2. 职位文本预处理不一致（如大小写、标点）。 3. 用户输入了过于复杂或歧义的表达式。	1.单元测试：为表达式解析器编写全面的单元测试，覆盖各种边界情况（嵌套括号、NOT运算符等）。 2.调试输出：在匹配时，打印出表达式AST、待匹配的文本以及中间匹配结果，进行人工复核。 3.文本规范化：在匹配前，对职位文本和关键词都进行相同的处理（如转为小写、移除多余空格和标点）。 4.提供表达式验证：在用户提交过滤器时，提供一个预览或测试功能，让用户用样例文本测试其表达式。
   系统内存/CPU占用过高	1. 并发工作者(-w)数量设置过多。 2. Playwright浏览器实例未正确关闭导致内存泄漏。 3. 数据库连接未释放。	1.资源监控：使用htop或docker stats监控资源使用情况。 2.减少并发数：根据机器性能（CPU核心数、内存）调整-w参数，每个Playwright实例都消耗不小内存。 3.确保资源释放：在try...finally块或异步上下文管理器中确保browser.close()和page.close()被调用。 4.使用连接池：确保数据库连接使用连接池，并在使用后及时归还。

   4.4 性能优化与扩展思路

   当系统稳定运行后，可以考虑以下优化和扩展方向：

   1. 引入任务队列（如Celery + Redis）：目前工作者可能直接从数据库读取关键词。引入任务队列可以将爬取任务解耦，实现更好的任务调度、重试和优先级管理。FastAPI接收到新的关键词后，将其作为任务发布到队列，工作者作为消费者从队列拉取任务执行。
   2. 分布式爬取：如果单机性能成为瓶颈，可以将工作者部署到多台服务器上。它们共享同一个任务队列和数据库，协同工作。需要确保代理池和去重机制（如基于职位ID）在分布式环境下正常工作。
   3. 更智能的调度策略：根据关键词的热度、历史爬取频率、目标国家的时区等因素，动态调整爬取任务的优先级和时间。例如，在目标地区的办公时间增加爬取频率。
   4. 数据去重与更新：建立基于职位唯一ID（如LinkedIn的>