#!/usr/bin/env python3 """牟平区通知公告爬虫 (JCMS 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 = "牟平区通知公告" BASE_URL = "https://www.muping.gov.cn" API_URL = "https://www.muping.gov.cn/api-gateway/jpaas-publish-server/front/page/build/unit" PAGES = 5 PER_PAGE = 25 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 = { "parseType": "bulidstatic", "webId": "149", "pageId": "13058", "pageType": "column", "tplSetId": "QJQ18GsL67Dutf4OTidwX", "tagId": "当前栏目列表", "paramJson": json.dumps({"pageNo": page_num, "pageSize": PER_PAGE}), } 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.muping.gov.cn/col/col13058/index.html", }) resp = urllib.request.urlopen(req, timeout=30, context=ssl_ctx) data = json.loads(resp.read().decode("utf-8")) html_content = data.get('data', {}).get('html', '') soup = BeautifulSoup(html_content, 'html.parser') items = [] for li in soup.find_all('li'): a_tag = li.find('a') span = li.find('span') if a_tag: href = a_tag.get('href', '') title = a_tag.get_text(strip=True) or a_tag.get('title', '') title = re.sub(r'\s+', ' ', title).strip() date_str = span.get_text(strip=True) if span else '' if href and title and not href.startswith('javascript'): items.append({ 'url': urljoin(BASE_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: print(f" [ERROR] {e}") return "", [] soup = BeautifulSoup(html, 'html.parser') # 正文 zoom = soup.find('div', class_='zoom') or soup.find('div', id='zoom') or soup.find('div', class_=lambda c: c and 'zoom' in str(c).split() if c else False) if not zoom: zoom = soup.find('div', class_='art_con') or soup.find('div', class_='article-content') content = "" attachments = [] if zoom: parts = [] for el in zoom.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])[ \t\u3000]+(?=[\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(BASE_URL, src) parts.append(f'![{alt}]({full_src})') content = '\n\n'.join(parts) # 附件 for a_tag in zoom.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(BASE_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/muping_tzgg.json', 'w', encoding='utf-8') as f: json.dump(db_items, f, ensure_ascii=False, indent=2) print(" 数据已保存到 /root/muping_tzgg.json") if __name__ == '__main__': main()