#!/usr/bin/env python3
"""
龙潭区人民政府-政策法规 爬虫
站点: http://www.longtan.gov.cn/gzms/zcfg/
CMS: TRS CMS（自定义模板）
分页: index.html(第1页), index_1.html(第2页) ... index_8.html(第9页)
详情: ./YYYYMM/tYYYYMMDD_NNNNNNN.html
正文: div.detail_content > div.TRS_Editor > p/table/img
"""
import os, re, sys, json, time, sqlite3
from bs4 import BeautifulSoup
from urllib import request
from urllib.parse import urljoin
from html import unescape
import urllib.parse
SITE_NAME = "龙潭区人民政府-政策法规"
BASE_URL = "http://www.longtan.gov.cn"
LIST_PATH = "/gzms/zcfg/"
DB_PATH = "/mnt/data/search.db"
DELAY = 1.5
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
}
def fetch(url, timeout=30):
req = request.Request(url, headers=HEADERS)
try:
resp = request.urlopen(req, timeout=timeout)
return resp.read().decode("utf-8", errors="ignore")
except Exception as e:
print(" [ERROR] %s: %s" % (url, e), flush=True)
return None
def get_total_pages(html):
for n in range(1, 50):
url = BASE_URL + LIST_PATH + "index_%d.html" % n
h = fetch(url, timeout=10)
if not h:
return n - 1
if 'href="./20' not in h:
return n - 1
return 49
def parse_list(html):
items = []
seen = set()
for m in re.finditer(
r']*>\s*]*>(.*?)
\s*(\d{4}-\d{2}-\d{2})
',
html, re.DOTALL
):
year_prefix = m.group(1)
month = m.group(2).zfill(2)
file_id = m.group(3)
article_id = m.group(4)
title = unescape(m.group(5).strip())
date = m.group(6)
year = "20" + year_prefix
href = "./%s%s/t%s_%s.html" % (year, month, file_id, article_id)
full_url = urljoin(BASE_URL + LIST_PATH, href)
if full_url in seen:
continue
seen.add(full_url)
title = re.sub(r'\s+', ' ', title).strip()
items.append((title, full_url, date))
return items
def html_table_to_html(table, base_url=""):
"""保留 HTML 表格结构，仅将相对链接/图片转绝对 URL"""
from bs4 import BeautifulSoup
tbl = BeautifulSoup(str(table), 'html.parser')
for a in tbl.find_all('a'):
href = a.get('href', '')
if href and not href.startswith(('http', 'javascript', '#')):
a['href'] = urllib.parse.urljoin(base_url, href) if base_url else href
for img in tbl.find_all('img'):
src = img.get('src', '')
if src and not src.startswith(('http', '//', 'data:')):
img['src'] = urllib.parse.urljoin(base_url, src) if base_url else src
return str(tbl)
def parse_detail(html, url):
content_parts = []
attachments = []
m = re.search(r'(.*?)
\s*(.*?)
', html, re.DOTALL)
if not m:
return "", []
container = m.group(1)
container = re.sub(r']*>.*?', '', container, flags=re.DOTALL)
container = re.sub(r']*>.*?', '', container, flags=re.DOTALL)
table_ranges = []
for tm in re.finditer(r"]*>.*?", container, re.DOTALL):
table_ranges.append((tm.start(), tm.end()))
def inside_table(pos):
return any(ts <= pos <= te for ts, te in table_ranges)
def has_block_children(el_html):
body = re.sub(r'^]*>', '', el_html)
body = re.sub(r'\s*$', '', body)
return bool(re.search(r'<(div|p|table|ul|ol|h[1-6])\b', body))
elements = []
for m in re.finditer(r"<(p|table|img)\b[^>]*>", container, re.DOTALL):
start, m_end, tag = m.start(), m.end(), m.group(1)
if tag in ("p", "table"):
close_tag = ""
end_pos = container.find(close_tag, m_end)
if end_pos >= 0:
el_html = container[start:end_pos + len(close_tag)]
else:
ns = container.find("<", m_end)
el_html = container[start:ns] if ns >= 0 else container[start:]
elif tag == "img":
cp = container.find(">", m_end)
el_html = container[start:cp + 1] if cp >= 0 else None
if el_html is None:
continue
else:
continue
if tag == "p":
if inside_table(start):
continue
if has_block_children(el_html):
continue
if "]+>", "", el_html).strip()
text = re.sub(r"\s+", " ", text)
text = unescape(text)
if text:
elements.append((start, text))
elif tag == "table":
if el_html.count(" 0:
md = html_table_to_html(el_html)
if md:
elements.append((start, md))
elif tag == "img":
src_m = re.search(r'src="([^"]+)"', el_html)
alt_m = re.search(r'alt="([^"]*)"', el_html)
if src_m:
src = urljoin(url, src_m.group(1))
alt = alt_m.group(1) if alt_m else ""
elements.append((start, "" % (alt, src)))
elements.sort(key=lambda x: x[0])
content = "\n\n".join(text for _, text in elements)
for a_href, a_text_raw in re.findall(
r']*href="([^"]+\.(?:doc|docx|pdf|xls|xlsx|ppt|pptx|zip|rar|7z|txt|wps|et|DOC))"[^>]*>(.*?)',
html, re.DOTALL
):
a_text = re.sub(r'<[^>]+>', '', a_text_raw).strip()
a_text = unescape(a_text)
if a_text:
attachments.append({"title": a_text, "url": urljoin(url, a_href)})
if (not content or not content.strip()) and attachments:
content = "【该文档内容详见附件PDF/文档】\n"
for att in attachments:
content += "附件: " + att["title"] + " - " + att["url"] + "\n"
return content, attachments
def crawl(max_pages=5):
conn = sqlite3.connect(DB_PATH, timeout=60)
conn.execute("PRAGMA busy_timeout=60000")
cur = conn.cursor()
# QC20260925 去掉整站清空再重灌(抢锁+中途死掉会清空整站; page_url 有 UNIQUE 索引，插入本就幂等) cur.execute("DELETE FROM gov_search WHERE site_name=?", (SITE_NAME,))
# QC20260925 去掉整站清空再重灌(抢锁+中途死掉会清空整站; page_url 有 UNIQUE 索引，插入本就幂等) cur.execute("DELETE FROM gov_raw WHERE site_name=?", (SITE_NAME,))
conn.commit()
print("[清理] 已清理旧数据", flush=True)
print("[分析] 扫描总页数...", flush=True)
html = fetch(BASE_URL + LIST_PATH)
if not html:
print("[失败] 无法获取列表", flush=True)
conn.close()
return
total_pages = get_total_pages(html)
pages_to_crawl = min(max_pages, total_pages) if max_pages > 0 else total_pages
print("[分析] 共%d页，本次爬取%d页" % (total_pages, pages_to_crawl), flush=True)
all_items = []
seen_urls = set()
# 第1页
items = parse_list(html)
for item in items:
if item[1] not in seen_urls:
seen_urls.add(item[1])
all_items.append(item)
print(" 第1页: %d条(去重后%d条)" % (len(items), len(all_items)), flush=True)
# 第2页+
for page in range(1, pages_to_crawl):
page_url = BASE_URL + LIST_PATH + "index_%d.html" % page
print("\n[分页] 第%d页: %s" % (page + 1, page_url), flush=True)
time.sleep(DELAY)
html = fetch(page_url)
if not html:
print(" [完成] 获取失败", flush=True)
break
items = parse_list(html)
new_items = 0
for item in items:
if item[1] not in seen_urls:
seen_urls.add(item[1])
all_items.append(item)
new_items += 1
print(" 找到%d条，去重后新增%d条" % (len(items), new_items), flush=True)
print("\n[爬取] 共%d条待爬详情页" % len(all_items), flush=True)
new_count = 0
for title, detail_url, date in all_items:
final_title = title
time.sleep(DELAY)
detail_html = fetch(detail_url)
if not detail_html:
print(" [跳过] %s -> 获取失败" % final_title[:30], flush=True)
continue
meta_m = re.search(r']*content="([^"]*)"', detail_html)
if meta_m:
meta_title = unescape(meta_m.group(1).strip())
if meta_title and len(meta_title) > len(final_title):
final_title = meta_title
meta_d = re.search(r']*content="([^"]*)"', detail_html)
if meta_d:
date_str = meta_d.group(1).strip()
date_m = re.match(r'(\d{4}-\d{2}-\d{2})', date_str)
if date_m:
date = date_m.group(1)
content, attachments = parse_detail(detail_html, detail_url)
aj = json.dumps(attachments, ensure_ascii=False)
summary = content[:200] if content else final_title
cur.execute(
"INSERT OR IGNORE INTO gov_raw (site_name,page_url,title,content,publish_date,attachments,summary) VALUES(?,?,?,?,?,?,?)",
(SITE_NAME, detail_url, final_title, content, date, aj, summary)
)
if cur.rowcount > 0:
new_count += 1
att_count = len(attachments)
status = "[正文]" if content else "[无文本]"
print(" %s %s -> %d chars, %d附件" % (status, final_title[:35], len(content), att_count), flush=True)
conn.commit()
conn.close()
print("\n[DONE] %s: 新增=%d" % (SITE_NAME, new_count), flush=True)
if __name__ == "__main__":
_pages_args = [int(a.split('=', 1)[1]) for a in sys.argv if a.startswith('--pages=')]
_pages_args = _pages_args or [int(a) for a in sys.argv[1:] if a.isdigit()]
pages = _pages_args[0] if _pages_args else 5
crawl(pages)