去后缀）日期: div.label 「日期：YYYY年MM月DD日」正文: div#vsb_content_500（内含 div.v_news_content，保留 <p>/<table>） """ import sys, os, re, time, json, html as html_lib, urllib.parse from datetime import datetime, timedelta try: from bs4 import BeautifulSoup except ImportError: BeautifulSoup = None JSONL_PATH = os.environ.get("JSONL_PATH", "") ENC = "utf-8" SITE_NAME = "武定县-项目环评审批结果" GROUP_NAME = "云南" SCRIPT_NAME = "crawl_ynwd.py" CATEGORY = "项目环评审批结果" BASE_URL = "https://ynwd.gov.cn" LIST_URL = BASE_URL + "/zfxxgk/fdzdgknr/zdmsjgysyjsxx/sthj/xmhpspjg.htm" CUTOFF = (datetime.now() - timedelta(days=365 * 3)).strftime("%Y-%m-%d") 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", "Referer": LIST_URL, } _MAX_PAGES = 5 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"<[^>]+>", "", t) t = re.sub(r"^\s*(?:·|·|\u00b7)?\s*(?: |\u00a0)?\s*", "", t) t = re.sub(r"^[•·]\s*", "", t) t = re.sub(r"\s+", " ", t) return t.strip() def fetch(url, timeout=30): h = dict(HEADERS) last_err = None for attempt in range(3): try: import requests, urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) r = requests.get(url, headers=h, timeout=timeout, verify=False) if r.status_code == 200: r.encoding = ENC return r.text last_err = "HTTP %s" % r.status_code except Exception as e: last_err = str(e)[:80] time.sleep(2) print(" [fetch fail] %s -> %s" % (url[:90], last_err), flush=True) return None def html_to_text(content, page_url): """正文 HTML 清洗：保留 <p>/<table>，附件转 <p><a>绝对URL</a></p> 段落""" if not content: return "", 0, [] content = re.sub(r"", "", content, flags=re.S) content = re.sub(r"<meta[^>]*>", "", content, flags=re.I) content = re.sub(r"<script[^>]*>.*?</script>", "", content, flags=re.S | re.I) content = re.sub(r"<style[^>]*>.*?</style>", "", content, flags=re.S | re.I) attachments = [] link_protect = {} content = re.sub(r"【字号：[^】]*】", "", content) if BeautifulSoup: try: _soup = BeautifulSoup(content, "html.parser") for a in _soup.find_all("a", href=True): href = a["href"] if (a.get("appendix") or a.get("data-appendix") or re.search(r"\.(pdf|docx?|xlsx?|zip|rar|wps|et|ofd)", href, re.I) or "/module/download/" in href): # TRS 站附件: href="./P020...xlsx" 可用; oldsrc="/protect/..." 是 CMS 内部路径(直访 404) → 优先 href _href = (href or "").strip() _oldsrc = (a.get("oldsrc") or a.get("OLDSRC") or "").strip() if _href and not re.match(r"^(?:javascript:|mailto:|tel:|#)", _href, re.I): real_href = _href else: real_href = _oldsrc or _href txt = (a.get_text(strip=True) or a.get("_title") or a.get("title") or os.path.basename(real_href.split("?")[0]) or "附件") abs_url = urllib.parse.urljoin(page_url, real_href) attachments.append((abs_url, txt)) key = "__ATTACH__%d__" % len(link_protect) link_protect[key] = '<a href="%s" target="_blank">%s</a>' % (abs_url, txt) a.replace_with(key) for img in _soup.find_all("img"): src = (img.get("src") or img.get("oldsrc") or "") if re.search(r"(icons?/|fileTypeImages/icon_)", src, re.I): img.decompose() 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 img.get("oldsrc") or "" if not src: continue abs_url = urllib.parse.urljoin(page_url, src) txt = (img.get("alt") or img.get("title") or "").strip() \ or os.path.basename(src.split("?")[0]) or "图片" srcs.append('<a href="%s" target="_blank">%s</a>' % (abs_url, txt)) if srcs: key = "__IMG__%d__" % len(link_protect) link_protect[key] = "<p>" + "<br/>".join(srcs) + "</p>" p.replace_with(key) for img in _soup.find_all("img"): src = img.get("src") or img.get("oldsrc") or "" if not src: continue abs_url = urllib.parse.urljoin(page_url, src) txt = (img.get("alt") or img.get("title") or "").strip() \ or os.path.basename(src.split("?")[0]) or "图片" key = "__IMG__%d__" % len(link_protect) link_protect[key] = '<p><a href="%s" target="_blank">%s</a></p>' % (abs_url, txt) img.replace_with(key) content = str(_soup) except Exception: pass def _link_repl(m): href = m.group(1) if re.match(r"^(?:javascript:|mailto:|tel:|s?ftp:|#)", href, re.I): return "" if href.lower().startswith("javascript:") else m.group(0) 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 = "__LINK__%d__" % len(link_protect) link_protect[key] = '<a href="%s" target="_blank">%s</a>' % (abs_url, txt) return key content = re.sub(r'<a\s[^>]*href="([^"]+)"[^>]*>(.*?)</a>', _link_repl, content, flags=re.S | re.I) 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) for d in _soup.find_all("div"): if not d.get_text(strip=True) and not d.find("table") \ and not d.find("img") and not d.find("a"): d.decompose() else: d.unwrap() content = str(_soup) except Exception: pass table_protect = [] def _tbl_repl(m): table_protect.append(m.group(0)) return "__TBL__%d__" % (len(table_protect) - 1) has_table = 1 if re.search(r"<table[^>]*>", content, re.I) else 0 prev = None while prev != content: prev = content content = re.sub(r"<table(?:(?!<table).)*?</table>", _tbl_repl, content, flags=re.S | re.I) content = re.sub(r"</p>", "</p>\n\n", content, flags=re.I) content = re.sub(r"<br\s*/?>", "\n", content, flags=re.I) content = re.sub(r"</?(?:span|font|o:p|st1?:[a-z]+|strong|em|b|i|u)\b[^>]*>", "", content, flags=re.I) parts = [] for block in content.split("\n\n"): b = block.strip() if not b: continue b = re.sub(r"(?<=>)\s*[\r\n\t]+\s*", "", b) b = re.sub(r"\s*[\r\n\t]+\s*(?=<)", "", b) b = b.replace("\r", "").replace("\t", " ") b = re.sub(r"[ \t]{2,}", " ", b) 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 # UI button blocks (back-to-top / print / close / share) -> drop if re.match(r"^[【\[（(]?\s*(?:返回顶部|回到顶部|返回列表|打印(?:本页)?|关闭(?:窗口)?|分享|扫一扫|上一篇|下一篇)\s*[】\]）)]?$", plain): continue for k, v in link_protect.items(): b = b.replace(k, v) for i, tbl in enumerate(table_protect): b = b.replace("__TBL__%d__" % i, tbl) if not re.search(r"<p[ >]", b, re.I) and re.search(r"<a[ >]", b, re.I): b = "<p>%s</p>" % b 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("__TBL__%d__" % i, tbl) out = re.sub(r"__(?:LINK|TBL|IMG|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, final = set(), [] for p in out.split("\n\n"): key = re.sub(r"\s+", "", re.sub(r"<[^>]+>", "", p)) if not key or key in seen: continue seen.add(key) final.append(p) out = "\n\n".join(final).strip() dedup, seen_u = [], set() for u, t in attachments: if u in seen_u: continue seen_u.add(u) dedup.append((u, t)) return out, has_table, dedup def parse_list(html_text, page_url): """返回 (items, next_page_url)""" items = [] next_url = None if not BeautifulSoup: return items, next_url soup = BeautifulSoup(html_text, "html.parser") for li in soup.find_all("li", id=re.compile(r"^line_u6_")): a = li.find("a", href=True) if not a: continue href = (a.get("href") or "").strip() title = clean_title(a.get("title") or "") if not title: title = clean_title(a.get_text(" ", strip=True)) if not title or len(title) < 2: continue if "/info/" not in href: continue abs_url = urllib.parse.urljoin(page_url, href) date = "" sp = li.find("span") if sp: dm = re.search(r"(\d{4})[年/-](\d{1,2})[月/-](\d{1,2})", sp.get_text(strip=True)) if dm: date = "%s-%s-%s" % (dm.group(1), dm.group(2).zfill(2), dm.group(3).zfill(2)) items.append((abs_url, title, date)) # 下页链接 nxt = soup.find("span", class_=re.compile(r"p_next")) if nxt: a = nxt.find("a", href=True) if a and a.get("href") and not a["href"].lower().startswith("javascript"): next_url = urllib.parse.urljoin(page_url, a["href"]) return items, next_url def parse_detail(html_text, page_url): title = date = content_html = "" soup = None if BeautifulSoup: try: soup = BeautifulSoup(html_text, "html.parser") mt = soup.find("meta", attrs={"name": re.compile(r"ArticleTitle", re.I)}) if mt and mt.get("content"): title = clean_title(mt["content"]) if not title: cw = soup.select_one("div.contentwarp") h1 = cw.find("h1") if cw else soup.find("h1") if h1: title = clean_title(h1.get_text(strip=True)) lab = soup.find("div", class_="label") if lab: dm = re.search(r"(\d{4})[年/-](\d{1,2})[月/-](\d{1,2})", lab.get_text(" ")) if dm: date = "%s-%s-%s" % (dm.group(1), dm.group(2).zfill(2), dm.group(3).zfill(2)) el = soup.find(id="vsb_content_500") or soup.select_one("div.v_news_content") if el: content_html = "".join(str(c) for c in el.contents) except Exception: pass if not title: m = re.search(r"<title>(.*?)

