标题 + div.content-box 正文(TRS_UEDITOR) """ import sys, os, requests, re, argparse from datetime import datetime, timedelta sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from crawler_lib import push_to_searchdb CUTOFF = (datetime.now() - timedelta(days=365*3)).strftime('%Y-%m-%d') HEADERS = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0'} BASE = 'https://sthjt.shanxi.gov.cn' COLUMNS = { 'xmslgs': {'name': '山西省生态环境厅-项目受理公示', 'path': 'zwgk/hjpj/xmslgs'}, 'nspgs': {'name': '山西省生态环境厅-拟审批公示', 'path': 'zwgk/hjpj/nspgs'}, 'hpwjspjdgg': {'name': '山西省生态环境厅-环评文件审批决定', 'path': 'zwgk/hjpj/hpwjspjdgg'}, 'nzcspyjgs': {'name': '山西省生态环境厅-拟作出审批意见公示', 'path': 'zwgk/hjpj/nzcspyjgs'}, 'sljsxmhjyxpjwjqk': {'name': '山西省生态环境厅-受理建设项目环评文件情况', 'path': 'zwgk/hjpj/sljsxmhjyxpjwjqk'}, } def fetch_list(col, page_idx): path = COLUMNS[col]['path'] if page_idx == 1: url = f'{BASE}/{path}/' else: url = f'{BASE}/{path}/index_{page_idx-1}.shtml' try: r = requests.get(url, timeout=(5, 15), headers=HEADERS, verify=False) r.encoding = 'utf-8' except: return [] items = re.findall( r"<li>\s*<a href='\./(\d{6}/t\d+_\d+\.shtml)' title='([^']*)'>(.*?)</a>\s*<span>\[(\d{4}-\d{2}-\d{2})\]</span>", r.text, re.DOTALL ) results = [] for href, title_attr, title_text, pd in items: title = re.sub(r'\s+', '', title_attr or title_text).strip() if len(title) < 4: continue url = f"{BASE}/{path}/{href}" results.append({'url': url, 'pub_date': pd, 'title': title}) if results: return results # 结构B: table.list-table (nzcspyjgs/sljsxmhjyxpjwjqk) rows = re.findall( r"<tr>\s*<td><a[^>]*href='\./(\d{6}/t\d+_\d+\.shtml)'[^>]*title='([^']*)'>(.*?)</a></td>\s*<td[^>]*>.*?</td>\s*<td[^>]*>.*?</td>\s*<td[^>]*>.*?</td>\s*<td[^>]*>(\d{4}-\d{2}-\d{2})</td>", r.text, re.DOTALL ) for href, title_attr, title_text, pd in rows: title = re.sub(r'\s+', '', title_attr or title_text).strip() if len(title) < 4: continue url = f"{BASE}/{path}/{href}" results.append({'url': url, 'pub_date': pd, 'title': title}) return results def extract_balanced_div(html, open_tag_re): """平衡 div 匹配: 从匹配到的 <div...> 的 '>' 之后, 到对应 </div>, 返回内部 HTML。修复非贪婪 (.*?)</div> 在正文含嵌套 div(表格/附件区)时截断的问题""" m = re.search(open_tag_re, html) if not m: return '' gt = html.find('>', m.start()) if gt == -1: return '' i = gt + 1 depth = 1 for mm in re.finditer(r'<div[\s>]|</div>', html[i:]): if mm.group(0).startswith('<div'): depth += 1 else: depth -= 1 if depth == 0: return html[i:i + mm.start()] return html[i:] def fetch_detail(url): for _attempt in range(2): try: r = requests.get(url, timeout=(5, 15), headers=HEADERS, verify=False) r.encoding = 'utf-8' break except: r = None if r is None: return '', '' title = '' m = re.search(r'<title>\s*([^<]+?)\s*

#!/usr/bin/env python3 """crawl_shanxi_sthjt.py - 山西省生态环境厅-环境影响评价公示 (5栏目) 站点: https://sthjt.shanxi.gov.cn/zwgk/hjpj/{col}/ CMS: TRS WCM (UTF-8, 服务端渲染) 栏目(--col): xmslgs = 项目受理公示 nspgs = 拟审批公示 hpwjspjdgg = 环评文件审批决定 nzcspyjgs = 拟作出审批意见公示 sljsxmhjyxpjwjqk = 受理建设项目环评文件情况列表: 两种结构 — A) ul.list-details > li > a[title=标题](href=./YYYYMM/tYYYYMMDD_xxx.shtml) + span[YYYY-MM-DD] (xmslgs/nspgs/hpwjspjdgg) B) table.list-table > tr > td > a[title=标题] + ... + td 日期 (nzcspyjgs/sljsxmhjyxpjwjqk, 列: 项目名称/建设地点/建设单位/环评机构/日期) 分页: index_N.shtml (首页=index.shtml, 第2页=index_1.shtml, 每页10~20条) 详情: 标题 + div.content-box 正文(TRS_UEDITOR) """ import sys, os, requests, re, argparse from datetime import datetime, timedelta sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from crawler_lib import push_to_searchdb CUTOFF = (datetime.now() - timedelta(days=365*3)).strftime('%Y-%m-%d') HEADERS = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0'} BASE = 'https://sthjt.shanxi.gov.cn' COLUMNS = { 'xmslgs': {'name': '山西省生态环境厅-项目受理公示', 'path': 'zwgk/hjpj/xmslgs'}, 'nspgs': {'name': '山西省生态环境厅-拟审批公示', 'path': 'zwgk/hjpj/nspgs'}, 'hpwjspjdgg': {'name': '山西省生态环境厅-环评文件审批决定', 'path': 'zwgk/hjpj/hpwjspjdgg'}, 'nzcspyjgs': {'name': '山西省生态环境厅-拟作出审批意见公示', 'path': 'zwgk/hjpj/nzcspyjgs'}, 'sljsxmhjyxpjwjqk': {'name': '山西省生态环境厅-受理建设项目环评文件情况', 'path': 'zwgk/hjpj/sljsxmhjyxpjwjqk'}, } def fetch_list(col, page_idx): path = COLUMNS[col]['path'] if page_idx == 1: url = f'{BASE}/{path}/' else: url = f'{BASE}/{path}/index_{page_idx-1}.shtml' try: r = requests.get(url, timeout=(5, 15), headers=HEADERS, verify=False) r.encoding = 'utf-8' except: return [] items = re.findall( r"<li>\s*<a href='\./(\d{6}/t\d+_\d+\.shtml)' title='([^']*)'>(.*?)</a>\s*<span>\[(\d{4}-\d{2}-\d{2})\]</span>", r.text, re.DOTALL ) results = [] for href, title_attr, title_text, pd in items: title = re.sub(r'\s+', '', title_attr or title_text).strip() if len(title) < 4: continue url = f"{BASE}/{path}/{href}" results.append({'url': url, 'pub_date': pd, 'title': title}) if results: return results # 结构B: table.list-table (nzcspyjgs/sljsxmhjyxpjwjqk) rows = re.findall( r"<tr>\s*<td><a[^>]*href='\./(\d{6}/t\d+_\d+\.shtml)'[^>]*title='([^']*)'>(.*?)</a></td>\s*<td[^>]*>.*?</td>\s*<td[^>]*>.*?</td>\s*<td[^>]*>.*?</td>\s*<td[^>]*>(\d{4}-\d{2}-\d{2})</td>", r.text, re.DOTALL ) for href, title_attr, title_text, pd in rows: title = re.sub(r'\s+', '', title_attr or title_text).strip() if len(title) < 4: continue url = f"{BASE}/{path}/{href}" results.append({'url': url, 'pub_date': pd, 'title': title}) return results def extract_balanced_div(html, open_tag_re): """平衡 div 匹配: 从匹配到的 <div...> 的 '>' 之后, 到对应 </div>, 返回内部 HTML。修复非贪婪 (.*?)</div> 在正文含嵌套 div(表格/附件区)时截断的问题""" m = re.search(open_tag_re, html) if not m: return '' gt = html.find('>', m.start()) if gt == -1: return '' i = gt + 1 depth = 1 for mm in re.finditer(r'<div[\s>]|</div>', html[i:]): if mm.group(0).startswith('<div'): depth += 1 else: depth -= 1 if depth == 0: return html[i:i + mm.start()] return html[i:] def fetch_detail(url): for _attempt in range(2): try: r = requests.get(url, timeout=(5, 15), headers=HEADERS, verify=False) r.encoding = 'utf-8' break except: r = None if r is None: return '', '' title = '' m = re.search(r'<title>\s*([^<]+?)\s*', r.text, re.DOTALL) if m: title = re.sub(r'\s+', '', m.group(1)).strip() title = re.split(r'[-–—|_]', title)[0].strip() if not title or title == '山西省生态环境厅': m = re.search(r']*>(.*?)', r.text, re.DOTALL) if m: title = re.sub(r'<[^>]+>', '', m.group(1)).strip() title = re.sub(r'\s+', '', title).strip() content = '' # 容器优先级: content-box → text-details(含表格完整正文) → trs_editor_view(嵌套正文) # 全部平衡 div 匹配, 非贪婪 (.*?) 会在表格/嵌套 div 处截断 for pat in [r']*id="content-box"[^>]*>', r']*class="[^"]*text-details[^"]*"[^>]*>', r']*class="[^"]*trs_editor_view[^"]*"[^>]*>']: c = extract_balanced_div(r.text, pat) if c.strip(): content = c.strip() break # 清理: 附件 JS / text-details 内重复 h2 标题 / 编辑时间行 content = re.sub(r'', '', content, flags=re.DOTALL | re.I) content = re.sub(r']*>.*?', '', content, flags=re.DOTALL | re.I) content = re.sub(r']*class="[^"]*td-title-2[^"]*"[^>]*>.*?

