\s*
', '', body) # 兜底: 尾部页脚关键词截断 (仅用不会出现在正文的词, 勿用 关闭/打印 等) for kw in ('中央政府网站', '省级政府网站', '市级政府网站', '关于我们', '主办单位：', '网站标识码', '本网站支持IPv6', '政务新媒体矩阵'): i = body.find(kw) if i > 0: body = body[:i] # 附件链接绝对化 def abs_url(match): href = match.group(1) if href.startswith('//'): href = 'https:' + href elif href.startswith('/'): href = BASE + href return f']*href="([^"]+)"', abs_url, body) # 微信分享/打印等工具条 body = re.sub(r'\s*
', '', body) body = body.strip() return {"title": title, "pub_date": pub_date, "source": source, "body": body} async def run_async(pages=5): async with async_playwright() as p: browser = await p.chromium.launch(headless=True, args=[ "--disable-blink-features=AutomationControlled", "--no-sandbox", ]) ctx = await browser.new_context( user_agent=UA, ignore_https_errors=True, viewport={"width": 1366, "height": 900}, ) stealth = Stealth() await stealth.apply_stealth_async(ctx) page = await ctx.new_page() # 1. 过盾 ok = False for attempt in range(3): try: resp = await page.goto(f"{BASE}/public/column/19081?type=4&catId={CAT_ID}&action=list", wait_until="domcontentloaded", timeout=60000) await page.wait_for_timeout(8000) title = await page.title() print(f"[LOAD] attempt={attempt+1} status={resp.status if resp else 'N/A'} title={title}") ok = True break except Exception as e: print(f"[LOAD-ERR] attempt={attempt+1}: {e}") await page.wait_for_timeout(5000) if not ok: await browser.close() return 0 items_all = [] cutoff_hit = False total_pages = pages for pg in range(1, pages + 1): api = list_api(pg) try: html = await page.evaluate("""async (url) => { const r = await fetch(url, {credentials: 'include'}); return await r.text(); }""", api) except Exception as e: print(f" [API-ERR] page={pg}: {e}") continue items = parse_list(html) print(f" 第{pg}页: {len(items)} 条") if not items: # 可能到末页 if pg > 1: break continue # 提取 pageCount (从 API 尾部 Ls.pagination) m_pc = re.search(r'pageCount:(\d+)', html) if m_pc: total_pages = max(total_pages, int(m_pc.group(1))) for it in items: # CUTOFF 检查 if it["date"] and it["date"] < CUTOFF: cutoff_hit = True break items_all.append(it) if cutoff_hit: print(f" [CUTOFF] 第{pg}页出现 < {CUTOFF} 日期, 停止翻页") break await page.wait_for_timeout(random.randint(2000, 4000)) # 用 pageCount 停止 if pg >= total_pages: break print(f" [LIST] 共收集 {len(items_all)} 条, total_pages={total_pages}") if not items_all: await browser.close() return 0 # 2. 详情页 records = [] for i, it in enumerate(items_all, 1): url = it["url"] try: html = await page.evaluate("""async (url) => { const r = await fetch(url, {credentials: 'include'}); return await r.text(); }""", url) except Exception as e: print(f" [{i}/{len(items_all)}] [ERR] {e}") records.append({ "site_name": SITE_NAME, "title": it["title"], "pub_date": it["date"], "content": f"{it['title']}
", "source_url": url, "url": url, }) continue d = parse_detail(html) if d and d["body"]: title = d["title"] or it["title"] content = d["body"] pub_date = d["pub_date"] or it["date"] records.append({ "site_name": SITE_NAME, "title": title, "pub_date": pub_date, "content": content, "source_url": url, "url": url, }) else: print(f" [{i}/{len(items_all)}] [SKIP-EMPTY] {it['title'][:50]}...") records.append({ "site_name": SITE_NAME, "title": it["title"], "pub_date": it["date"], "content": f"{it['title']}
", "source_url": url, "url": url, }) if i % 10 == 0: print(f" [{i}/{len(items_all)}] 最新: {it['title'][:60]}...") await page.wait_for_timeout(random.randint(1500, 3000)) await browser.close() # 3. 入库 new_c = push_to_searchdb(records, batch_label="lj_hpgs") return new_c def main(): pages = 5 for i, a in enumerate(sys.argv): if a.startswith("--pages="): pages = int(a.split("=")[1]) elif a == "--pages" and i + 1 < len(sys.argv): pages = int(sys.argv[i + 1]) n = asyncio.run(run_async(pages=pages)) print(f"Done: {n} records") if __name__ == "__main__": main()