#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ crawl_ynwd.py — 武定县（楚雄）项目环评审批结果 https://ynwd.gov.cn/zfxxgk/fdzdgknr/zdmsjgysyjsxx/sthj/xmhpspjg.htm CMS: 拓尔思(TRS) 政务公开平台（静态 .htm，静态分页）列表:

标题 2026年09月08日

20 条/页分页: 静态「下页」链接（数字倒序），当前第 1 页 = xmhpspjg.htm，第 2 页 = xmhpspjg/7.htm …（共 151 条 / 8 页）→ 逐页跟随「下页」链接详情: /info/2554/.htm 标题: div.contentwarp > h1（回退去后缀）日期: div.label 「日期：YYYY年MM月DD日」正文: div#vsb_content_500（内含 div.v_news_content，保留 <p>/<table>） """ import sys, os, re, time, json, html as html_lib, urllib.parse from datetime import datetime, timedelta try: from bs4 import BeautifulSoup except ImportError: BeautifulSoup = None JSONL_PATH = os.environ.get("JSONL_PATH", "") ENC = "utf-8" SITE_NAME = "武定县-项目环评审批结果" GROUP_NAME = "云南" SCRIPT_NAME = "crawl_ynwd.py" CATEGORY = "项目环评审批结果" BASE_URL = "https://ynwd.gov.cn" LIST_URL = BASE_URL + "/zfxxgk/fdzdgknr/zdmsjgysyjsxx/sthj/xmhpspjg.htm" CUTOFF = (datetime.now() - timedelta(days=365 * 3)).strftime("%Y-%m-%d") 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", "Referer": LIST_URL, } _MAX_PAGES = 5 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"<[^>]+>", "", t) t = re.sub(r"^\s*(?:·|·|\u00b7)?\s*(?: |\u00a0)?\s*", "", t) t = re.sub(r"^[•·]\s*", "", t) t = re.sub(r"\s+", " ", t) return t.strip() def fetch(url, timeout=30): h = dict(HEADERS) last_err = None for attempt in range(3): try: import requests, urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) r = requests.get(url, headers=h, timeout=timeout, verify=False) if r.status_code == 200: r.encoding = ENC return r.text last_err = "HTTP %s" % r.status_code except Exception as e: last_err = str(e)[:80] time.sleep(2) print(" [fetch fail] %s -> %s" % (url[:90], last_err), flush=True) return None def html_to_text(content, page_url): """正文 HTML 清洗：保留 <p>/<table>，附件转 <p><a>绝对URL</a></p> 段落""" if not content: return "", 0, [] content = re.sub(r"", "", content, flags=re.S) content = re.sub(r"<meta[^>]*>", "", content, flags=re.I) content = re.sub(r"<script[^>]*>.*?</script>", "", content, flags=re.S | re.I) content = re.sub(r"<style[^>]*>.*?</style>", "", content, flags=re.S | re.I) attachments = [] link_protect = {} content = re.sub(r"【字号：[^】]*】", "", content) if BeautifulSoup: try: _soup = BeautifulSoup(content, "html.parser") for a in _soup.find_all("a", href=True): href = a["href"] if (a.get("appendix") or a.get("data-appendix") or re.search(r"\.(pdf|docx?|xlsx?|zip|rar|wps|et|ofd)", href, re.I) or "/module/download/" in href): # TRS 站附件: href="./P020...xlsx" 可用; oldsrc="/protect/..." 是 CMS 内部路径(直访 404) → 优先 href _href = (href or "").strip() _oldsrc = (a.get("oldsrc") or a.get("OLDSRC") or "").strip() if _href and not re.match(r"^(?:javascript:|mailto:|tel:|#)", _href, re.I): real_href = _href else: real_href = _oldsrc or _href txt = (a.get_text(strip=True) or a.get("_title") or a.get("title") or os.path.basename(real_href.split("?")[0]) or "附件") abs_url = urllib.parse.urljoin(page_url, real_href) attachments.append((abs_url, txt)) key = "__ATTACH__%d__" % len(link_protect) link_protect[key] = '<a href="%s" target="_blank">%s</a>' % (abs_url, txt) a.replace_with(key) for img in _soup.find_all("img"): src = (img.get("src") or img.get("oldsrc") or "") if re.search(r"(icons?/|fileTypeImages/icon_)", src, re.I): img.decompose() 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 img.get("oldsrc") or "" if not src: continue abs_url = urllib.parse.urljoin(page_url, src) txt = (img.get("alt") or img.get("title") or "").strip() \ or os.path.basename(src.split("?")[0]) or "图片" srcs.append('<a href="%s" target="_blank">%s</a>' % (abs_url, txt)) if srcs: key = "__IMG__%d__" % len(link_protect) link_protect[key] = "<p>" + "<br/>".join(srcs) + "</p>" p.replace_with(key) for img in _soup.find_all("img"): src = img.get("src") or img.get("oldsrc") or "" if not src: continue abs_url = urllib.parse.urljoin(page_url, src) txt = (img.get("alt") or img.get("title") or "").strip() \ or os.path.basename(src.split("?")[0]) or "图片" key = "__IMG__%d__" % len(link_protect) link_protect[key] = '<p><a href="%s" target="_blank">%s</a></p>' % (abs_url, txt) img.replace_with(key) content = str(_soup) except Exception: pass def _link_repl(m): href = m.group(1) if re.match(r"^(?:javascript:|mailto:|tel:|s?ftp:|#)", href, re.I): return "" if href.lower().startswith("javascript:") else m.group(0) 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 = "__LINK__%d__" % len(link_protect) link_protect[key] = '<a href="%s" target="_blank">%s</a>' % (abs_url, txt) return key content = re.sub(r'<a\s[^>]*href="([^"]+)"[^>]*>(.*?)</a>', _link_repl, content, flags=re.S | re.I) 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) for d in _soup.find_all("div"): if not d.get_text(strip=True) and not d.find("table") \ and not d.find("img") and not d.find("a"): d.decompose() else: d.unwrap() content = str(_soup) except Exception: pass table_protect = [] def _tbl_repl(m): table_protect.append(m.group(0)) return "__TBL__%d__" % (len(table_protect) - 1) has_table = 1 if re.search(r"<table[^>]*>", content, re.I) else 0 prev = None while prev != content: prev = content content = re.sub(r"<table(?:(?!<table).)*?</table>", _tbl_repl, content, flags=re.S | re.I) content = re.sub(r"</p>", "</p>\n\n", content, flags=re.I) content = re.sub(r"<br\s*/?>", "\n", content, flags=re.I) content = re.sub(r"</?(?:span|font|o:p|st1?:[a-z]+|strong|em|b|i|u)\b[^>]*>", "", content, flags=re.I) parts = [] for block in content.split("\n\n"): b = block.strip() if not b: continue b = re.sub(r"(?<=>)\s*[\r\n\t]+\s*", "", b) b = re.sub(r"\s*[\r\n\t]+\s*(?=<)", "", b) b = b.replace("\r", "").replace("\t", " ") b = re.sub(r"[ \t]{2,}", " ", b) 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 # UI button blocks (back-to-top / print / close / share) -> drop if re.match(r"^[【\[（(]?\s*(?:返回顶部|回到顶部|返回列表|打印(?:本页)?|关闭(?:窗口)?|分享|扫一扫|上一篇|下一篇)\s*[】\]）)]?$", plain): continue for k, v in link_protect.items(): b = b.replace(k, v) for i, tbl in enumerate(table_protect): b = b.replace("__TBL__%d__" % i, tbl) if not re.search(r"<p[ >]", b, re.I) and re.search(r"<a[ >]", b, re.I): b = "<p>%s</p>" % b 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("__TBL__%d__" % i, tbl) out = re.sub(r"__(?:LINK|TBL|IMG|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, final = set(), [] for p in out.split("\n\n"): key = re.sub(r"\s+", "", re.sub(r"<[^>]+>", "", p)) if not key or key in seen: continue seen.add(key) final.append(p) out = "\n\n".join(final).strip() dedup, seen_u = [], set() for u, t in attachments: if u in seen_u: continue seen_u.add(u) dedup.append((u, t)) return out, has_table, dedup def parse_list(html_text, page_url): """返回 (items, next_page_url)""" items = [] next_url = None if not BeautifulSoup: return items, next_url soup = BeautifulSoup(html_text, "html.parser") for li in soup.find_all("li", id=re.compile(r"^line_u6_")): a = li.find("a", href=True) if not a: continue href = (a.get("href") or "").strip() title = clean_title(a.get("title") or "") if not title: title = clean_title(a.get_text(" ", strip=True)) if not title or len(title) < 2: continue if "/info/" not in href: continue abs_url = urllib.parse.urljoin(page_url, href) date = "" sp = li.find("span") if sp: dm = re.search(r"(\d{4})[年/-](\d{1,2})[月/-](\d{1,2})", sp.get_text(strip=True)) if dm: date = "%s-%s-%s" % (dm.group(1), dm.group(2).zfill(2), dm.group(3).zfill(2)) items.append((abs_url, title, date)) # 下页链接 nxt = soup.find("span", class_=re.compile(r"p_next")) if nxt: a = nxt.find("a", href=True) if a and a.get("href") and not a["href"].lower().startswith("javascript"): next_url = urllib.parse.urljoin(page_url, a["href"]) return items, next_url def parse_detail(html_text, page_url): title = date = content_html = "" soup = None if BeautifulSoup: try: soup = BeautifulSoup(html_text, "html.parser") mt = soup.find("meta", attrs={"name": re.compile(r"ArticleTitle", re.I)}) if mt and mt.get("content"): title = clean_title(mt["content"]) if not title: cw = soup.select_one("div.contentwarp") h1 = cw.find("h1") if cw else soup.find("h1") if h1: title = clean_title(h1.get_text(strip=True)) lab = soup.find("div", class_="label") if lab: dm = re.search(r"(\d{4})[年/-](\d{1,2})[月/-](\d{1,2})", lab.get_text(" ")) if dm: date = "%s-%s-%s" % (dm.group(1), dm.group(2).zfill(2), dm.group(3).zfill(2)) el = soup.find(id="vsb_content_500") or soup.select_one("div.v_news_content") if el: content_html = "".join(str(c) for c in el.contents) except Exception: pass if not title: m = re.search(r"<title>(.*?)", html_text, re.S | re.I) if m: t = clean_title(m.group(1)) t = re.sub(r"[-|_].*(?:武定|人民政府).*$", "", t).strip() title = t if not date: dm = re.search(r"(\d{4})[年/-](\d{1,2})[月/-](\d{1,2})", html_text) if dm: date = "%s-%s-%s" % (dm.group(1), dm.group(2).zfill(2), dm.group(3).zfill(2)) return title, date, content_html def main(): import requests # noqa sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from crawler_lib import push_to_searchdb print("=== %s (%s) pages=%d cutoff=%s ===" % (SITE_NAME, SCRIPT_NAME, _PAGES, CUTOFF), flush=True) new_count = skip_count = cutoff_count = 0 seen_urls = set() jsonl_rows = [] batch = [] url = LIST_URL for page in range(1, _PAGES + 1): if not url: print("Page %d: no next link, stop" % page, flush=True) break try: html_text = fetch(url) except Exception as e: print("Page %d fetch error: %s" % (page, e), flush=True) break if not html_text: print("Page %d: empty, stop" % page, flush=True) break items, next_url = parse_list(html_text, url) if not items: print("Page %d: no items, stop" % page, flush=True) break print("Page %d: found %d items (%s)" % (page, len(items), url[-40:]), flush=True) for abs_url, title, date in items: if abs_url in seen_urls: continue seen_urls.add(abs_url) if date and date < CUTOFF: cutoff_count += 1 continue dhtml = fetch(abs_url) if not dhtml: skip_count += 1 continue dtitle, ddate, content_html = parse_detail(dhtml, abs_url) if not dtitle: dtitle = title if not ddate: ddate = date content, has_table, atts = html_to_text(content_html, abs_url) plain_len = len(re.sub(r"<[^>]+>", "", content).strip()) no_para = (not re.search(r"]", content, re.I) and not re.search(r"]", content, re.I) and not has_table) if plain_len < 5 or no_para: print(" skip empty/placeholder: %s" % dtitle[:40], flush=True) skip_count += 1 continue item = { "site_name": SITE_NAME, "source_url": abs_url, "url": abs_url, "title": dtitle, "pub_date": ddate, "summary": "", "content": content, "category": CATEGORY, "tags": "", "group_name": GROUP_NAME, "attachments": json.dumps( [{"title": t, "url": u} for u, t in atts], ensure_ascii=False) if atts else "", } if JSONL_PATH: jsonl_rows.append(item) else: batch.append(item) new_count += 1 print(" + %s (%s, %d字%s)" % (dtitle[:38], ddate, plain_len, ", 表格" if has_table else ""), flush=True) time.sleep(0.2) if not JSONL_PATH and batch: push_to_searchdb(batch, batch_label=SCRIPT_NAME) batch = [] url = next_url time.sleep(0.3) 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("JSONL written: %s (%d rows)" % (JSONL_PATH, len(jsonl_rows)), flush=True) else: if batch: push_to_searchdb(batch, batch_label=SCRIPT_NAME) print("新增: %d | 跳过: %d | CUTOFF截断: %d" % (new_count, skip_count, cutoff_count), flush=True) print("=== %s done ===" % SITE_NAME, flush=True) if __name__ == "__main__": main()