494 lines
18 KiB
Python
494 lines
18 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
政策法规检索与整理系统
|
||
自动化从中国税务相关部门网站抓取、筛选、下载和整理政策法规文件
|
||
"""
|
||
|
||
import argparse
|
||
import logging
|
||
import os
|
||
import sys
|
||
import json
|
||
import hashlib
|
||
import time
|
||
import re
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import List, Dict, Optional, Tuple
|
||
from urllib.parse import urljoin, urlparse
|
||
import subprocess
|
||
|
||
import yaml
|
||
import requests
|
||
from bs4 import BeautifulSoup
|
||
from apscheduler.schedulers.blocking import BlockingScheduler
|
||
from apscheduler.triggers.cron import CronTrigger
|
||
import pandas as pd
|
||
|
||
from notifier import EmailNotifier
|
||
|
||
|
||
class PolicyRetrievalSystem:
|
||
"""政策法规检索与整理系统主类"""
|
||
|
||
def __init__(self, config_path: str = None):
|
||
self.base_dir = Path(__file__).parent
|
||
self.config_path = config_path or str(self.base_dir / "config.yaml")
|
||
self.config = self._load_config()
|
||
self.setup_logging()
|
||
self.logger = logging.getLogger(__name__)
|
||
self.scheduler = None
|
||
self.results = []
|
||
self.notifier = EmailNotifier(self.config)
|
||
self.recipients = self.config.get('notification', {}).get('email', {}).get('to_addrs', [])
|
||
|
||
def _load_config(self) -> dict:
|
||
"""加载配置文件"""
|
||
try:
|
||
with open(self.config_path, 'r', encoding='utf-8') as f:
|
||
return yaml.safe_load(f)
|
||
except FileNotFoundError:
|
||
return self._default_config()
|
||
|
||
def _default_config(self) -> dict:
|
||
"""默认配置"""
|
||
return {
|
||
'scheduler': {'enabled': False, 'time': '09:00', 'days': ['mon', 'tue', 'wed', 'thu', 'fri']},
|
||
'targets': [{'name': '国家税务总局', 'url': 'https://www.chinatax.gov.cn/', 'enabled': True}],
|
||
'download': {'path': './downloads', 'formats': ['pdf', 'doc', 'docx', 'txt']},
|
||
'deduplication': {'title_similarity': 0.8, 'content_similarity': 0.9},
|
||
'categories': [{'name': '税收政策', 'keywords': ['税收', '税务']}]
|
||
}
|
||
|
||
def setup_logging(self):
|
||
"""设置日志"""
|
||
log_config = self.config.get('logging', {})
|
||
log_dir = self.base_dir / 'logs'
|
||
log_dir.mkdir(exist_ok=True)
|
||
|
||
logging.basicConfig(
|
||
level=getattr(logging, log_config.get('level', 'INFO')),
|
||
format=log_config.get('format', '%(asctime)s - %(name)s - %(levelname)s - %(message)s'),
|
||
handlers=[
|
||
logging.FileHandler(log_config.get('file', './logs/policy_retrieval.log')),
|
||
logging.StreamHandler()
|
||
]
|
||
)
|
||
|
||
def run(self, send_email: bool = True):
|
||
"""执行一次完整的检索流程
|
||
|
||
Args:
|
||
send_email: 是否发送邮件通知，默认为True
|
||
"""
|
||
self.logger.info("=" * 60)
|
||
self.logger.info("开始执行政策法规检索任务")
|
||
self.logger.info("=" * 60)
|
||
|
||
self.results = []
|
||
|
||
targets = [t for t in self.config.get('targets', []) if t.get('enabled', False)]
|
||
for target in targets:
|
||
self.logger.info(f"正在检索: {target['name']}")
|
||
try:
|
||
articles = self.fetch_articles(target)
|
||
self.logger.info(f"从 {target['name']} 获取到 {len(articles)} 条记录")
|
||
self.results.extend(articles)
|
||
except Exception as e:
|
||
self.logger.error(f"检索 {target['name']} 时出错: {e}")
|
||
|
||
self.logger.info(f"共获取 {len(self.results)} 条原始记录")
|
||
|
||
filtered_results = self.filter_content(self.results)
|
||
self.logger.info(f"筛选后保留 {len(filtered_results)} 条记录")
|
||
|
||
deduplicated = self.deduplicate(filtered_results)
|
||
self.logger.info(f"去重后保留 {len(deduplicated)} 条记录")
|
||
|
||
categorized = self.categorize(deduplicated)
|
||
self.logger.info(f"分类完成，共 {len(categorized)} 个类别")
|
||
|
||
downloaded = self.download_files(categorized)
|
||
self.logger.info(f"文件下载完成，{len(downloaded)} 个文件")
|
||
|
||
report_file = self.generate_report(downloaded)
|
||
|
||
self.logger.info("=" * 60)
|
||
self.logger.info("政策法规检索任务完成")
|
||
self.logger.info("=" * 60)
|
||
|
||
if send_email and self.recipients:
|
||
self.logger.info(f"正在发送邮件报告到: {self.recipients}")
|
||
for article in downloaded:
|
||
article['category'] = self.get_category(article)
|
||
|
||
success = self.notifier.send_policy_report(
|
||
articles=downloaded,
|
||
to_addrs=self.recipients,
|
||
report_file=str(report_file) if report_file else None
|
||
)
|
||
if success:
|
||
self.logger.info("邮件报告发送成功")
|
||
else:
|
||
self.logger.warning("邮件报告发送失败")
|
||
|
||
return downloaded
|
||
|
||
def fetch_articles(self, target: Dict) -> List[Dict]:
|
||
"""从目标网站获取文章列表"""
|
||
articles = []
|
||
keywords = target.get('keywords', [])
|
||
base_url = target['url']
|
||
|
||
try:
|
||
headers = {
|
||
'User-Agent': self.config.get('download', {}).get('user_agent', 'Mozilla/5.0')
|
||
}
|
||
response = requests.get(base_url, headers=headers, timeout=30)
|
||
response.encoding = 'utf-8'
|
||
soup = BeautifulSoup(response.text, 'html.parser')
|
||
|
||
links = soup.find_all('a', href=True)
|
||
for link in links:
|
||
href = link.get('href', '')
|
||
text = link.get_text(strip=True)
|
||
|
||
if any(kw in text for kw in keywords):
|
||
full_url = urljoin(base_url, href)
|
||
article = {
|
||
'title': text,
|
||
'url': full_url,
|
||
'source': target['name'],
|
||
'fetch_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
|
||
'keywords': [kw for kw in keywords if kw in text]
|
||
}
|
||
articles.append(article)
|
||
|
||
for article in articles:
|
||
try:
|
||
detail = self.fetch_article_detail(article['url'], headers)
|
||
article.update(detail)
|
||
except Exception as e:
|
||
self.logger.warning(f"获取详情失败: {article['url']} - {e}")
|
||
|
||
except Exception as e:
|
||
self.logger.error(f"抓取 {target['name']} 失败: {e}")
|
||
|
||
return articles
|
||
|
||
def fetch_article_detail(self, url: str, headers: Dict) -> Dict:
|
||
"""获取文章详情"""
|
||
detail = {'publish_date': '', 'content': '', 'summary': '', 'file_url': ''}
|
||
|
||
try:
|
||
response = requests.get(url, headers=headers, timeout=30)
|
||
response.encoding = 'utf-8'
|
||
soup = BeautifulSoup(response.text, 'html.parser')
|
||
|
||
date_pattern = r'(\d{4}[-/年]\d{1,2}[-/月]\d{1,2}[日]?)'
|
||
text_content = soup.get_text()
|
||
date_match = re.search(date_pattern, text_content)
|
||
if date_match:
|
||
detail['publish_date'] = date_match.group(1).replace('年', '-').replace('月', '-').replace('日', '')
|
||
|
||
main_content = soup.find('div', class_=re.compile('content|article|text'))
|
||
if main_content:
|
||
detail['content'] = main_content.get_text(strip=True)[:500]
|
||
detail['summary'] = detail['content'][:200] + '...' if len(detail['content']) > 200 else detail['content']
|
||
|
||
file_links = soup.find_all('a', href=re.compile(r'\.(pdf|doc|docx|xls|xlsx|txt)$', re.I))
|
||
if file_links:
|
||
detail['file_url'] = file_links[0].get('href', '')
|
||
|
||
except Exception as e:
|
||
self.logger.warning(f"解析详情失败: {url} - {e}")
|
||
|
||
return detail
|
||
|
||
def filter_content(self, articles: List[Dict]) -> List[Dict]:
|
||
"""筛选相关内容"""
|
||
filter_keywords = ['最新', '通知', '公告', '政策', '法规']
|
||
filtered = []
|
||
|
||
for article in articles:
|
||
title = article.get('title', '')
|
||
if any(kw in title for kw in filter_keywords):
|
||
filtered.append(article)
|
||
|
||
return filtered
|
||
|
||
def deduplicate(self, articles: List[Dict]) -> List[Dict]:
|
||
"""内容去重"""
|
||
dedup_config = self.config.get('deduplication', {})
|
||
title_threshold = dedup_config.get('title_similarity', 0.8)
|
||
|
||
seen = {}
|
||
unique_articles = []
|
||
|
||
for article in articles:
|
||
title_hash = hashlib.md5(article.get('title', '').encode()).hexdigest()
|
||
|
||
is_duplicate = False
|
||
for seen_title, seen_data in seen.items():
|
||
similarity = self.calculate_similarity(article.get('title', ''), seen_title)
|
||
if similarity >= title_threshold:
|
||
if article.get('publish_date') < seen_data.get('publish_date'):
|
||
del seen[seen_title]
|
||
seen[article.get('title', '')] = article
|
||
is_duplicate = True
|
||
break
|
||
|
||
if not is_duplicate:
|
||
seen[article.get('title', '')] = article
|
||
unique_articles.append(article)
|
||
|
||
return unique_articles
|
||
|
||
def calculate_similarity(self, text1: str, text2: str) -> float:
|
||
"""计算文本相似度"""
|
||
if not text1 or not text2:
|
||
return 0.0
|
||
|
||
set1 = set(text1)
|
||
set2 = set(text2)
|
||
intersection = len(set1 & set2)
|
||
union = len(set1 | set2)
|
||
|
||
return intersection / union if union > 0 else 0.0
|
||
|
||
def categorize(self, articles: List[Dict]) -> Dict[str, List[Dict]]:
|
||
"""分类整理"""
|
||
categories_config = self.config.get('categories', [])
|
||
categorized = {}
|
||
|
||
for category in categories_config:
|
||
categorized[category['name']] = []
|
||
|
||
categorized['其他政策'] = []
|
||
|
||
for article in articles:
|
||
content = article.get('title', '') + ' ' + article.get('content', '')
|
||
assigned = False
|
||
|
||
for category in sorted(categories_config, key=lambda x: x.get('priority', 99)):
|
||
keywords = category.get('keywords', [])
|
||
if any(kw in content for kw in keywords):
|
||
categorized[category['name']].append(article)
|
||
assigned = True
|
||
break
|
||
|
||
if not assigned:
|
||
categorized['其他政策'].append(article)
|
||
|
||
return categorized
|
||
|
||
def download_files(self, categorized: Dict[str, List[Dict]]) -> List[Dict]:
|
||
"""下载文件"""
|
||
download_config = self.config.get('download', {})
|
||
download_path = Path(download_config.get('path', './downloads'))
|
||
download_path.mkdir(parents=True, exist_ok=True)
|
||
|
||
formats = download_config.get('formats', ['pdf', 'doc', 'docx', 'txt'])
|
||
downloaded = []
|
||
|
||
for category, articles in categorized.items():
|
||
category_path = download_path / category
|
||
category_path.mkdir(exist_ok=True)
|
||
|
||
for article in articles:
|
||
file_url = article.get('file_url', '')
|
||
if not file_url:
|
||
continue
|
||
|
||
if any(file_url.lower().endswith(f'.{fmt}') for fmt in formats):
|
||
try:
|
||
filename = self.download_file(file_url, category_path)
|
||
article['local_path'] = str(category_path / filename)
|
||
downloaded.append(article)
|
||
except Exception as e:
|
||
self.logger.warning(f"下载失败: {file_url} - {e}")
|
||
|
||
return downloaded
|
||
|
||
def download_file(self, url: str, save_path: Path) -> str:
|
||
"""下载单个文件"""
|
||
headers = {'User-Agent': self.config.get('download', {}).get('user_agent', 'Mozilla/5.0')}
|
||
|
||
response = requests.get(url, headers=headers, timeout=60, stream=True)
|
||
response.raise_for_status()
|
||
|
||
filename = Path(urlparse(url).path).name
|
||
if not filename:
|
||
filename = f"document_{int(time.time())}.pdf"
|
||
|
||
filepath = save_path / filename
|
||
with open(filepath, 'wb') as f:
|
||
for chunk in response.iter_content(chunk_size=8192):
|
||
f.write(chunk)
|
||
|
||
return filename
|
||
|
||
def generate_report(self, articles: List[Dict]) -> str:
|
||
"""生成汇总报告
|
||
|
||
Returns:
|
||
str: 报告文件路径
|
||
"""
|
||
output_dir = self.base_dir / 'output'
|
||
output_dir.mkdir(exist_ok=True)
|
||
|
||
today = datetime.now().strftime('%Y%m%d')
|
||
report_file = output_dir / f'summary_{today}.xlsx'
|
||
|
||
if not articles:
|
||
self.logger.warning("没有数据生成报告")
|
||
return ""
|
||
|
||
df_data = []
|
||
for article in articles:
|
||
df_data.append({
|
||
'标题': article.get('title', ''),
|
||
'发布时间': article.get('publish_date', ''),
|
||
'来源': article.get('source', ''),
|
||
'类别': self.get_category(article),
|
||
'摘要': article.get('summary', ''),
|
||
'下载链接': article.get('local_path', article.get('file_url', '')),
|
||
'关键词': ', '.join(article.get('keywords', [])),
|
||
'抓取时间': article.get('fetch_time', '')
|
||
})
|
||
|
||
df = pd.DataFrame(df_data)
|
||
df.to_excel(report_file, index=False, engine='openpyxl')
|
||
|
||
json_file = output_dir / f'deduplicated_data_{today}.json'
|
||
with open(json_file, 'w', encoding='utf-8') as f:
|
||
json.dump(articles, f, ensure_ascii=False, indent=2)
|
||
|
||
self.logger.info(f"报告已生成: {report_file}")
|
||
self.logger.info(f"数据已保存: {json_file}")
|
||
|
||
return str(report_file)
|
||
|
||
def get_category(self, article: Dict) -> str:
|
||
"""获取文章类别"""
|
||
content = article.get('title', '') + ' ' + article.get('content', '')
|
||
categories = self.config.get('categories', [])
|
||
|
||
for category in sorted(categories, key=lambda x: x.get('priority', 99)):
|
||
keywords = category.get('keywords', [])
|
||
if any(kw in content for kw in keywords):
|
||
return category['name']
|
||
|
||
return '其他政策'
|
||
|
||
def start_scheduler(self):
|
||
"""启动定时任务"""
|
||
scheduler_config = self.config.get('scheduler', {})
|
||
if not scheduler_config.get('enabled', False):
|
||
self.logger.info("定时任务未启用")
|
||
return
|
||
|
||
self.scheduler = BlockingScheduler()
|
||
|
||
time_parts = scheduler_config.get('time', '09:00').split(':')
|
||
hour, minute = int(time_parts[0]), int(time_parts[1])
|
||
|
||
days_map = {'mon': '0', 'tue': '1', 'wed': '2', 'thu': '3', 'fri': '4', 'sat': '5', 'sun': '6'}
|
||
days = [days_map.get(d, '0') for d in scheduler_config.get('days', ['mon', 'tue', 'wed', 'thu', 'fri'])]
|
||
|
||
trigger = CronTrigger(
|
||
day_of_week=','.join(days),
|
||
hour=hour,
|
||
minute=minute
|
||
)
|
||
|
||
self.scheduler.add_job(self.run, trigger, id='policy_retrieval')
|
||
|
||
self.logger.info(f"定时任务已启动，将在每天 {scheduler_config['time']} 执行")
|
||
|
||
try:
|
||
self.scheduler.start()
|
||
except (KeyboardInterrupt, SystemExit):
|
||
self.logger.info("定时任务已停止")
|
||
self.scheduler.shutdown()
|
||
|
||
def init_config(self):
|
||
"""初始化配置文件"""
|
||
self.logger.info("配置文件已就绪")
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
parser = argparse.ArgumentParser(description='政策法规检索与整理系统')
|
||
parser.add_argument('command', choices=['init', 'run', 'schedule', 'report', 'help'],
|
||
help='命令: init=初始化, run=立即执行, schedule=定时任务, report=查看报告, help=帮助')
|
||
parser.add_argument('--config', '-c', help='配置文件路径')
|
||
parser.add_argument('--time', '-t', help='定时任务时间 (如: 09:00)')
|
||
parser.add_argument('--enable', action='store_true', help='启用定时任务')
|
||
parser.add_argument('--disable', action='store_true', help='禁用定时任务')
|
||
parser.add_argument('--no-email', action='store_true', help='不发送邮件报告')
|
||
parser.add_argument('--email-to', '-e', help='指定收件人邮箱（可多次使用）', action='append')
|
||
|
||
args = parser.parse_args()
|
||
|
||
system = PolicyRetrievalSystem(config_path=args.config)
|
||
|
||
if args.email_to:
|
||
system.recipients = args.email_to
|
||
system.config.setdefault('notification', {}).setdefault('email', {})['to_addrs'] = args.email_to
|
||
system.logger.info(f"邮件将发送到: {system.recipients}")
|
||
|
||
send_email = not args.no_email
|
||
|
||
if args.command == 'init':
|
||
system.init_config()
|
||
print("初始化完成，配置文件: config.yaml")
|
||
|
||
elif args.command == 'run':
|
||
try:
|
||
system.run(send_email=send_email)
|
||
except Exception as e:
|
||
error_msg = f"任务执行失败: {str(e)}"
|
||
system.logger.error(error_msg)
|
||
if system.notifier.is_enabled() and system.recipients:
|
||
system.notifier.send_error_alert(error_msg, system.recipients)
|
||
raise
|
||
|
||
elif args.command == 'schedule':
|
||
if args.time:
|
||
system.config['scheduler']['time'] = args.time
|
||
if args.enable:
|
||
system.config['scheduler']['enabled'] = True
|
||
elif args.disable:
|
||
system.config['scheduler']['enabled'] = False
|
||
print("定时任务已禁用")
|
||
return
|
||
|
||
with open(system.config_path, 'w', encoding='utf-8') as f:
|
||
yaml.dump(system.config, f, allow_unicode=True)
|
||
|
||
print(f"定时任务时间: {system.config['scheduler']['time']}")
|
||
print("启动定时任务...")
|
||
system.start_scheduler()
|
||
|
||
elif args.command == 'report':
|
||
output_dir = Path(__file__).parent / 'output'
|
||
if output_dir.exists():
|
||
reports = list(output_dir.glob('summary_*.xlsx'))
|
||
if reports:
|
||
latest = max(reports, key=lambda x: x.stat().st_mtime)
|
||
print(f"最新报告: {latest}")
|
||
df = pd.read_excel(latest)
|
||
print(df.to_string())
|
||
else:
|
||
print("暂无报告")
|
||
else:
|
||
print("暂无报告")
|
||
|
||
elif args.command == 'help':
|
||
parser.print_help()
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|