写了个笔趣阁爬书的小程序

shadowmage

如题。笔趣阁的服务器应该挺多，我这个是“https://www.22biqu.com/”，为了把书爬下来放手机里离线看准备的，每个章节1个TXT，用的时候合并到一起，传手机上就可以了。

1. """
   
2. 笔趣阁爬书小程序
   
3. python>=3.8
   
4. 依赖: pip install playwright tqdm
   
5.      playwright install        # 仅需第一次
   
6. Author：shadowmage
   
7. """
   
8. import os
   
9. import re
   
10. import time
    
11. from urllib.parse import urljoin
    
12. from playwright.sync_api import sync_playwright
    
13. 
    
14. SAVE_DIR  = "output"
    
15. 
    
16. # ---------- 工具 ----------
    
17. def sanitize_filename(name):
    
18.     return re.sub(r'[\\/*?:"<>|]', "_", name)
    
19. 
    
20. def get_last_page(base_url):
    
21.     with sync_playwright() as p:
    
22.         browser = p.chromium.launch(headless=True)
    
23.         ctx = browser.new_context(ignore_https_errors=True)
    
24.         page = ctx.new_page()
    
25.         page.goto(base_url, timeout=60000)
    
26.         last_option = page.locator("#indexselect option:last-child").get_attribute("value")
    
27.         browser.close()
    
28.         return int(last_option.split("/")[-2])
    
29. 
    
30. def get_chapter_links(base_url, last_page):
    
31.     links = []
    
32.     with sync_playwright() as p:
    
33.         browser = p.chromium.launch(headless=True)
    
34.         ctx = browser.new_context(ignore_https_errors=True)
    
35.         page = ctx.new_page()
    
36.         for i in range(1, last_page + 1):
    
37.             url = f"{base_url.rstrip('/')}/{i}/"
    
38.             print(f"正在获取目录页：{url}")
    
39.             page.goto(url, timeout=60000)
    
40. 
    
41.             # 只拿第二个 section-box 里的章节
    
42.             section = page.locator(".section-box").nth(1)
    
43.             section.wait_for()
    
44.             lis = section.locator("li").all()
    
45. 
    
46.             for li in lis:
    
47.                 a = li.locator("a")
    
48.                 href = a.get_attribute("href")
    
49.                 title = a.inner_text()
    
50.                 full_url = urljoin("https://www.22biqu.com", href)
    
51.                 links.append((title, full_url))
    
52.         browser.close()
    
53.     return links
    
54. 
    
55. def get_chapter_content(url):
    
56.     with sync_playwright() as p:
    
57.         browser = p.chromium.launch(headless=True)
    
58.         ctx = browser.new_context(ignore_https_errors=True)
    
59.         page = ctx.new_page()
    
60.         page.goto(url, timeout=60000)
    
61.         page.wait_for_selector("#content")
    
62.         title = page.locator("h1.title").inner_text()
    
63.         content = page.locator("#content").inner_text()
    
64.         browser.close()
    
65.         return title, content
    
66. 
    
67. # ---------- 保存txt ----------
    
68. def get_output_dir(base_url: str) -> str:
    
69.     """
    
70.     根据小说首页返回专属输出目录路径：
    
71.     1. 先打开页面抓取书名
    
72.     2. 生成合法文件夹名
    
73.     3. 确保目录存在并返回绝对路径
    
74.     """
    
75.     with sync_playwright() as p:
    
76.         browser = p.chromium.launch(headless=True)
    
77.         ctx = browser.new_context(ignore_https_errors=True)
    
78.         page = ctx.new_page()
    
79.         page.goto(base_url, timeout=60000)
    
80.         # 站点如把书名放在 h1，可按实际改选择器
    
81.         book_name = page.locator("h1:not(.logo)").inner_text().strip()
    
82.         browser.close()
    
83. 
    
84.     safe_name = re.sub(r'[\\/:*?"<>|]', "", book_name) or "book"
    
85.     out_dir = os.path.join(SAVE_DIR, safe_name)
    
