\s*
\s*
(.*?)
', html_text, re.S)
if not m:
m = re.search(r'
(.*?)
', html_text, re.S)
if m:
t = re.sub(r'<[^>]+>', '', m.group(1))
t = clean_title(t)
if t:
return t
return fallback
def parse_detail(html_text, url):
"""解析详情页, 返回 (content_html, attachments)"""
# 正文: div.article-con > div.trs_editor_view (排除注释块)
m = re.search(r'
', html_text, re.S)
if not m:
# 回退: 任意 trs_editor_view
m = re.search(r'
]*>(.*?)
', html_text, re.S)
if not m:
return "", []
body = m.group(1)
# 移除注释块(如果有)
body = re.sub(r'', '', body, flags=re.S)
# 附件
atts = re.findall(r'
]+href="([^"]+)"[^>]*>', body)
atts = [urljoin(url, a) for a in atts if re.search(r'\.(?:pdf|docx?|xlsx?|zip|rar|wps|et|dps)(?:\?|$)', a, re.I)]
return body, atts
def html_to_text(html_text):
"""HTML -> 文本, 保留表格、\n\n 分段"""
tables = []
def _save_tbl(m):
tables.append(m.group(0))
return f"@@TABLE{len(tables)-1}@@"
html_text = re.sub(r'', _save_tbl, html_text)
html_text = re.sub(r']*>', '\n', html_text)
html_text = re.sub(r']*>', '', html_text)
html_text = re.sub(r'\n{3,}', '\n\n', html_text)
lines = [ln.strip() for ln in html_text.split('\n')]
out = [ln for ln in lines if ln]
text = '\n\n'.join(out)
for i, t in enumerate(tables):
text = text.replace(f"@@TABLE{i}@@", t)
text = re.sub(r'\n{3,}', '\n\n', text)
return text.strip()
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--pages", type=int, default=1, help="抓取页数")
ap.add_argument("--db", default="/mnt/data/search.db", help="SQLite 数据库路径")
ap.add_argument("--script-name", default=SCRIPT_NAME)
args = ap.parse_args()
db_path = args.db
script_name = args.script_name
if not db_path.startswith("/mnt/"):
db_path = os.path.expanduser(db_path)
conn = sqlite3.connect(db_path, timeout=60)
cur = conn.cursor()
new_count = 0
dup_count = 0
err_count = 0
for page in range(1, args.pages + 1):
if page == 1:
url = LIST_URL
else:
url = f"http://www.lc.gov.cn/ztzl/33025/9652/index_{page-1}.html"
try:
r = requests.get(url, headers=HEADERS, timeout=30)
r.encoding = "utf-8"
except Exception as e:
print(f"[{page}] 请求失败 {url}: {e}")
err_count += 1
continue
items = parse_list(r.text, url)
print(f"[{page}] 列表 {len(items)} 条 {url}")
if not items:
continue
for title, art_url, pub_date in items:
try:
cur.execute("SELECT COUNT(*) FROM gov_raw WHERE source_url=? AND script_name=?", (art_url, script_name))
if cur.fetchone()[0] > 0:
dup_count += 1
continue
dr = requests.get(art_url, headers=HEADERS, timeout=30)
dr.encoding = "utf-8"
# 完整标题优先从详情页 h2 提取(列表页标题被站点截断带省略号)
full_title = extract_full_title(dr.text, title)
content_html, atts = parse_detail(dr.text, art_url)
if not content_html.strip():
print(f" 跳过空正文: {title[:30]}")
continue
content = html_to_text(content_html)
att_txt = "\n\n".join(atts) if atts else ""
md5 = hashlib.md5((full_title + content[:500]).encode("utf-8")).hexdigest()
summary = content[:500] if content else ""
cur.execute(
"INSERT INTO gov_raw (site_name, source_url, page_url, title, publish_date, date_rank, summary, status, category, visits, content, tags, industry, attachments, group_name, has_table, script_name) "
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
(SITE_NAME, art_url, art_url, full_title, pub_date, 0, summary, "", "", 0, content, "",
"other", att_txt, GROUP_NAME, 1 if "