"""遍历 herb_list.csv，爬取每味中药的详情，保存到 中药/中药名.txt。
支持断点续爬：已经存在对应 txt 的中药直接跳过，中断后重跑即可继续。
页面结构（2026-07 实测）：
div.p_content 下有 3~4 个 div.card-panel，每个 panel 一个 h2 大节：
名称 / xx的种植和炮制 / xx的效果 / xx的药方（部分药材没有药方）
"种植和炮制" 和 "效果" 两节内部再用 h3 分小节（来源/性味/炮制/性状/
功效/经脉/主治/用法用量/注意禁忌），h3 集合每味药都不一样，所以按
文档顺序动态解析，不写死字段名。
用法（在项目根目录执行）：
python -m _001__clawler.02_herb_detail # 跑全部，5 线程
python -m _001__clawler.02_herb_detail --limit 5 # 只跑 5 条看效果
python -m _001__clawler.02_herb_detail --workers 10 # 提到 10 线程
python -m _001__clawler.02_herb_detail --workers 1 # 退回单线程
"""
import argparse
import csv
import os
import random
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
from bs4 import BeautifulSoup
from common.path_utils import get_file_path
INPUT_CSV = get_file_path("_001__clawler/herb_list.csv")
OUTPUT_DIR = get_file_path("_001__clawler/中药")
HEADERS = {
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
)
}
# requests.Session 不是线程安全的，每个线程各用一个（连接复用也还在）
_local = threading.local()
# print 在多线程下会串行，加锁保证一条日志不被撕开
_print_lock = threading.Lock()
def log(message):
with _print_lock:
print(message, flush=True)
def get_session():
session = getattr(_local, "session", None)
if session is None:
session = requests.Session()
session.headers.update(HEADERS)
_local.session = session
return session
def fetch_html(url, retries=3, timeout=30):
"""下载页面 html，失败时退避重试。"""
for attempt in range(1, retries + 1):
try:
response = get_session().get(url, timeout=timeout)
response.raise_for_status()
response.encoding = "utf-8"
return response.text
except requests.RequestException as exc:
log(f" 第 {attempt}/{retries} 次请求失败: {exc}")
if attempt == retries:
return None
time.sleep(2 * attempt)
def parse_detail(html):
"""解析详情页，返回 [(标题, [段落, ...]), ...]，标题层级用 h2/h3 区分。
跳过两类噪音：
1. 图片 figure，其文字内容是无意义的 "1 #"
2. ol.p_alternative_list，别名已经在上一个 <p> 里以顿号形式出现过
"""
soup = BeautifulSoup(html, "html.parser")
content = soup.find("div", class_="p_content")
if content is None:
return None
sections = []
for panel in content.find_all("div", class_="card-panel"):
# 药方那一节的正文在 ul > li 里，不是 p，所以 li 也要收
for element in panel.find_all(["h2", "h3", "p", "li"]):
if element.name == "p":
# 图片块和别名 ol 都挂在 <p> 里面，先剔掉
if element.find("div", class_="p_image_center") or element.find("ol"):
continue
elif element.name == "li":
# 别名列表 ol.p_alternative_list 的 li 是重复内容，跳过
if element.find_parent("ol", class_="p_alternative_list"):
continue
text = element.get_text(" ", strip=True)
if not text:
continue
if element.name in ("h2", "h3"):
# h2 后面多了空格的情况（如 "毛车藤 的效果"）顺手清掉
sections.append((element.name, " ".join(text.split()), []))
elif sections:
sections[-1][2].append(" ".join(text.split()))
return sections
def format_detail(name, url, category, sections):
"""把解析结果拼成可读的 txt 文本。"""
lines = [f"中药名：{name}", f"分类：{category}", f"链接：{url}", ""]
for level, title, paragraphs in sections:
lines.append(("## " if level == "h2" else "### ") + title)
lines.extend(paragraphs)
lines.append("")
return "\n".join(lines).rstrip() + "\n"
def crawl_one(herb, min_delay, max_delay):
"""抓一味药并落盘。返回 True 成功 / False 失败。由线程池并发调用。"""
name, url = herb["中药名"], herb["链接"]
output_path = os.path.join(OUTPUT_DIR, f"{name}.txt")
html = fetch_html(url)
if html is None:
return False
sections = parse_detail(html)
if not sections:
log(f" 解析失败，跳过 {name}（页面结构可能不同）")
return False
text = format_detail(name, url, herb["分类"], sections)
# 先写临时文件再改名，避免中断时留下半个文件被误判成已完成。
# 带线程名后缀，防止两个线程刚好撞到同名临时文件。
temp_path = f"{output_path}.{threading.get_ident()}.part"
with open(temp_path, "w", encoding="utf-8") as f:
f.write(text)
os.replace(temp_path, output_path)
# 每个线程自己睡，等效于把整体 QPS 压到 workers / 平均延时
time.sleep(random.uniform(min_delay, max_delay))
return True
def crawl(limit=0, workers=5, min_delay=1.0, max_delay=2.0):
os.makedirs(OUTPUT_DIR, exist_ok=True)
with open(INPUT_CSV, encoding="utf-8-sig") as f:
herbs = list(csv.DictReader(f))
# 断点续爬：先把已抓过的过滤掉，剩下的才交给线程池
todo = [
herb
for herb in herbs
if not os.path.exists(os.path.join(OUTPUT_DIR, f"{herb['中药名']}.txt"))
]
skipped = len(herbs) - len(todo)
if limit:
todo = todo[:limit]
print(f"共 {len(herbs)} 条，已存在 {skipped} 条，本次待抓 {len(todo)} 条")
print(f"并发线程数: {workers}")
done = failed = 0
if todo:
with ThreadPoolExecutor(max_workers=workers) as pool:
futures = {
pool.submit(crawl_one, herb, min_delay, max_delay): herb
for herb in todo
}
for finished, future in enumerate(as_completed(futures), start=1):
herb = futures[future]
try:
ok = future.result()
except Exception as exc: # 单条异常不该拖垮整轮
log(f" {herb['中药名']} 异常: {exc}")
ok = False
if ok:
done += 1
else:
failed += 1
log(f"[{finished}/{len(todo)}] {herb['中药名']} {'✓' if ok else '✗'}")
print(f"\n完成 {done} 条，跳过（已存在）{skipped} 条，失败 {failed} 条")
print(f"输出目录: {OUTPUT_DIR}")
def main():
parser = argparse.ArgumentParser(description="爬取中药详情")
parser.add_argument(
"--limit",
type=int,
default=0,
help="本次最多新抓多少条，0 表示不限制（默认 0）",
)
parser.add_argument(
"--workers",
type=int,
default=5,
help="并发线程数（默认 5，设 1 就是单线程）",
)
parser.add_argument(
"--min-delay",
type=float,
default=1.0,
help="每个线程每条之间的最小延时秒数（默认 1.0）",
)
parser.add_argument(
"--max-delay",
type=float,
default=2.0,
help="每个线程每条之间的最大延时秒数（默认 2.0）",
)
args = parser.parse_args()
if args.workers < 1:
parser.error("--workers 至少为 1")
crawl(
limit=args.limit,
workers=args.workers,
min_delay=args.min_delay,
max_delay=args.max_delay,
)
if __name__ == "__main__":
main()