...Word转换逐段 ... |
无 meta ArticleTitle/PubDate — 标题取 前缀或 strong 区, 日期取列表页 font 日期
无独立附件区 (环评公示纯文本, 无 <a href> 下载链接)
"""
import sys
import os
import re
import time
import html as html_lib
import urllib.parse
import sqlite3
try:
from bs4 import BeautifulSoup
except ImportError:
BeautifulSoup = None
# ---------------- config ----------------
BASE_URL = "http://www.jxjrxhb.com"
LIST_PATH = "/index.php/news/hjkx/1/cn/15410/15410.html"
LIST_URL = BASE_URL + LIST_PATH
SITE_NAME = "江西景瑞祥环保-项目公告"
GROUP_NAME = "企业"
SCRIPT_NAME = "crawl_jxjrxhb_tzgg.py"
DB_PATH = "/mnt/data/search.db"
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",
}
# default: daily incremental = 1 page (single static page, no real pagination)
_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
# ---------------- helpers ----------------
def clean_title(t):
"""unescape entities + strip zero-width + collapse whitespace."""
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 fetch(url, session=None, timeout=30, retries=3):
import requests
last = None
for i in range(retries):
try:
r = (session or requests).get(url, headers=HEADERS, timeout=timeout)
if r.status_code == 200:
r.encoding = r.apparent_encoding or "utf-8"
return r
last = r
except Exception as e:
last = e
time.sleep(1.5 * (i + 1))
if isinstance(last, Exception):
raise last
return last
def html_to_text(content, page_url):
"""Convert article-body inner HTML to stored format:
- keep <table> HTML intact
- \n\n between <p> blocks
- attachment <a> links embedded with absolute URL
- strip <script>/<style>/noise + style attrs via BS4
"""
if not content:
return "", 0, []
content = re.sub(r"<!--.*?-->", "", content, flags=re.S)
content = re.sub(r"<script[^>]*>.*?</script>", "", content, flags=re.S | re.I)
content = re.sub(r"<style[^>]*>.*?</style>", "", content, flags=re.S | re.I)
# strip style/class attributes via BS4 BEFORE paragraph splitting (合浦 hepu 教训)
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
# 1) FIRST protect attachment <a> links (before table protection, so table
# regex can't swallow links that sit right after a <table> close tag)
content2 = content
# remove attachment-file-type icon imgs (icon_doc.gif / icon_xls.gif etc.)
content2 = re.sub(
r'<img[^>]*src="[^"]*fileTypeImages/icon_[a-z]+\.gif"[^>]*/?>', "", content2,
flags=re.I,
)
attachments = []
link_protect = {}
def _link_repl(m):
href = m.group(1)
inner = m.group(2)
txt = re.sub(r"<[^>]+>", "", inner)
txt = html_lib.unescape(txt).strip()
txt = re.sub(r"^附件[:：]\s*", "", txt).strip()
if not txt:
txt = os.path.basename(href.split("?")[0]) or "附件"
abs_url = urllib.parse.urljoin(page_url, href)
attachments.append((abs_url, txt))
key = f"__LINK__{len(link_protect)}__"
link_protect[key] = f'<a href="{abs_url}" target="_blank">{txt}</a>'
return key
content2 = re.sub(r'<a\s[^>]*href="([^"]+)"[^>]*>(.*?)</a>', _link_repl, content2, flags=re.S | re.I)
# 2) THEN protect tables
table_protect = []
def _tbl_repl(m):
table_protect.append(m.group(0))
return f"__TBL__{len(table_protect)-1}__"
content2 = re.sub(r"<table[^>]*>.*?</table>", _tbl_repl, content2, flags=re.S | re.I)
has_table = 1 if re.search(r"<table[^>]*>", content, re.I) else 0
content2 = re.sub(r"</p>", "</p>\n\n", content2, flags=re.I)
content2 = re.sub(r"<br\s*/?>", "\n", content2, flags=re.I)
content2 = re.sub(r"</?(?:span|font|o:p|st1?:[a-z]+)\b[^>]*>", "", content2, flags=re.I)
parts = []
for block in content2.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 = 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)
# dedupe paragraphs
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 parse_list(html_text, page_url):
"""Extract (url, title, date) from list page.
zhasp list: <td style="border-bottom:#D7E5F2 solid 1px"><img src="/tp/newslist0021_box/alloon2.gif" width="11" height="16" align="absmiddle" /> <a href="/userlist/hjkx/newshow-18202.html" target="_blank">标题</a> <font color="#89CAFE">2026-07-22</font></td>
"""
items = []
if BeautifulSoup:
soup = BeautifulSoup(html_text, "html.parser")
for a in soup.find_all("a", href=True):
href = a.get("href", "").strip()
if not re.search(r"/userlist/hjkx/newshow-\d+\.html$", href):
continue
abs_url = urllib.parse.urljoin(LIST_URL, href)
title = clean_title(a.get_text(" ", strip=True))
# date: <font color="#89CAFE">YYYY-MM-DD</font> right after the </a>
date = ""
nxt = a.find_next_sibling()
if nxt is not None and nxt.name == "font":
date = clean_title(nxt.get_text(" ", strip=True))
if not re.match(r"^20\d{2}-\d{2}-\d{2}$", date):
# fallback: search in parent td
parent = a.find_parent("td")
if parent:
fm = re.search(r"(20\d{2}-\d{2}-\d{2})", parent.get_text(" ", strip=True))
if fm:
date = fm.group(1)
items.append((abs_url, title, date))
if not items:
# regex fallback: <a href="/userlist/hjkx/newshow-18202.html"...>标题</a> <font color="#89CAFE">2026-07-22</font>
for m in re.finditer(
r'<a[^>]+href="(/userlist/hjkx/newshow-\d+\.html)"[^>]*>(.*?)</a>\s* \s* \s*<font[^>]*>([^<]*)</font>',
html_text, re.S,
):
abs_url = urllib.parse.urljoin(LIST_URL, m.group(1))
items.append((abs_url, clean_title(m.group(2)), m.group(3).strip()))
return items
def parse_detail(html_text, page_url):
"""Extract title, date, content from detail page.
zhasp detail: 标题区 <font color="#1B8CD1"><strong>标题</strong></font>
正文: <table width="99%" height="25"...><tr><td><P align=center><B><FONT face=宋体>...</FONT></B></P>...</span></td>
无 meta — title 从 <title>前缀 或 strong 区, date 从列表页.
"""
title = ""
content_html = ""
if BeautifulSoup:
soup = BeautifulSoup(html_text, "html.parser")
# <title>完整标题 - 项目公告 - 公司名
mt = soup.find("title")
if mt:
t = mt.get_text(" ", strip=True)
if " - " in t:
t = t.split(" - ")[0]
title = clean_title(t)
# fallback: strong title area
if not title:
st = soup.find("font", attrs={"color": "#1B8CD1"})
if st:
strong = st.find("strong")
if strong:
title = clean_title(strong.get_text(" ", strip=True))
# content: table width=99% height=25 ... ... | (Word converted )
tbl = None
for tb in soup.find_all("table"):
w = tb.get("width", "")
h = tb.get("height", "")
if w == "99%" and h == "25":
tbl = tb
break
if tbl:
td = tbl.find("td")
if td:
# decode_contents() to drop the outer
wrapper (邗江 hj 教训)
content_html = td.decode_contents()
if not title or not content_html:
# regex fallback: 标题
m = re.search(r'(.*?)', html_text, re.S)
if m:
title = clean_title(m.group(1))
if not title:
m = re.search(r"([^<]*?)\s*-\s*项目公告", html_text, re.S)
if m:
title = clean_title(m.group(1))
if not content_html:
m = re.search(r'<table width="99%" height="25"[^>]*>\s*<tr>\s*<td>(.*?)</span>\s*</td>', html_text, re.S)
if m:
content_html = m.group(1) + "</span>"
return title, content_html
def store_record(conn, url, title, content, date, has_table):
cur = conn.execute(
"SELECT id FROM gov_raw WHERE page_url=?",
(url,),
)
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 = sqlite3.connect(DB_PATH, timeout=60)
conn.execute("PRAGMA busy_timeout=60000")
conn.execute("PRAGMA journal_mode=WAL")
new_count = 0
skip_count = 0
seen_urls = set()
# zhasp: single static list page — no real pagination; page > 1 returns the
# same content, so dedup via seen_urls makes extra pages harmless.
for page in range(1, _PAGES + 1):
url = LIST_URL
try:
r = fetch(url, session=s)
html_text = r.content.decode("utf-8", errors="replace")
except Exception as e:
print(f"Page {page} fetch error: {e}", flush=True)
break
items = parse_list(html_text, url)
print(f"Page {page}: found {len(items)} items", flush=True)
if not items and page > 1:
print("Empty page, stop pagination", flush=True)
break
for abs_url, title, date in items:
if abs_url in seen_urls:
continue
seen_urls.add(abs_url)
try:
r = fetch(abs_url, session=s)
dhtml = r.content.decode("utf-8", errors="replace")
except Exception as e:
print(f" detail error {abs_url}: {e}", flush=True)
skip_count += 1
continue
dtitle, content_html = parse_detail(dhtml, abs_url)
if not dtitle:
dtitle = title
content, has_table, _ = html_to_text(content_html, abs_url)
plain_len = len(re.sub(r"<[^>]+>", "", content).strip())
if plain_len < 10:
print(f" skip empty: {dtitle[:30]}", flush=True)
skip_count += 1
continue
ok = store_record(conn, abs_url, dtitle, content, date, has_table)
if ok:
new_count += 1
print(f" + {dtitle[:40]} ({date})", flush=True)
else:
skip_count += 1
time.sleep(0.3)
time.sleep(0.5)
conn.commit()
conn.close()
print(f"新增: {new_count}", flush=True)
print(f"跳过: {skip_count}", flush=True)
print(f"=== {SITE_NAME} done ===", flush=True)
if __name__ == "__main__":
main()
|