#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ crawl_jingxi_tzgg.py - 靖西市-通知公告 (TRS CMS) http://www.jingxi.gov.cn/xxgk/fdzdgkmr/jcxxgk/zwdt/tzgg/ 注意: 用户给的 /xxgk/jcxxgk/zwdt/tzgg/ 是空壳(0条), 正确路径多一级 fdzdgkmr 列表: ul.more-list > li > span(日期) + a[href=./tXXX.shtml][title=完整标题] 分页: index.shtml(第1页) + index_{1..24}.shtml (25页共742条, 每页29条) 详情: h1(标题) + .article-con/.trs_editor_view(正文HTML) + #article-file/#relative-file(附件) """ import sys import os import re import time import html as html_lib import urllib.parse import sqlite3 import json try: from bs4 import BeautifulSoup except ImportError: BeautifulSoup = None # ---------------- config ---------------- BASE_URL = "http://www.jingxi.gov.cn" LIST_DIR = "/xxgk/fdzdgkmr/jcxxgk/zwdt/tzgg/" SITE_NAME = "靖西市-通知公告" GROUP_NAME = "广西" SCRIPT_NAME = "crawl_jingxi_tzgg.py" DB_PATH = os.environ.get("DB_PATH", "/root/search.db") JSONL_PATH = os.environ.get("JSONL_PATH", "") ENC = "utf-8" TOTAL_PAGES = 25 # 742 条 / 29 条每页 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", } _MAX_PAGES = 1 for i, a in enumerate(sys.argv): if a.startswith("--pages="): try: _MAX_PAGES = int(a.split("=", 1)[1]) except ValueError: pass elif a == "--pages" and i + 1 < len(sys.argv): try: _MAX_PAGES = int(sys.argv[i + 1]) except ValueError: pass _PAGES = _MAX_PAGES if _MAX_PAGES >= 1 else 1 def clean_title(t): if not t: return "" t = html_lib.unescape(t) t = t.replace("\u200b", "").replace("\u200c", "").replace("\u200d", "").replace("\ufeff", "") t = re.sub(r"\s+", " ", t) return t.strip() def http_get(session, url, retries=3, timeout=30): for i in range(retries): try: r = session.get(url, headers=HEADERS, timeout=timeout) if r.status_code == 200: return r except Exception: pass time.sleep(2 + i * 2) return None def fetch_list(session, page): if page == 1: url = BASE_URL + LIST_DIR else: url = f"{BASE_URL}{LIST_DIR}index_{page - 1}.shtml" r = http_get(session, url) if r is None: return None return r.content.decode(ENC, errors="replace") def parse_list(html_text): """ul.more-list > li > span(日期) + a[href=./tXXX.shtml][title=完整标题]""" items = [] if BeautifulSoup: soup = BeautifulSoup(html_text, "html.parser") for li in soup.select("ul.more-list li"): a = li.find("a", href=True) if not a: continue href = a["href"] title = a.get("title") or a.get_text(strip=True) title = clean_title(title) date = "" sp = li.find("span") if sp: date = sp.get_text(strip=True) if not title or len(title) < 4: continue abs_url = urllib.parse.urljoin(BASE_URL + LIST_DIR, href) items.append((abs_url, title, date)) return items def parse_detail(html_text, page_url): """h1 完整标题 + .article-con 正文HTML + #article-file/#relative-file 附件""" title = "" date = "" content_html = "" attachments = [] if BeautifulSoup: soup = BeautifulSoup(html_text, "html.parser") h1 = soup.find("h1") if h1: title = clean_title(h1.get_text(" ", strip=True)) # 日期: .article-inf-left 或 meta inf = soup.select_one(".article-inf-left") if inf: dm = re.search(r"(\d{4}-\d{2}-\d{2})", inf.get_text()) if dm: date = dm.group(1) if not date: md = soup.find("meta", attrs={"name": re.compile(r"PubDate|publishdate|pubdate", re.I)}) if md and md.get("content"): dm = re.match(r"(\d{4}-\d{2}-\d{2})", md["content"].strip()) if dm: date = dm.group(1) # 正文 .article-con (保留表格 HTML, 过滤注释节点) con = soup.select_one(".article-con") if con: parts = [] for c in con.contents: if isinstance(c, str): continue # 纯文本(如注释内容)跳过 if getattr(c, "name", None) is None: continue parts.append(str(c)) content_html = "".join(parts) # 附件区 #article-file + #relative-file for fid in ("article-file", "relative-file", "downloadfile", "relfile"): fd = soup.find(id=fid) if fd: for a in fd.find_all("a", href=True): href = a["href"] txt = clean_title(a.get_text(" ", strip=True)) or os.path.basename(href.split("?")[0]) abs_url = urllib.parse.urljoin(page_url, href) if abs_url not in [x[0] for x in attachments]: attachments.append((abs_url, txt)) # 正文内附件链接 if con: for a in con.find_all("a", href=True): href = a["href"] if re.search(r"\.(pdf|docx?|xlsx?|zip|rar|wps|ofd)", href, re.I): txt = clean_title(a.get_text(" ", strip=True)) or os.path.basename(href.split("?")[0]) abs_url = urllib.parse.urljoin(page_url, href) if abs_url not in [x[0] for x in attachments]: attachments.append((abs_url, txt)) return title, date, content_html, attachments def html_to_text(content, page_url): """清理正文, 保留表格 HTML, 附件链接转绝对 URL 段落""" if not content: return "", 0, [] content = re.sub(r"", "", content, flags=re.S) content = re.sub(r"]*>", "", content, flags=re.I) content = re.sub(r"]*>.*?", "", content, flags=re.S | re.I) content = re.sub(r"]*>.*?", "", content, flags=re.S | re.I) attachments = [] link_protect = {} if BeautifulSoup: try: _soup = BeautifulSoup(content, "html.parser") for a in _soup.find_all("a", href=True): if re.search(r"\.(pdf|docx?|xlsx?|zip|rar|wps|ofd)", a.get("href", ""), re.I): href = a["href"] txt = a.get_text(strip=True) or os.path.basename(href.split("?")[0]) abs_url = urllib.parse.urljoin(page_url, href) attachments.append((abs_url, txt)) key = f"__ATTACH__{len(link_protect)}__" link_protect[key] = f'

