目录一、项目说明二、源码总结一、项目说明1、该项目仅用于学习参考。2、项目使用的是playwright。采集字段包含序号、公司名称、统一社会信用代码、法定代表人、注册资本、成立日期、电话、邮箱、官网、地址、企查查行业、国标行业、企业规模、员工人数、营业收入、简介、股东、对外投资、所属地区、公司类型、经营状态、经营范围、实际控制人、受益所有人、标签。二、源码功能按关键词搜索企查查爬取公司详情页的可见信息依赖playwright, requests安装pip install playwright playwright install chromiumimport asyncio import csv import json import os import random import re import time from playwright.async_api import async_playwright OUTPUT_CSV qcc_company_data.csv SEARCH_DELAY (1, 3) DETAIL_DELAY (1, 5) MAX_CONCURRENT 2 CSV_FIELDNAMES [ 序号, 公司名称, 统一社会信用代码, 法定代表人, 注册资本, 成立日期, 电话, 邮箱, 官网, 地址, 企查查行业, 国标行业, 企业规模, 员工人数, 营业收入, 简介, 股东, 对外投资, 所属地区, 公司类型, 经营状态, 经营范围, 实际控制人, 受益所有人, 标签, ] USER_AGENTS [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.1 Safari/605.1.15, ] COOKIE_FILE qcc_cookies.json def load_cookies(): 从文件加载已保存的Cookies if os.path.exists(COOKIE_FILE): try: with open(COOKIE_FILE, r, encodingutf-8) as f: cookies json.load(f) print(f已加载 {len(cookies)} 个Cookie) return cookies except Exception as e: print(f加载Cookie失败: {e}) return None def save_cookies(cookies): 保存Cookies到文件 with open(COOKIE_FILE, w, encodingutf-8) as f: json.dump(cookies, f, ensure_asciiFalse, indent2) print(fCookie已保存到 {COOKIE_FILE}) def get_next_index(): 获取CSV文件中下一个可用的序号 if not os.path.exists(OUTPUT_CSV) or os.path.getsize(OUTPUT_CSV) 0: return 1 try: with open(OUTPUT_CSV, r, encodingutf-8-sig) as f: reader csv.DictReader(f) max_idx 0 for row in reader: try: idx int(row.get(序号, 0)) if idx max_idx: max_idx idx except (ValueError, TypeError): continue return max_idx 1 except Exception: return 1 def append_to_csv(data): 追加单条数据到CSV文件自动分配序号 file_exists os.path.exists(OUTPUT_CSV) and os.path.getsize(OUTPUT_CSV) 0 with open(OUTPUT_CSV, a, encodingutf-8-sig, newline) as f: writer csv.DictWriter(f, fieldnamesCSV_FIELDNAMES) if not file_exists: writer.writeheader() row {k: data.get(k, ) for k in CSV_FIELDNAMES} writer.writerow(row) def save_to_csv(results): 保存全部数据到CSV追加写入 for data in results: append_to_csv(data) def get_random_delay(delay_range): 获取随机延时秒 return random.uniform(*delay_range) def extract_field(text, patterns): 从文本中提取字段值尝试多个正则模式 for pattern in patterns: match re.search(pattern, text, re.DOTALL) if match: value match.group(1).strip() value re.sub(r\s, , value) return value return async def wait_for_page_stable(page, timeout15000): 等待页面稳定network idle try: await page.wait_for_load_state(networkidle, timeouttimeout) except Exception: pass async def check_captcha(page): 检测页面是否出现验证码/滑块 captcha_selectors [ #nc_1_n1z, .nc_iconfont, iframe[src*captcha], iframe[src*verify], .captcha, .slider, div[class*captcha], div[class*verify], div[class*slider], ] for sel in captcha_selectors: try: elem await page.query_selector(sel) if elem: return True except Exception: continue return False async def debug_save_page(page, namedebug): 保存页面调试信息截图HTML try: screenshot_path fdebug_{name}.png await page.screenshot(pathscreenshot_path) html_path fdebug_{name}.html content await page.content() with open(html_path, w, encodingutf-8) as f: f.write(content) print(f 调试文件已保存: {screenshot_path}, {html_path}) except Exception as e: print(f 保存调试文件失败: {e}) async def search_companies(page, keyword, max_pages3): 按关键词搜索公司返回公司URL列表 company_urls [] print(f前往企查查首页...) try: await page.goto(https://www.qcc.com, wait_untildomcontentloaded, timeout30000) await asyncio.sleep(3) except Exception as e: print(f首页加载失败: {e}) search_input None try: search_input await page.query_selector(input[typesearch], input[namekey], input[placeholder*搜索], input.search, #key) if not search_input: search_input await page.query_selector(input[typetext]) except Exception: pass if search_input: print(f使用首页搜索框搜索: {keyword}) try: await search_input.click() await search_input.fill() await search_input.type(keyword, delay50) await asyncio.sleep(0.5) search_btn await page.query_selector(button[typesubmit], .search-btn, button:has-text(搜索), .btn-search) if search_btn: await search_btn.click() else: await search_input.press(Enter) await asyncio.sleep(5) current_url page.url print(f 跳转后URL: {current_url}) if login in current_url or passport in current_url: print(f 跳转到登录页需要Cookie) return company_urls page_text await page.inner_text(body) if 访问过于频繁 in page_text: print( 触发频率限制等待30秒后重试...) print( 如需立即继续请在浏览器中手动操作) await asyncio.sleep(30) if 验证码 in page_text or 安全验证 in page_text: print( 页面需要安全验证等待处理...) input( 请在浏览器中完成验证然后按 Enter 键继续 ) await asyncio.sleep(3) results await page.query_selector_all(a[href*/company/]) if not results: results await page.query_selector_all(.search-item a, .list-item a) if not results: results await page.query_selector_all(a.name) if not results: results await page.query_selector_all(.title a, .name a) if not results: all_links await page.query_selector_all(a[href]) results [] for link in all_links: try: href await link.get_attribute(href) if href and (/company/ in href or /firm/ in href): results.append(link) except Exception: continue if results: page_urls [] seen_urls set() for elem in results: try: href await elem.get_attribute(href) if href and href not in seen_urls: if not href.startswith(http): href https://www.qcc.com href page_urls.append(href) seen_urls.add(href) except Exception: continue if page_urls: print(f 找到 {len(page_urls)} 条结果) company_urls.extend(page_urls) else: print(f 解析后无有效链接) else: print(f 无搜索结果) except Exception as e: print(f 首页搜索出错: {str(e)[:80]}) else: print(f未找到搜索框尝试直接URL搜索...) search_urls [ fhttps://www.qcc.com/web/search?key{keyword}, fhttps://www.qcc.com/search?key{keyword}, ] for search_url in search_urls: print(f尝试搜索URL: {search_url}) try: await page.goto(search_url, wait_untildomcontentloaded, timeout30000) await asyncio.sleep(5) current_url page.url print(f 跳转后URL: {current_url}) if login in current_url or passport in current_url: print(f 跳转到登录页) continue page_text await page.inner_text(body) if 访问过于频繁 in page_text: print( 触发频率限制等待30秒...) await asyncio.sleep(30) continue results await page.query_selector_all(a[href*/company/]) if not results: results await page.query_selector_all(.search-item a, .list-item a) if not results: results await page.query_selector_all(a.name) if not results: results await page.query_selector_all(.title a, .name a) if not results: all_links await page.query_selector_all(a[href]) results [] for link in all_links: try: href await link.get_attribute(href) if href and (/company/ in href or /firm/ in href): results.append(link) except Exception: continue if results: page_urls [] seen_urls set() for elem in results: try: href await elem.get_attribute(href) if href and href not in seen_urls: if not href.startswith(http): href https://www.qcc.com href page_urls.append(href) seen_urls.add(href) except Exception: continue if page_urls: print(f 找到 {len(page_urls)} 条结果) company_urls.extend(page_urls) break else: print(f 解析后无有效链接) else: print(f 无搜索结果) continue except Exception as e: print(f 搜索出错: {str(e)[:80]}) continue if not company_urls: print(\n搜索无结果可能原因) print( 1. Cookie失效请重新运行并手动登录) print( 2. 关键词无匹配结果) print( 3. 触发了反爬频率限制请等待几分钟后重试) return company_urls async def parse_company_detail(page, url, index): 解析公司详情页返回完整数据 data { 序号: index, } try: await page.goto(url, wait_untildomcontentloaded, timeout30000) await wait_for_page_stable(page) await asyncio.sleep(get_random_delay(DETAIL_DELAY)) current_url page.url if login in current_url or passport in current_url: print(f 跳转到登录页跳过: {url}) return None if await check_captcha(page): print(f 检测到验证码等待处理...) input( 请在浏览器中完成验证码然后按 Enter 键继续 ) await asyncio.sleep(3) full_text await page.inner_text(body) data[公司名称] try: name_elem await page.query_selector(h1) if name_elem: data[公司名称] (await name_elem.text_content()).strip() except Exception: pass if not data[公司名称]: try: name_elem await page.query_selector(.company-name, .name) if name_elem: data[公司名称] (await name_elem.text_content()).strip() except Exception: pass if not data[公司名称]: data[公司名称] extract_field(full_text, [ rcompany-name[^]*([^]), rclassname[^]*[^]*([^]), ]) fields_map { 统一社会信用代码: [ r统一社会信用代码[:]\s*([A-Z0-9]), r91[A-Z0-9]{18}, ], 法定代表人: [ r法定代表人[:]\s*([^\n]?)(?:\n|$), r法定代表人[\s\S]*?[^]*([^]), ], 注册资本: [ r注册资本[:]\s*([^\n]?)(?:\n|$), r注册资本[\s\S]*?[^]*([^]), ], 成立日期: [ r成立日期[:]\s*(\d{4}[-/年]\d{1,2}[-/月]\d{1,2}日?), r成立日期[\s\S]*?(\d{4}[-/年]\d{1,2}[-/月]\d{1,2}日?), ], 电话: [ r电话[:]\s*([0-9\-]), r电话[\s\S]*?([0-9\-]{7,}), ], 邮箱: [ r邮箱[:]\s*([a-zA-Z0-9._%\-][a-zA-Z0-9.\-]), r邮箱[\s\S]*?([a-zA-Z0-9._%\-][a-zA-Z0-9.\-]), ], 官网: [ r官网[:]\s*([a-zA-Z0-9.\-]), r官网[\s\S]*?([a-zA-Z0-9.\-]), ], 地址: [ r地址[:]\s*([^\n]?)(?:\n|$), r注册地址[\s\S]*?([^\n]), ], 企查查行业: [ r企查查行业[:]\s*([^\n]?)(?:\n|$), r企查查行业[\s\S]*?([^\n]), ], 国标行业: [ r国标行业[:]\s*([^\n]?)(?:\n|$), r国标行业[\s\S]*?([^\n]), ], 企业规模: [ r企业规模[:]\s*([^\n]?)(?:\n|$), r企业规模[\s\S]*?([^\n]), ], 员工人数: [ r员工人数[:]\s*([^\n]?)(?:\n|$), r员工人数[\s\S]*?([^\n]), ], 营业收入: [ r营业收入[:]\s*([^\n]?)(?:\n|$), r营业收入[\s\S]*?([^\n]), ], 简介: [ r简介[:]\s*([^\n](?:\n(?!股东|对外|风险)[^\n])*), ], 所属地区: [ r所属地区[:]\s*([^\n]?)(?:\n|$), r所属地区[\s\S]*?([^\n]), ], 公司类型: [ r公司类型[:]\s*([^\n]?)(?:\n|$), r公司类型[\s\S]*?([^\n]), ], 经营状态: [ r经营状态[:]\s*([^\n]?)(?:\n|$), r经营状态[\s\S]*?([^\n]), ], 经营范围: [ r经营范围[:]\s*([^\n](?:\n[^\n])*), ], 实际控制人: [ r实际控制人[:]\s*([^\n]?)(?:\n|$), r实际控制人[\s\S]*?([^\n]), ], 受益所有人: [ r受益所有人[:]\s*([^\n]?)(?:\n|$), r受益所有人[\s\S]*?([^\n]), ], } for field_name, patterns in fields_map.items(): if not data.get(field_name): value extract_field(full_text, patterns) if value: data[field_name] value try: info_items await page.query_selector_all(.basic-info .item, .info .item, .base-info .item) for item in info_items: try: text await item.text_content() text text.strip() if not text: continue for field_name in CSV_FIELDNAMES: if field_name in text and not data.get(field_name): value text.replace(field_name, ).strip(:).strip() if value: data[field_name] value except Exception: continue except Exception: pass if not data.get(公司名称): print(f 未能解析公司名称) return None for key in data: if isinstance(data[key], str): data[key] re.sub(r\s, , data[key]).strip() print(f [{index}] {data[公司名称]} - 获取成功) except Exception as e: error_str str(e) print(f [{index}] 解析出错: {error_str[:80]}) if 403 in error_str or 429 in error_str: print(f IP可能被封) return None return data async def handle_no_results(page, context, keyword): 搜索无结果时的处理返回用户选择的操作retry/change/login/menu/quit print(\n * 50) print( 未找到任何公司) print( 可能原因Cookie失效、关键词无结果、频率限制) print( * 50) print(\n请选择操作:) print( 1. 重试搜索相同关键词) print( 2. 换一个关键词) print( 3. 检查/重新登录) print( 4. 返回主菜单) print( 5. 退出程序) while True: choice input(\n请选择 (1/2/3/4/5): ).strip() if choice 1: return retry elif choice 2: new_kw input(输入新关键词: ).strip() if new_kw: return (change, new_kw) else: print(关键词不能为空) continue elif choice 3: print(正在跳转首页请在浏览器中完成登录...) await page.goto(https://www.qcc.com, wait_untildomcontentloaded) await asyncio.sleep(3) input( 登录完成后按 Enter 键继续 ) new_cookies await context.cookies() if new_cookies: save_cookies(new_cookies) print(fCookie已保存共 {len(new_cookies)} 个) verify input(是否已登录成功(y是 / n继续等待): ).strip().lower() if verify y: return retry else: print(请在浏览器中完成登录后再继续...) continue elif choice 4: return menu elif choice 5: print(程序即将退出...) return quit else: print(请输入 1-5) async def concurrent_crawl(context, page, company_urls, max_workers2, start_index1): 多页面并发爬取公司详情 results [] results_lock asyncio.Lock() url_index 0 index_lock asyncio.Lock() async def worker(worker_id, work_page): nonlocal url_index while True: async with index_lock: if url_index len(company_urls): break current_idx url_index current_url company_urls[current_idx] url_index 1 global_idx start_index current_idx print(f[Worker-{worker_id}] [{current_idx 1}/{len(company_urls)}] 爬取: {current_url}) try: data await parse_company_detail(work_page, current_url, global_idx) if data: async with results_lock: results.append(data) append_to_csv(data) delay get_random_delay(DETAIL_DELAY) print(f[Worker-{worker_id}] 等待 {delay:.1f} 秒...) await asyncio.sleep(delay) except Exception as e: print(f[Worker-{worker_id}] 出错: {str(e)[:80]}) continue workers [] for i in range(max_workers): if i 0: work_page page else: work_page await context.new_page() workers.append(asyncio.create_task(worker(i, work_page))) await asyncio.gather(*workers) return results async def crawl_with_browser(browser, context, page, keyword, max_search_pages3, max_companiesNone): 主爬取流程搜索 逐页解析Cookie失效时支持手动登录 results [] cookies load_cookies() if cookies: try: await context.add_cookies(cookies) print(f已注入 {len(cookies)} 个Cookie) except Exception as e: print(fCookie注入失败: {e}) print(\n 验证登录状态 ) await page.goto(https://www.qcc.com, wait_untildomcontentloaded) await asyncio.sleep(3) current_url page.url need_login login in current_url or passport in current_url if need_login: print(Cookie已失效需要重新登录) print(浏览器将保持打开请在浏览器中手动登录企查查) while True: input(\n 登录完成后按 Enter 键继续 ) new_cookies await context.cookies() if new_cookies: save_cookies(new_cookies) print(f新Cookie已保存共 {len(new_cookies)} 个) need_login_verify input(是否已确认登录成功(y是 / n重新检查 / q退出): ).strip().lower() if need_login_verify y: break elif need_login_verify q: print(程序即将退出...) return results, quit elif need_login_verify n: print(请在浏览器中完成登录后再继续...) continue else: print(请输入 y、n 或 q) continue search_keyword keyword search_pages max_search_pages search_max max_companies while True: print(f\n 开始搜索关键词: {search_keyword} ) company_urls await search_companies(page, search_keyword, search_pages) print(f共找到 {len(company_urls)} 家公司) if not company_urls: print(未找到任何公司) action await handle_no_results(page, context, search_keyword) if action retry: continue elif isinstance(action, tuple) and action[0] change: search_keyword action[1] continue elif action menu: return results, menu elif action quit: return results, quit else: return results, menu if search_max and len(company_urls) search_max: company_urls company_urls[:search_max] print(f限制爬取前 {search_max} 家) start_index get_next_index() print(f\n 开始爬取公司详情{MAX_CONCURRENT}并发序号从{start_index}开始 ) try: batch_results await concurrent_crawl(context, page, company_urls, MAX_CONCURRENT, start_index) results.extend(batch_results) print(f\n 本轮完成共获取 {len(batch_results)} 家公司信息 ) print(f累计已获取 {len(results)} 家公司) print(\n请选择下一步操作:) print( 1. 继续爬取使用相同关键词重新搜索) print( 2. 换关键词继续爬取) print( 3. 修改搜索页数和数量后继续) print( 4. 返回主菜单) print( 5. 退出程序) choice input(\n请选择 (1/2/3/4/5): ).strip() if choice 1: print(继续使用当前关键词搜索...) search_max input(最多爬取公司数0不限: ).strip() search_max int(search_max) if search_max.isdigit() and int(search_max) 0 else None continue elif choice 2: search_keyword input(输入新关键词: ).strip() if not search_keyword: return results, menu continue elif choice 3: new_pages input(搜索页数: ).strip() search_pages int(new_pages) if new_pages.isdigit() else search_pages new_max input(最多爬取公司数0不限: ).strip() search_max int(new_max) if new_max.isdigit() and int(new_max) 0 else None continue elif choice 4: return results, menu elif choice 5 or choice : print(程序即将退出...) return results, quit else: return results, menu except Exception as e: print(f爬取出错: {e}) break try: new_cookies await context.cookies() if new_cookies: save_cookies(new_cookies) except Exception: pass return results, menu async def interactive_login(browser, context, page, keyword, max_search_pages3, max_companiesNone): 交互式登录模式打开浏览器让用户手动登录企查查 cookies load_cookies() if cookies: try: await context.add_cookies(cookies) print(f已注入 {len(cookies)} 个Cookie) except Exception: pass print(\n * 50) print( 登录模式) print( * 50) print( 1. 浏览器即将打开企查查网页) print( 2. 请在浏览器中完成登录) print( 3. 登录成功能看到用户名后) print( 4. 回到此终端按 Enter 键继续) print( * 50) await page.goto(https://www.qcc.com, wait_untildomcontentloaded) await asyncio.sleep(2) login_prompt_shown False while True: if not login_prompt_shown: input(\n 登录完成后按 Enter 键继续 ) login_prompt_shown True else: choice input(\n 是否已登录成功(y已登录 / n重新检查 / q退出): ).strip().lower() if choice q: print(程序已退出) return [], quit elif choice y: break elif choice n: print(正在检查登录状态...) await page.reload(wait_untildomcontentloaded) await asyncio.sleep(2) continue else: print(请输入 y、n 或 q) continue new_cookies await context.cookies() save_cookies(new_cookies) print(fCookie已保存共 {len(new_cookies)} 个) results [] search_keyword keyword search_pages max_search_pages search_max max_companies while True: print(f\n 开始搜索关键词: {search_keyword} ) company_urls await search_companies(page, search_keyword, search_pages) print(f共找到 {len(company_urls)} 家公司) if not company_urls: print(未找到任何公司) action await handle_no_results(page, context, search_keyword) if action retry: continue elif isinstance(action, tuple) and action[0] change: search_keyword action[1] continue elif action menu: return results, menu elif action quit: return results, quit else: return results, menu if search_max and len(company_urls) search_max: company_urls company_urls[:search_max] print(f限制爬取前 {search_max} 家) start_index get_next_index() print(f\n 开始爬取公司详情{MAX_CONCURRENT}并发序号从{start_index}开始 ) batch_results await concurrent_crawl(context, page, company_urls, MAX_CONCURRENT, start_index) results.extend(batch_results) print(f\n 本轮完成共获取 {len(batch_results)} 家公司 ) print(f累计已获取 {len(results)} 家公司) print(\n请选择下一步操作:) print( 1. 继续爬取使用相同关键词重新搜索) print( 2. 换关键词继续爬取) print( 3. 修改搜索页数和数量后继续) print( 4. 返回主菜单) print( 5. 退出程序) choice input(\n请选择 (1/2/3/4/5): ).strip() if choice 1: search_max input(最多爬取公司数0不限: ).strip() search_max int(search_max) if search_max.isdigit() and int(search_max) 0 else None continue elif choice 2: search_keyword input(输入新关键词: ).strip() if not search_keyword: return results, menu continue elif choice 3: new_pages input(搜索页数: ).strip() search_pages int(new_pages) if new_pages.isdigit() else search_pages new_max input(最多爬取公司数0不限: ).strip() search_max int(new_max) if new_max.isdigit() and int(new_max) 0 else None continue elif choice 4: return results, menu elif choice 5 or choice : print(程序即将退出...) return results, quit else: return results, menu try: new_cookies await context.cookies() if new_cookies: save_cookies(new_cookies) except Exception: pass return results, menu async def main(): async with async_playwright() as p: browser await p.chromium.launch( headlessFalse, args[--disable-blink-featuresAutomationControlled] ) context await browser.new_context( user_agentrandom.choice(USER_AGENTS), localezh-CN, viewport{width: 1920, height: 1080} ) page await context.new_page() cookies load_cookies() if cookies: try: await context.add_cookies(cookies) print(f已注入 {len(cookies)} 个Cookie) except Exception as e: print(fCookie注入失败: {e}) else: print(\n * 50) print( 未检测到登录Cookie文件!) print( 首次使用请选择【模式2】手动登录企查查) print( 登录成功后Cookie将自动保存到 qcc_cookies.json) print( 之后即可使用【模式1】直接爬取) print( * 50) print(\n浏览器已启动全程保持打开只有手动选择退出才会关闭。\n) while True: print( * 50) print(企查查公司信息爬虫) print( * 50) print(\n请选择运行模式:) print(1. 直接爬取使用已有Cookie或无Cookie) print(2. 手动登录后爬取推荐数据更完整) print(3. 仅测试搜索功能) print(4. 退出程序关闭浏览器) mode input(\n请选择 (1/2/3/4): ).strip() if mode 4 or mode : try: final_cookies await context.cookies() if final_cookies: save_cookies(final_cookies) except Exception: pass print(程序已退出浏览器即将关闭...) await browser.close() return keyword input(请输入搜索关键词如动力: ).strip() if not keyword: print(关键词不能为空) continue max_pages input(搜索页数默认3: ).strip() max_pages int(max_pages) if max_pages.isdigit() else 3 max_comp input(最多爬取公司数0不限: ).strip() max_companies int(max_comp) if max_comp.isdigit() and int(max_comp) 0 else None results [] if mode 2: results, action await interactive_login(browser, context, page, keyword, max_pages, max_companies) if action quit: try: final_cookies await context.cookies() if final_cookies: save_cookies(final_cookies) except Exception: pass print(程序已退出浏览器即将关闭...) await browser.close() return continue elif mode 3: test_keyword keyword test_pages max_pages while True: urls await search_companies(page, test_keyword, test_pages) print(f\n搜索结果预览前10条:) for i, url in enumerate(urls[:10], 1): print(f {i}. {url}) print(f\n共 {len(urls)} 条结果) if not urls: action await handle_no_results(page, context, test_keyword) if action retry: continue elif isinstance(action, tuple) and action[0] change: test_keyword action[1] continue elif action menu: break elif action quit: try: final_cookies await context.cookies() if final_cookies: save_cookies(final_cookies) except Exception: pass print(程序已退出浏览器即将关闭...) await browser.close() return else: break else: print(\n请选择下一步:) print( 1. 测试其他关键词) print( 2. 返回主菜单) print( 3. 退出) ch input(请选择 (1/2/3): ).strip() if ch 1: test_keyword input(输入新关键词: ).strip() if not test_keyword: break continue elif ch 2: break else: try: final_cookies await context.cookies() if final_cookies: save_cookies(final_cookies) except Exception: pass print(程序已退出浏览器即将关闭...) await browser.close() return continue elif mode 1: results, action await crawl_with_browser(browser, context, page, keyword, max_pages, max_companies) if action quit: try: final_cookies await context.cookies() if final_cookies: save_cookies(final_cookies) except Exception: pass print(程序已退出浏览器即将关闭...) await browser.close() return continue else: print(无效选项请重新选择) continue if results: print(f\n数据已追加保存到 {OUTPUT_CSV}) else: print(\n未获取到任何数据) print(\n请选择下一步:) print( 1. 返回主菜单) print( 2. 退出) next_choice input(请选择 (1/2): ).strip() if next_choice 2 or next_choice : try: final_cookies await context.cookies() if final_cookies: save_cookies(final_cookies) except Exception: pass print(程序已退出浏览器即将关闭...) await browser.close() return elif next_choice 1: continue else: break if __name__ __main__: asyncio.run(main())总结该项目仅用于学习参考。有点慢改进方案 。方案一需要给ip-cookie池避免反爬调小时间区间多开几个并发.方案二本代码采用的是playwright,可改成使用cookiexhr接口ip。方案三如果用这个代码运行登录下的账号是会员账号这个好像没有限制。可以根据实际需求修改代码为适合自己的版本。