#!/usr/bin/env python3 """明光市生态环境行政审批爬虫 (政府信息公开API版)""" import sys, re, os, json sys.path.insert(0, '/root') sys.path.insert(0, '/root/gov_crawler') from urllib.parse import urljoin import urllib.request, ssl from bs4 import BeautifulSoup SITE_NAME = "明光市生态环境行政审批" API_URL = "https://www.mingguang.gov.cn/chuzhou/site/label/8888" PAGES = 5 PER_PAGE = 20 ssl_ctx = ssl.create_default_context() ssl_ctx.check_hostname = False ssl_ctx.verify_mode = ssl.CERT_NONE def html_table_to_html(table, base_url=""): """保留 HTML 表格结构，仅将相对链接/图片转绝对 URL""" import urllib.parse tbl = BeautifulSoup(str(table), 'html.parser') for a in tbl.find_all('a'): href = a.get('href', '') if href and not href.startswith(('http', 'javascript', '#')): a['href'] = urllib.parse.urljoin(base_url, href) if base_url else href for img in tbl.find_all('img'): src = img.get('src', '') if src and not src.startswith(('http', '//', 'data:')): img['src'] = urllib.parse.urljoin(base_url, src) if base_url else src return str(tbl) def fetch_list_page(page_num): """调用列表API""" params = { "labelName": "publicInfoList", "siteId": "2653861", "pageSize": str(PER_PAGE), "pageIndex": str(page_num), "action": "list", "isDate": "true", "dateFormat": "yyyy-MM-dd", "length": "50", "organId": "161056936", "organName": "", "type": "4", "catId": "170077439", "result": "暂无相关信息", } query = '&'.join(f'{k}={urllib.request.quote(str(v))}' for k, v in params.items()) url = f"{API_URL}?{query}" req = urllib.request.Request(url, headers={ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Referer": "https://www.mingguang.gov.cn/site/tpl/162748858", }) resp = urllib.request.urlopen(req, timeout=30, context=ssl_ctx) html = resp.read().decode("utf-8", errors="replace") soup = BeautifulSoup(html, 'html.parser') items = [] for navli in soup.find_all('div', class_='xxgk_navli'): mc = navli.find('li', class_='mc') rq = navli.find('li', class_='rq') if mc: a_tag = mc.find('a') if a_tag: href = a_tag.get('href', '') title = re.sub(r'\s+', ' ', a_tag.get_text(strip=True)).strip() date_str = rq.get_text(strip=True) if rq else '' if href and title: items.append({ 'url': href if href.startswith('http') else urljoin(API_URL, href), 'title': title, 'date': date_str, }) return items def fetch_detail(url): """抓取详情页提取正文""" try: req = urllib.request.Request(url, headers={ "User-Agent": "Mozilla/5.0", "Accept-Language": "zh-CN,zh;q=0.9", }) resp = urllib.request.urlopen(req, timeout=30, context=ssl_ctx) html = resp.read().decode("utf-8", errors="replace") except Exception as e: return "", [] soup = BeautifulSoup(html, 'html.parser') # 正文: gkwz_contnet j-fontContent content_div = soup.find('div', class_='gkwz_contnet') or \ soup.find('div', class_=lambda c: c and 'gkwz_contnet' in str(c) if c else False) if not content_div: content_div = soup.find('div', class_='xxgk_contnet') if not content_div: content_div = soup.find('div', class_=lambda c: c and 'xxgkcontent' in str(c) if c else False) content = "" attachments = [] if content_div: parts = [] for el in content_div.find_all(['p', 'table', 'img']): if el.name == 'p': if el.find_parent('table'): continue txt = el.get_text(' ', strip=True) txt = re.sub(r'\s+', ' ', txt) txt = re.sub(r'(?<=[\u4e00-\u9fff])\s+(?=[\u4e00-\u9fff])', '', txt) if txt: parts.append(txt) elif el.name == 'table': tbl_html = html_table_to_html(el, url) if tbl_html: parts.append(tbl_html) elif el.name == 'img': src = el.get('src', '') alt = el.get('alt', '图片') if src and not src.startswith('data:'): full_src = urljoin(url, src) parts.append(f'![{alt}]({full_src})') content = '\n\n'.join(parts) # 附件 for a_tag in content_div.find_all('a', href=True): href = a_tag['href'] if re.search(r'\.(pdf|doc|docx|xls|xlsx|zip|rar)$', href, re.I): full_url = urljoin(url, href) name = a_tag.get_text(strip=True) or os.path.basename(href) attachments.append({'name': name, 'url': full_url}) # 内嵌附件链接 if attachments: if content: content += '\n\n' content += '\n'.join(f'[{att["name"]}]({att["url"]})' for att in attachments) return content, attachments def main(): all_items = [] print(f"=== {SITE_NAME} ===") for page in range(1, PAGES + 1): print(f" [Page {page}/{PAGES}] fetching...", end=' ') items = fetch_list_page(page) print(f"{len(items)} items") all_items.extend(items) print(f"\n=== 共 {len(all_items)} 条列表项 ===") # 去重 seen_urls = set() unique_items = [] for item in all_items: if item['url'] not in seen_urls: seen_urls.add(item['url']) unique_items.append(item) print(f"去重后: {len(unique_items)} 条") # 爬详情 db_items = [] for i, item in enumerate(unique_items): print(f" [{i+1}/{len(unique_items)}] {item['title'][:40]}...") content, attachments = fetch_detail(item['url']) summary = re.sub(r'\s+', '', content)[:200] if content else '' att_json = json.dumps(attachments, ensure_ascii=False) if attachments else '' db_items.append({ 'site_name': SITE_NAME, 'source_url': item['url'], 'url': item['url'], 'title': item['title'], 'pub_date': item.get('date', ''), 'summary': summary, 'content': content, 'attachments': att_json, }) # 入库 try: from crawler_lib import push_to_searchdb push_to_searchdb(db_items, SITE_NAME) print(f"\n✅ 入库完成: {len(db_items)} 条") except Exception as e: print(f"\n❌ 入库异常: {e}") import traceback traceback.print_exc() with open('/root/mingguang_hj.json', 'w', encoding='utf-8') as f: json.dump(db_items, f, ensure_ascii=False, indent=2) print(" 数据已保存到 /root/mingguang_hj.json") if __name__ == '__main__': main()