86.     os.makedirs(out_dir, exist_ok=True)
    
87.     return out_dir
    
88. 
    
89. # ---------- 补漏逻辑 ----------
    
90. def repair_mode():
    
91.     if not os.path.exists("failed.txt"):
    
92.         return []                       # 无失败记录，直接回主流程
    
93.     with open("failed.txt", "r", encoding="utf-8") as f:
    
94.         tasks = [l.strip().split("\t") for l in f if l.strip()]
    
95.     if not tasks:
    
96.         return []
    
97.     print(f"检测到 failed.txt，共 {len(tasks)} 章需要补漏。")
    
98.     still_failed = []
    
99.     for idx, (title, url) in enumerate(tasks, 1):
    
100.         try:
     
101.             print(f"[{idx}/{len(tasks)}] 补漏：{title}")
     
102.             chapter_title, content = get_chapter_content(url)
     
103.             filename = f"{sanitize_filename(chapter_title)}.txt"
     
104.             with open(os.path.join("novel_output", filename), "w", encoding="utf-8") as f:
     
105.                 f.write(f"{chapter_title}\n\n{content}")
     
106.             time.sleep(15)
     
107.         except Exception as e:
     
108.             print(f"仍失败：{title}，{e}")
     
109.             still_failed.append((title, url))
     
110.     # 重写失败文件
     
111.     if still_failed:
     
112.         with open("failed.txt", "w", encoding="utf-8") as f:
     
113.             for title, url in still_failed:
     
114.                 f.write(f"{title}\t{url}\n")
     
115.         print(f"还有 {len(still_failed)} 章未成功，可再次运行本脚本补漏。")
     
116.         return still_failed
     
117.     else:
     
118.         os.remove("failed.txt")
     
119.         print("补漏完成，failed.txt 已清除。")
     
120.         return []
     
121. 
     
122. # ---------- 主流程 ----------
     
123. def main():
     
124.     base_url = input("请输入小说首页地址：").strip()
     
125.     if not base_url.startswith("http"):
     
126.         print("请输入完整的网址，以 http 或 https 开头")
     
127.         return
     
128.     base_url = base_url.rstrip("/") + "/"
     
129. 
     
130.     # 1. 先进入补漏模式（如果有 failed.txt）
     
131.     failed = repair_mode()
     
132.     if failed:
     
133.         print("请先处理完失败章节再执行完整下载。")
     
134.         return
     
135. 
     
136.     # 2. 正常流程：获取全部目录
     
137.     last_page = get_last_page(base_url)
     
138.     print(f"检测到目录共 {last_page} 页")
     
139.     chapters = get_chapter_links(base_url, last_page)
     
140.     print(f"共获取到 {len(chapters)} 个章节")
     
141.     out_dir = get_output_dir(base_url)
     
142. 
     
143.     failed = []
     
144.     for idx, (title, url) in enumerate(chapters, 1):
     
145.         try:
     
146.             print(f"[{idx}/{len(chapters)}] 正在下载：{title}")
     
147.             chapter_title, content = get_chapter_content(url)
     
148.             filename = f"{idx:03}_{sanitize_filename(chapter_title)}.txt"
     
149.             with open(os.path.join(out_dir, filename), "w", encoding="utf-8") as f:
     
150.                 f.write(f"{chapter_title}\n\n{content}")
     
151.             time.sleep(15)
     
152.         except Exception as e:
     
153.             print(f"下载失败：{title}，{e}")
     
154.             failed.append((title, url))
     
155. 
     
156.     # 3. 把失败记录落盘
     
157.     if failed:
     
158.         with open("failed.txt", "w", encoding="utf-8") as f:
     
159.             for title, url in failed:
     
160.                 f.write(f"{title}\t{url}\n")
     
161.         print(f"共有 {len(failed)} 章下载失败，已写入 failed.txt，下次运行本脚本将自动补漏。")
     
162.     else:
     
163.         print("全部下载完成！")
     
164. 
     
165. if __name__ == "__main__":
     
166.     main()

复制代码