{txt}

' a.replace_with(key) for p in _soup.find_all("p"): imgs = p.find_all("img") if imgs and not p.get_text(strip=True): srcs = [] for img in imgs: src = img.get("src") or "" if src: abs_url = urllib.parse.urljoin(page_url, src) srcs.append(f'{os.path.basename(src.split("?")[0]) or "图片"}') if srcs: key = f"__IMG__{len(link_protect)}__" link_protect[key] = "

" + "
".join(srcs) + "

" p.replace_with(key) content = str(_soup) except Exception: pass table_protect = [] def _tbl_repl(m): table_protect.append(m.group(0)) return f"__TBL__{len(table_protect)-1}__" content = re.sub(r"]*>.*?", _tbl_repl, content, flags=re.S | re.I) has_table = 1 if re.search(r"]*>", content, re.I) else 0 if BeautifulSoup: try: _soup = BeautifulSoup(content, "html.parser") for tag in _soup.find_all(True): for attr in ("style", "class", "lang", "dir", "align", "valign", "width", "height", "border", "cellpadding", "cellspacing"): tag.attrs.pop(attr, None) content = str(_soup) except Exception: pass content = re.sub(r"

", "

\n\n", content, flags=re.I) content = re.sub(r"", "\n", content, flags=re.I) content = re.sub(r"]*>", "", content, flags=re.I) parts = [] for block in content.split("\n\n"): b = block.strip() if not b: continue plain = re.sub(r"<[^>]+>", "", b) plain = html_lib.unescape(plain) plain = plain.replace("\xa0", " ").replace(" ", " ") plain = re.sub(r"[\s\u200b\u200c\u200d\ufeff]+", "", plain) if not plain: continue for k, v in link_protect.items(): b = b.replace(k, v) for i, tbl in enumerate(table_protect): b = b.replace(f"__TBL__{i}__", tbl) parts.append(b.strip()) out = "\n\n".join(parts) for k, v in link_protect.items(): out = out.replace(k, v) for i, tbl in enumerate(table_protect): out = out.replace(f"__TBL__{i}__", tbl) out = re.sub(r"__LINK__\d+__", "", out) out = re.sub(r"__TBL__\d+__", "", out) out = re.sub(r"__IMG__\d+__", "", out) out = re.sub(r"__ATTACH__\d+__", "", out) out = html_lib.unescape(out) out = re.sub(r"[ \t]+\n", "\n", out) out = re.sub(r"\n{3,}", "\n\n", out) out = re.sub(r"[ \t]{2,}", " ", out) seen = set() final = [] for p in out.split("\n\n"): key = re.sub(r"\s+", "", re.sub(r"<[^>]+>", "", p)) if not key: continue if key in seen: continue seen.add(key) final.append(p) out = "\n\n".join(final).strip() return out, has_table, attachments def append_attachments(content, attachments): """附件区转

