]*>(.*?)', html, re.DOTALL)
if tm:
title = re.sub(r'<[^>]+>', '', tm.group(1)).strip()
# 只去掉尾部 " — 网站名"（em-dash）或 " - 网站名"（最后一个 - 段）
title = re.sub(r'\s*[—]\s*.*$', '', title).strip()
title = re.sub(r'\s*-\s+[^-]+$', '', title).strip()
# ── 正文提取 ──
raw = _extract_detail_via_patterns(html, content_patterns)
content = ''
if raw:
content = clean_content(raw)
# 可选：去掉开头重复标题和元数据
if strip_metadata:
content = re.sub(
r'^.*?
\s*.*?
\s*\s*',
'', content, flags=re.DOTALL
)
# 后备方案：对于没有标准正文容器的CMS（如Jcms1），直接用标签提取
if not content:
raw = extract_body_via_p_tags(html)
if raw:
if len(raw) > 15000:
raw = raw[:15000] + '...'
content = raw
# ── 日期提取 ──
date = ''
dm = re.search(r'(\d{4})-(\d{2})-(\d{2})', html)
if dm:
date = dm.group(0)
# ── 附件提取 ──
attachments = []
# 1. 从正文中提取标准扩展名附件
if content:
for am in re.finditer(r']*href="([^"]+\.(?:pdf|doc|docx|xls|xlsx|zip|rar))"[^>]*>([^<]+)',
content, re.I):
attachments.append({'name': am.group(2).strip(), 'url': am.group(1)})
# 2. 从原始HTML中提取 virtual_attach_file 等非标准路径附件（Visual SiteBuilder CMS）
# 这些文件在 content 外，URL 不含标准扩展名，需额外处理
if not attachments:
for am in re.finditer(
r'href="([^"]*virtual_attach_file[^"]*)"[^>]*>[^<]*<[^>]+>([^<]+\.pdf)',
html, re.I
):
name = am.group(2).strip()
url = am.group(1).strip()
# 处理 ../../ 相对路径
if url.startswith('../../'):
url = '/' + '/'.join(url.split('/')[2:])
elif url.startswith('../'):
url = url[2:]
attachments.append({'name': name, 'url': url})
# ── 摘要 ──
summary = extract_plain_text(content, noise_lines)
if attachments:
for a in attachments:
if a['url'] not in summary:
summary += f"\n📎 {a['name']}: {a['url']}"
return {
'title': title,
'content': content,
'date': date,
'attachments': attachments,
'summary': summary,
}
def clean_content(raw_html):
"""
清洗原始HTML：保留语义标签（ ![]()
等），只去 script/style。
这是存到 content 列的格式。
"""
if not raw_html:
return ''
cleaned = re.sub(r']*>.*?', '', raw_html, flags=re.DOTALL|re.I)
cleaned = re.sub(r']*>.*?', '', cleaned, flags=re.DOTALL|re.I)
cleaned = re.sub(r'\s*\s*
', '', cleaned)
if len(cleaned) > 15000:
cleaned = cleaned[:15000] + '...'
# 清理多余内联样式（保留图片的 style="max-width:100%"）
img_styles = {}
for m in re.finditer(r'(]+)style="([^"]*max-width[^"]*)"', cleaned):
img_styles[m.start()] = m.group(2)
cleaned = re.sub(r'\s+style="[^"]*"', '', cleaned)
# 恢复图片的响应式style
if img_styles:
def restore_img_style(m):
tag = m.group(0)
for pos, style in img_styles.items():
if abs(pos - m.start()) < 200 and 'img' in tag[:10].lower():
return tag.rstrip('/>') + f' style="{style}"' + ('/>' if tag.endswith('/>') else '>')
return tag
cleaned = re.sub(r']*/?>', restore_img_style, cleaned)
return cleaned.strip()
def clean_noisy_html(html):
"""
清洗论坛/富文本编辑器的冗余 HTML。
专门处理 Discuz!、CKEditor 等产生的噪音标签如：
字符
...
...
保留语义标签： ![]()
-
"""
if not html:
return ''
# 清空空font
html = re.sub(r']*>\s*', '', html)
# 剥离font包裹，保留内容
html = re.sub(r']*>(.*?)', r'\1', html, flags=re.DOTALL)
# 剥离嵌套 → 内容
html = re.sub(r'\s*]*>(.*?)\s*', r'\1', html, flags=re.DOTALL)
# 去掉多余的
但保留内容
html = re.sub(r'
]*align="[^"]*"[^>]*>', '', html)
html = re.sub(r'段落。
返回：拼接的HTML（含
标签）或空字符串
"""
# 先去除banner/header等装饰区
body = strip_banner_section(html)
# 找第一个有实质中文内容的
（跳过非语义标签）
paragraphs = []
in_content = False
for m in re.finditer(r']*>(.*?)
', body, re.DOTALL):
text = re.sub(r'<[^>]+>', '', m.group(1)).strip()
# 附件链接也算有效
has_link = '外）
for am in re.finditer(r']*href="([^"]*.(?:pdf|docx?|xlsx?|zip|rar))"[^>]*>([^<]+)',
body, re.I):
a_tag = am.group(0)
if a_tag not in html_content:
html_content += '\n' + a_tag
return html_content
return ''
def extract_plain_text(html_content, noise_lines=None):
"""
从HTML提取纯文本，保留段落结构，过滤噪声行。
自定义噪声：
extract_plain_text(html_content, noise_lines=['扫一扫', '浏览量：', 'Copyright'])
"""
if not html_content:
return ''
text = re.sub(r'', '\n\n', html_content)
text = re.sub(r'', '\n', text)
text = re.sub(r'<[^>]+>', '', text)
text = html_mod.unescape(text)
text = re.sub(r'[ \t]+', ' ', text)
text = re.sub(r'\n{3,}', '\n\n', text)
text = text.strip()[:500]
DEFAULT_NOISE = [
'扫一扫', '打印|', '关闭', '相关链接', '相关文档',
'相关图片', '相关音频', '相关视频', '相关附件',
]
noise_lines = noise_lines or DEFAULT_NOISE
clean_lines = []
for line in text.split('\n'):
ls = line.strip()
if not ls:
clean_lines.append('')
elif not any(n in ls for n in noise_lines):
clean_lines.append(ls)
text = '\n'.join(clean_lines)
text = re.sub(r'\n{3,}', '\n\n', text)
return text.strip()
# ════════════════════════════════════════════
# 日期提取工具
# ════════════════════════════════════════════
def extract_date_from_url(url):
"""
从URL路径提取日期。
支持模式:
/2026/05/19/402155.html → 2026-05-19
/20260519/402155.html → 2026-05-19
_2026-05-19_ → 2026-05-19
"""
# /YYYY/MM/DD/ 或 /YYYYMMDD/
dm = re.search(r'/(\d{4})/(\d{2})/(\d{2})/', url)
if dm:
return f"{dm.group(1)}-{dm.group(2)}-{dm.group(3)}"
dm = re.search(r'/(\d{4})(\d{2})(\d{2})/', url)
if dm:
return f"{dm.group(1)}-{dm.group(2)}-{dm.group(3)}"
return ''
def extract_date_from_text(text):
"""
从文本中提取第一个日期。
支持:
"2026年5月19日" → 2026-05-19
"2026-05-19" → 2026-05-19
"2026/05/19" → 2026-05-19
"2026.05.19" → 2026-05-19
"""
# 中文格式
dm = re.search(r'(\d{4})\s*年\s*(\d{1,2})\s*月\s*(\d{1,2})\s*日', text)
if dm:
return f"{dm.group(1)}-{int(dm.group(2)):02d}-{int(dm.group(3)):02d}"
# 标准格式
dm = re.search(r'(\d{4})[-/.](\d{2})[-/.](\d{2})', text)
if dm:
return f"{dm.group(1)}-{dm.group(2)}-{dm.group(3)}"
return ''
# ════════════════════════════════════════════
# 列表页抓取模式
# ════════════════════════════════════════════
def fetch_list_via_api(api_url, page=1, page_size=20, data_key='data.list', **kwargs):
"""
模式A：通过API JSON获取列表页。
用法:
data = fetch_list_via_api(API_URL, page=1, page_size=20,
data_key='data.list',
dept_id=90, column_id=35663)
for item in data:
print(item['title'], item['pub_url'])
参数:
data_key: 点号分隔的JSON路径，如 'data.list' 或 'data.data.list'
"""
headers = {**HEADERS, 'Content-Type': 'application/json;charset=UTF-8'}
body = {"page_index": page, "page_size": page_size, **kwargs}
req = urllib.request.Request(api_url, data=json.dumps(body).encode('utf-8'), headers=headers)
try:
resp = urllib.request.urlopen(req, timeout=20, context=CTX)
result = json.loads(resp.read().decode('utf-8'))
# 按 data_key 路径取值
if data_key:
for key in data_key.split('.'):
if isinstance(result, dict):
result = result.get(key, [])
else:
return result
return result if isinstance(result, list) else []
except Exception as e:
return None
def fetch_list_via_html(list_url, link_pattern, exclude_titles=None):
"""
模式B：通过HTML解析列表页链接。
用法:
items = fetch_list_via_html(
'https://www.example.com/list',
r'(.*?)',
exclude_titles=['法律声明', '隐私保护'],
)
"""
html = http_get(list_url)
if not html:
return []
exclude_titles = exclude_titles or ['法律声明', '隐私保护', '网站声明', '关于我们', '联系我们']
items = []
seen = set()
for m in re.finditer(link_pattern, html, re.DOTALL):
href = m.group(1).strip()
title = re.sub(r'<[^>]+>', '', m.group(2)).strip()
if not title or not href:
continue
if title in exclude_titles:
continue
if not href.startswith('http'):
href = urljoin(list_url, href)
if href in seen:
continue
seen.add(href)
items.append({'title': title, 'url': href})
return items
def fetch_list_paginated(list_url=None, api_url=None, page_param='page',
link_pattern=None, max_pages=100, page_size=20,
**api_kwargs):
"""
模式C：分页抓取（自动翻页）。
HTML分页用法:
for items in fetch_list_paginated(
list_url='https://example.com/list_{page}.html',
link_pattern=r'(.*?)',
max_pages=20
):
for item in items:
...
API分页用法:
for items in fetch_list_paginated(
api_url='https://example.com/api/list',
page_param='page_index', page_size=20,
max_pages=100, dept_id=90
):
for item in items:
...
"""
page = 1
yielded_any = False
while page <= max_pages:
items = []
if api_url:
params = {**api_kwargs}
params.setdefault(page_param or 'page', page)
params.setdefault('page_size', page_size)
raw = fetch_list_via_api(api_url, **params)
if raw is None:
break
items = raw if isinstance(raw, list) else []
elif list_url:
url = list_url.format(page=page) if '{page}' in list_url else \
f"{list_url.rstrip('/')}_{page}.html" if '{' not in list_url and page > 1 \
else list_url
items = fetch_list_via_html(url, link_pattern) if link_pattern else []
else:
break
if not items:
break
yielded_any = True
yield items
page += 1
time.sleep(0.3)
if not yielded_any:
yield [] # 空列表表示无数据
# ════════════════════════════════════════════
# DB 工具
# ════════════════════════════════════════════
def init_db(db_path):
"""确保 DB 有所需的表结构"""
conn = sqlite3.connect(db_path)
conn.execute("PRAGMA journal_mode=WAL")
conn.executescript("""
CREATE TABLE IF NOT EXISTS crawl_results (
id INTEGER PRIMARY KEY AUTOINCREMENT,
site_id INTEGER NOT NULL,
title TEXT NOT NULL,
url TEXT UNIQUE NOT NULL,
content TEXT,
publish_date TEXT,
summary TEXT,
domain TEXT,
category TEXT DEFAULT 'gov',
crawled_at TEXT DEFAULT (datetime('now','localtime')),
content_hash TEXT
);
CREATE INDEX IF NOT EXISTS idx_results_url ON crawl_results(url);
CREATE INDEX IF NOT EXISTS idx_results_domain ON crawl_results(domain);
CREATE INDEX IF NOT EXISTS idx_results_date ON crawl_results(publish_date);
CREATE INDEX IF NOT EXISTS idx_results_site ON crawl_results(site_id);
CREATE TABLE IF NOT EXISTS crawl_sites (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
url TEXT NOT NULL,
domain TEXT,
crawl_method TEXT DEFAULT 'py_script',
crawl_script TEXT,
status TEXT DEFAULT '待爬取',
last_crawl_at TEXT,
total_crawled INTEGER DEFAULT 0,
enabled INTEGER DEFAULT 1,
created_at TEXT DEFAULT (datetime('now','localtime')),
updated_at TEXT DEFAULT (datetime('now','localtime'))
);
""")
conn.commit()
conn.close()
# ════════════════════════════════════════════
# GovCrawler 基类
# ════════════════════════════════════════════
class GovCrawler:
"""
政府站点爬虫基类（佛冈模式 v3）。
所有子类自动获得以下能力：
# 详情页
extract_detail() — 4种正文容器自动尝试 + 标题/日期/附件提取
clean_content() — HTML清洗（去script/style）
extract_plain_text() — HTML→纯文本 + 噪声过滤
clean_noisy_html() — 清洗论坛HTML（font/div align噪音）
# 列表页
fetch_list_via_api() — API JSON 模式
fetch_list_via_html() — HTML 链接提取模式
fetch_list_paginated() — 分页自动翻页（生成器）
# 日期
extract_date_from_url() — 从URL路径提取
extract_date_from_text() — 从文本中提取
# 数据处理
store_item() — 按 URL 去重存储
sync_to_server() — 增量同步到服务器 + 重启 search_app
http_get() — HTTP GET（自动重试）
"""
extract_detail = staticmethod(extract_detail)
clean_content = staticmethod(clean_content)
extract_plain_text = staticmethod(extract_plain_text)
clean_noisy_html = staticmethod(clean_noisy_html)
strip_banner_section = staticmethod(strip_banner_section)
extract_body_via_p_tags = staticmethod(extract_body_via_p_tags)
fetch_list_via_api = staticmethod(fetch_list_via_api)
fetch_list_via_html = staticmethod(fetch_list_via_html)
fetch_list_paginated = staticmethod(fetch_list_paginated)
extract_date_from_url = staticmethod(extract_date_from_url)
extract_date_from_text = staticmethod(extract_date_from_text)
http_get = staticmethod(http_get)
def __init__(self, site_name, domain, url="", site_id=None, db_name=None, sync_delay=0.3):
self.site_name = site_name
self.domain = domain
self.url = url
self.site_id = site_id
self.session = None
self._stats = {"new": 0, "skip": 0, "errors": 0}
self._crawler_name = Path(sys.argv[0]).stem
self.sync_delay = sync_delay # 请求间隔
if db_name:
self.db_path = str(BASE_DIR / db_name)
init_db(self.db_path)
else:
self.db_path = DEFAULT_DB
# ── DB 连接 ──
def _get_db(self):
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
return conn
# ── 站点注册 ──
def ensure_site(self):
db = self._get_db()
try:
if self.site_id:
row = db.execute("SELECT id FROM crawl_sites WHERE id=?", (self.site_id,)).fetchone()
if row:
return row["id"]
row = db.execute(
"SELECT id FROM crawl_sites WHERE domain=? ORDER BY id LIMIT 1",
(self.domain,),
).fetchone()
if row:
self.site_id = row["id"]
return self.site_id
cur = db.execute(
"INSERT INTO crawl_sites (name, url, domain, crawl_method, crawl_script, status) VALUES (?,?,?,?,?,?)",
(self.site_name, self.url, self.domain, "py_script", self._crawler_name, "待爬取"),
)
db.commit()
self.site_id = cur.lastrowid
return self.site_id
finally:
db.close()
def set_status(self, status):
if not self.site_id:
return
db = self._get_db()
try:
db.execute("UPDATE crawl_sites SET status=?, updated_at=datetime('now','localtime') WHERE id=?",
(status, self.site_id))
db.commit()
finally:
db.close()
def update_stats(self):
if not self.site_id:
return
db = self._get_db()
try:
db.execute(
"UPDATE crawl_sites SET status='已完成', last_crawl_at=datetime('now','localtime'), "
"total_crawled=total_crawled+?, crawl_script=?, updated_at=datetime('now','localtime') WHERE id=?",
(self._stats["new"], self._crawler_name, self.site_id),
)
db.commit()
finally:
db.close()
# ── 数据存储（佛冈模式） ──
def store_item(self, title, url, content, date, summary=None):
"""
存储单条记录。
- content = 原始HTML
- summary = 自动生成（如果未提供）
- 按 URL 去重
"""
if not self.site_id:
raise RuntimeError("请先调用 ensure_site()")
# 清洗: 3+连续换行 -> 段落分隔
if content:
content = re.sub(r'\n{3,}', '\n\n', content).strip()
if summary is None:
summary = self.extract_plain_text(content)[:500] if content else (title[:300] if title else "")
ch = hashlib.md5((content or "").encode()).hexdigest()
db = self._get_db()
try:
db.execute(
"INSERT OR IGNORE INTO crawl_results "
"(site_id, title, url, content, publish_date, summary, domain, category, content_hash) "
"VALUES (?,?,?,?,?,?,?,?,?)",
(self.site_id, title, url, content or "", date, summary, self.domain, "gov", ch),
)
db.commit()
if db.total_changes > 0:
self._stats["new"] += 1
else:
self._stats["skip"] += 1
except Exception as e:
self._stats["errors"] += 1
finally:
db.close()
# ── 服务器同步 ──
def sync_to_server(self):
"""增量同步本站点数据到服务器 search.db + 重启 search_app"""
print(f"\n📤 同步到服务器 ({SERVER_SSH})...")
db = self._get_db()
rows = db.execute(
"SELECT title, url, content, publish_date, summary FROM crawl_results WHERE site_id=? ORDER BY id",
(self.site_id,),
).fetchall()
db.close()
if not rows:
print(" ⏭️ 无数据，跳过同步")
return
def esc(v):
if v is None: return "''"
return "'" + str(v).replace("'", "''") + "'"
sql_lines = []
for title, page_url, content, pub_date, summary in rows:
content = (content or '')[:8000]
summary = (summary or '')[:300]
sql_lines.append(
f"INSERT OR IGNORE INTO gov_raw "
f"(title, page_url, content, publish_date, summary, site_name, tags) "
f"VALUES ({esc(title)},{esc(page_url)},{esc(content)},{esc(pub_date or '')},{esc(summary)},'{self.site_name}','环评阶段');"
)
sql_lines.append("SELECT 1 /* noop: gov_search 由触发器维护, 无需 rebuild */;")
sql = '\n'.join(sql_lines)
# 写入临时SQL文件，避免命令行过长
import tempfile, os
tmp = tempfile.mktemp(suffix='.sql', dir='/tmp')
with open(tmp, 'w') as f:
f.write(sql)
# scp到服务器
subprocess.run(
["scp", "-o", "StrictHostKeyChecking=no", tmp, f"{SERVER_SSH}:{tmp}"],
capture_output=True, timeout=30,
)
# 执行
result = subprocess.run(
["ssh", "-o", "ConnectTimeout=10", "-o", "StrictHostKeyChecking=no", SERVER_SSH,
f"sqlite3 {SERVER_SEARCH_DB} < {tmp} && echo OK"],
capture_output=True, text=True, timeout=120,
)
os.unlink(tmp)
subprocess.run(
["ssh", "-o", "StrictHostKeyChecking=no", "-o", "ConnectTimeout=10", SERVER_SSH,
f"rm -f {tmp}"],
capture_output=True, timeout=10,
)
if "OK" in result.stdout:
print(f" ✅ {len(rows)} 条同步成功")
subprocess.run(
["ssh", "-o", "StrictHostKeyChecking=no", "-o", "ConnectTimeout=10", SERVER_SSH,
"pkill -f search_app; sleep 1; cd /root && nohup python3 search_app.py > /var/log/srch.log 2>&1 &"],
capture_output=True, timeout=30,
)
print(" ✅ search_app 已重启")
else:
print(f" ✗ 同步失败: {result.stderr[:200] or result.stdout[:200]}")
# ── 主流程 ──
def run(self):
start = time.time()
print(f"\n{'='*50}")
print(f"🏠 {self.site_name} ({self.domain})")
print(f" DB: {self.db_path}")
print(f"{'='*50}")
sid = self.ensure_site()
print(f"📋 site_id={sid}")
self.set_status("爬取中")
try:
self.do_crawl()
except Exception as e:
import traceback
print(f"\n❌ 爬取异常: {e}")
traceback.print_exc()
self.set_status("失败")
self._stats["errors"] += 1
return
self.update_stats()
elapsed = time.time() - start
s = self._stats
print(f"\n{'─'*30}")
print(f"✅ 新增: {s['new']} | 跳过: {s['skip']} | 错误: {s['errors']}")
print(f"⏱️ 耗时: {elapsed:.1f}s")
if s['new'] > 0:
self.sync_to_server()
else:
print(" ⏭️ 无新数据，跳过同步")
def do_crawl(self):
raise NotImplementedError("子类必须实现 do_crawl()")
# backup test marker Tue May 19 03:29:13 PDT 2026