本文目录导读：
我来为您创建一个定时检查并清理无效链接的脚本,这个脚本可以检查文件中的链接是否有效，并自动清理无效的链接。
基础版：检查单个文件中的链接
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
定时检查并清理无效链接脚本
"""
import os
import re
import json
import requests
import time
import logging
from datetime import datetime, timedelta
from urllib.parse import urlparse
from concurrent.futures import ThreadPoolExecutor, as_completed
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('link_checker.log', encoding='utf-8'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
class LinkChecker:
def __init__(self, config_file='config.json'):
"""初始化链接检查器"""
self.config = self.load_config(config_file)
self.valid_links = []
self.invalid_links = []
self.timeout = self.config.get('timeout', 5)
self.retry_times = self.config.get('retry_times', 2)
self.max_workers = self.config.get('max_workers', 10)
def load_config(self, config_file):
"""加载配置文件"""
default_config = {
'files_to_check': ['README.md', 'links.txt'], # 要检查的文件
'backup_dir': 'backup', # 备份目录
'cleanup_mode': 'comment', # 'comment' 或 'delete'
'timeout': 5, # 请求超时时间
'retry_times': 2, # 重试次数
'max_workers': 10, # 并发线程数
'check_interval': 3600, # 检查间隔（秒）
'report_file': 'link_report.json', # 报告文件
'link_patterns': [ # 支持多种链接格式
r'https?://[^\s<>"\'()]+\.[^\s<>"\'()]+',
r'https?://[^\s<>"\'()\[\]]+'
]
}
try:
if os.path.exists(config_file):
with open(config_file, 'r', encoding='utf-8') as f:
loaded_config = json.load(f)
default_config.update(loaded_config)
logger.info(f"配置文件 {config_file} 加载成功")
else:
self.save_config(config_file, default_config)
logger.info(f"已创建默认配置文件 {config_file}")
except Exception as e:
logger.error(f"加载配置文件失败: {e}")
return default_config
def save_config(self, config_file, config):
"""保存配置文件"""
with open(config_file, 'w', encoding='utf-8') as f:
json.dump(config, f, indent=4, ensure_ascii=False)
def extract_links(self, file_path):
"""从文件中提取所有链接"""
links = []
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
for pattern in self.config['link_patterns']:
found_links = re.findall(pattern, content)
links.extend(found_links)
# 去重
links = list(set(links))
logger.info(f"从 {file_path} 中提取到 {len(links)} 个链接")
return links
except Exception as e:
logger.error(f"提取链接失败 {file_path}: {e}")
return []
def check_link(self, url):
"""检查单个链接是否有效"""
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}
for attempt in range(self.retry_times + 1):
try:
response = requests.get(
url,
headers=headers,
timeout=self.timeout,
allow_redirects=True
)
if response.status_code == 200:
return url, True, "OK"
elif response.status_code == 403:
return url, True, "Forbidden (可能有效)"
elif response.status_code == 404:
return url, False, f"404 Not Found"
else:
return url, False, f"HTTP {response.status_code}"
except requests.ConnectionError:
if attempt < self.retry_times:
time.sleep(1)
continue
return url, False, "Connection Error"
except requests.Timeout:
if attempt < self.retry_times:
time.sleep(1)
continue
return url, False, "Timeout"
except Exception as e:
if attempt < self.retry_times:
time.sleep(1)
continue
return url, False, str(e)
return url, False, "Max retries exceeded"
def check_all_links(self, file_path):
"""检查文件中的所有链接"""
links = self.extract_links(file_path)
if not links:
logger.warning(f"文件 {file_path} 中没有找到链接")
return [], []
self.valid_links = []
self.invalid_links = []
logger.info(f"开始检查 {len(links)} 个链接...")
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
future_to_url = {executor.submit(self.check_link, url): url for url in links}
for future in as_completed(future_to_url):
url, is_valid, message = future.result()
if is_valid:
self.valid_links.append(url)
logger.info(f"✓ {url[:50]}... - {message}")
else:
self.invalid_links.append({'url': url, 'reason': message})
logger.warning(f"✗ {url[:50]}... - {message}")
return self.valid_links, self.invalid_links
def backup_file(self, file_path):
"""备份文件"""
if not os.path.exists(self.config['backup_dir']):
os.makedirs(self.config['backup_dir'])
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
filename = os.path.basename(file_path)
backup_path = os.path.join(
self.config['backup_dir'],
f"{filename}.{timestamp}.bak"
)
try:
import shutil
shutil.copy2(file_path, backup_path)
logger.info(f"已备份文件到 {backup_path}")
return True
except Exception as e:
logger.error(f"备份失败: {e}")
return False
def cleanup_links(self, file_path, invalid_links):
"""清理无效链接"""
if not invalid_links:
logger.info("没有发现无效链接，无需清理")
return
# 创建备份
self.backup_file(file_path)
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
cleanup_mode = self.config.get('cleanup_mode', 'comment')
invalid_urls = [link['url'] for link in invalid_links]
for url in invalid_urls:
if cleanup_mode == 'delete':
# 删除包含无效链接的整行
lines = content.split('\n')
content = '\n'.join([
line for line in lines
if url not in line
])
logger.info(f"已删除包含 {url[:50]}... 的行")
else:
# 注释掉包含无效链接的行
lines = content.split('\n')
new_lines = []
for line in lines:
if url in line and not line.strip().startswith('#'):
new_lines.append(f"# INVALID LINK: {line}")
else:
new_lines.append(line)
content = '\n'.join(new_lines)
logger.info(f"已注释包含 {url[:50]}... 的行")
with open(file_path, 'w', encoding='utf-8') as f:
f.write(content)
logger.info(f"清理完成，已处理 {len(invalid_urls)} 个无效链接")
except Exception as e:
logger.error(f"清理链接失败: {e}")
def generate_report(self, file_path):
"""生成检查报告"""
report = {
'timestamp': datetime.now().isoformat(),
'file_checked': file_path,
'total_links': len(self.valid_links) + len(self.invalid_links),
'valid_links': len(self.valid_links),
'invalid_links': len(self.invalid_links),
'valid_urls': self.valid_links,
'invalid_urls': self.invalid_links,
'cleanup_mode': self.config['cleanup_mode']
}
report_file = self.config['report_file']
try:
with open(report_file, 'w', encoding='utf-8') as f:
json.dump(report, f, indent=2, ensure_ascii=False)
logger.info(f"报告已保存到 {report_file}")
except Exception as e:
logger.error(f"保存报告失败: {e}")
return report
def run_check(self, file_path=None):
"""运行链接检查"""
if file_path is None:
file_paths = self.config['files_to_check']
else:
file_paths = [file_path]
for fp in file_paths:
if not os.path.exists(fp):
logger.warning(f"文件 {fp} 不存在，跳过")
continue
logger.info(f"="*50)
logger.info(f"开始检查文件: {fp}")
valid_links, invalid_links = self.check_all_links(fp)
if invalid_links:
logger.info(f"\n发现 {len(invalid_links)} 个无效链接:")
for link in invalid_links:
logger.info(f" {link['url'][:60]}... - {link['reason']}")
# 清理无效链接
self.cleanup_links(fp, invalid_links)
else:
logger.info("所有链接看起来都有效！")
# 生成报告
self.generate_report(fp)
def run_scheduled(self):
"""定时运行检查"""
logger.info(f"启动定时链接检查，间隔: {self.config['check_interval']}秒")
while True:
try:
self.run_check()
next_check = datetime.now() + timedelta(seconds=self.config['check_interval'])
logger.info(f"下次检查时间: {next_check.strftime('%Y-%m-%d %H:%M:%S')}")
time.sleep(self.config['check_interval'])
except KeyboardInterrupt:
logger.info("用户中断，停止检查")
break
except Exception as e:
logger.error(f"检查过程出错: {e}")
time.sleep(60) # 出错后等待一分钟再试
def main():
"""主函数"""
import argparse
parser = argparse.ArgumentParser(description='定时检查并清理无效链接')
parser.add_argument('--config', '-c', default='config.json', help='配置文件路径')
parser.add_argument('--file', '-f', help='要检查的特定文件')
parser.add_argument('--once', '-o', action='store_true', help='只运行一次检查')
parser.add_argument('--mode', '-m', choices=['comment', 'delete'], help='清理模式')
args = parser.parse_args()
checker = LinkChecker(args.config)
if args.mode:
checker.config['cleanup_mode'] = args.mode
if args.once:
checker.run_check(args.file)
else:
checker.run_scheduled()
if __name__ == '__main__':
main()
使用说明
安装依赖
pip install requests
创建配置文件 config.json
{
"files_to_check": ["README.md", "links.txt", "docs/*.md"],
"backup_dir": "backup",
"cleanup_mode": "comment",
"timeout": 5,
"retry_times": 2,
"max_workers": 10,
"check_interval": 3600,
"report_file": "link_report.json",
"link_patterns": [
"https?://[^\\s<>\"'()]+\\.[^\\s<>\"'()]+",
"https?://[^\\s<>\"'()\\[\\]]+"
]
}
运行脚本
# 单次运行 python link_checker.py --once # 定时运行（默认每小时检查一次） python link_checker.py # 检查特定文件 python link_checker.py --file README.md --once # 设置清理模式为删除 python link_checker.py --mode delete --once # 使用自定义配置 python link_checker.py --config my_config.json
高级功能
添加链接白名单
def should_skip(self, url):
"""检查链接是否应该跳过"""
skip_patterns = [
'localhost',
'127.0.0.1',
'example.com',
]
for pattern in skip_patterns:
if pattern in url:
return True
return False
增强的链接检测
def extract_all_links(self, file_path):
"""从文件中提取所有类型的链接"""
import markdown
from bs4 import BeautifulSoup
# 读取文件
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
# 提取 Markdown 链接
md_links = re.findall(r'\[([^\]]+)\]\(([^\)]+)\)', content)
# 提取 HTML 链接
soup = BeautifulSoup(markdown.markdown(content), 'html.parser')
html_links = [a.get('href') for a in soup.find_all('a') if a.get('href')]
return list(set(md_links + html_links))
定时任务调度
# 使用 systemd 或 crontab 实现定时运行 # 添加到 crontab # 0 */2 * * * /usr/bin/python3 /path/to/link_checker.py --once
这个脚本提供了完整的链接检查和清理功能,可以根据需要调整配置和功能。
