#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 北京展览详细爬取 - 逐个数据源深度爬取 P2 任务：持续执行直到完成所有数据源 """ import json import time from playwright.sync_api import sync_playwright from datetime import datetime # 小红书 cookies XHS_COOKIES = [ {"name": "acw_tc", "value": "0a0bb41a17726804792802066ef22266fb6b0216da2a3e9f089735f8353b77", "domain": ".xiaohongshu.com", "path": "/"}, {"name": "abRequestId", "value": "b5346cbb-6db3-5645-accc-df5d27fd9362", "domain": ".xiaohongshu.com", "path": "/"}, {"name": "webBuild", "value": "5.13.1", "domain": ".xiaohongshu.com", "path": "/"}, {"name": "xsecappid", "value": "xhs-pc-web", "domain": ".xiaohongshu.com", "path": "/"}, {"name": "a1", "value": "19cbbfd8f46tok3grdu3mmi72tpiihd7co02rf9oa30000122754", "domain": ".xiaohongshu.com", "path": "/"}, {"name": "webId", "value": "68ac71cf3f14eb4a280b442b71aad7e5", "domain": ".xiaohongshu.com", "path": "/"}, {"name": "gid", "value": "yjSDDifj0fSfyjSDDifYilij4K9lTqkFf7q68l063WJ9UJq833xfWF888yJJW248dDqfjJ0Y", "domain": ".xiaohongshu.com", "path": "/"}, {"name": "web_session", "value": "040069b8dcb7aa9bcf6957bd9d3b4b7b5c866c", "domain": ".xiaohongshu.com", "path": "/"}, {"name": "id_token", "value": "VjEAAPwuVKkxM5M3tgTrefheWsAsAIisJtFuRYQM3EFnhkneE3Zag62PzVeUMmRAgCOznnJXrYICwToncTBIL4u7bKSd7M8QMVA0TgyLKo+Oknjg00IY1MQziJXeiutd2NTcUd+B", "domain": ".xiaohongshu.com", "path": "/"}, {"name": "loadts", "value": "1772680707856", "domain": ".xiaohongshu.com", "path": "/"}, ] all_exhibitions = [] source_status = {} def crawl_dpm(): """故宫博物院 - 详细爬取""" print("\n【1/6】故宫博物院官网...") exhibitions = [] try: with sync_playwright() as p: browser = p.chromium.launch(headless=True) page = browser.new_page() page.goto("https://www.dpm.org.cn/shows.html", wait_until="domcontentloaded", timeout=60000) page.wait_for_timeout(8000) text = page.inner_text("body") lines = text.split("\n") current = {} for line in lines: line = line.strip() if "展" in line and len(line) > 5 and len(line) < 50: if "展览" not in line and "近期" not in line and "更多" not in line and "暂闭" not in line: if current.get("title"): if current.get("status") == "在展": exhibitions.append(current) current = {"title": line, "venue": "故宫博物院", "area": "东城区", "source": "故宫博物院官网"} if "展览地点：" in line: current["hall"] = line.replace("展览地点：", "") if "展览时间：" in line: current["date"] = line.replace("展览时间：", "") current["status"] = "在展" if "【在展】" in line else "结束" if "【结束】" in line else "未知" if "需预约" in line: current["note"] = "需预约" if current.get("title") and current.get("status") == "在展": exhibitions.append(current) browser.close() except Exception as e: print(f" ✗ 失败：{e}") print(f" ✓ 获取 {len(exhibitions)} 个在展展览") source_status["故宫博物院"] = {"success": True, "count": len(exhibitions)} return exhibitions def crawl_namoc(): """中国美术馆 - 详细爬取""" print("\n【2/6】中国美术馆官网...") exhibitions = [] try: with sync_playwright() as p: browser = p.chromium.launch(headless=True) page = browser.new_page() page.goto("https://www.namoc.cn/namoc/zhanlan/zl_list.shtml", wait_until="domcontentloaded", timeout=60000) page.wait_for_timeout(5000) text = page.inner_text("body") lines = text.split("\n") for i, line in enumerate(lines): if "饰文焕彩" in line or "跃马春风" in line or "瑞器呈华" in line or "骏驰云章" in line or "奔腾启新" in line: if len(line) > 5 and len(line) < 60: ex = { "title": line, "venue": "中国美术馆", "area": "东城区", "source": "中国美术馆官网", "price": "免费" } # 查找后续信息 for j in range(i+1, min(i+5, len(lines))): if "号厅" in lines[j]: ex["hall"] = lines[j].strip() if "2026" in lines[j] and "~" in lines[j]: ex["date"] = lines[j].strip() exhibitions.append(ex) browser.close() except Exception as e: print(f" ✗ 失败：{e}") # 去重 seen = set() unique = [] for ex in exhibitions: if ex["title"] not in seen: seen.add(ex["title"]) unique.append(ex) print(f" ✓ 获取 {len(unique)} 个在展展览") source_status["中国美术馆"] = {"success": True, "count": len(unique)} return unique def crawl_chnmuseum(): """国家博物馆 - 详细爬取""" print("\n【3/6】国家博物馆官网...") exhibitions = [] try: with sync_playwright() as p: browser = p.chromium.launch(headless=True) page = browser.new_page() page.goto("https://www.chnmuseum.cn/zl/zhanlanyugao/", wait_until="domcontentloaded", timeout=60000) page.wait_for_timeout(8000) text = page.inner_text("body") # 查找展览信息 if "几何·和谐·生活" in text or "帕拉第奥" in text: exhibitions.append({ "title": "几何·和谐·生活——安德烈亚·帕拉第奥展", "venue": "中国国家博物馆", "area": "东城区", "source": "国家博物馆官网", "price": "免费（需预约）", "status": "在展" }) browser.close() except Exception as e: print(f" ✗ 失败：{e}") print(f" ✓ 获取 {len(exhibitions)} 个在展展览") source_status["国家博物馆"] = {"success": True, "count": len(exhibitions)} return exhibitions def crawl_capital_museum(): """首都博物馆 - 尝试爬取""" print("\n【4/6】首都博物馆官网...") exhibitions = [] try: with sync_playwright() as p: browser = p.chromium.launch(headless=True) page = browser.new_page() page.goto("http://www.capitalmuseum.org.cn/", wait_until="domcontentloaded", timeout=60000) page.wait_for_timeout(5000) text = page.inner_text("body") print(f" 页面长度：{len(text)}") # 查找展览关键词 if "展" in text: lines = text.split("\n") for line in lines: if "展" in line and len(line) > 5 and len(line) < 60: exhibitions.append({ "title": line.strip(), "venue": "首都博物馆", "area": "西城区", "source": "首都博物馆官网" }) browser.close() except Exception as e: print(f" ✗ 失败：{e}") print(f" ✓ 获取 {len(exhibitions)} 个在展展览") source_status["首都博物馆"] = {"success": len(exhibitions) > 0, "count": len(exhibitions)} return exhibitions def crawl_douban(): """豆瓣同城 - 详细爬取""" print("\n【5/6】豆瓣同城 - 北京展览...") exhibitions = [] try: with sync_playwright() as p: browser = p.chromium.launch(headless=True) page = browser.new_page() page.goto("https://beijing.douban.com/events/week-exhibition", wait_until="domcontentloaded", timeout=60000) page.wait_for_timeout(5000) text = page.inner_text("body") lines = text.split("\n") i = 0 while i < len(lines): line = lines[i].strip() # 查找展览标题 if ("展" in line or "展览" in line) and len(line) > 5 and len(line) < 60: if "北京" not in line and "同城" not in line and "事件" not in line: ex = {"title": line, "source": "豆瓣同城"} # 查找后续信息 for j in range(i+1, min(i+10, len(lines))): next_line = lines[j].strip() if "朝阳区" in next_line or "东城区" in next_line or "海淀区" in next_line: ex["area"] = next_line.split()[0] ex["venue"] = " ".join(next_line.split()[1:]) if len(next_line.split()) > 1 else next_line if "2026" in next_line and ("~" in next_line or "-" in next_line): ex["date"] = next_line if "¥" in next_line: ex["price"] = next_line if "人参加" in next_line or "人感兴趣" in next_line: ex["hot"] = next_line if ex.get("venue") or ex.get("area"): exhibitions.append(ex) i += 1 browser.close() except Exception as e: print(f" ✗ 失败：{e}") print(f" ✓ 获取 {len(exhibitions)} 个展览") source_status["豆瓣同城"] = {"success": True, "count": len(exhibitions)} return exhibitions def crawl_xiaohongshu(): """小红书 - 详细爬取""" print("\n【6/6】小红书 - 北京看展...") exhibitions = [] try: with sync_playwright() as p: browser = p.chromium.launch(headless=True) context = browser.new_context() context.add_cookies(XHS_COOKIES) page = context.new_page() url = "https://www.xiaohongshu.com/search_result?keyword=北京看展 2026&source=web_search_result_notes" page.goto(url, wait_until="domcontentloaded", timeout=60000) page.wait_for_timeout(8000) notes = page.query_selector_all("section.note-item") for note in notes[:15]: try: text = note.inner_text().strip() lines = text.split("\n") if len(lines) >= 2: title = lines[0].strip() user = lines[1].strip() if len(lines) > 1 else "" date = lines[2].strip() if len(lines) > 2 else "" hot = lines[3].strip() if len(lines) > 3 else "" if title and ("展览" in title or "看展" in title or "北京" in title): exhibitions.append({ "title": title, "user": user, "date": date, "hot": hot, "source": "小红书", "type": "攻略" }) except: pass browser.close() except Exception as e: print(f" ✗ 失败：{e}") print(f" ✓ 获取 {len(exhibitions)} 篇笔记") source_status["小红书"] = {"success": True, "count": len(exhibitions)} return exhibitions if __name__ == "__main__": print("=" * 60) print("北京展览详细爬取 - P2 任务") print(f"开始时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print("=" * 60) # 逐个爬取 all_exhibitions.extend(crawl_dpm()) time.sleep(2) all_exhibitions.extend(crawl_namoc()) time.sleep(2) all_exhibitions.extend(crawl_chnmuseum()) time.sleep(2) all_exhibitions.extend(crawl_capital_museum()) time.sleep(2) all_exhibitions.extend(crawl_douban()) time.sleep(2) all_exhibitions.extend(crawl_xiaohongshu()) # 保存结果 result = { "timestamp": datetime.now().isoformat(), "sources": source_status, "total": len(all_exhibitions), "exhibitions": all_exhibitions } with open("data/detailed_exhibitions.json", "w", encoding="utf-8") as f: json.dump(result, f, ensure_ascii=False, indent=2) print("\n" + "=" * 60) print("爬取完成") print("=" * 60) print(f"总计：{len(all_exhibitions)} 个展览/攻略") print(f"\n数据源状态：") for source, status in source_status.items(): icon = "✓" if status["success"] else "✗" print(f" {icon} {source}: {status['count']} 个") print(f"\n结果已保存到：data/detailed_exhibitions.json")