"""爬虫流程编排：搜索笑话站点 → 深度翻页采集 → AI 提取 → 入库""" import asyncio import hashlib import random import re import httpx from urllib.parse import urljoin, urlparse from crawler.ai_service import AiService from crawler.crawler_service import CrawlerService class Processor: def __init__(self, api_base: str, username: str, password: str, headless: bool = True): self.api_base = api_base.rstrip("/") self.username = username self.password = password self.token = None self.ai = None self.crawler = CrawlerService(headless=headless) self.types = [] self.crowds = [] # === API 认证 === def _login(self) -> str: resp = httpx.post( f"{self.api_base}/api/auth/login", json={"username": self.username, "password": self.password}, timeout=30, ) resp.raise_for_status() return resp.json()["access_token"] def _get(self, path: str) -> dict | list: resp = httpx.get( f"{self.api_base}{path}", headers={"Authorization": f"Bearer {self.token}"}, timeout=30, ) resp.raise_for_status() return resp.json() def _post(self, path: str, data: dict) -> dict: resp = httpx.post( f"{self.api_base}{path}", json=data, headers={"Authorization": f"Bearer {self.token}"}, timeout=30, ) resp.raise_for_status() return resp.json() # === 初始化 === def setup(self): print("[*] 正在登录...") self.token = self._login() print("[*] 登录成功") self.ai = AiService( api_base=self.api_base, username=self.username, password=self.password, ) self.ai.setup() print(f"[*] AI 配置: {self.ai.model}") self.types = self._get("/api/categories/types") self.crowds = self._get("/api/categories/crowds") print(f"[*] 分类: {len(self.types)} 种类型, {len(self.crowds)} 种人群") def _get_existing_hashes(self) -> set[str]: """获取库里已有笑话的 content hash，用于去重""" try: data = self._get("/api/admin/jokes?page=1&page_size=1000") hashes = set() for j in data.get("items", []): content = j.get("content", "") if content: hashes.add(hashlib.md5(content.encode()).hexdigest()) return hashes except Exception: return set() def _submit_joke(self, joke: dict) -> bool: """提交单条笑话至 API（status=pending 待审核）""" try: # Support both old format (type/crowd) and new format (types/crowds arrays) type_names = joke.get("types", [joke.get("type", "")]) crowd_names = joke.get("crowds", [joke.get("crowd", "")]) if isinstance(type_names, str): type_names = [type_names] if type_names else [] if isinstance(crowd_names, str): crowd_names = [crowd_names] if crowd_names else [] type_ids = [] crowd_ids = [] for n in type_names: for t in self.types: if t.get("name") == n: type_ids.append(t.get("id")) break for n in crowd_names: for c in self.crowds: if c.get("name") == n: crowd_ids.append(c.get("id")) break payload = { "title": joke.get("title", "无标题"), "content": joke.get("content", ""), "type_ids": type_ids if type_ids else None, "crowd_ids": crowd_ids if crowd_ids else None, "status": "pending", } self._post("/api/admin/jokes", payload) return True except Exception as e: print(f" [!] 提交失败: {e}") return False # === 主循环 === async def run_continuous(self, keywords: list[str], max_pages: int = 3, batch_size: int = 20): """持续深度采集循环""" print(f"\n>> 深度采集模式启动") print(f" 搜索词: {keywords}") print(f" 每批目标: {batch_size} 条") print(f" API: {self.api_base}") print(f" 按 Ctrl+C 中断\n") self.setup() existing_hashes = self._get_existing_hashes() print(f"[*] 当前库中已有 {len(existing_hashes)} 条笑话（用于去重）") type_names = [t.get("name", "") for t in self.types] crowd_names = [c.get("name", "") for c in self.crowds] total_saved = 0 visited_urls: set[str] = set() try: while True: batch_saved = 0 print(f"\n{'='*50}") print(f" 开始新一批深度采集 (已累计 {total_saved} 条)") print(f"{'='*50}") try: # Step 1: 搜索笑话站点 print(f"\n[*] 搜索笑话站点...") sites = await self.crawler.search_joke_sites(keywords, max_results=8) if not sites: print(f" [!] 未找到笑话站点，休息后重试") rest = random.randint(120, 600) print(f" 休息 {rest//60} 分 {rest%60} 秒...") await asyncio.sleep(rest) continue # Step 2: 逐个站点深度采集 for site in sites: if batch_saved >= batch_size: break domain = site["domain"] site_url = site["url"] print(f"\n{'─'*40}") print(f" 开始采集站点: {domain}") print(f"{'─'*40}") # 检查该站点失败次数 fail_count = self.crawler.site_failures.get(domain, 0) if fail_count >= 3: print(f" [!] 站点 {domain} 已连续失败 {fail_count} 次，跳过") continue saved_from_site = await self._crawl_site_deep( site_url=site_url, domain=domain, type_names=type_names, crowd_names=crowd_names, existing_hashes=existing_hashes, visited_urls=visited_urls, target=batch_size - batch_saved, ) batch_saved += saved_from_site total_saved += saved_from_site # Step 3: 休息 rest = random.randint(120, 600) print(f"\n[OK] 本批入库 {batch_saved} 条，休息 {rest//60} 分 {rest%60} 秒...") print(f" 按 Ctrl+C 中断\n") except Exception as e: rest = random.randint(180, 660) print(f"\n[!] 出错: {e}") print(f" 休息 {rest//60} 分 {rest%60} 秒后重试...") await asyncio.sleep(rest) except asyncio.CancelledError: print("\n用户中断，退出") finally: await self.crawler.close() async def _crawl_site_deep( self, site_url: str, domain: str, type_names: list[str], crowd_names: list[str], existing_hashes: set[str], visited_urls: set[str], target: int, ) -> int: """深度采集一个站点：抓取首页 → 发现翻页链接 → 逐页抓取提取笑话""" saved = 0 pages_to_crawl = [] # 1. 抓取首页 print(f" [*] 抓取首页: {site_url[:60]}") html, ok = await self.crawler.crawl_page_with_retry(site_url) if not ok: self.crawler.site_failures[domain] = self.crawler.site_failures.get(domain, 0) + 1 print(f" [!] 首页抓取失败 ({self.crawler.site_failures[domain]}/3)") return 0 visited_urls.add(site_url) # 2. 从首页提取笑话 try: jokes = self.ai.extract_jokes(html[:8000], type_names, crowd_names) saved += self._save_jokes(jokes, existing_hashes, target - saved) print(f" [+] 首页提取 {len(jokes)} 条，入库 {saved} 条") except Exception as e: print(f" [!] 首页 AI 提取失败: {e}") if saved >= target: self.crawler.site_failures[domain] = 0 return saved # 3. 发现翻页链接 page_links = self.crawler.discover_page_links(html, site_url) # 过滤已访问的链接 page_links = [l for l in page_links if l not in visited_urls] # 按页码排序 page_links.sort(key=lambda l: self.crawler._extract_page_number(l)) # 限制翻页深度，避免无限抓取 max_pages_per_site = 30 page_links = page_links[:max_pages_per_site] print(f" [*] 发现 {len(page_links)} 个翻页链接，开始逐页采集...") # 4. 逐页翻页采集 for idx, page_url in enumerate(page_links): if saved >= target: break # 检查站点是否已失效 if self.crawler.site_failures.get(domain, 0) >= 3: print(f" [!] 站点 {domain} 失败过多，跳过") break print(f" [*] 翻页 {idx+1}/{len(page_links)}: {page_url[:60]}") html, ok = await self.crawler.crawl_page_with_retry(page_url) visited_urls.add(page_url) if not ok: self.crawler.site_failures[domain] = self.crawler.site_failures.get(domain, 0) + 1 print(f" [!] 抓取失败 ({self.crawler.site_failures[domain]}/3)") continue # 重置失败计数 self.crawler.site_failures[domain] = 0 try: jokes = self.ai.extract_jokes(html[:8000], type_names, crowd_names) new_saved = self._save_jokes(jokes, existing_hashes, target - saved) if new_saved > 0: saved += new_saved print(f" [+] 提取 {len(jokes)} 条，入库 {new_saved} 条 (累计 {saved}/{target})") else: print(f" [*] 提取 {len(jokes)} 条（均为重复）") except Exception as e: print(f" [!] AI 提取失败: {e}") await asyncio.sleep(random.uniform(1, 3)) self.crawler.site_failures[domain] = 0 # 5. 发现分类链接并逐个深度采集 cat_links = self.crawler.discover_category_links(html, site_url) cat_links = [l for l in cat_links if l not in visited_urls] # 限制分类数量 max_categories = 20 cat_links = cat_links[:max_categories] if cat_links: print(f" [*] 发现 {len(cat_links)} 个分类链接，开始逐类采集...") for cat_url in cat_links: if saved >= target: break if self.crawler.site_failures.get(domain, 0) >= 3: print(f" [!] 站点 {domain} 失败过多，跳过分类") break saved += await self._crawl_category( cat_url=cat_url, domain=domain, type_names=type_names, crowd_names=crowd_names, existing_hashes=existing_hashes, visited_urls=visited_urls, target=target - saved, ) print(f" [*] 站点 {domain} 采集完成，共入库 {saved} 条") return saved async def _crawl_category( self, cat_url: str, domain: str, type_names: list[str], crowd_names: list[str], existing_hashes: set[str], visited_urls: set[str], target: int, ) -> int: """深度采集一个分类页及其翻页""" saved = 0 cat_name = cat_url.split("/")[-1].split(".")[0] print(f"\n {'─'*36}") print(f" 分类采集 [{cat_name}]: {cat_url[:60]}") print(f" {'─'*36}") # 1. 抓取分类首页 html, ok = await self.crawler.crawl_page_with_retry(cat_url) visited_urls.add(cat_url) if not ok: self.crawler.site_failures[domain] = self.crawler.site_failures.get(domain, 0) + 1 print(f" [!] 分类首页抓取失败 ({self.crawler.site_failures[domain]}/3)") return 0 # 2. AI 提取笑话 try: jokes = self.ai.extract_jokes(html[:8000], type_names, crowd_names) saved += self._save_jokes(jokes, existing_hashes, target - saved) print(f" [+] 分类首页提取 {len(jokes)} 条，入库 {saved} 条") except Exception as e: print(f" [!] 分类首页 AI 提取失败: {e}") if saved >= target: return saved # 3. 发现该分类的翻页链接 # 先尝试通用翻页模式，再尝试分类特定翻页（category-5_2.html） page_links = self.crawler.discover_page_links(html, cat_url) # 从当前分类 URL 派生出分类翻页模式（e.g. category-5 → category-5_2.html） cat_base = cat_url.split("/")[-1].replace(".html", "") cat_page_pattern = re.compile( rf'href="([^"]*{re.escape(cat_base)}[-_]?(\d+)\.html?)"', re.IGNORECASE, ) raw = self.crawler._last_raw_html or html for m in cat_page_pattern.finditer(raw): full_url = urljoin(cat_url, m.group(1)) page_links.append(full_url) page_links = [l for l in page_links if l not in visited_urls] page_links = list(set(page_links)) # 去重 page_links.sort(key=lambda l: self.crawler._extract_page_number(l)) # 限制翻页深度 max_pages_per_cat = 20 page_links = page_links[:max_pages_per_cat] if page_links: print(f" [*] 发现 {len(page_links)} 个翻页链接，开始逐页采集...") # 4. 逐页抓取 for idx, page_url in enumerate(page_links): if saved >= target: break if self.crawler.site_failures.get(domain, 0) >= 3: print(f" [!] 站点 {domain} 失败过多，跳过本分类") break print(f" [*] 翻页 {idx+1}/{len(page_links)}: {page_url[:60]}") html, ok = await self.crawler.crawl_page_with_retry(page_url) visited_urls.add(page_url) if not ok: self.crawler.site_failures[domain] = self.crawler.site_failures.get(domain, 0) + 1 print(f" [!] 分类翻页抓取失败 ({self.crawler.site_failures[domain]}/3)") continue self.crawler.site_failures[domain] = 0 try: jokes = self.ai.extract_jokes(html[:8000], type_names, crowd_names) new_saved = self._save_jokes(jokes, existing_hashes, target - saved) if new_saved > 0: saved += new_saved print(f" [+] 提取 {len(jokes)} 条，入库 {new_saved} 条 (累计 {saved}/{target})") else: print(f" [*] 提取 {len(jokes)} 条（均为重复）") except Exception as e: print(f" [!] AI 提取失败: {e}") await asyncio.sleep(random.uniform(1, 3)) print(f" [*] 分类 [{cat_name}] 采集完成，入库 {saved} 条") return saved def _save_jokes(self, jokes: list[dict], existing_hashes: set[str], limit: int) -> int: """去重并入库笑话，返回成功入库数""" saved = 0 for joke in jokes: if saved >= limit: break content_text = joke.get("content", "") if not content_text: continue h = hashlib.md5(content_text.encode()).hexdigest() if h in existing_hashes: continue existing_hashes.add(h) if self._submit_joke(joke): saved += 1 print(f" [+] 入库: {joke.get('title', '')[:30]}") return saved # === 单站点采集（供 site_crawler.py 调用） === async def crawl_site(self, site_url: str, batch_size: int = 9999): """初始化后深度采集单个站点""" print(f"\n>> 单站点采集: {site_url}\n") self.setup() existing_hashes = self._get_existing_hashes() print(f"[*] 当前库中已有 {len(existing_hashes)} 条笑话（用于去重）") type_names = [t.get("name", "") for t in self.types] crowd_names = [c.get("name", "") for c in self.crowds] domain = urlparse(site_url).netloc.lower() visited_urls: set[str] = set() try: saved = await self._crawl_site_deep( site_url=site_url, domain=domain, type_names=type_names, crowd_names=crowd_names, existing_hashes=existing_hashes, visited_urls=visited_urls, target=batch_size, ) print(f"\n[OK] 站点采集完成，共入库 {saved} 条笑话") finally: await self.crawler.close() # === 旧接口兼容 === async def run(self, keywords: list[str], max_pages: int = 3): """单轮爬取（旧接口，内部调用 run_continuous）""" await self.run_continuous(keywords, max_pages, batch_size=9999)