HTML intact
- \n\n between blocks
- attachment links embedded with absolute URL
- strip ", "", content, flags=re.S | re.I)
content = re.sub(r"]*>.*?", "", content, flags=re.S | re.I)
# strip style/class attributes via BS4 BEFORE paragraph splitting (合浦 hepu 教训)
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
# 1) FIRST protect attachment links (before table protection, so table
# regex can't swallow links that sit right after a close tag)
content2 = content
# remove attachment-file-type icon imgs (icon_doc.gif / icon_xls.gif etc.)
content2 = re.sub(
r']*src="[^"]*fileTypeImages/icon_[a-z]+\.gif"[^>]*/?>', "", content2,
flags=re.I,
)
attachments = []
link_protect = {}
def _link_repl(m):
href = m.group(1)
inner = m.group(2)
txt = re.sub(r"<[^>]+>", "", inner)
txt = html_lib.unescape(txt).strip()
txt = re.sub(r"^附件[:：]\s*", "", txt).strip()
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 = f"__LINK__{len(link_protect)}__"
link_protect[key] = f'{txt}'
return key
content2 = re.sub(r']*href="([^"]+)"[^>]*>(.*?)', _link_repl, content2, flags=re.S | re.I)
# 2) THEN protect tables
table_protect = []
def _tbl_repl(m):
table_protect.append(m.group(0))
return f"__TBL__{len(table_protect)-1}__"
content2 = re.sub(r"]*>.*?
", _tbl_repl, content2, flags=re.S | re.I)
has_table = 1 if re.search(r"]*>", content, re.I) else 0
content2 = re.sub(r"
", "\n\n", content2, flags=re.I)
content2 = re.sub(r"", "\n", content2, flags=re.I)
content2 = re.sub(r"]*>", "", content2, flags=re.I)
parts = []
for block in content2.split("\n\n"):
b = block.strip()
if not b:
continue
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(f"__TBL__{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(f"__TBL__{i}__", tbl)
out = re.sub(r"__LINK__\d+__", "", out)
out = re.sub(r"__TBL__\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)
# dedupe paragraphs
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 parse_list(html_text, page_url):
"""Extract (url, title, date) from list page."""
items = []
if BeautifulSoup:
soup = BeautifulSoup(html_text, "html.parser")
# precise sub-column: ul.r_ico inside div.r_bottom
ul = soup.find("ul", class_=re.compile(r"r_ico"))
if ul:
for li in ul.find_all("li"):
a = li.find("a", href=True)
if not a:
continue
href = a.get("href", "").strip()
if not re.search(r"/20\d{4}/t\d+_\d+\.html$", href):
continue
abs_url = urllib.parse.urljoin(LIST_URL, href)
title = clean_title(a.get("title", "") or a.get_text(" ", strip=True))
span = li.find("span")
date = span.get_text(strip=True) if span else ""
if not re.match(r"^20\d{2}-\d{2}-\d{2}$", date):
date = ""
items.append((abs_url, title, date))
if not items:
# regex fallback: li > a + span
for m in re.finditer(
r'\s*]+href="([^"]*?/20\d{4}/t\d+_\d+\.html)"[^>]*>(.*?)\s*([^<]*)',
html_text, re.S,
):
abs_url = urllib.parse.urljoin(LIST_URL, m.group(1))
items.append((abs_url, clean_title(m.group(2)), m.group(3).strip()))
return items
def parse_detail(html_text, page_url):
"""Extract title, date, content from detail page."""
title = ""
date = ""
content_html = ""
if BeautifulSoup:
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"])
h1 = soup.find("h1")
if not title and h1:
title = clean_title(h1.get_text(" ", strip=True))
md = soup.find("meta", attrs={"name": re.compile(r"PubDate", re.I)})
if md and md.get("content"):
d = md["content"].strip()
dm = re.match(r"(\d{4})-(\d{2})-(\d{2})", d)
if dm:
date = dm.group(0)
# content: div.trs_editor_view (innermost)
view = soup.find("div", class_=re.compile(r"trs_editor_view"))
if view:
content_html = str(view)
else:
# fallback: div.text > div content
text = soup.find("div", class_=re.compile(r"\btext\b"))
if text:
content_html = str(text)
if not title or not content_html:
# regex fallback
m = re.search(r']*class="trs_editor_view[^"]*"[^>]*>(.*?)\s*\s*', html_text, re.S)
if m:
content_html = m.group(0)
return title, date, content_html
def store_record(conn, url, title, content, date, has_table):
cur = conn.execute(
"SELECT id FROM gov_raw WHERE page_url=?",
(url,),
)
if cur.fetchone():
return False
try:
cur = conn.execute(
"INSERT OR IGNORE INTO gov_raw "
"(page_url, source_url, title, content, publish_date, site_name, group_name, script_name, has_table, industry) "
"VALUES (?,?,?,?,?,?,?,?,?,?)",
(url, url, title, content, date, SITE_NAME, GROUP_NAME, SCRIPT_NAME, has_table, "other"),
)
if cur.rowcount > 0:
summary = re.sub(r"<[^>]+>", "", content)
summary = html_lib.unescape(summary)
summary = re.sub(r"\s+", " ", summary).strip()[:200]
conn.execute(
"INSERT OR IGNORE INTO gov_search (rowid, title, site_name, summary) VALUES (?,?,?,?)",
(cur.lastrowid, title, SITE_NAME, summary),
)
return True
return False
except sqlite3.OperationalError as e:
if "locked" in str(e):
time.sleep(3)
return store_record(conn, url, title, content, date, has_table)
raise
def main():
import requests
s = requests.Session()
s.headers.update(HEADERS)
conn = sqlite3.connect(DB_PATH, timeout=60)
conn.execute("PRAGMA busy_timeout=60000")
conn.execute("PRAGMA journal_mode=WAL")
new_count = 0
skip_count = 0
seen_urls = set()
for page in range(1, _PAGES + 1):
if page == 1:
url = LIST_URL
else:
url = f"{LIST_URL}index_{page-1}.html"
try:
r = fetch(url, session=s)
html_text = r.content.decode("utf-8", errors="replace")
except Exception as e:
print(f"Page {page} fetch error: {e}", flush=True)
break
items = parse_list(html_text, url)
print(f"Page {page}: found {len(items)} items", flush=True)
if not items and page > 1:
print("Empty page, stop pagination", flush=True)
break
for abs_url, title, date in items:
if abs_url in seen_urls:
continue
seen_urls.add(abs_url)
try:
r = fetch(abs_url, session=s)
dhtml = r.content.decode("utf-8", errors="replace")
except Exception as e:
print(f" detail error {abs_url}: {e}", flush=True)
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, _ = html_to_text(content_html, abs_url)
plain_len = len(re.sub(r"<[^>]+>", "", content).strip())
if plain_len < 10:
print(f" skip empty: {dtitle[:30]}", flush=True)
skip_count += 1
continue
ok = store_record(conn, abs_url, dtitle, content, ddate, has_table)
if ok:
new_count += 1
print(f" + {dtitle[:40]} ({ddate})", flush=True)
else:
skip_count += 1
time.sleep(0.3)
time.sleep(0.5)
conn.commit()
conn.close()
print(f"新增: {new_count}", flush=True)
print(f"跳过: {skip_count}", flush=True)
print(f"=== {SITE_NAME} done ===", flush=True)
if __name__ == "__main__":
main()