手动模式
This commit is contained in:
@@ -0,0 +1,399 @@
|
||||
"""
|
||||
Markdown生成器模块
|
||||
生成公告的Markdown格式输出文件
|
||||
"""
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import List, Dict, Any, Optional
|
||||
from datetime import datetime
|
||||
from collections import defaultdict
|
||||
|
||||
try:
|
||||
from ..core.models import Announcement
|
||||
from ..core.config_manager import get_config
|
||||
from ..core.logger import get_logger
|
||||
except ImportError:
|
||||
from core.models import Announcement
|
||||
from core.config_manager import get_config
|
||||
from core.logger import get_logger
|
||||
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class MarkdownGenerator:
|
||||
"""Markdown生成器"""
|
||||
|
||||
def __init__(self, output_file: Optional[str] = None):
|
||||
"""
|
||||
初始化Markdown生成器
|
||||
|
||||
Args:
|
||||
output_file: 输出文件路径
|
||||
"""
|
||||
self.config = get_config()
|
||||
self.output_file = output_file or self.config.markdown.output_file
|
||||
self.max_entries = self.config.markdown.max_entries
|
||||
self.include_today_highlight = self.config.markdown.include_today_highlight
|
||||
|
||||
# 确保输出目录存在
|
||||
output_path = Path(self.output_file)
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def generate_markdown(self, announcements: List[Announcement],
|
||||
title: str = "广西政府采购网公告监控") -> str:
|
||||
"""
|
||||
生成Markdown内容
|
||||
|
||||
Args:
|
||||
announcements: 公告列表
|
||||
title: 文档标题
|
||||
|
||||
Returns:
|
||||
str: Markdown格式的文本
|
||||
"""
|
||||
if not announcements:
|
||||
return self._generate_empty_markdown(title)
|
||||
|
||||
# 按来源分组
|
||||
grouped_announcements = self._group_announcements_by_source(announcements)
|
||||
|
||||
# 生成Markdown
|
||||
lines = []
|
||||
lines.append(f"# {title}")
|
||||
lines.append("")
|
||||
lines.append(f"**更新时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||||
lines.append(f"**总公告数**: {len(announcements)}")
|
||||
lines.append("")
|
||||
|
||||
# 生成目录
|
||||
lines.extend(self._generate_toc(grouped_announcements))
|
||||
lines.append("---")
|
||||
lines.append("")
|
||||
|
||||
# 生成各来源的公告
|
||||
for source_name, source_announcements in grouped_announcements.items():
|
||||
lines.extend(self._generate_source_section(source_name, source_announcements))
|
||||
lines.append("")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
def _group_announcements_by_source(self, announcements: List[Announcement]) -> Dict[str, List[Announcement]]:
|
||||
"""按来源分组公告"""
|
||||
grouped = defaultdict(list)
|
||||
|
||||
for announcement in announcements:
|
||||
grouped[announcement.source_name].append(announcement)
|
||||
|
||||
# 对每个组内的公告按时间倒序排列
|
||||
for source_name in grouped:
|
||||
grouped[source_name].sort(key=lambda x: x.publish_date, reverse=True)
|
||||
|
||||
return dict(grouped)
|
||||
|
||||
def _generate_toc(self, grouped_announcements: Dict[str, List[Announcement]]) -> List[str]:
|
||||
"""生成目录"""
|
||||
lines = ["## 目录", ""]
|
||||
|
||||
for source_name, announcements in grouped_announcements.items():
|
||||
# 创建锚点链接
|
||||
anchor = self._create_anchor(source_name)
|
||||
count = len(announcements)
|
||||
lines.append(f"- [{source_name}](#{anchor}) ({count}条)")
|
||||
|
||||
return lines
|
||||
|
||||
def _generate_source_section(self, source_name: str, announcements: List[Announcement]) -> List[str]:
|
||||
"""生成来源章节"""
|
||||
lines = []
|
||||
anchor = self._create_anchor(source_name)
|
||||
|
||||
lines.append(f"## {source_name}")
|
||||
lines.append("")
|
||||
lines.append(f"**共 {len(announcements)} 条公告**")
|
||||
lines.append("")
|
||||
|
||||
# 生成公告列表
|
||||
for i, announcement in enumerate(announcements, 1):
|
||||
lines.extend(self._generate_announcement_item(announcement, i))
|
||||
|
||||
return lines
|
||||
|
||||
def _generate_announcement_item(self, announcement: Announcement, index: int) -> List[str]:
|
||||
"""生成单个公告项"""
|
||||
lines = []
|
||||
|
||||
# 公告标题
|
||||
title_prefix = ""
|
||||
if self.include_today_highlight and announcement.is_today:
|
||||
title_prefix = "🆕 **【今日】**"
|
||||
|
||||
title_line = f"### {index}. {title_prefix}[{announcement.title}]({announcement.content_url})"
|
||||
lines.append(title_line)
|
||||
lines.append("")
|
||||
|
||||
# 公告信息
|
||||
info_items = []
|
||||
|
||||
if announcement.publish_date:
|
||||
publish_date = announcement.publish_date.strftime("%Y-%m-%d")
|
||||
info_items.append(f"📅 发布时间: {publish_date}")
|
||||
|
||||
if announcement.purchase_name:
|
||||
info_items.append(f"🏢 发布单位: {announcement.purchase_name}")
|
||||
|
||||
info_items.append(f"📄 来源: {announcement.source_name}")
|
||||
|
||||
if announcement.crawled_at:
|
||||
crawled_time = announcement.crawled_at.strftime("%m-%d %H:%M")
|
||||
info_items.append(f"🤖 爬取时间: {crawled_time}")
|
||||
|
||||
if info_items:
|
||||
lines.append(" | ".join(info_items))
|
||||
lines.append("")
|
||||
|
||||
# 如果是新公告，添加标记
|
||||
if announcement.is_new:
|
||||
lines.append("*🚀 新公告*")
|
||||
lines.append("")
|
||||
|
||||
lines.append("---")
|
||||
lines.append("")
|
||||
|
||||
return lines
|
||||
|
||||
def _generate_empty_markdown(self, title: str) -> str:
|
||||
"""生成空内容的Markdown"""
|
||||
lines = [
|
||||
f"# {title}",
|
||||
"",
|
||||
f"**更新时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
|
||||
"",
|
||||
"## 无新公告",
|
||||
"",
|
||||
"当前时间范围内没有找到符合条件的公告。",
|
||||
"",
|
||||
"---",
|
||||
"",
|
||||
f"*由广西政府采购网公告监控系统生成*"
|
||||
]
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
def _create_anchor(self, text: str) -> str:
|
||||
"""创建锚点链接"""
|
||||
# 移除特殊字符，替换空格为连字符，转为小写
|
||||
import re
|
||||
anchor = re.sub(r'[^\w\s-]', '', text)
|
||||
anchor = re.sub(r'[-\s]+', '-', anchor)
|
||||
return anchor.lower().strip('-')
|
||||
|
||||
def save_to_file(self, announcements: List[Announcement],
|
||||
title: Optional[str] = None) -> bool:
|
||||
"""
|
||||
保存Markdown到文件
|
||||
|
||||
Args:
|
||||
announcements: 公告列表
|
||||
title: 文档标题
|
||||
|
||||
Returns:
|
||||
bool: 保存是否成功
|
||||
"""
|
||||
try:
|
||||
markdown_content = self.generate_markdown(announcements, title)
|
||||
|
||||
with open(self.output_file, 'w', encoding='utf-8') as f:
|
||||
f.write(markdown_content)
|
||||
|
||||
logger.info(f"Markdown文件已保存到: {self.output_file} (共 {len(announcements)} 条公告)")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"保存Markdown文件失败: {str(e)}")
|
||||
return False
|
||||
|
||||
def append_to_file(self, new_announcements: List[Announcement]) -> bool:
|
||||
"""
|
||||
追加新公告到现有文件
|
||||
|
||||
Args:
|
||||
new_announcements: 新公告列表
|
||||
|
||||
Returns:
|
||||
bool: 追加是否成功
|
||||
"""
|
||||
if not new_announcements:
|
||||
return True
|
||||
|
||||
try:
|
||||
# 读取现有文件
|
||||
existing_content = ""
|
||||
if os.path.exists(self.output_file):
|
||||
with open(self.output_file, 'r', encoding='utf-8') as f:
|
||||
existing_content = f.read()
|
||||
|
||||
# 如果文件不存在或为空，创建新文件
|
||||
if not existing_content.strip():
|
||||
return self.save_to_file(new_announcements)
|
||||
|
||||
# 解析现有公告（这里简化处理，实际可能需要更复杂的解析）
|
||||
# 为简单起见，我们重新生成完整文件
|
||||
logger.info("重新生成完整Markdown文件")
|
||||
return self.save_to_file(new_announcements)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"追加公告到Markdown文件失败: {str(e)}")
|
||||
return False
|
||||
|
||||
def get_file_stats(self) -> Dict[str, Any]:
|
||||
"""获取文件统计信息"""
|
||||
stats = {
|
||||
"file_exists": False,
|
||||
"file_size": 0,
|
||||
"last_modified": None,
|
||||
"announcement_count": 0
|
||||
}
|
||||
|
||||
try:
|
||||
if os.path.exists(self.output_file):
|
||||
file_stat = os.stat(self.output_file)
|
||||
stats["file_exists"] = True
|
||||
stats["file_size"] = file_stat.st_size
|
||||
stats["last_modified"] = datetime.fromtimestamp(file_stat.st_mtime).isoformat()
|
||||
|
||||
# 尝试统计公告数量（简单计数）
|
||||
with open(self.output_file, 'r', encoding='utf-8') as f:
|
||||
content = f.read()
|
||||
# 统计###开头的行（每个公告的标题行）
|
||||
stats["announcement_count"] = content.count("### ")
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"获取文件统计信息失败: {str(e)}")
|
||||
|
||||
return stats
|
||||
|
||||
|
||||
class AnnouncementMarkdownFormatter:
|
||||
"""公告Markdown格式化器"""
|
||||
|
||||
@staticmethod
|
||||
def format_announcement_card(announcement: Announcement) -> str:
|
||||
"""格式化单个公告为卡片样式"""
|
||||
lines = []
|
||||
|
||||
# 标题
|
||||
emoji = "🆕" if announcement.is_today else "📄"
|
||||
lines.append(f"### {emoji} {announcement.title}")
|
||||
lines.append("")
|
||||
|
||||
# 链接
|
||||
lines.append(f"[查看详情]({announcement.content_url})")
|
||||
lines.append("")
|
||||
|
||||
# 信息表格
|
||||
lines.append("| 属性 | 值 |")
|
||||
lines.append("|------|-----|")
|
||||
|
||||
if announcement.publish_date:
|
||||
lines.append(f"| 发布时间 | {announcement.publish_date.strftime('%Y-%m-%d %H:%M')} |")
|
||||
|
||||
lines.append(f"| 发布单位 | {announcement.purchase_name or 'N/A'} |")
|
||||
lines.append(f"| 来源 | {announcement.source_name} |")
|
||||
lines.append(f"| 公告类型 | {announcement.announcement_type.value} |")
|
||||
|
||||
if announcement.crawled_at:
|
||||
lines.append(f"| 爬取时间 | {announcement.crawled_at.strftime('%m-%d %H:%M')} |")
|
||||
|
||||
lines.append("")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
@staticmethod
|
||||
def format_announcement_list(announcements: List[Announcement]) -> str:
|
||||
"""格式化公告列表"""
|
||||
if not announcements:
|
||||
return "*暂无公告*"
|
||||
|
||||
lines = []
|
||||
for announcement in announcements:
|
||||
emoji = "🆕" if announcement.is_today else "•"
|
||||
publish_date = announcement.publish_date.strftime("%m-%d") if announcement.publish_date else "N/A"
|
||||
line = f"{emoji} [{announcement.title}]({announcement.content_url}) - {publish_date}"
|
||||
lines.append(line)
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
@staticmethod
|
||||
def format_notification_message(announcements: List[Announcement],
|
||||
max_count: int = 10) -> str:
|
||||
"""格式化为通知消息"""
|
||||
if not announcements:
|
||||
return "暂无新公告"
|
||||
|
||||
# 只显示前N条
|
||||
display_announcements = announcements[:max_count]
|
||||
remaining_count = len(announcements) - max_count
|
||||
|
||||
lines = [f"🔔 发现 {len(announcements)} 条新公告：", ""]
|
||||
|
||||
for announcement in display_announcements:
|
||||
title = announcement.title[:50] + "..." if len(announcement.title) > 50 else announcement.title
|
||||
publish_date = announcement.publish_date.strftime("%m-%d") if announcement.publish_date else "N/A"
|
||||
lines.append(f"• {title} ({publish_date})")
|
||||
|
||||
if remaining_count > 0:
|
||||
lines.append(f"... 还有 {remaining_count} 条公告")
|
||||
|
||||
lines.append("")
|
||||
lines.append("*点击公告标题查看详情*")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def create_markdown_generator(output_file: Optional[str] = None) -> MarkdownGenerator:
|
||||
"""
|
||||
创建Markdown生成器实例
|
||||
|
||||
Args:
|
||||
output_file: 输出文件路径
|
||||
|
||||
Returns:
|
||||
MarkdownGenerator: 生成器实例
|
||||
"""
|
||||
return MarkdownGenerator(output_file)
|
||||
|
||||
|
||||
def generate_onu_md(announcements: List[Announcement]) -> bool:
|
||||
"""
|
||||
生成onu.md文件
|
||||
|
||||
Args:
|
||||
announcements: 公告列表
|
||||
|
||||
Returns:
|
||||
bool: 生成是否成功
|
||||
"""
|
||||
generator = create_markdown_generator()
|
||||
return generator.save_to_file(announcements, "广西政府采购网公告监控")
|
||||
|
||||
|
||||
def update_onu_md(new_announcements: List[Announcement]) -> bool:
|
||||
"""
|
||||
更新onu.md文件，追加新公告
|
||||
|
||||
Args:
|
||||
new_announcements: 新公告列表
|
||||
|
||||
Returns:
|
||||
bool: 更新是否成功
|
||||
"""
|
||||
generator = create_markdown_generator()
|
||||
|
||||
# 如果文件不存在，创建新文件
|
||||
if not os.path.exists(generator.output_file):
|
||||
return generator.save_to_file(new_announcements)
|
||||
|
||||
# 否则追加新公告
|
||||
return generator.append_to_file(new_announcements)
|
||||
Reference in New Issue
Block a user