(.*?)

分段） # → 改名

保住分段；否则 unwrap 会丢段落边界，正文糊成一坨 if d.find("div") is None and d.find("table") is None: if not d.get_text(strip=True) and not d.find("img") and not d.find("a"): d.decompose() else: d.name = "p" elif 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"]*>", content, re.I) else 0 prev = None while prev != content: prev = content content = re.sub(r"", _tbl_repl, content, flags=re.S | re.I) 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 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"]", b, re.I) and not re.search(r"<(?:table|ul|ol|h[1-6])\b", b, re.I)): b = "

" % 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) # 最终兜底：去嵌套/空的

（源站

常见的产物） for _ in range(3): out = re.sub(r"]*>\s*]*>", "

", out, flags=re.I) out = re.sub(r"\s*", "

", out, flags=re.I) out = re.sub(r"]*>\s*", "", out, flags=re.I) 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 list_page_url(page): """createPageHTML(19,0,"index","html") → 第1页=目录, 第N页=index_(N-1).html""" if page <= 1: return LIST_URL return urllib.parse.urljoin(LIST_URL, "index_%d.html" % (page - 1)) def parse_list(html_text): """

……[YYYY-MM-DD]

""" items = [] for m in re.finditer(r"]*>(.*?)", html_text, re.S | re.I): li = m.group(1) am = re.search(r']*href="([^"]+)"[^>]*title="([^"]*)"', li, re.I) if not am: am = re.search(r']*href="([^"]+)"[^>]*>(.*?)', li, re.S | re.I) if not am: continue href = html_lib.unescape(am.group(1)).strip() if not re.search(r"/t\d+\.html$", href, re.I): continue title = clean_title(am.group(2) if am.lastindex and am.lastindex >= 2 else "") dm = re.search(r"\[?(\d{4})-(\d{1,2})-(\d{1,2})\]?", li) if not title or len(title) < 2 or not dm: continue date = "%s-%s-%s" % (dm.group(1), dm.group(2).zfill(2), dm.group(3).zfill(2)) items.append((urllib.parse.urljoin(LIST_URL, href), title, date)) return items def extract_body(soup, html_text): """正文 = div.trs_editor_view.TRS_UEDITOR（TRS 富文本容器）""" el = soup.select_one("div.trs_editor_view") or soup.select_one("div[class*=TRS_UEDITOR]") if el is not None and el.get_text(strip=True): return "".join(str(c) for c in el.contents) el = soup.select_one("div.content") or soup.select_one("div.aticle_box") if el is not None: return "".join(str(c) for c in el.contents) return "" 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"]) md = soup.find("meta", attrs={"name": re.compile(r"PubDate", re.I)}) if md and md.get("content"): dm = re.search(r"(\d{4})-(\d{1,2})-(\d{1,2})", md["content"]) if dm: date = "%s-%s-%s" % (dm.group(1), dm.group(2).zfill(2), dm.group(3).zfill(2)) if not title: h1 = soup.find("h1") if h1: title = clean_title(h1.get_text(strip=True)) content_html = extract_body(soup, html_text) except Exception: soup = None if not title: m = re.search(r"(.*?)", html_text, re.S | re.I) if m: title = clean_title(m.group(1)) 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(): 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 = [] for page in range(1, _PAGES + 1): url = list_page_url(page) html_text = fetch(url) if not html_text: print("Page %d fetch error, stop" % page, flush=True) break items = parse_list(html_text) if not items: print("Page %d: empty, stop" % page, flush=True) break print("Page %d: found %d items" % (page, len(items)), 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 = [] 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()