...

段落

... ⚠️ div.content 后紧跟上/下一篇导航 (dreamer-prevnext) → 正文截到该标记前正文内附件相对路径 urljoin(详情页URL) 绝对化用法: python3 crawl_huojia_hjgl.py --pages=1 python3 crawl_huojia_hjgl.py --pages=13 # CUTOFF 3年窗口 """ import os, sys, re, time, json, html as html_mod import requests from urllib.parse import urljoin sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from crawler_lib import push_to_searchdb BASE_URL = "https://www.huojia.gov.cn" LIST_TPL = BASE_URL + "/htmls/huanjingguanli/list-%d.html" SITE_NAME = "获嘉县人民政府-环境管理" CATEGORY = "环境管理" CUTOFF = "2023-08-11" # 3 年窗口 HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36", "Accept-Language": "zh-CN,zh;q=0.9", "Referer": BASE_URL + "/htmls/huanjingguanli/list-1.html", } TIMEOUT = 30 def parse_args(): pages = 0 for a in sys.argv[1:]: if a.startswith("--pages="): pages = int(a.split("=")[1]) elif a.startswith("--pages") and len(sys.argv) > sys.argv.index(a) + 1: pages = int(sys.argv[sys.argv.index(a) + 1]) return pages def clean_title(t): t = html_mod.unescape(t or "") t = re.sub(r"[\u200b\u200e\u200f\ufeff\xa0]", "", t) t = re.sub(r"\s+", " ", t) t = re.sub(r"^[•·\-—]\s*", "", t) return t.strip() def http_get(url, timeout=TIMEOUT): for attempt in range(3): try: r = requests.get(url, headers=HEADERS, timeout=timeout) if r.status_code == 200: r.encoding = "utf-8" return r.text except Exception as e: if attempt == 2: print(f" [WARN] GET failed {url}: {e}", file=sys.stderr) return None time.sleep(1.5) return None def _abs_url(u, page_url): if u.startswith(("http://", "https://", "javascript:", "mailto:", "tel:", "data:", "#")): return u return urljoin(page_url, u) def clean_content(raw_html, page_url): """正文清洗: 保留

段落 HTML, 剥 span/font 内联样式, 附件/图片绝对化""" if not raw_html: return "" raw_html = re.sub(r"", "", raw_html) raw_html = re.sub(r"", "", raw_html) # 附件/图片相对路径绝对化 (以详情页 URL 为基址) raw_html = re.sub(r' 标签 raw_html = re.sub(r"]*>", "

", raw_html) # 剥 span 标签 (保留文本) raw_html = re.sub(r"]*>", "", raw_html) # 其余标签去属性 raw_html = re.sub(r"<(table|tbody|thead|tr|td|th|br)([^>]*)>", lambda m: f"<{m.group(1)}>", raw_html) raw_html = re.sub(r"", lambda m: f"", raw_html) # 剥残留标签 (⚠️ \b 边界防误伤 ) raw_html = re.sub(r"]*>", "", raw_html, flags=re.I) # 空白清理 raw_html = re.sub(r"[\u3000\xa0]", " ", raw_html) raw_html = re.sub(r"

\s*(?: )?\s*

", "", raw_html) raw_html = re.sub(r"\r\n", "\n", raw_html) return raw_html.strip() def extract_attachments(content, page_url): """从正文提取附件链接, 返回 JSON 数组字符串""" atts = [] seen = set() for am in re.finditer(r']*>(.*?)', content, re.S): ahref, atext = am.group(1), re.sub(r"<[^>]+>", "", am.group(2)).strip() if re.search(r"\.(pdf|doc|docx|xls|xlsx|zip|rar|wps|7z|txt)(\?|$)", ahref, re.I) or "download" in ahref.lower(): u = _abs_url(ahref, page_url) if u not in seen: seen.add(u) atts.append({"name": atext or u.split("/")[-1], "url": u}) return json.dumps(atts, ensure_ascii=False) if atts else "[]" def fetch_detail(url): """抓详情页: 返回 (正文HTML, 附件列表)""" html_text = http_get(url) if not html_text: return "", [] atts = [] # 正文: div.content 截到上/下一篇导航前 body = "" m = re.search(r'

([\s\S]*?)(?:dreamer-prevnext|]*class="dis"|]*class="share|分享：)', html_text) if m: body = clean_content(m.group(1), url) else: m2 = re.search(r'

([\s\S]*?)

\s*

', html_text) if m2: body = clean_content(m2.group(1), url) # 正文内附件 for a in re.finditer(r']*>(.*?)', body, re.S): ahref = a.group(1) if re.search(r"\.(pdf|doc|docx|xls|xlsx|zip|rar|wps|7z)(\?|$)", ahref, re.I): atext = re.sub(r"<[^>]+>", "", a.group(2)).strip() or ahref.split("/")[-1] atts.append({"name": atext, "url": ahref}) return body, atts def main(): pages = parse_args() if not pages: pages = 5 all_items = [] seen = set() for pg in range(1, pages + 1): html_text = http_get(LIST_TPL % pg) if not html_text: print(f"[PAGE {pg}] fetch failed, stopping") break items = re.findall( r'