#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
crawl_jslygc_hpgs.py - 江苏绿源-环评公示
http://www.jslygc.com/list_info.aspx?cid=74 (iframe: list_news_iframe.aspx?cid=74)
列表: ASP.NET WebForms, #news_list ul li > a[href=show.aspx?id=N] + 日期
分页: POST __EVENTTARGET=AspNetPager1&__EVENTARGUMENT=N (需携带最新 VIEWSTATE + Cookie)
详情: show.aspx?id=N → h1(完整标题) + #txt1(正文HTML, 保留表格) + #fj(附件)
注意: 列表标题被截断(无省略号), 完整标题必须从详情页 h1 提取
"""
import sys
import os
import re
import time
import html as html_lib
import urllib.parse
import sqlite3
import json
try:
from bs4 import BeautifulSoup
except ImportError:
BeautifulSoup = None
# ---------------- config ----------------
BASE_URL = "http://www.jslygc.com"
LIST_URL = "http://www.jslygc.com/list_news_iframe.aspx?cid=74"
SITE_NAME = "江苏绿源-环评公示"
GROUP_NAME = "江苏"
SCRIPT_NAME = "crawl_jslygc_hpgs.py"
DB_PATH = os.environ.get("DB_PATH", "/root/search.db")
JSONL_PATH = os.environ.get("JSONL_PATH", "")
ENC = "utf-8"
TOTAL_PAGES = 57 # 734 条 / 13 条每页
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.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",
}
_MAX_PAGES = 1
for i, a in enumerate(sys.argv):
if a.startswith("--pages="):
try:
_MAX_PAGES = int(a.split("=", 1)[1])
except ValueError:
pass
elif a == "--pages" and i + 1 < len(sys.argv):
try:
_MAX_PAGES = int(sys.argv[i + 1])
except ValueError:
pass
_PAGES = _MAX_PAGES if _MAX_PAGES >= 1 else 1
def clean_title(t):
if not t:
return ""
t = html_lib.unescape(t)
t = t.replace("\u200b", "").replace("\u200c", "").replace("\u200d", "").replace("\ufeff", "")
t = re.sub(r"\s+", " ", t)
return t.strip()
def http_get(session, url, retries=3, timeout=30):
"""GET with retry (站点间歇性超时)"""
for i in range(retries):
try:
r = session.get(url, headers=HEADERS, timeout=timeout)
if r.status_code == 200:
return r
except Exception:
pass
time.sleep(2 + i * 2)
return None
def http_post(session, url, data, retries=3, timeout=30):
for i in range(retries):
try:
r = session.post(url, data=data, headers=HEADERS, timeout=timeout)
if r.status_code == 200:
return r
except Exception:
pass
time.sleep(2 + i * 2)
return None
def fetch_list(session, page, last_vs="", last_vg="0C85029D"):
"""获取列表页. page=1 GET; page>1 POST AspNetPager 回发"""
if page == 1:
r = http_get(session, LIST_URL)
if r is None:
return None, "", ""
else:
data = {
"__VIEWSTATE": last_vs,
"__VIEWSTATEGENERATOR": last_vg,
"__EVENTTARGET": "AspNetPager1",
"__EVENTARGUMENT": str(page),
}
r = http_post(session, LIST_URL, data=data)
if r is None:
return None, "", ""
text = r.content.decode(ENC, errors="replace") if ENC == "utf-8" else r.text
vs = re.search(r'id="__VIEWSTATE" value="([^"]*)"', text)
vg = re.search(r'id="__VIEWSTATEGENERATOR" value="([^"]*)"', text)
return text, (vs.group(1) if vs else last_vs), (vg.group(1) if vg else last_vg)
def parse_list(html_text):
"""#news_list ul li: 标题 2025-07-09"""
items = []
for m in re.finditer(r']*href="show\.aspx\?id=(\d+)"[^>]*>(.*?)\s*([0-9]{4}-[0-9]{2}-[0-9]{2})', html_text, re.S | re.I):
nid = m.group(1)
title = clean_title(m.group(2))
date = m.group(3)
if not title or len(title) < 4:
continue
abs_url = f"{BASE_URL}/show.aspx?id={nid}"
items.append((abs_url, title, date))
return items
def parse_detail(html_text, page_url):
"""h1 完整标题 + #txt1 正文HTML + #fj 附件列表"""
title = ""
date = ""
content_html = ""
attachments = []
if BeautifulSoup:
soup = BeautifulSoup(html_text, "html.parser")
h1 = soup.find("h1")
if h1:
title = clean_title(h1.get_text(" ", strip=True))
# 正文 #txt1（保留表格 HTML 原文）
txt1 = soup.find("div", id="txt1")
if txt1:
content_html = "".join(str(c) for c in txt1.contents)
# 附件区 #fj:
fj = soup.find("div", id="fj")
if fj:
for a in fj.find_all("a", href=True):
href = a["href"]
txt = clean_title(a.get_text(" ", strip=True)) or os.path.basename(href.split("?")[0])
abs_url = urllib.parse.urljoin(page_url, href)
attachments.append((abs_url, txt))
# 正文内附件链接也抓
if txt1:
for a in txt1.find_all("a", href=True):
href = a["href"]
if re.search(r"\.(pdf|docx?|xlsx?|zip|rar|txt)", href, re.I):
txt = clean_title(a.get_text(" ", strip=True)) or os.path.basename(href.split("?")[0])
abs_url = urllib.parse.urljoin(page_url, href)
if abs_url not in [x[0] for x in attachments]:
attachments.append((abs_url, txt))
# 日期: 优先 h2 里的 "日期：YYYY-MM-DD" (不全局匹配, 避免顶栏今天日期)
if not date:
h2 = soup.find("h2")
if h2:
dm = re.search(r"日期[：:]\s*(\d{4}-\d{2}-\d{2})", h2.get_text())
if dm:
date = dm.group(1)
return title, date, content_html, attachments
def html_to_text(content, page_url):
"""清理正文, 保留表格 HTML, 附件链接转绝对 URL 段落"""
if not content:
return "", 0, []
content = re.sub(r"", "", content, flags=re.S)
content = re.sub(r"]*>", "", content, flags=re.I)
content = re.sub(r"]*>.*?", "", content, flags=re.S | re.I)
content = re.sub(r"]*>.*?", "", content, flags=re.S | re.I)
attachments = []
link_protect = {}
if BeautifulSoup:
try:
_soup = BeautifulSoup(content, "html.parser")
for a in _soup.find_all("a", href=True):
if re.search(r"\.(pdf|docx?|xlsx?|zip|rar|txt)", a.get("href", ""), re.I):
href = a["href"]
txt = a.get_text(strip=True) or os.path.basename(href.split("?")[0])
abs_url = urllib.parse.urljoin(page_url, href)
attachments.append((abs_url, txt))
key = f"__ATTACH__{len(link_protect)}__"
link_protect[key] = f'{txt}
'
a.replace_with(key)
for p in _soup.find_all("p"):
imgs = p.find_all("img")
if imgs and not p.get_text(strip=True):
srcs = []
for img in imgs:
src = img.get("src") or ""
if src:
abs_url = urllib.parse.urljoin(page_url, src)
srcs.append(f'{os.path.basename(src.split("?")[0]) or "图片"}')
if srcs:
key = f"__IMG__{len(link_protect)}__"
link_protect[key] = "" + "
".join(srcs) + "
"
p.replace_with(key)
content = str(_soup)
except Exception:
pass
table_protect = []
def _tbl_repl(m):
table_protect.append(m.group(0))
return f"__TBL__{len(table_protect)-1}__"
content = re.sub(r"]*>.*?", _tbl_repl, content, flags=re.S | re.I)
has_table = 1 if re.search(r"]*>", content, re.I) else 0
if BeautifulSoup:
try:
_soup = BeautifulSoup(content, "html.parser")
for tag in _soup.find_all(True):
for attr in ("style", "class", "lang", "dir", "align", "valign", "width", "height", "border", "cellpadding", "cellspacing"):
tag.attrs.pop(attr, None)
content = str(_soup)
except Exception:
pass
content = re.sub(r"", "\n\n", content, flags=re.I)
content = re.sub(r"", "\n", content, flags=re.I)
content = re.sub(r"]*>", "", content, flags=re.I)
parts = []
for block in content.split("\n\n"):
b = block.strip()
if not b:
continue
plain = re.sub(r"<[^>]+>", "", b)
plain = html_lib.unescape(plain)
plain = plain.replace("\xa0", " ").replace(" ", " ")
plain = re.sub(r"[\s\u200b\u200c\u200d\ufeff]+", "", plain)
if not plain:
continue
for k, v in link_protect.items():
b = b.replace(k, v)
for i, tbl in enumerate(table_protect):
b = b.replace(f"__TBL__{i}__", tbl)
parts.append(b.strip())
out = "\n\n".join(parts)
for k, v in link_protect.items():
out = out.replace(k, v)
for i, tbl in enumerate(table_protect):
out = out.replace(f"__TBL__{i}__", tbl)
out = re.sub(r"__LINK__\d+__", "", out)
out = re.sub(r"__TBL__\d+__", "", out)
out = re.sub(r"__IMG__\d+__", "", out)
out = re.sub(r"__ATTACH__\d+__", "", out)
out = html_lib.unescape(out)
out = re.sub(r"[ \t]+\n", "\n", out)
out = re.sub(r"\n{3,}", "\n\n", out)
out = re.sub(r"[ \t]{2,}", " ", out)
seen = set()
final = []
for p in out.split("\n\n"):
key = re.sub(r"\s+", "", re.sub(r"<[^>]+>", "", p))
if not key:
continue
if key in seen:
continue
seen.add(key)
final.append(p)
out = "\n\n".join(final).strip()
return out, has_table, attachments
def append_attachments(content, attachments):
"""附件区转 绝对URL
段落, 不保留 ul/li 包装"""
if not attachments:
return content
parts = [content] if content.strip() else []
for abs_url, txt in attachments:
parts.append(f'{txt}
')
return "\n\n".join(parts)
def load_existing_urls(conn):
"""加载已入库 page_url 集合用于列表级预查重"""
try:
cur = conn.execute("SELECT page_url FROM gov_raw WHERE site_name=?", (SITE_NAME,))
return {row[0] for row in cur.fetchall()}
except Exception:
return set()
def store_record(conn, url, title, content, date, has_table):
cur = conn.execute("SELECT id FROM gov_raw WHERE page_url=? AND site_name=?", (url, SITE_NAME))
if cur.fetchone():
return False
try:
industry = "other"
try:
from crawler_lib import classify_industry
industry = classify_industry(title)
except Exception:
pass
cur = conn.execute(
"INSERT OR IGNORE INTO gov_raw "
"(page_url, source_url, title, content, publish_date, site_name, group_name, script_name, has_table, industry) "
"VALUES (?,?,?,?,?,?,?,?,?,?)",
(url, url, title, content, date, SITE_NAME, GROUP_NAME, SCRIPT_NAME, has_table, industry),
)
if cur.rowcount > 0:
summary = re.sub(r"<[^>]+>", "", content)
summary = html_lib.unescape(summary)
summary = re.sub(r"\s+", " ", summary).strip()[:200]
conn.execute(
"INSERT OR IGNORE INTO gov_search (rowid, title, site_name, summary) VALUES (?,?,?,?)",
(cur.lastrowid, title, SITE_NAME, summary),
)
return True
return False
except sqlite3.OperationalError as e:
if "locked" in str(e):
time.sleep(3)
return store_record(conn, url, title, content, date, has_table)
raise
def main():
import requests
s = requests.Session()
s.headers.update(HEADERS)
conn = None
if not JSONL_PATH:
conn = sqlite3.connect(DB_PATH, timeout=60)
conn.execute("PRAGMA busy_timeout=60000")
conn.execute("PRAGMA journal_mode=WAL")
existing = load_existing_urls(conn)
print(f"已入库 {len(existing)} 条, 用于列表级预查重", flush=True)
new_count = 0
skip_count = 0
jsonl_rows = []
last_vs = ""
last_vg = "0C85029D"
for page in range(1, _PAGES + 1):
try:
html_text, last_vs, last_vg = fetch_list(s, page, last_vs, last_vg)
if html_text is None:
print(f"Page {page} fetch error (all retries)", flush=True)
break
except Exception as e:
print(f"Page {page} fetch error: {e}", flush=True)
break
items = parse_list(html_text)
print(f"Page {page}: found {len(items)} items", flush=True)
if not items:
print("Empty page, stop pagination", flush=True)
break
for abs_url, title, date in items:
if not JSONL_PATH and abs_url in existing:
skip_count += 1
continue # 列表级预查重: 已入库跳过详情
try:
dhtml_r = http_get(s, abs_url)
if dhtml_r is None:
print(f" detail error (all retries): {abs_url}", flush=True)
skip_count += 1
continue
dhtml = dhtml_r.content.decode(ENC, errors="replace")
except Exception as e:
print(f" detail error {abs_url}: {e}", flush=True)
skip_count += 1
continue
dtitle, ddate, content_html, atts = parse_detail(dhtml, abs_url)
if not dtitle:
dtitle = title # fallback 列表标题(截断)
if not ddate:
ddate = date
content, has_table, atts_in = html_to_text(content_html, abs_url)
all_atts = atts + atts_in if atts_in else atts
content = append_attachments(content, all_atts)
plain_len = len(re.sub(r"<[^>]+>", "", content).strip())
if plain_len < 10 and not content_html:
print(f" skip empty: {dtitle[:30]}", flush=True)
skip_count += 1
continue
if JSONL_PATH:
jsonl_rows.append({
"page_url": abs_url, "title": dtitle, "content": content,
"publish_date": ddate, "has_table": has_table,
"site_name": SITE_NAME, "group_name": GROUP_NAME,
"script_name": SCRIPT_NAME,
})
print(f" JSONL + {dtitle[:40]} ({ddate})", flush=True)
new_count += 1
continue
ok = store_record(conn, abs_url, dtitle, content, ddate, has_table)
if ok:
new_count += 1
existing.add(abs_url)
print(f" + {dtitle[:40]} ({ddate})", flush=True)
else:
skip_count += 1
time.sleep(0.3)
time.sleep(0.5)
if JSONL_PATH:
with open(JSONL_PATH, "w", encoding="utf-8") as f:
for row in jsonl_rows:
f.write(json.dumps(row, ensure_ascii=False) + "\n")
print(f"JSONL written: {JSONL_PATH} ({len(jsonl_rows)} rows)", flush=True)
if conn is not None:
conn.commit()
conn.close()
print(f"DONE: new={new_count} skip={skip_count}", flush=True)
return 0 if new_count > 0 or skip_count > 0 else 1
if __name__ == "__main__":
sys.exit(main())