HTML, 附件/图片绝对化"""
# 附件/图片相对路径绝对化 (以详情页 URL 为基址)
def _abs_url(u):
if u.startswith(("http://", "https://", "javascript:", "mailto:", "tel:", "data:")):
return u
return urljoin(page_url, u)
soup = BeautifulSoup(raw_html, "html.parser")
# 剥 script/style
for tag in soup.find_all(["script", "style"]):
tag.decompose()
# 附件/图片绝对化
for a in soup.find_all("a", href=True):
a["href"] = _abs_url(a["href"])
for img in soup.find_all("img", src=True):
img["src"] = _abs_url(img["src"])
# 剥 二维码/分享/打印 容器
for sel in [".ewm", ".ewm-1", ".ewm-2", ".share", ".print", ".qr", ".weixin-share"]:
for el in soup.select(sel):
el.decompose()
# 规范化 标签
for p in soup.find_all("p"):
p.attrs = {}
# 保留结构: table/tr/td 去属性, 其他标签剥除
for tag in soup.find_all(True):
if tag.name in ("table", "tbody", "thead", "tr", "td", "th", "p", "div", "span", "br", "a", "img"):
if tag.name not in ("a", "img"):
tag.attrs = {}
else:
tag.unwrap()
# 空段落删除
for p in soup.find_all("p"):
if not p.get_text(strip=True) and not p.find("img") and not p.find("a"):
p.decompose()
# 序列化
body = soup.decode_contents()
body = re.sub(r"[\u3000\xa0]", " ", body)
body = re.sub(r"\r\n", "\n", body)
return body.strip()
def extract_detail(html_text, url):
"""详情: 标题/日期/正文/附件"""
soup = BeautifulSoup(html_text, "html.parser")
# 标题: meta ArticleTitle
title = ""
m = re.search(r'ArticleTitle"\s+content="([^"]*)"', html_text)
if m:
title = clean_title(m.group(1))
# 日期: meta PubDate
pub_date = ""
m = re.search(r'PubDate"\s+content="([^"]*)"', html_text)
if m:
pm = re.search(r"(\d{4}-\d{2}-\d{2})", m.group(1))
pub_date = pm.group(1) if pm else ""
# 正文: div#zoom.detail_content
body = ""
zoom = soup.find("div", id="zoom")
if zoom:
body = clean_content(str(zoom), url)
# 附件: 正文内
attachments = []
if body:
for am in re.finditer(r']*>([^<]*)', body):
ahref, atext = am.group(1), am.group(2).strip()
if re.search(r"\.(pdf|doc|docx|xls|xlsx|zip|rar|wps|7z)(\?|$)", ahref, re.I) or "download" in ahref.lower():
attachments.append({"name": atext or ahref.split("/")[-1], "url": ahref})
return {"title": title, "pub_date": pub_date, "content": body, "attachments": attachments}
def main():
pages = parse_args()
if not pages:
pages = 5
all_items = []
seen = set()
for pg in range(1, pages + 1):
list_url = LIST_TPL % pg
html_text = fetch(list_url)
if not html_text:
print(f"[PAGE {pg}] fetch failed, stopping")
break
items = extract_list_items(html_text)
if not items:
print(f"[PAGE {pg}] 0 items, stopping")
break
new_items = [i for i in items if i["url"] not in seen]
seen.update(i["url"] for i in items)
all_items.extend(new_items)
print(f"[PAGE {pg}] +{len(new_items)} items (total {len(all_items)})")
time.sleep(1.0)
print(f"\n列表共 {len(all_items)} 条, 开始抓详情...")
valid = []
for i, item in enumerate(all_items, 1):
detail_html = fetch(item["url"])
if not detail_html:
print(f" [{i}/{len(all_items)}] SKIP fetch: {item['title'][:40]}")
continue
d = extract_detail(detail_html, item["url"])
if not d["content"]:
print(f" [{i}/{len(all_items)}] SKIP 空正文: {item['title'][:40]}")
continue
title = d["title"] or item["title"]
pub_date = d["pub_date"] or item["date"]
content = d["content"]
att = d["attachments"]
# 附件嵌入正文尾部 (若正文未含)
if att:
has_att = any(re.search(r"\.(pdf|doc|docx|xls|xlsx|zip|rar|wps|7z)(\?|$)", a.get("href", ""), re.I)
for a in re.finditer(r'{a["name"]}
' for a in att]
content = content + "\n" + "\n".join(parts)
valid.append({
"site_name": SITE_NAME,
"title": title,
"pub_date": pub_date,
"content": content,
"source_url": item["url"],
"url": item["url"],
"attachments": json.dumps(att, ensure_ascii=False) if att else "[]",
"industry": "",
})
if i % 10 == 0:
print(f" [{i}/{len(all_items)}] 已处理...")
time.sleep(0.5)
print(f"\n有效 {len(valid)} 条, 入库...")
push_to_searchdb(valid, CATEGORY)
print(f"完成! 新增/更新 {len(valid)} 条")
if __name__ == "__main__":
main()