#!/usr/bin/env python3 # -*- coding: utf-8 -*- """渌口区人民政府 - 通知公告 (c18458) 爬虫 CMS: zzcms/DFS (株洲政府网站群) 列表: /c18458/index.html + /c18458/pages/{N}.html 15条/页, 共149页≈2235条 详情: /c18458/{YYYYMMDD}/i{id}.html 详情页: meta ArticleTitle/PubDate/ContentSource + div.details-content.article-content-body 正文 附件: 正文外, 协议相对 //img.zhuzhou.gov.cn/zzcms/DFS/file/... (urljoin 处理) """ import sys, re, time, sqlite3, ssl, urllib.request from urllib.parse import urljoin from bs4 import BeautifulSoup SITE_NAME = "渌口区-通知公告" SCRIPT_NAME = "crawl_lukou_tzgg.py" GROUP_NAME = "湖南" BASE_URL = "http://www.lukou.gov.cn" LIST_FIRST = BASE_URL + "/c18458/index.html" TOTAL_PAGES = 149 DB_PATH = "/mnt/data/search.db" OUTPUT_FILE = "/root/gov_crawler/lukou_tzgg_output.jsonl" ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE 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,en;q=0.8", } def http_get(url, timeout=25, retries=3): for i in range(retries): try: req = urllib.request.Request(url, headers=HEADERS) resp = urllib.request.urlopen(req, timeout=timeout, context=ctx) return resp.read().decode("utf-8", errors="replace") except Exception as e: if i == retries - 1: print(f" [WARN] 请求失败 {url}: {e}", file=sys.stderr) return "" time.sleep(1.5) return "" def clean_title(title): """strip ·  实体前缀 和省略号截断后缀""" title = title.replace("·", "").replace(" ", "").replace("\u00b7", "") title = re.sub(r"^[\s\xa0·\u00b7]+", "", title) title = re.sub(r"\s*\.{3,}\s*$", "", title) return title.strip() def parse_list(html): """从列表页提取 (url, title, date)""" items = [] # 列表条目: href="/c18458/20260730/i2517805.html" 文本 for m in re.finditer(r'href="([^"]*?/c18458/\d{8}/i\d+\.html)"[^>]*>\s*([^<]{2,80})<', html): href, t = m.group(1), m.group(2) url = urljoin(BASE_URL, href) title = clean_title(t.strip()) # 日期从 URL 提取 (20260730 -> 2026-07-30) dm = re.search(r"/(\d{4})(\d{2})(\d{2})/i\d+\.html$", href) date = f"{dm.group(1)}-{dm.group(2)}-{dm.group(3)}" if dm else "" if title: items.append((url, title, date)) return items def extract_detail(html, page_url): """提取 (title, date, content_html, attachments)""" soup = BeautifulSoup(html, "html.parser") # 标题 (meta ArticleTitle 优先) title = "" mt = re.search(r']+>", "", p)) if key and key not in seen: seen.add(key) final_parts.append(p) # \n\n 分段; 未入流的附件追加 content = "\n\n".join(final_parts) for a, txt, abs_url in attachments: if abs_url not in content: content += f'\n\n

{txt}

' # 清理 content = re.sub(r"打印本页[^\n]*", "", content) content = re.sub(r"\n{3,}", "\n\n", content).strip() return title, date, content, attachments def main(): max_pages = 1 args = sys.argv[1:] i = 0 while i < len(args): a = args[i] if a.startswith("--pages="): max_pages = int(a.split("=", 1)[1]) elif a == "--pages" and i + 1 < len(args): max_pages = int(args[i + 1]) i += 1 i += 1 conn = sqlite3.connect(DB_PATH, timeout=30) c = conn.cursor() total_new = 0 total_dup = 0 total_skip = 0 for page in range(1, max_pages + 1): if page == 1: list_url = LIST_FIRST else: list_url = f"{BASE_URL}/c18458/pages/{page}.html" html = http_get(list_url) if not html: print(f" [WARN] 第{page}页获取失败, 跳过", file=sys.stderr) continue items = parse_list(html) print(f" 第{page}页: 找到 {len(items)} 条") for url, title, date in items: c.execute("SELECT id FROM gov_raw WHERE page_url=?", (url,)) if c.fetchone(): total_dup += 1 continue dhtml = http_get(url) if not dhtml: total_skip += 1 continue d_title, d_date, content, attachments = extract_detail(dhtml, url) if not content or len(re.sub(r"<[^>]+>", "", content).strip()) < 10: total_skip += 1 continue if not d_title: d_title = title if not d_date: d_date = date summary = re.sub(r"<[^>]+>", "", content)[:200] summary = re.sub(r"\s+", " ", summary).strip() has_table = 1 if "