求助:如何使下载的古诗词内容正常显示(不错字、不漏字、按网页页面排版)
下列代码有点问题:有错字、漏字、排版不规范等现象,请高手赐教,万分感谢!!!待修改代码:
import re
import urllib.parse
import aiohttp
import asyncio
from bs4 import BeautifulSoup
import aiofiles
from pathlib import Path
# 常量定义
BASE_URL = 'https://www.'
TANGSHI_URL = 'https://www.'
MORE_URL_TPL = f'{BASE_URL}/nocdn/ajax{{}}.aspx'
PATTERN = r"(\w+)Show\((\d+),'([\w\d]+)'\)"
HEADERS = {
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 Edg/120.0.0.0',
'Referer': 'https://www.'
}
OUTPUT_DIR = Path('tangshi_txt')
OUTPUT_DIR.mkdir(exist_ok=True)
# 仅保留非唐诗黑名单,移除作者白名单(核心:不再靠作者丢唐诗)
BLACK_HREF = ['songci', 'yuanqu', 'guwen', 'shijing', 'chuci', 'sanwen']
async def fetch_all_tangshi_urls(session):
try:
async with session.get(TANGSHI_URL, headers=HEADERS) as response:
response.raise_for_status()
html = await response.text()
soup = BeautifulSoup(html, 'html.parser')
poem_urls = []
# 修复:左侧全部栏目,不再限制单个.sons,所有分类区块全部遍历
left_all = soup.select('.main3 .left')
for block in left_all:
for a in block.find_all('a', href=True):
href = a['href'].strip()
full_url = urllib.parse.urljoin(BASE_URL, href)
# 剔除其它文体链接
if any(b in href.lower() for b in BLACK_HREF):
continue
# 只保留诗词详情页
if '/shiwenv_' in href or '/shiwen_' in href:
if full_url not in poem_urls:
poem_urls.append(full_url)
print(f"筛选后唐诗总链接:{len(poem_urls)} 首")
return poem_urls
except Exception as e:
print(f"获取唐诗列表时出错: {e}")
return []
def format_poems_content(content):
if not content:
return ""
lines = content.splitlines()
res = []
for line in lines:
strip_line = line.strip()
res.append(strip_line)
return "\n".join(res)
def parse_translation(translation_elem):
if not translation_elem:
return ""
for br in translation_elem.find_all('br'):
br.replace_with('\n')
raw = translation_elem.get_text(strip=False)
block_list = [i.strip() for i in raw.splitlines() if i.strip()]
return "\n\n".join(block_list)
def parse_annotations(base_elem):
if not base_elem:
return ""
anno_list = []
for p in base_elem.find_all('p'):
txt = p.get_text(strip=True)
if not txt or txt.startswith("▲"):
continue
anno_list.append(txt)
return "\n".join(anno_list) if anno_list else ""
async def fetch_one_poem(session, url):
try:
async with session.get(url, headers=HEADERS) as response:
response.raise_for_status()
html = await response.text()
soup = BeautifulSoup(html, 'html.parser')
base_info = soup.select_one('#sonsyuanwen .cont div:nth-child(2)')
if not base_info:
print(f"页面异常跳过:{url}")
return False
poems_name = base_info.find('h1').text.strip()
author_elem = base_info.find('p', attrs={'class': 'source'})
author_name = author_elem.get_text(strip=True) if author_elem else "未知作者"
# 正文
poems_content_elem = base_info.find('div', attrs={'class': 'contson'})
poems_content_raw = poems_content_elem.get_text(strip=False) if poems_content_elem else ""
poems_content = format_poems_content(poems_content_raw)
if not poems_content:
return False
txt_content = []
txt_content.append(f"【标题】{poems_name}")
txt_content.append(f"【作者】{author_name}")
txt_content.append("【正文】")
txt_content.append("")
txt_content.append(poems_content)
txt_content.append("")
translation_content = ""
annotation_content = ""
background_content = ""
brief_analysis_content = ""
appreciation_content = ""
# 折叠内容AJAX完整加载(保留修复后的替换逻辑)
for item in soup.select('.sons .contyishang'):
item_copy = item
div_elem = item_copy.select_one('div:nth-child(1)')
if div_elem and div_elem.has_attr('onclick'):
onclick_txt = div_elem['onclick']
match = re.match(PATTERN, onclick_txt)
if match:
type_, id_, idjm = match.groups()
params = {'id': id_, 'idjm': idjm}
try:
ajax_url = MORE_URL_TPL.format(type_)
async with session.get(ajax_url, params=params, headers=HEADERS) as more_response:
more_response.raise_for_status()
more_html = await more_response.text()
more_soup = BeautifulSoup(more_html, 'html.parser')
new_item = more_soup.select_one('.contyishang')
if new_item:
item_copy = new_item
except Exception as e:
print(f"折叠内容加载失败 {url}: {e}")
title_elem = item_copy.find('h2')
if not title_elem:
continue
title = title_elem.text.strip()
title_elem.decompose()
raw_text = item_copy.get_text(strip=False).rstrip("▲").strip()
if '译文' in title:
translation_content = parse_translation(item_copy)
elif '注释' in title:
annotation_content = parse_annotations(item_copy)
elif '创作背景' in title:
background_content = raw_text
elif '简析' in title:
brief_analysis_content = raw_text
elif '赏析' in title:
appreciation_content = raw_text
if translation_content:
txt_content.append("【译文与注释】")
txt_content.append("")
txt_content.append(translation_content)
txt_content.append("")
if annotation_content:
txt_content.append("【注释】")
txt_content.append("")
txt_content.append(annotation_content)
txt_content.append("")
if background_content:
txt_content.append("【创作背景】")
txt_content.append("")
txt_content.append(background_content)
txt_content.append("")
if brief_analysis_content:
txt_content.append("【简析】")
txt_content.append("")
txt_content.append(brief_analysis_content)
txt_content.append("")
if appreciation_content:
txt_content.append("【赏析】")
txt_content.append("")
txt_content.append(appreciation_content)
await save_poem_to_txt(poems_name, author_name, txt_content)
return True
except Exception as e:
print(f"诗文解析异常 {url}: {e}")
return False
async def save_poem_to_txt(poems_name, author_name, content_list):
try:
filename = f"{author_name}_{poems_name}"
filename = re.sub(r'[\\/*?:"<>|]', '_', filename)
filepath = OUTPUT_DIR / f"{filename}.txt"
async with aiofiles.open(filepath, 'w', encoding='utf-8') as f:
for line in content_list:
await f.write(line + '\n')
print(f'✓ 已保存:《{poems_name}》')
except Exception as e:
print(f"保存失败 {poems_name}: {e}")
async def process_poems(session, urls, semaphore):
tasks = []
for url in urls:
async with semaphore:
tasks.append(fetch_one_poem(session, url))
results = await asyncio.gather(*tasks)
success_count = sum(1 for r in results if r)
print(f"\n本轮结束:成功{success_count}/{len(urls)}")
async def async_main():
semaphore = asyncio.Semaphore(5)
async with aiohttp.ClientSession() as session:
poem_urls = await fetch_all_tangshi_urls(session)
if not poem_urls:
print("无可用唐诗链接,程序退出")
return
await process_poems(session, poem_urls, semaphore)
print(f"\n爬取完毕,文件路径:{OUTPUT_DIR.resolve()}")
def main():
asyncio.run(async_main())
if __name__ == '__main__':
main()







