#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ crawl_gzqz_tzgg.py - 清镇市人民政府-通知公告 列表: https://www.gzqz.gov.cn/xwdt/tzgg/ (TRS createPageHTML, 共113页 index_N.html) 详情: https://www.gzqz.gov.cn/xwdt/tzgg/YYYYMM/tYYYYMMDD_ID.html 正文: div.Article_zw > font#Zoom > .trs_editor_view 附件: 正文内相对路径 ./P0xxx -> 转绝对 URL, 独立成段

附件名

图片: 转绝对 URL,

查看图片

用法: python3 crawl_gzqz_tzgg.py --pages 1 # 测试 1 页 python3 crawl_gzqz_tzgg.py --pages 5 # 前 5 页 python3 crawl_gzqz_tzgg.py --pages 113 # 全量 python3 crawl_gzqz_tzgg.py --incremental # 增量 (断点续传) """ import argparse import json import os import re import sqlite3 import ssl import sys import time import urllib.request from urllib.parse import urljoin BASE = "https://www.gzqz.gov.cn" LIST_URL = "https://www.gzqz.gov.cn/xwdt/tzgg/" SITE_NAME = "清镇市人民政府-通知公告" GROUP_NAME = "贵州" TOTAL_PAGES = 113 DB_PATH = "/root/search.db" DONE_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "gzqz_tzgg_done.json") HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9", } ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE def safe_get(url, timeout=25, retries=3): for i in range(retries): try: req = urllib.request.Request(url, headers=HEADERS) with urllib.request.urlopen(req, timeout=timeout, context=ctx) as r: return r.read().decode("utf-8", errors="replace") except Exception as e: if i == retries - 1: raise time.sleep(2 + i * 2) return None def load_done(): if os.path.exists(DONE_FILE): try: return set(json.load(open(DONE_FILE, encoding="utf-8"))) except Exception: return set() return set() def save_done(done): json.dump(sorted(done), open(DONE_FILE, "w", encoding="utf-8"), ensure_ascii=False) def parse_list(html): """提取列表项: (url, title, date)""" items = [] # li 结构:
  • 标题日期
    • for m in re.finditer(r']*>\s*]+href="([^"]+)"[^>]*title="([^"]*)"[^>]*>(.*?)\s*([^<]*)', html, re.DOTALL): href, title, inner, date = m.group(1), m.group(2), m.group(3), m.group(4).strip() if not title: title = re.sub(r'<[^>]+>', '', inner).strip() url = urljoin(LIST_URL, href) date = date.strip() # 只保留真正的通知公告详情页: /xwdt/tzgg/YYYYMM/tYYYYMMDD_数字.html if not re.search(r'/xwdt/tzgg/\d{6}/t\d{8}_\d+\.html', url): continue items.append({"url": url, "title": title.strip(), "date": date}) return items def parse_detail(html, url): """提取详情: title/publish_date/content""" title = "" m = re.search(r'([^<]*)', html) if m: title = re.sub(r'[-\s_].*$', '', m.group(1)).strip() pub_date = "" m = re.search(r' font#Zoom content = "" m = re.search(r']*class="Article_zw"[^>]*>(.*?)\s*', html, re.DOTALL) if not m: m = re.search(r']*id="Zoom"[^>]*>(.*?)', html, re.DOTALL) if m: content = m.group(1).strip() if not content: # 兜底: trs_editor_view m = re.search(r']*class="trs_editor_view[^"]*"[^>]*>(.*?)', html, re.DOTALL) if m: content = m.group(1).strip() return title, pub_date, content def clean_content(content, detail_url): """清洗正文: 附件/图片转绝对URL独立成段, 去脚本/样式""" if not content: return "" # 附件: ./P0xxx ->

    附件名

    def fix_attach(m): href = m.group(1) name = m.group(2).strip() abs_url = urljoin(detail_url, href) return f'

    {name}

    ' content = re.sub(r']+href="([^"]+)"[^>]*>\s*(附件[^<]{0,60}?)\s*', fix_attach, content, flags=re.I) # 图片: src ->

    查看图片

    def fix_img(m): src = m.group(1) # 过滤编辑器资源/图标 if re.search(r'ueditor|editor|icon|emoji|placeholder|data:', src, re.I): return '' abs_url = urljoin(detail_url, src) return f'

    查看图片

    ' content = re.sub(r']+src="([^"]+)"[^>]*>', fix_img, content) # 其他图片引用 (src=) def fix_src(m): src = m.group(1) # 过滤编辑器资源/图标 (ueditor_demo 等) if re.search(r'ueditor|editor|icon|emoji|placeholder', src, re.I): return '' if re.match(r'^(https?://|data:)', src): return f'

    查看图片

    ' return f'

    查看图片

    ' content = re.sub(r'(?:src|href)="([^"]+\.(?:jpg|jpeg|png|gif|bmp|webp))"', fix_src, content, flags=re.I) # 删除脚本/样式/iframe content = re.sub(r']*>.*?', '', content, flags=re.DOTALL | re.I) content = re.sub(r']*>.*?', '', content, flags=re.DOTALL | re.I) content = re.sub(r']*>.*?', '', content, flags=re.DOTALL | re.I) # 清理空白 content = re.sub(r'\n\s*\n+', '\n', content) content = content.strip() return content def push_to_db(items, incremental=False): """写 search.db (INSERT OR IGNORE by page_url) + FTS""" conn = sqlite3.connect(DB_PATH, timeout=60) conn.execute("PRAGMA busy_timeout=60000") added = 0 skipped = 0 cur = conn.cursor() for it in items: exists = cur.execute("SELECT 1 FROM gov_raw WHERE page_url=?", (it["url"],)).fetchone() if exists: skipped += 1 continue cur.execute( """INSERT OR IGNORE INTO gov_raw (title, content, publish_date, source_url, page_url, site_name, summary, group_name, category, script_name) VALUES (?,?,?,?,?,?,?,?,?,?)""", (it["title"], it["content"], it["pub_date"], it["url"], it["url"], SITE_NAME, it.get("summary", ""), GROUP_NAME, "通知公告", "crawl_gzqz_tzgg.py"), ) added += 1 conn.commit() # FTS 增量 rows = [] if items: urls = [it["url"] for it in items] placeholders = ",".join("?" * len(urls)) rows = cur.execute( f"SELECT id, title, content, site_name FROM gov_raw WHERE page_url IN ({placeholders})", urls, ).fetchall() for rid, title, content, site_name in rows: summary = re.sub(r'<[^>]+>', '', content or '')[:200] cur.execute( "INSERT OR IGNORE INTO gov_search(rowid, title, site_name, summary) VALUES(?,?,?,?)", (rid, title, site_name, summary), ) conn.commit() conn.close() return added, skipped def main(): ap = argparse.ArgumentParser() ap.add_argument("--pages", type=int, default=5, help="抓取页数") ap.add_argument("--incremental", action="store_true", help="增量模式(断点续传)") ap.add_argument("--out", default=None, help="输出 JSONL 路径(默认直写DB)") args = ap.parse_args() done = load_done() if args.incremental else set() all_items = [] pages = min(args.pages, TOTAL_PAGES) for pg in range(pages): page_url = LIST_URL if pg == 0 else f"{LIST_URL}index_{pg}.html" try: html = safe_get(page_url) except Exception as e: print(f"[list] p{pg+1} ERR: {str(e)[:80]}") break if not html: print(f"[list] p{pg+1} 空响应") break items = parse_list(html) new = [it for it in items if it["url"] not in done] print(f"[list] p{pg+1}: {len(items)} 条(新{len(new)})") all_items.extend(new) if args.incremental: done.update(it["url"] for it in items) time.sleep(0.5) print(f"[*] 列表共 {len(all_items)} 条, 开始抓详情") records = [] for i, it in enumerate(all_items, 1): try: html = safe_get(it["url"]) except Exception as e: print(f"[detail] {i}/{len(all_items)} ERR {it['url']}: {str(e)[:60]}") continue if not html: continue title, pub_date, content = parse_detail(html, it["url"]) content = clean_content(content, it["url"]) if not title: title = it["title"] if not pub_date: pub_date = it["date"] records.append({ "title": title, "url": it["url"], "pub_date": pub_date, "content": content, "summary": "", }) if i % 20 == 0: print(f"[detail] +{len(records)} 条") time.sleep(0.3) if args.out: with open(args.out, "w", encoding="utf-8") as f: for r in records: f.write(json.dumps(r, ensure_ascii=False) + "\n") print(f"[✓] 完成: {len(records)} 条 -> {args.out}") else: added, skipped = push_to_db(records, incremental=args.incremental) print(f"[✓] 完成: 本次新增 {added} 条, 跳过 {skipped} 条") if args.incremental: save_done(done) if __name__ == "__main__": main()