#!/usr/bin/env python3 """湘乡经济开发区 - 通知公告""" import json, re, sys, requests from bs4 import BeautifulSoup from urllib.parse import urljoin DB_PATH = "/root/search.db" SITE_NAME = "湘乡经济开发区-通知公告" CATEGORY = "通知公告" GROUP = "湘乡市" BASE_URL = "http://www.xxs.gov.cn/1941/1943/1975/4193" MAX_PAGES = 17 HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Accept-Language": "zh-CN,zh;q=0.9", } SQL = """INSERT OR IGNORE INTO gov_raw (site_name, source_url, page_url, title, publish_date, summary, content, category, attachments, group_name) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""" def init_db(): import sqlite3 conn = sqlite3.connect(DB_PATH, timeout=60) conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA busy_timeout=5000") return conn def decode_resp(r): raw = r.content raw_lower = raw[:2000].lower() if b"gb2312" in raw_lower or b"gbk" in raw_lower or b"charset=gb" in raw_lower: return raw.decode("gbk", errors="replace") return raw.decode("utf-8", errors="replace") def table_to_md(table, *args, **kwargs): """保留 HTML 表格结构（不转 md）""" return str(table) def extract_content(content_div, detail_url): parts = [] processed = set() for el in content_div.find_all(["p", "table", "img"], recursive=True): if id(el) in processed: continue processed.add(id(el)) if el.name == "table": md = table_to_md(el) if md: parts.append(md) elif el.name == "img": src = el.get("src", "") if src: full_src = urljoin(detail_url, src) if not src.startswith("http") else src parts.append(f"![]({full_src})") elif el.name == "p": txt = el.get_text(" ", strip=True) if txt: txt = re.sub(r'[^\S\n]+', ' ', txt) parts.append(txt) return "\n\n".join(parts) def extract_attachments(soup, detail_url): attachments = [] for a_tag in soup.find_all("a", href=True): href = a_tag["href"].strip().lower() if any(href.endswith(ext) for ext in [".pdf", ".doc", ".docx", ".xls", ".xlsx", ".rar", ".zip"]): name = a_tag.get_text(strip=True) or href.split("/")[-1] full_url = urljoin(detail_url, a_tag["href"]) if not a_tag["href"].startswith("http") else a_tag["href"] if not any(a["url"] == full_url for a in attachments): attachments.append({"name": name, "url": full_url}) return attachments def extract_list_page(page_num): if page_num == 1: url = f"{BASE_URL}/index.htm" else: url = f"{BASE_URL}/index_{page_num-1}.htm" try: r = requests.get(url, headers=HEADERS, timeout=30) r.raise_for_status() except Exception as e: print(f" [ERR] page {page_num}: {e}", file=sys.stderr) return [] html = decode_resp(r) soup = BeautifulSoup(html, "html.parser") items = [] table = soup.select_one("div.xxgk-right table.table tbody") if not table: print(f" [WARN] page {page_num}: no table found", file=sys.stderr) return [] for tr in table.find_all("tr", recursive=False): tds = tr.find_all("td") if len(tds) < 3: continue a_tag = tds[1].find("a") if not a_tag or not a_tag.get("href"): continue href = a_tag["href"].strip() title = a_tag.get("title", "") or a_tag.get_text(strip=True) if not title: continue date_str = tds[2].get_text(strip=True) if len(tds) > 2 else "" detail_url = urljoin(BASE_URL + "/", href) items.append({"title": title, "url": detail_url, "date": date_str}) print(f" Page {page_num}: {len(items)} items") return items def extract_detail(url): try: r = requests.get(url, headers=HEADERS, timeout=30) r.raise_for_status() except Exception: return None, None, None, [] html = decode_resp(r) soup = BeautifulSoup(html, "html.parser") # Title: div.xlnk > h2 title = "" xlnk = soup.find("div", class_="xlnk") if xlnk: h2 = xlnk.find("h2") if h2: title = h2.get_text(strip=True) if not title: mt = soup.find("title") if mt: title = mt.get_text(strip=True) title = re.sub(r'[-_|]\s*通知公告.*$', '', title) # Date: xlnk > h6 > span "发布时间：YYYY-MM-DD" date_str = "" if xlnk: h6 = xlnk.find("h6") if h6: for span in h6.find_all("span"): txt = span.get_text(strip=True) m = re.search(r'发布时间[：:]\s*(\d{4}-\d{2}-\d{2})', txt) if m: date_str = m.group(1) break if not date_str: body = soup.get_text(" ", strip=True) m = re.search(r'发布时间[：:]\s*(\d{4}-\d{2}-\d{2})', body) if m: date_str = m.group(1) # Content: div.xl-xqnr content = "" xqnr = soup.find("div", class_="xl-xqnr") if xqnr: content = extract_content(xqnr, url) attachments = extract_attachments(soup, url) if len(content.strip()) < 20: fallback = f"[{title or url.split('/')[-1]}]({url})" if attachments: for a in attachments: fallback += f"\n\n附件：[{a['name']}]({a['url']})" content = fallback return title, date_str, content, attachments def crawl(test_mode=False, max_pages=MAX_PAGES): conn = init_db() cur = conn.cursor() total = 0 errors = 0 total_listed = 0 for pn in range(1, max_pages + 1): items = extract_list_page(pn) total_listed += len(items) if not items: break for item in items: url = item["url"] cur.execute("SELECT id FROM gov_raw WHERE page_url = ?", (url,)) if cur.fetchone(): continue title, date_str, content, attachments = extract_detail(url) if not title: errors += 1 continue final_title = title or item["title"] final_date = date_str or item["date"] att_json = json.dumps(attachments, ensure_ascii=False) if attachments else "" summary = (content[:200] if content else final_title).strip() if not summary: summary = final_title try: cur.execute(SQL, ( SITE_NAME, url, url, final_title.strip(), final_date, summary, content, CATEGORY, att_json, GROUP, )) conn.commit() if cur.rowcount > 0: total += 1 except Exception as e: print(f" DB error: {e}", file=sys.stderr) errors += 1 if test_mode and total >= 5: break if test_mode and total >= 5: break conn.close() print(f"\n[DONE] {SITE_NAME}") print(f" New: {total}, Errors: {errors}, Listed total: {total_listed}") return total if __name__ == "__main__": mode = "test" if "--test" in sys.argv else "full" max_p = 5 if mode == "test" else MAX_PAGES if len(sys.argv) > 1 and sys.argv[1].isdigit(): max_p = int(sys.argv[1]) crawl(test_mode=(mode == "test"), max_pages=max_p)