u6j. d Z ddlZej j d Zd ZdZd Zedk( r* edd d
d ed d
dd edddd e d yy)uQ 批量生成爬虫脚本 Part6 — trs×3 (铜仁) + fgw (淄博发改 dataproxy) Nz ~/Crawler/gov_crawler/gen_outc t j j t | }t |dd 5 }|j | d d d t
d| y # 1 sw Y xY w)Nwzutf-8)encodingWROTE)ospathjoinOUTopenwriteprint)namecontentr fs gen_crawlers_p6.pyr r sO
77<<T "D
dC' * a '4 s AA#us) #!/usr/bin/env python3
"""铜仁市生态环境局 - {display}"""
import re, sys, os, json
import urllib.request, urllib.error
from bs4 import BeautifulSoup
import sqlite3
BASE_URL = "https://sthjj.trs.gov.cn"
LIST_URL = BASE_URL + "{list_path}"
SITE_NAME = "{site_name}"
SITE_DISPLAY = "铜仁市生态环境局"
GROUP_NAME = "环评平台"
DB = os.environ.get("SEARCH_DB", "/root/search.db")
HEADERS = {{"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}}
_MAX_PG = None
for i, a in enumerate(sys.argv):
if a == "--pages" and i + 1 < len(sys.argv):
_MAX_PG = int(sys.argv[i + 1])
break
def html_table_to_html(table, base_url=""):
"""保留 HTML 表格结构，仅将相对链接/图片转绝对 URL"""
import urllib.parse
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 fetch(url):
req = urllib.request.Request(url, headers=HEADERS)
try:
return urllib.request.urlopen(req, timeout=30).read().decode("utf-8", errors="replace")
except Exception as e:
print(f"[WARN] fetch failed: {url} - {e}", file=sys.stderr)
return ""
def extract_items(html):
"""Extract items from a list page"""
soup = BeautifulSoup(html, "html.parser")
items = []
for li in soup.select("ul.NewsList > li"):
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_span = li.find("span")
date_text = date_span.get_text(strip=True) if date_span else ""
if not href.startswith("http"):
href = BASE_URL + href
items.append({{"title": title.strip(), "url": href, "date": date_text}})
return items
def get_total_pages(html):
"""Get total pages from createPageHTML script"""
m = re.search(r'createPageHTML\((\d+),\s*\d+,\s*"index', html)
if m:
return int(m.group(1))
return None
def parse_detail(html, url):
"""Parse detail page content"""
soup = BeautifulSoup(html, "html.parser")
title = ""
title_div = soup.select_one("div.ArticleTitle")
if title_div:
title = title_div.get_text(strip=True)
if not title:
meta_title = soup.find("meta", attrs={{"name": "ArticleTitle"}})
if meta_title and meta_title.get("content"):
title = meta_title["content"].strip()
if not title and soup.title:
title = soup.title.string.strip()
date_text = ""
meta_date = soup.find("meta", attrs={{"name": "PubDate"}})
if meta_date and meta_date.get("content"):
m = re.search(r"(\d{{4}}-\d{{2}}-\d{{2}})", meta_date["content"])
if m:
date_text = m.group(1)
zoom = soup.find("font", id="Zoom")
if not zoom:
zoom = soup.select_one("font#Zoom")
if not zoom:
content_div = soup.select_one("div.Article_Con")
if content_div:
zoom = content_div.find("font", id="Zoom")
if not zoom:
zoom = soup.select_one("div.trs_editor_view")
if not zoom:
return {{"title": title, "date": date_text, "content": "", "attachments": []}}
attachments = []
parts = []
for el in zoom.find_all(["p", "table", "img"], recursive=True):
if el.name == "p" and el.find_parent("table"):
continue
if el.name == "table" and el.find_parent("table"):
continue
if el.name == "p":
text = el.get_text(" ", strip=True)
for a in el.find_all("a", href=True, appendix=True):
href = a["href"].strip()
atitle = a.get_text(strip=True) or a.get("download", "")
oldsrc = a.get("OLDSRC", "")
if oldsrc:
if oldsrc.startswith("/"):
ahref = BASE_URL + oldsrc
elif oldsrc.startswith("http"):
ahref = oldsrc
else:
ahref = url.rsplit("/", 1)[0] + "/" + oldsrc
else:
if href.startswith("./"):
ahref = url.rsplit("/", 1)[0] + "/" + href[2:]
elif href.startswith("/"):
ahref = BASE_URL + href
elif href.startswith("http"):
ahref = href
else:
ahref = url.rsplit("/", 1)[0] + "/" + href
attachments.append({{"title": atitle, "url": ahref}})
if text:
parts.append(text)
elif el.name == 'table':
tbl_html = html_table_to_html(el, url)
if tbl_html:
parts.append(tbl_html)
elif el.name == "img":
src = el.get("src", "")
alt = el.get("alt", "")
if src:
if src.startswith("//"):
src = "https:" + src
elif src.startswith("/"):
src = BASE_URL + src
elif not src.startswith("http"):
src = url.rsplit("/", 1)[0] + "/" + src
if alt:
parts.append(f'{alt}')
else:
parts.append(f'图片')
for a in zoom.find_all("a", href=True):
href = a["href"].strip()
if re.search(r'\.(pdf|docx?|xlsx?|rar|zip)$', href, re.I) and not a.get("appendix"):
atitle = a.get_text(strip=True) or a.get("download", "")
if atitle and href != "./" and not any(att["url"].endswith(href.split("/")[-1]) for att in attachments):
oldsrc = a.get("OLDSRC", "")
if oldsrc:
ahref = (BASE_URL + oldsrc) if oldsrc.startswith("/") else (url.rsplit("/", 1)[0] + "/" + oldsrc)
else:
if href.startswith("./"):
ahref = url.rsplit("/", 1)[0] + "/" + href[2:]
elif href.startswith("/"):
ahref = BASE_URL + href
elif href.startswith("http"):
ahref = href
else:
ahref = url.rsplit("/", 1)[0] + "/" + href
attachments.append({{"title": atitle, "url": ahref}})
seen = set()
unique_attachments = []
for att in attachments:
if att["url"] not in seen:
seen.add(att["url"])
unique_attachments.append(att)
content = "\n\n".join(parts)
return {{"title": title, "date": date_text, "content": content, "attachments": unique_attachments}}
def main():
html = fetch(LIST_URL)
if not html:
print("[ERROR] Cannot fetch list page", file=sys.stderr)
sys.exit(1)
total_pages = get_total_pages(html)
if total_pages is None:
soup = BeautifulSoup(html, "html.parser")
total_pages = 1
for a in soup.select("div.page a"):
m = re.search(r'index_(\d+)\.html', a.get("href", ""))
if m:
p = int(m.group(1)) + 1
if p > total_pages:
total_pages = p
print(f"[INFO] Total pages: {total_pages}", file=sys.stderr)
if _MAX_PG and total_pages > _MAX_PG:
print(f"[INFO] Limiting to {_MAX_PG} pages (--pages={_MAX_PG})", file=sys.stderr)
total_pages = _MAX_PG
all_items = []
for pg in range(total_pages):
if pg == 0:
url = LIST_URL
else:
url = f"{{LIST_URL}}index_{{pg}}.html"
print(f"[INFO] Fetching page {{pg+1}}/{{total_pages}}: {{url}}", file=sys.stderr)
html = fetch(url)
if not html:
print(f"[WARN] Empty response for page {{pg+1}}", file=sys.stderr)
continue
items = extract_items(html)
print(f"[INFO] Found {{len(items)}} items on page {{pg+1}}", file=sys.stderr)
all_items.extend(items)
print(f"[INFO] Total items from list: {{len(all_items)}}", file=sys.stderr)
conn = sqlite3.connect(DB)
c = conn.cursor()
new_count = 0
empty_content = 0
total_attachments = 0
for item in all_items:
c.execute("SELECT id FROM gov_raw WHERE page_url=?", (item["url"],))
if c.fetchone():
continue
print(f"[INFO] Fetching detail: {{item['title'][:50]}}...", file=sys.stderr)
html = fetch(item["url"])
if not html:
print(f"[WARN] Cannot fetch detail, skipping: {{item['url']}}", file=sys.stderr)
continue
detail = parse_detail(html, item["url"])
title = detail["title"] or item["title"]
date_text = detail["date"] or item["date"]
content = detail["content"]
attachments = detail["attachments"]
plain = re.sub(r'<[^>]+>', '', content)
plain = re.sub(r'\s+', ' ', plain).strip()
summary = plain[:200] if plain else ""
if not content:
empty_content += 1
print(f"[WARN] Empty content: {title}", file=sys.stderr)
date_rank = 0
if date_text:
m = re.search(r"(\d{{4}})-(\d{{2}})-(\d{{2}})", date_text)
if m:
date_rank = int(m.group(1) + m.group(2) + m.group(3))
att_json = json.dumps(attachments, ensure_ascii=False) if attachments else "[]"
try:
c.execute(
"INSERT OR IGNORE INTO gov_raw (title, site_name, group_name, page_url, publish_date, content, summary, attachments, date_rank) VALUES (?,?,?,?,?,?,?,?,?)",
(title, SITE_NAME, GROUP_NAME, item["url"], date_text, content, summary, att_json, date_rank)
)
if c.rowcount > 0:
new_count += 1
total_attachments += len(attachments)
except Exception as e:
print(f"[ERROR] DB insert failed: {{title[:30]}} - {{e}}", file=sys.stderr)
conn.commit()
conn.close()
print(f"\n[RESULT] {{SITE_NAME}}: {{new_count}} new, {{empty_content}} empty content, {{total_attachments}} attachments")
if __name__ == "__main__":
main()
c t }|j d| }|j d| }|j d| }|j dd j dd }t | | y )Nz {list_path}z {site_name}z {display}z{{{z}}})TRS_TPLreplacer )r list_path site_namedisplayss r gen_trs3r 1 s] A -+A -+A +w'A $ $$T3/A $N __main__zcrawl_trs_nspgs.pyz /xzsp/nspgs/u 铜仁市-拟审批公示u 拟审批公示zcrawl_trs_sphggl.pyz
/xzsp/sphggl/u 铜仁市-审批后公告u 审批后公告栏zcrawl_trs_bypzgg.pyz
/xzsp/bypzgg/u 铜仁市-不予批准公告u 不予批准公告u DONE part6 (trs×3))
__doc__r r
expanduserr
r r r __name__r
r r r" s~ W gg89_ B z
!>3NPab
"O5PRfg
"O5SUij
r