绝对URL

段落, 不保留 ul/li 包装""" if not attachments: return content parts = [content] if content.strip() else [] for abs_url, txt in attachments: parts.append(f'

{txt}

') return "\n\n".join(parts) def load_existing_urls(conn): try: cur = conn.execute("SELECT page_url FROM gov_raw WHERE site_name=?", (SITE_NAME,)) return {row[0] for row in cur.fetchall()} except Exception: return set() def store_record(conn, url, title, content, date, has_table): cur = conn.execute("SELECT id FROM gov_raw WHERE page_url=? AND site_name=?", (url, SITE_NAME)) if cur.fetchone(): return False try: industry = "other" try: from crawler_lib import classify_industry industry = classify_industry(title) except Exception: pass cur = conn.execute( "INSERT OR IGNORE INTO gov_raw " "(page_url, source_url, title, content, publish_date, site_name, group_name, script_name, has_table, industry) " "VALUES (?,?,?,?,?,?,?,?,?,?)", (url, url, title, content, date, SITE_NAME, GROUP_NAME, SCRIPT_NAME, has_table, industry), ) if cur.rowcount > 0: summary = re.sub(r"<[^>]+>", "", content) summary = html_lib.unescape(summary) summary = re.sub(r"\s+", " ", summary).strip()[:200] conn.execute( "INSERT OR IGNORE INTO gov_search (rowid, title, site_name, summary) VALUES (?,?,?,?)", (cur.lastrowid, title, SITE_NAME, summary), ) return True return False except sqlite3.OperationalError as e: if "locked" in str(e): time.sleep(3) return store_record(conn, url, title, content, date, has_table) raise def main(): import requests s = requests.Session() s.headers.update(HEADERS) conn = None if not JSONL_PATH: conn = sqlite3.connect(DB_PATH, timeout=60) conn.execute("PRAGMA busy_timeout=60000") conn.execute("PRAGMA journal_mode=WAL") existing = load_existing_urls(conn) print(f"已入库 {len(existing)} 条, 用于列表级预查重", flush=True) new_count = 0 skip_count = 0 jsonl_rows = [] for page in range(1, _PAGES + 1): try: html_text = fetch_list(s, page) if html_text is None: print(f"Page {page} fetch error (all retries)", flush=True) break except Exception as e: print(f"Page {page} fetch error: {e}", flush=True) break items = parse_list(html_text) print(f"Page {page}: found {len(items)} items", flush=True) if not items: print("Empty page, stop pagination", flush=True) break for abs_url, title, date in items: if not JSONL_PATH and abs_url in existing: skip_count += 1 continue # 列表级预查重: 已入库跳过详情 try: r = http_get(s, abs_url) if r is None: print(f" detail error (all retries): {abs_url}", flush=True) skip_count += 1 continue dhtml = r.content.decode(ENC, errors="replace") except Exception as e: print(f" detail error {abs_url}: {e}", flush=True) skip_count += 1 continue dtitle, ddate, content_html, atts = parse_detail(dhtml, abs_url) if not dtitle: dtitle = title if not ddate: ddate = date content, has_table, atts_in = html_to_text(content_html, abs_url) all_atts = atts + atts_in if atts_in else atts content = append_attachments(content, all_atts) plain_len = len(re.sub(r"<[^>]+>", "", content).strip()) if plain_len < 10 and not content_html: print(f" skip empty: {dtitle[:30]}", flush=True) skip_count += 1 continue if JSONL_PATH: jsonl_rows.append({ "page_url": abs_url, "title": dtitle, "content": content, "publish_date": ddate, "has_table": has_table, "site_name": SITE_NAME, "group_name": GROUP_NAME, "script_name": SCRIPT_NAME, }) print(f" JSONL + {dtitle[:40]} ({ddate})", flush=True) new_count += 1 continue ok = store_record(conn, abs_url, dtitle, content, ddate, has_table) if ok: new_count += 1 existing.add(abs_url) print(f" + {dtitle[:40]} ({ddate})", flush=True) else: skip_count += 1 time.sleep(0.3) time.sleep(0.5) if JSONL_PATH: with open(JSONL_PATH, "w", encoding="utf-8") as f: for row in jsonl_rows: f.write(json.dumps(row, ensure_ascii=False) + "\n") print(f"JSONL written: {JSONL_PATH} ({len(jsonl_rows)} rows)", flush=True) if conn is not None: conn.commit() conn.close() print(f"DONE: new={new_count} skip={skip_count}", flush=True) return 0 if new_count > 0 or skip_count > 0 else 1 if __name__ == "__main__": sys.exit(main())