', '', content, flags=re.DOTALL | re.I) content = re.sub(r'src="(?!https?://|//)(/[^"]+)"', lambda mm: f'src="{BASE}{mm.group(1)}"', content) return title, content def run(col, max_pages=5): records = [] seen = set() for pg in range(max_pages): items = fetch_list(col, pg + 1) if not items: print(f" Page {pg+1}: 0 items (stop)") break print(f" Page {pg+1}: {len(items)} items") for it in items: if it['pub_date'] and it['pub_date'] < CUTOFF: continue if it['url'] in seen: continue seen.add(it['url']) title, content = fetch_detail(it['url']) if not title: title = it['title'] if not content.strip(): continue records.append({ 'title': title, 'url': it['url'], 'source_url': it['url'], 'pub_date': it['pub_date'], 'site_name': COLUMNS[col]['name'], 'content': content, 'summary': '', }) print(f" Total: {len(records)} items") valid = [r for r in records if r['content'].strip()] if valid: push_to_searchdb(valid, f"shx_sthjt_{col}") return len(valid) if __name__ == '__main__': ap = argparse.ArgumentParser() ap.add_argument('--col', default='xmslgs', choices=list(COLUMNS.keys())) ap.add_argument('--pages', type=int, default=5) args = ap.parse_args() cnt = run(args.col, max_pages=args.pages) print(f"Done: {cnt} records")