#!/usr/bin/env python3 """辽宁昌鑫环境工程咨询有限公司-公示信息 爬虫 (portal-saas 中企动力)""" import re, requests, sqlite3, os, sys, time from bs4 import BeautifulSoup from urllib.parse import urljoin SITE_NAME = '辽宁昌鑫-环评报告公示' BASE = 'http://www.liaoningchangxin.com' # 环评报告公示: /news/9/ → detailId=1690997738015166464, size=5 # 验收报告公示: /news/10/ → detailId?... # 企业信息公开: /news/11/ → detailId?... # 用参数可切换 DETAIL_ID = '1690997738015166464' PAGE_SIZE = 5 DB_PATH = os.environ.get('DB_PATH', '/root/search.db') 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', } SECTIONS = { '9': {'name': '辽宁昌鑫-环评报告公示', 'detail_id': '1690997738015166464', 'size': 5}, '10': {'name': '辽宁昌鑫-验收报告公示', 'detail_id': '1690997738493317120', 'size': 5}, '11': {'name': '辽宁昌鑫-企业信息公开', 'detail_id': '1690997738975662080', 'size': 5}, } def discover_section(cat_id): """从列表页发现 detailId""" url = f'{BASE}/news/{cat_id}/' try: r = requests.get(url, headers=HEADERS, timeout=30) r.encoding = 'utf-8' except Exception as e: print(f' ⚠️ 无法访问 {url}: {e}') return None, 0 # 从 _config 中提取 detailId 和 totalCount m = re.search(r'"totalCount"\s*:\s*(\d+)', r.text) total = int(m.group(1)) if m else 0 # 从分页链接提取 detailId m2 = re.search(r'/news_lsit/(\d+)-(\d+)-\d+\.html', r.text) if m2: return m2.group(1), total # 备用: 从 _config 中提取 _detailId m3 = re.search(r'_detailId[^}]*?"(\d+)"', r.text) if m3: return m3.group(1), total return None, total def get_total_pages(detail_id, size, cat_id): """从列表页获取总页数""" url = f'{BASE}/news/{cat_id}/' try: r = requests.get(url, headers=HEADERS, timeout=30) r.encoding = 'utf-8' except Exception as e: print(f' ⚠️ 请求失败 {url}: {e}') return 1 # 从分页链接中找最大页数 pages = re.findall(r'/news_lsit/\d+-(\d+)-\d+\.html', r.text) max_from = 0 for p in pages: try: f = int(p) if f > max_from: max_from = f except: pass if max_from > 0: return (max_from // size) + 1 # 从 totalCount 推算 m = re.search(r'"totalCount"\s*:\s*(\d+)', r.text) if m: total = int(m.group(1)) return (total + size - 1) // size return 1 def parse_list(page_html, base_url): """解析列表页，提取条目""" soup = BeautifulSoup(page_html, 'lxml') items = [] for item in soup.select('div.p_loopitem'): title_tag = item.select_one('.e_text-11 a') if not title_tag: continue href = title_tag.get('href', '') if not href.startswith('http'): href = urljoin(base_url, href) title = title_tag.get_text(strip=True) # 日期: day, month, year 分别在不同标签 day_el = item.select_one('.e_timeFormat-17') month_el = item.select_one('.e_timeFormat-20') year_el = item.select_one('.e_timeFormat-21') day = day_el.get_text(strip=True).zfill(2) if day_el else '01' month = month_el.get_text(strip=True).zfill(2) if month_el else '01' year = year_el.get_text(strip=True) if year_el else '2020' date = f'{year}-{month}-{day}' items.append({ 'title': title, 'url': href, 'date': date, }) return items def parse_detail(url): """解析详情页正文""" try: r = requests.get(url, headers=HEADERS, timeout=30) r.encoding = 'utf-8' soup = BeautifulSoup(r.text, 'lxml') # portal-saas 详情页内容: e_richText-{n} s_title (类名含编号后缀) div = soup.find('div', class_=lambda c: c and 'e_richText' in str(c)) if div: for tag in div(['script', 'style', 'iframe']): tag.decompose() html = str(div) html = re.sub(r'\s*style="[^"]*"', '', html) html = re.sub(r'\s*class="[^"]*"', '', html) return html for selector in ['.p_articles', '#c_detail_desc', '.e_text-1', '.detail-content']: div = soup.select_one(selector) if div: for tag in div(['script', 'style', 'iframe']): tag.decompose() html = str(div) html = re.sub(r'\s*style="[^"]*"', '', html) html = re.sub(r'\s*class="[^"]*"', '', html) return html # 备用: 找内容区 div = soup.find('div', class_=lambda c: c and ('content' in str(c).lower() or 'detail' in str(c).lower() or 'article' in str(c).lower())) if div: for tag in div(['script', 'style', 'iframe']): tag.decompose() html = str(div) html = re.sub(r'\s*style="[^"]*"', '', html) html = re.sub(r'\s*class="[^"]*"', '', html) return html except Exception as e: print(f' ⚠️ 详情解析失败 {url}: {e}') return '' def run(cat_id='9', max_pages=None): if cat_id not in SECTIONS: print(f'❌ 未知分类: {cat_id}, 可选: {", ".join(SECTIONS.keys())}') return site_name = SECTIONS[cat_id]['name'] size = SECTIONS[cat_id]['size'] # 发现 detailId detail_id, _ = discover_section(cat_id) if not detail_id: print(f'❌ 无法从 {cat_id} 发现 detailId') return print(f'📋 {site_name} (detailId={detail_id})') # 从首页分页链接获取真实总页数 total_pages = get_total_pages(detail_id, size, cat_id) if max_pages and max_pages < total_pages: total_pages = max_pages print(f'📋 共 {total_pages} 页') conn = sqlite3.connect(DB_PATH) new, skip = 0, 0 for pg in range(1, total_pages + 1): from_val = (pg - 1) * size list_url = f'{BASE}/news_lsit/{detail_id}-{from_val}-{size}.html' try: r = requests.get(list_url, headers=HEADERS, timeout=30) r.encoding = 'utf-8' except Exception as e: print(f' ⚠️ 第{pg}页请求失败: {e}') continue items = parse_list(r.text, list_url) if not items: print(f' ⚠️ 第{pg}页无数据') continue print(f'📄 第{pg}/{total_pages}页 ({len(items)}条)') for item in items: body = parse_detail(item['url']) try: conn.execute(""" INSERT OR IGNORE INTO gov_raw (site_name, source_url, page_url, title, publish_date, content, summary) VALUES (?, ?, ?, ?, ?, ?, ?) """, (site_name, item['url'], item['url'], item['title'], item['date'], body, item['title'])) if conn.total_changes > 0: new += 1 else: skip += 1 except Exception as e: print(f' ❌ {item["title"][:30]} - {e}') time.sleep(0.3) conn.commit() conn.close() print(f'\n✅ {site_name}: 新增{new}, 跳过{skip}, 共{total_pages}页') if __name__ == '__main__': cat = sys.argv[1] if len(sys.argv) > 1 else 'all' max_p = int(sys.argv[2]) if len(sys.argv) > 2 and sys.argv[2].isdigit() else None if cat == 'all': for c in SECTIONS: name = SECTIONS[c]['name'] print(f'\n===== {name} =====') run(c, max_p) else: run(cat, max_p)