#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 北海市 铁山港（临海）工业区管理委员会 - 通知公告 爬虫 列表: http://www.beihai.gov.cn/xxgkbm/bhstsglhgyqglwyh/tzgg_24/ createPageHTML(12,0,"index","shtml","229") → 12 页 / 229 条，20 条/页 第 2 页起 index_N.shtml（N=1 即第 2 页） 条目:
  • 标题2026-09-15
    • ⚠️ 日期 在 **之后**（TRS 常见形），标题取 title 属性最稳 详情: /xxgkbm/bhstsglhgyqglwyh/tzgg_24/t{id}.shtml meta ArticleTitle / PubDate / ContentSource 正文 div.article-con > div.trs_editor_view.TRS_UEDITOR 附件 div.article-file (本栏目是石化/临海工业区公示：规划环评、询价公告、节能审查等) 用法: python3 crawl_beihai_tsg_tzgg.py --pages=1 python3 crawl_beihai_tsg_tzgg.py --pages=5 """ import os import re import sys import time import json import html as html_mod from urllib.parse import urljoin import requests from bs4 import BeautifulSoup sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from crawler_lib import push_to_searchdb BASE_URL = "http://www.beihai.gov.cn" LIST_DIR = "/xxgkbm/bhstsglhgyqglwyh/tzgg_24" LIST_URL = BASE_URL + LIST_DIR + "/" SITE_NAME = "北海市-铁山港临海工业区管委会通知公告" CATEGORY = "通知公告" GROUP = "广西" CUTOFF = "2023-09-18" UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36") TIMEOUT = 45 CONTENT_SELECTORS = [ "div.article-con > div.trs_editor_view.TRS_UEDITOR", "div.article-con div.trs_editor_view", "div.article-con", "div.trs_editor_view", "div#zoom", ] SESSION = requests.Session() SESSION.headers.update({"User-Agent": UA, "Accept-Language": "zh-CN,zh;q=0.9"}) def parse_args(): pages = 0 for a in sys.argv[1:]: if a.startswith("--pages="): pages = int(a.split("=")[1]) elif a == "--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, referer=None): for attempt in range(3): try: r = SESSION.get(url, headers={"Referer": referer or LIST_URL}, timeout=TIMEOUT, verify=False) if r.status_code == 200: if not r.encoding or r.encoding.lower() in ("iso-8859-1", "latin-1", "ascii", "cp1252"): r.encoding = r.apparent_encoding or "utf-8" return r.text print(" [WARN] HTTP %s %s" % (r.status_code, url[:90]), file=sys.stderr) return None except Exception as e: if attempt == 2: print(" [WARN] GET failed %s: %s" % (url[:80], str(e)[:80]), file=sys.stderr) return None time.sleep(1.0 * (attempt + 1)) return None def parse_list(html_text, list_page_url): soup = BeautifulSoup(html_text, "html.parser") out = [] seen = set() for li in soup.find_all("li"): a = li.find("a", href=True) if not a: continue href = a["href"].strip() if not re.search(r"t\d{6,}\.shtml$", href): continue u = urljoin(list_page_url, href) if u in seen: continue seen.add(u) title = clean_title(a.get("title") or a.get_text(" ", strip=True)) date = "" sp = li.find("span") if sp: m = re.search(r"(\d{4}-\d{2}-\d{2})", sp.get_text(" ", strip=True)) if m: date = m.group(1) if not date: m = re.search(r"(\d{4}-\d{2}-\d{2})", li.get_text(" ", strip=True)) if m: date = m.group(1) if title: out.append((u, title, date)) return out def total_pages(html_text): m = re.search(r"createPageHTML\(\s*(\d+)", html_text) return int(m.group(1)) if m else 0 def absolutize(soup, page_url): for tag, attr in (("a", "href"), ("img", "src")): for t in soup.find_all(tag): v = (t.get(attr) or "").strip() if not v or v.startswith(("javascript:", "mailto:", "tel:", "#", "data:")): continue if v.startswith("//"): t[attr] = "https:" + v elif not v.startswith("http"): t[attr] = urljoin(page_url, v) def clean_content(node, page_url): if node is None: return "" for bad in node.find_all(["script", "style", "iframe", "noscript"]): bad.decompose() for b in node.select("div.article-share-group, div.article-share-more, div.favoriteContainer, div.article-inf"): b.decompose() absolutize(node, page_url) parts = [] for el in node.find_all(["p", "table", "h1", "h2", "h3", "h4"], recursive=True): if el.name == "table": txt = str(el) else: txt = clean_title(el.get_text(" ", strip=True)) if txt: parts.append(txt) if not parts: t = clean_title(node.get_text("\n", strip=True)) if t: parts = [t] body = "\n".join(p if p.strip().startswith("%s

    " % p for p in parts) body = re.sub(r"

    \s*

    ", "", body) return body def extract_atts(html_text, page_url): """优先解析 div.article-file 区（TRS 附件区在正文容器外）""" atts = [] seen = set() soup = BeautifulSoup(html_text, "html.parser") for cont in soup.select("div.article-file, div#downloadfile, div#article-file, div#relfile, div.fjxz"): for a in cont.find_all("a", href=True): u = a["href"].strip() if not u or u.startswith(("javascript:", "#", "mailto:")): continue if u.startswith("//"): u = "https:" + u elif not u.startswith("http"): u = urljoin(page_url, u) if u in seen: continue seen.add(u) atts.append({"name": clean_title(a.get_text(" ", strip=True)) or os.path.basename(u.split("?")[0])[:120], "url": u}) # 兜底：全文扫常见附件后缀 for m in re.finditer(r'href="([^"]+\.(?:pdf|docx?|xlsx?|zip|rar|wps|7z|txt))"', html_text, re.I): u = m.group(1).strip() if u.startswith("//"): u = "https:" + u elif not u.startswith("http"): u = urljoin(page_url, u) if u in seen: continue seen.add(u) atts.append({"name": os.path.basename(u.split("?")[0])[:120], "url": u}) return atts def fetch_detail(url, list_title, list_date): html_text = http_get(url) if not html_text: return list_title, list_date, "", [] soup = BeautifulSoup(html_text, "html.parser") title = list_title mt = soup.find("meta", {"name": "ArticleTitle"}) if mt and mt.get("content"): title = clean_title(mt["content"]) or title date = list_date mp = soup.find("meta", {"name": "PubDate"}) if mp and mp.get("content"): m = re.search(r"(\d{4})-(\d{1,2})-(\d{1,2})", mp["content"]) if m: date = "%04d-%02d-%02d" % (int(m.group(1)), int(m.group(2)), int(m.group(3))) node = None for sel in CONTENT_SELECTORS: cand = soup.select_one(sel) if cand and len(cand.get_text(" ", strip=True)) > 20: node = cand break body = clean_content(node, url) return title, date, body, extract_atts(html_text, url) def main(): pages = parse_args() if not pages: pages = 5 all_items = [] seen = set() for pg in range(1, pages + 1): if pg == 1: list_page_url = LIST_URL + "index.shtml" else: list_page_url = "%s%s/index_%d.shtml" % (BASE_URL, LIST_DIR, pg - 1) html_text = http_get(list_page_url, referer=LIST_URL) if not html_text: print("[PAGE %d] fetch failed, stopping" % pg) break if pg == 1: tp = total_pages(html_text) if tp: print("[INFO] 栏目共 %d 页" % tp) items = parse_list(html_text, list_page_url) if not items: print("[PAGE %d] 0 items (分页边界), stopping" % pg) break n_new = 0 for detail_url, title, date in items: if detail_url in seen: continue seen.add(detail_url) if date and date < CUTOFF: continue d_title, d_date, body, atts = fetch_detail(detail_url, title, date) content = body if atts: seg = "\n".join('

    %s

    ' % (a["url"], a["name"]) for a in atts) content = (content + "\n" + seg).strip() if content else seg if not content or len(re.sub(r"<[^>]*>?", "", content).strip()) < 10: content = "

    %s

    " % (d_title or title) all_items.append({ "site_name": SITE_NAME, "title": d_title or title, "pub_date": d_date or date, "content": content, "source_url": detail_url, "url": detail_url, "attachments": json.dumps(atts, ensure_ascii=False) if atts else "[]", "group_name": GROUP, "industry": "", }) n_new += 1 time.sleep(0.25) print("[PAGE %d] +%d items (total %d)" % (pg, n_new, len(all_items))) print("\n有效 %d 条, 入库..." % len(all_items)) push_to_searchdb(all_items, CATEGORY) print("完成! 新增/更新 %d 条" % len(all_items)) if __name__ == "__main__": main()