u6j*?dZddlZejjdZdZdZdZdZe dk(rWed ed d d d d ddd dddd ededddddddddddd ededdd d!d"d#d$d d%d&dd' e d(yy))uz批量生成爬虫脚本 — 基于 crawl_daixian_hjbh.py 规范模板（HTML表格保留+附件内嵌+push_to_searchdb）Nz~/Crawler/gov_crawler/gen_outu-#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ @DESC@ URL: @URL@ """ import sys, os, re, time, html as html_lib, urllib.parse, sqlite3, json BASE_URL = "@BASE@" LIST_URL = "@LIST_URL@" SITE_NAME = "@SITE_NAME@" GROUP_NAME = "@GROUP@" SCRIPT_NAME = "@SCRIPT@" CATEGORY = "@CATEGORY@" try: from bs4 import BeautifulSoup except ImportError: BeautifulSoup = None DB_PATH = os.environ.get("DB_PATH", "/root/search.db") JSONL_PATH = os.environ.get("JSONL_PATH", "") ENC = "utf-8" 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 _DAYS_CUTOFF = 365 * 3 _CUTOFF_DATE = time.strftime("%Y-%m-%d", time.localtime(time.time() - _DAYS_CUTOFF * 86400)) 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*(?:·|·|\u00b7)?\s*(?: |\u00a0)?\s*", "", t) t = re.sub(r"\s+", " ", t) return t.strip() def fetch(url, referer=None, timeout=30): h = dict(HEADERS) if referer: h["Referer"] = referer last_err = None for _ in range(2): 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(1) print(" [fetch fail] %s -> %s" % (url[:90], last_err), flush=True) return None def html_to_text(content, page_url): """HTML -> 纯HTML段落输出(保留表格/附件/图片链接)""" 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 = {} content = re.sub(r"【字号：[^】]*】", "", content) content = re.sub(r"发布(?:时间|日期)[:：][^<]{0,30}", "", content) if BeautifulSoup: try: _soup = BeautifulSoup(content, "html.parser") for sel in (".bdsharebuttonbox", ".share", ".sharebox", ".shares", "#share", ".fx", ".fenxiang", ".bshare-custom", ".article-share", ".social-share", ".printer", ".print", "div[class*=share]", "div[id*=share]", ".ewm", ".qrcode", ".erweima"): for el in _soup.select(sel): el.decompose() for a in _soup.find_all("a", href=True): if a.get("appendix") or a.get("data-appendix") or re.search(r"\.(pdf|docx?|xlsx?|zip|rar|wps)", a.get("href", ""), re.I): href = a["href"] real_href = a.get("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]) abs_url = urllib.parse.urljoin(page_url, real_href) attachments.append((abs_url, txt)) key = "__ATTACH__%d__" % len(link_protect) link_protect[key] = '%s' % (abs_url, 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 img.get("oldsrc") or "" if src: abs_url = urllib.parse.urljoin(page_url, src) srcs.append('%s' % (abs_url, os.path.basename(src.split("?")[0]) or "图片")) if srcs: key = "__IMG__%d__" % len(link_protect) link_protect[key] = "

" + "
".join(srcs) + "

" p.replace_with(key) content = str(_soup) except Exception: pass def _link_repl(m): href = m.group(1) if href.startswith("javascript:"): return "" inner = m.group(2) txt = re.sub(r"<[^>]+>", "", inner) txt = html_lib.unescape(txt).strip() txt = re.sub(r"^附件[:：]\s*", "", txt).strip() if re.search(r"\.(pdf|docx?|xlsx?|zip|rar|wps|et|dwg)", href, re.I) or "attachment" in href.lower() or "download" in href.lower(): 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] = '%s' % (abs_url, txt) return key abs_url = urllib.parse.urljoin(page_url, href) return '%s' % (abs_url, txt) content = re.sub(r']*href="([^"]+)"[^>]*>(.*?)', _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) 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 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 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) 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__\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 main(): import requests sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from crawler_lib import push_to_searchdb s = requests.Session() s.headers.update(HEADERS) new_count = 0 skip_count = 0 seen_urls = set() jsonl_rows = [] batch = [] for page in range(1, _PAGES + 1): try: url = build_list_url(page) html_text = fetch(url) if not html_text: print("Page %d fetch failed" % page, flush=True) break items = parse_list(html_text, url) except Exception as e: print("Page %d fetch error: %s" % (page, e), flush=True) break 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_DATE: 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 if ddate and ddate < _CUTOFF_DATE: skip_count += 1 continue 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_html, re.I) and not has_table if plain_len < 10 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": "; ".join("%s|%s" % (u, t) for u, t in atts) if atts else "", } if JSONL_PATH: jsonl_rows.append(item) print(" JSONL + %s (%s)" % (dtitle[:40], ddate), flush=True) new_count += 1 continue batch.append(item) new_count += 1 print(" + %s (%s)" % (dtitle[:40], ddate), flush=True) time.sleep(0.08) time.sleep(0.2) 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: push_to_searchdb(batch, batch_label=SCRIPT_NAME) print("新增: %d" % new_count, flush=True) print("跳过: %d" % skip_count, flush=True) print("=== %s done ===" % SITE_NAME, flush=True) a def build_list_url(page): if page == 1: return LIST_URL return BASE_URL + "@PAGE_TPL@" % page def parse_list(html_text, page_url): items = [] if not BeautifulSoup: return items soup = BeautifulSoup(html_text, "html.parser") for li in soup.select("@LIST_SEL@"): a = li.find("a", href=True) if not a: continue href = a["href"].strip() title = a.get("title", "") or a.get_text(strip=True) date = "" d = li.select_one("@DATE_SEL@") if d: date = d.get_text(strip=True) title = clean_title(title) if not title or len(title) < 4: continue if not href.startswith("http"): href = urllib.parse.urljoin(page_url, href) items.append((href, title, date)) return items def parse_detail(html_text, page_url): title = ""; date = ""; content_html = "" if BeautifulSoup: soup = BeautifulSoup(html_text, "html.parser") d = soup.select_one("@DETAIL_SEL@") if d: content_html = "".join(str(c) for c in d.contents) t = soup.find("meta", attrs={"name": re.compile("ArticleTitle", re.I)}) if t and t.get("content"): title = clean_title(t["content"]) md = soup.find("meta", attrs={"name": re.compile("PubDate", re.I)}) if md and md.get("content"): dm = re.search(r"([0-9]{4}-[0-9]{2}-[0-9]{2})", md["content"]) if dm: date = dm.group(1) if not title: h1 = soup.find("h1") or soup.find("h2") if h1: title = clean_title(h1.get_text()) return title, date, content_html if __name__ == "__main__": main() c tjd|jd|jd|jd|jd|jd|jd|jd|} tjd |jd | jd | jd | } | | zS) uTRS/CSS选择器型列表z@DESC@z@URL@z@BASE@z @LIST_URL@z @SITE_NAME@z@GROUP@z@SCRIPT@z @CATEGORY@z @PAGE_TPL@z @LIST_SEL@z @DATE_SEL@z @DETAIL_SEL@) COMMON_HEADreplaceBODY_TRS)namedescurl site_namegroupcategorybaselist_urlpage_tpl list_selectordate_seldetail_selectorheadbodys gen_crawlers.pygen_trs_selectorr;s  WXt $WWWc%:778T;R W\8 ,WW]I-N WY &wwz4'@W_A`   W\8 ,WW\=-Q W\8 ,WW^_-U  $;ctjjt|}tjtdt |dd5}|j |dddtd|y#1swYxYw)NT)exist_okwzutf-8)encodingWROTE)ospathjoinOUTmakedirsopenwriteprint)rcontentrfs rr#r#Gs] 77<<T "DKKd# dC' *a  '4s A55A>__main__zcrawl_gdkfq_ezhou_hbgs.pyu葛店开发区 - 环保公示z$http://gdkfq.ezhou.gov.cn/zwdt/hbgs/u葛店开发区-环保公示u湖北u 环境保护zhttp://gdkfq.ezhou.gov.cnz/zwdt/hbgs/index_%d.htmlzdiv.list ul lispanz3div#myTabContent.tab-content, div.content, div.listzcrawl_gxdmjkq_nanning.pyu+广西-东盟经开区 - 环境核查审批z/http://gxdmjkq.nanning.gov.cn/zwgk/hjbh/hjhcsp/u(广西东盟经开区-环境核查审批u广西zhttp://gxdmjkq.nanning.gov.cnz/zwgk/hjbh/hjhcsp/index_%d.htmlzul.has_line liz span.timez/div.content, div.article, .TRS_Editor, #contentzcrawl_youyang_gsgg.pyu酉阳县 - 公示公告z#https://youyang.gov.cn/sy_236/gsgg/u酉阳县-公示公告u重庆u 公示公告zhttps://youyang.gov.cnz/sy_236/gsgg/index_%d.htmlz ul.list liz>div.TRS_Editor, div.article-content, div.content, .content-boxzDONE gen part 1) __doc__rr expanduserr rrrr#__name__r$rrr-sA gg89p d U  z %'7#(.&.#%K"&"W( $&6"592Hn')Z)+'X' !#3"- (N "G$f^$ ;r