#!/usr/bin/env python3 """ G批 crawl_jian_hpsp.py - 吉安市生态环境局-环评审批公示 ================================== CMS: 自建PHP + AJAX分页 列表API: POST /api-ajax_list-{page}.html (jQuery嵌套数组格式) 详情: /news-show-{id}.html (div.text-main) CloudWAF防护，需浏览器UA + session cookies """ import sys, os, re, urllib.parse, json, warnings def _make_summary(body): """剔除 markdown 表格行后取前 300 字符，防止搜索页摘要表格断裂""" if not body: return "" lines = [l for l in body.split("\n") if not l.strip().startswith("|")] return re.sub(r"\s+", " ", " ".join(lines))[:300] warnings.filterwarnings("ignore") sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from crawler_lib import push_to_searchdb import requests from bs4 import BeautifulSoup SITE_NAME = "吉安市生态环境局-环评审批公示" BASE_URL = "http://sthj.jian.gov.cn" IP = "218.64.81.44" PAGE_SIZE = 20 HEADERS = { "Host": "sthj.jian.gov.cn", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.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,en;q=0.8", } # jQuery-style serialized data for AJAX API AJAX_DATA_TEMPLATE = ( 'ajax_type%5B%5D=53_news&ajax_type%5B%5D=18&ajax_type%5B%5D=53' '&ajax_type%5B%5D=news&ajax_type%5B%5D=Y-m-d&ajax_type%5B%5D=40' '&ajax_type%5B%5D=20&ajax_type%5B7%5D%5B%5D=is_top+DESC' '&ajax_type%5B7%5D%5B%5D=displayorder+DESC' '&ajax_type%5B7%5D%5B%5D=inputtime+DESC&ajax_type%5B%5D=&is_ds=1' ) def create_session(): """Create requests session with CloudWAF cookies""" s = requests.Session() s.headers.update(HEADERS) # Get page first to establish CloudWAF session try: s.get(f"{BASE_URL}/news-list-hpsp.html", timeout=30) except Exception: pass return s def fetch_detail(url): """Fetch detail page, try domain first, fallback to IP+Host""" s = requests.Session() s.headers.update(HEADERS) try: r = s.get(url, timeout=30) r.encoding = 'utf-8' return r.text except Exception: # Fallback to IP ip_url = re.sub(r'https?://sthj\.jian\.gov\.cn', f'http://{IP}', url) try: r = requests.get(ip_url, headers=HEADERS, timeout=30) r.encoding = 'utf-8' return r.text except Exception as e: return None def fetch_list_page(session, page_no): """Fetch list page via AJAX API""" data = AJAX_DATA_TEMPLATE # page number is in URL, not data api_headers = { "User-Agent": HEADERS["User-Agent"], "Accept": "application/json, text/javascript, */*", "X-Requested-With": "XMLHttpRequest", "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8", "Referer": f"{BASE_URL}/news-list-hpsp.html", } try: r = session.post(f"{BASE_URL}/api-ajax_list-{page_no}.html", headers=api_headers, data=data, timeout=30) if r.status_code != 200: return [], 0 result = r.json() except Exception: return [], 0 if not result.get("data"): return [], result.get("total", 0) items = [] for item in result["data"]: title = item.get("title", "").strip() url = item.get("url", "") date = item.get("inputtime", "") if url and not url.startswith("http"): url = urllib.parse.urljoin(BASE_URL, url) items.append({"title": title, "url": url, "date": date}) return items, result.get("total", 0) def html_table_to_html(table, base_url=""): """保留 HTML 表格结构，仅将相对链接/图片转绝对 URL""" from bs4 import BeautifulSoup 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 parse_detail(detail_url): """Extract content from detail page (div.text-main)""" text = fetch_detail(detail_url) if not text: return None, None, None, [] soup = BeautifulSoup(text, 'html.parser') # Title from title_tag = soup.find('title') title = title_tag.get_text(strip=True) if title_tag else "" title = re.sub(r'^吉安市生态环境局\s*', '', title).strip() # Date from "发布时间：" date_str = None date_m = re.search(r'发布时间[：:]\s*(\d{4}-\d{1,2}-\d{1,2})', text) if date_m: date_str = date_m.group(1) # Content container text_main = soup.find('div', class_='text-main') if not text_main: return title, date_str, None, [] body_parts = [] attachments = [] # Iterate direct children: p, table, img, div, span for el in text_main.find_all(['p', 'table', 'img', 'div', 'span'], recursive=False): if el.name in ['p', 'div', 'span']: # Check for <a> with attachments inside for a in el.find_all('a'): href = a.get('href', '') fname = a.get_text(strip=True) if href and '/uploadfile/' in href: full_url = urllib.parse.urljoin(BASE_URL, href) attachments.append({"name": fname, "url": full_url}) # Also add as markdown link in body if fname: body_parts.append(f"- [{fname}]({full_url})") # Brighter line breaks: replace <br> with delimiter, extract text, convert back DELIM = '\x00BR\x00' for br in el.find_all('br'): br.replace_with(DELIM) txt = el.get_text(" ", strip=True) txt = txt.replace(DELIM, '\n') txt = re.sub(r'\n\s*\n', '\n\n', txt) if txt and len(txt) > 5: # Skip nav/breadcrumb-like text if '首页' not in txt[:10] and '当前位置' not in txt[:10]: body_parts.append(txt) elif el.name == 'table': md = html_table_to_html(el, base_url=BASE_URL) if md: body_parts.append(md) elif el.name == 'img': src = el.get('src', '') alt = el.get('alt', '') if src and '/uploadfile/' in src: full_src = urllib.parse.urljoin(BASE_URL, src) body_parts.append(f"![{alt}]({full_src})") # Clean up: check attachments in the table too (for 建设项目环评文件 row) for a in text_main.find_all('a'): href = a.get('href', '') fname = a.get_text(strip=True) if href and '/uploadfile/' in href: full_url = urllib.parse.urljoin(BASE_URL, href) # Check if not already added if not any(a2["url"] == full_url for a2 in attachments): attachments.append({"name": fname, "url": full_url}) body = "\n\n".join(body_parts) if body_parts else None return title, date_str, body, attachments def crawl(pages=5): session = create_session() all_items = [] total_count = 0 for page_no in range(1, pages + 1): items, total = fetch_list_page(session, page_no) if not items: break total_count = total all_items.extend(items) print(f"Page {page_no}: {len(items)} items", flush=True) if len(all_items) >= total_count: break print(f"\nTotal list items: {len(all_items)}", flush=True) results = [] empty_body = 0 with_attach = 0 for i, item in enumerate(all_items): title = item["title"] detail_url = item["url"] list_date = item["date"] parsed_title, pub_date, body, attachments = parse_detail(detail_url) final_title = parsed_title or title final_date = pub_date or list_date attach_str = ";".join([f"[{a['name']}]({a['url']})" for a in attachments]) if attachments else "" if attachments: with_attach += 1 if not body: empty_body += 1 results.append({ "site_name": SITE_NAME, "title": final_title, "url": detail_url, "source_url": detail_url, "pub_date": final_date, "content": body or "", "summary": _make_summary(body), "attachments": attach_str, }) if (i + 1) % 10 == 0: print(f" Parsed: {i+1}/{len(all_items)}", flush=True) print(f"Parsed: {len(results)} items ({empty_body} empty body, {with_attach} with attachments)", flush=True) push_to_searchdb(results, "jian_hpsp") if __name__ == "__main__": pages = int(sys.argv[sys.argv.index("--pages") + 1]) if "--pages" in sys.argv else 5 crawl(pages=pages)