202 lines
6.7 KiB
Python
202 lines
6.7 KiB
Python
import re
|
||
import os
|
||
import requests
|
||
import time
|
||
from urllib.parse import urlparse, unquote
|
||
import hashlib
|
||
|
||
# 请求网址
|
||
url = 'https://mp.weixin.qq.com/s/DWsowFDU7D5baNyTGXDWGw'
|
||
|
||
# 请求头（添加更多头部信息以提高成功率）
|
||
headers = {
|
||
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36',
|
||
'referer': 'https://mp.weixin.qq.com/',
|
||
'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8',
|
||
'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8',
|
||
'cache-control': 'max-age=0',
|
||
'connection': 'keep-alive',
|
||
}
|
||
|
||
try:
|
||
response = requests.get(url=url, headers=headers, timeout=30)
|
||
response.raise_for_status()
|
||
response.encoding = response.apparent_encoding
|
||
except Exception as e:
|
||
print(f"请求失败: {e}")
|
||
exit()
|
||
|
||
# 改进的正则表达式，更精确地匹配图片链接
|
||
# 匹配微信文章中的图片链接格式
|
||
patterns = [
|
||
# 匹配微信图片链接（带wx_fmt参数）
|
||
r'data-src="(https?://mmbiz\.qpic\.cn/[^"]+?wx_fmt=(?:jpeg|png|gif|bmp|webp)[^"]*?)"',
|
||
r'src="(https?://mmbiz\.qpic\.cn/[^"]+?wx_fmt=(?:jpeg|png|gif|bmp|webp)[^"]*?)"',
|
||
# 匹配其他常见的图片链接
|
||
r'(https?://mmbiz\.qpic\.cn/[^"\s]+?\.(?:jpg|jpeg|png|gif|bmp|webp)[^\s"]*)',
|
||
]
|
||
|
||
all_image_urls = []
|
||
for pattern in patterns:
|
||
matches = re.findall(pattern, response.text, re.IGNORECASE)
|
||
all_image_urls.extend(matches)
|
||
|
||
# 过滤掉不需要的链接
|
||
def is_valid_image_url(url):
|
||
"""检查是否是有效的图片链接"""
|
||
# 排除JS文件、GIF占位图等
|
||
exclude_patterns = [
|
||
'js',
|
||
'pic_blank.gif',
|
||
'//res.wx.qq.com/',
|
||
'.concat',
|
||
'hit_biz_headimg',
|
||
'emoji_url',
|
||
]
|
||
|
||
for pattern in exclude_patterns:
|
||
if pattern in url:
|
||
return False
|
||
|
||
# 必须是有效的URL
|
||
if not url.startswith(('http://', 'https://')):
|
||
return False
|
||
|
||
# 检查是否是图片格式
|
||
img_extensions = ['.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp']
|
||
for ext in img_extensions:
|
||
if ext in url.lower():
|
||
return True
|
||
|
||
# 或者包含微信图片格式参数
|
||
if 'wx_fmt=' in url.lower():
|
||
return True
|
||
|
||
return False
|
||
|
||
# 过滤和去重
|
||
valid_image_urls = []
|
||
for img_url in all_image_urls:
|
||
if is_valid_image_url(img_url) and img_url not in valid_image_urls:
|
||
valid_image_urls.append(img_url)
|
||
|
||
print(f"找到 {len(valid_image_urls)} 个有效图片链接:")
|
||
for i, img_url in enumerate(valid_image_urls, 1):
|
||
print(f"{i:2d}. {img_url}")
|
||
|
||
# 指定保存路径
|
||
save_path = r"F:\测试\图片"
|
||
|
||
# 确保目录存在
|
||
if not os.path.exists(save_path):
|
||
os.makedirs(save_path)
|
||
print(f"创建目录: {save_path}")
|
||
|
||
# 下载图片
|
||
success_count = 0
|
||
fail_count = 0
|
||
|
||
for i, img_url in enumerate(valid_image_urls, 1):
|
||
try:
|
||
# 生成文件名
|
||
# 从URL中提取有用的部分作为文件名
|
||
parsed_url = urlparse(img_url)
|
||
|
||
# 尝试获取图片格式
|
||
fmt_match = re.search(r'wx_fmt=(\w+)', img_url, re.IGNORECASE)
|
||
if fmt_match:
|
||
ext = fmt_match.group(1).lower()
|
||
if ext == 'jpeg':
|
||
ext = 'jpg'
|
||
else:
|
||
# 从URL路径猜测扩展名
|
||
path_lower = parsed_url.path.lower()
|
||
if '.jpg' in path_lower or '.jpeg' in path_lower:
|
||
ext = 'jpg'
|
||
elif '.png' in path_lower:
|
||
ext = 'png'
|
||
elif '.gif' in path_lower:
|
||
ext = 'gif'
|
||
elif '.webp' in path_lower:
|
||
ext = 'webp'
|
||
else:
|
||
ext = 'jpg' # 默认
|
||
|
||
# 使用序号和URL哈希作为文件名（避免特殊字符）
|
||
url_hash = hashlib.md5(img_url.encode()).hexdigest()[:6]
|
||
filename = f"weixin_image_{i:03d}_{url_hash}.{ext}"
|
||
filepath = os.path.join(save_path, filename)
|
||
|
||
# 如果文件已存在，跳过
|
||
if os.path.exists(filepath):
|
||
print(f"{i:2d}. 跳过: {filename} (已存在)")
|
||
success_count += 1
|
||
continue
|
||
|
||
print(f"{i:2d}. 正在下载: {filename}")
|
||
|
||
# 下载图片（添加图片请求特定的headers）
|
||
img_headers = headers.copy()
|
||
img_headers['accept'] = 'image/webp,image/apng,image/*,*/*;q=0.8'
|
||
|
||
img_response = requests.get(img_url, headers=img_headers, timeout=30, stream=True)
|
||
img_response.raise_for_status()
|
||
|
||
# 检查是否是图片
|
||
content_type = img_response.headers.get('Content-Type', '').lower()
|
||
if 'image' not in content_type:
|
||
print(f" 警告: 响应不是图片，类型: {content_type}")
|
||
|
||
# 获取文件大小
|
||
content_length = img_response.headers.get('Content-Length')
|
||
if content_length:
|
||
file_size = int(content_length)
|
||
else:
|
||
file_size = len(img_response.content)
|
||
|
||
# 保存图片
|
||
with open(filepath, 'wb') as f:
|
||
if hasattr(img_response, 'iter_content'):
|
||
# 使用流式下载
|
||
for chunk in img_response.iter_content(chunk_size=8192):
|
||
if chunk:
|
||
f.write(chunk)
|
||
else:
|
||
f.write(img_response.content)
|
||
|
||
# 验证文件大小
|
||
actual_size = os.path.getsize(filepath)
|
||
if file_size > 0 and actual_size < file_size * 0.9: # 如果实际大小小于预期90%
|
||
print(f" 警告: 文件可能未完整下载 (预期: {file_size}, 实际: {actual_size})")
|
||
|
||
print(f" 保存成功: {filename} ({actual_size/1024:.1f} KB)")
|
||
success_count += 1
|
||
|
||
# 延迟，避免请求过快
|
||
time.sleep(0.3)
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"{i:2d}. 下载失败: {e}")
|
||
fail_count += 1
|
||
except Exception as e:
|
||
print(f"{i:2d}. 处理失败: {e}")
|
||
fail_count += 1
|
||
|
||
print(f"\n下载完成!")
|
||
print(f"成功: {success_count} 张")
|
||
print(f"失败: {fail_count} 张")
|
||
print(f"保存路径: {save_path}")
|
||
|
||
# 显示保存的文件
|
||
if success_count > 0:
|
||
print("\n保存的文件:")
|
||
img_files = [f for f in os.listdir(save_path)
|
||
if f.lower().endswith(('.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp'))]
|
||
|
||
for i, filename in enumerate(sorted(img_files)[:10], 1): # 只显示前10个
|
||
filepath = os.path.join(save_path, filename)
|
||
size_kb = os.path.getsize(filepath) / 1024
|
||
print(f"{i:2d}. {filename} ({size_kb:.1f} KB)")
|
||
|
||
if len(img_files) > 10:
|
||
print(f"... 还有 {len(img_files) - 10} 个文件") |