#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 七台河市人民政府 - 公示公告 爬虫 栏目: https://www.qth.gov.cn/qth/c100183/list.shtml CMS: 黑龙江政府站群 (UCMS) — /common/search/ JSON API 列表 API: /common/search/{channelId}?_isAgg=false&_isJson=true&_pageSize=N&_template=index&_rangeTimeGte=&_channelName=&page=N channelId: e9ff8bea997945789925c7aa8cb39f2a (c100183 公示公告) total: 518 条, 10条/页 (52页) API 直供 contentHtml 完整正文 (含

) → 无需抓详情页 详情: /qth/c100183/YYYYMM/UUID.shtml 标题: title 字段 (完整) 用法: python3 crawl_qth_gsgg.py --pages=5 python3 crawl_qth_gsgg.py --pages=1 """ 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.qth.gov.cn" CHANNEL_ID = "e9ff8bea997945789925c7aa8cb39f2a" API_TPL = (BASE_URL + "/common/search/" + CHANNEL_ID + "?_isAgg=false&_isJson=true&_pageSize=10&_template=index&_rangeTimeGte=&_channelName=&page=%d") SITE_NAME = "七台河市人民政府-公示公告" CATEGORY = "公示公告" MAX_PAGES = 52 # total=518, 10条/页 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": "application/json, text/javascript, */*; q=0.01", "Accept-Language": "zh-CN,zh;q=0.9", "Referer": BASE_URL + "/qth/c100183/list.shtml", "X-Requested-With": "XMLHttpRequest", } 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 fetch_api(page): url = API_TPL % page for attempt in range(3): try: r = requests.get(url, headers=HEADERS, timeout=TIMEOUT) if r.status_code == 200: return r.json() except Exception as e: if attempt == 2: print(f" [WARN] API p{page} failed: {e}", file=sys.stderr) return None time.sleep(1.5) return None def clean_content(raw_html, page_url): """正文清洗: 保留

/ HTML, 附件/图片绝对化 (以详情页 URL 为基址保留栏目路径)""" raw_html = re.sub(r"", "", raw_html) raw_html = re.sub(r"", "", raw_html) # 附件/图片相对路径绝对化 (⚠️ 用 urljoin(page_url, href) 而非 BASE_URL, 否则缺栏目路径段 404) def _abs_url(u): if u.startswith(("http://", "https://", "javascript:", "mailto:", "tel:", "data:", "#")): return u return urljoin(page_url, u) 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|div|br)([^>]*)>", lambda m: f"<{m.group(1)}>", raw_html) raw_html = re.sub(r"", lambda m: f"", raw_html) # 剥残留标签 (⚠️ 不能用 , o 会误伤 ) 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): """从正文提取附件链接 (pdf/doc/xls/zip 等), 返回 JSON 数组字符串""" atts = [] 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(): atts.append({"name": atext or ahref.split("/")[-1], "url": _abs_url_outer(ahref, page_url)}) return json.dumps(atts, ensure_ascii=False) if atts else "[]" def _abs_url_outer(u, page_url): if u.startswith(("http://", "https://", "javascript:", "mailto:", "tel:", "data:", "#")): return u return urljoin(page_url, u) def fetch_detail(url): """抓详情页补全正文 (API contentHtml 常截断, 详情页更完整含图片)""" for attempt in range(3): try: r = requests.get(url, headers=HEADERS, timeout=TIMEOUT) if r.status_code != 200: return "" r.encoding = "utf-8" html_text = r.text m = re.search(r'
(.*?)
', html_text, re.S) if m: return clean_content(m.group(1), url) # fallback: 任意 TRS_UEDITOR 容器 m = re.search(r'class="[^"]*TRS_UEDITOR[^"]*"[^>]*>(.*?)', html_text, re.S) if m: return clean_content(m.group(1), url) return "" except Exception as e: if attempt == 2: print(f" [WARN] detail failed {url}: {e}", file=sys.stderr) return "" time.sleep(1.0) return "" def main(): pages = parse_args() if not pages: pages = 5 all_items = [] seen = set() for pg in range(1, pages + 1): data = fetch_api(pg) if not data: print(f"[PAGE {pg}] API failed, stopping") break results = data.get("data", {}).get("results", []) if not results: print(f"[PAGE {pg}] 0 items, stopping") break for r in results: url = urljoin(BASE_URL, r.get("url", "")) if not url or url in seen: continue seen.add(url) title = clean_title(r.get("title", "")) if not title: continue pub_date = "" pm = re.search(r"(\d{4}-\d{2}-\d{2})", r.get("publishedTimeStr", "")) if pm: pub_date = pm.group(1) content = clean_content(r.get("contentHtml", ""), url) # 无正文或过短时回退 content 字段/抓详情页 if len(re.sub(r"<[^>]+>", "", content)) < 100: content = clean_content(r.get("content", ""), url) if len(re.sub(r"<[^>]+>", "", content)) < 100: detail_content = fetch_detail(url) if detail_content: content = detail_content attachments = extract_attachments(content, url) all_items.append({ "site_name": SITE_NAME, "title": title, "pub_date": pub_date, "content": content, "source_url": url, "url": url, "attachments": attachments, "industry": "", }) print(f"[PAGE {pg}] +{len(results)} items (total {len(all_items)})") time.sleep(0.8) print(f"\n有效 {len(all_items)} 条, 入库...") push_to_searchdb(all_items, CATEGORY) print(f"完成! 新增/更新 {len(all_items)} 条") if __name__ == "__main__": main()