hehaiguang1123
a6f05ab2d5
- Phase 0: AGENTS.md cleanup (dedup quotes, renumber sections, merge qmd) - Phase 1: typed relations (manage-relations.py, graph-search.py, check-staleness.py, detect-conflicts.py) - Phase 2: frontmatter validator, weekly lint, knowledge promotion, git hooks - Fix .gitignore to track tools/ and .githooks/ - Fix git remote URL (remove plaintext token) - New wiki pages: 504 pages, 34 raw sources
407 lines
14 KiB
Python
407 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
import sys
|
||
import io
|
||
import json
|
||
import re
|
||
import datetime
|
||
from pathlib import Path
|
||
from typing import Dict, List, Optional, Any
|
||
from dataclasses import dataclass, asdict
|
||
from enum import Enum
|
||
|
||
if sys.platform == "win32":
|
||
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
|
||
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8")
|
||
|
||
|
||
class Priority(Enum):
|
||
P0 = "P0"
|
||
P1 = "P1"
|
||
P2 = "P2"
|
||
|
||
|
||
class TaskStatus(Enum):
|
||
COMPLETED = "✅"
|
||
IN_PROGRESS = "⏳"
|
||
PENDING = "⏸️"
|
||
BLOCKED = "🔴"
|
||
|
||
|
||
@dataclass
|
||
class Task:
|
||
name: str
|
||
priority: Priority
|
||
status: TaskStatus
|
||
output_path: str
|
||
completed_date: Optional[str] = None
|
||
|
||
|
||
@dataclass
|
||
class InstitutionProfile:
|
||
name: str
|
||
type: str
|
||
quality_score: float
|
||
key_data: str
|
||
funding_amount: Optional[float] = None
|
||
category: Optional[str] = None
|
||
|
||
|
||
@dataclass
|
||
class Metric:
|
||
name: str
|
||
value: str
|
||
unit: Optional[str] = None
|
||
date: Optional[str] = None
|
||
|
||
|
||
@dataclass
|
||
class Discovery:
|
||
title: str
|
||
content: str
|
||
category: str
|
||
impact_level: str
|
||
|
||
|
||
@dataclass
|
||
class WeeklyReportData:
|
||
week_number: str
|
||
report_period: str
|
||
theme: str
|
||
author: str
|
||
completion_date: str
|
||
tasks: List[Task]
|
||
institutions: List[InstitutionProfile]
|
||
metrics: List[Metric]
|
||
discoveries: List[Discovery]
|
||
completion_rate: str
|
||
|
||
|
||
class WeeklyReportExtractor:
|
||
def __init__(self, reports_dir: Path):
|
||
self.reports_dir = reports_dir
|
||
self.reports_data: List[WeeklyReportData] = []
|
||
|
||
def extract_week_number(self, filename: str) -> Optional[str]:
|
||
match = re.search(r"W(\d+)", filename)
|
||
return match.group(1) if match else None
|
||
|
||
def extract_frontmatter(self, content: str) -> Dict[str, Any]:
|
||
frontmatter = {}
|
||
frontmatter_match = re.search(r"^---\n(.*?)\n---", content, re.DOTALL)
|
||
if frontmatter_match:
|
||
frontmatter_text = frontmatter_match.group(1)
|
||
for line in frontmatter_text.split("\n"):
|
||
if ":" in line:
|
||
key, value = line.split(":", 1)
|
||
key = key.strip().lower()
|
||
value = value.strip().strip('"')
|
||
if key == "tags":
|
||
frontmatter[key] = [
|
||
tag.strip() for tag in value.split(",") if tag.strip()
|
||
]
|
||
elif key == "created":
|
||
frontmatter[key] = value
|
||
elif key == "author":
|
||
frontmatter[key] = [
|
||
author.strip()
|
||
for author in value.split(",")
|
||
if author.strip()
|
||
]
|
||
return frontmatter
|
||
|
||
def extract_report_period(self, content: str) -> Optional[str]:
|
||
match = re.search(r"报告周期.*?：(.+?)(?:\n|$)", content)
|
||
return match.group(1).strip() if match else None
|
||
|
||
def extract_theme(self, content: str) -> Optional[str]:
|
||
match = re.search(r"主题.*?：(.+?)(?:\n|$)", content)
|
||
return match.group(1).strip() if match else None
|
||
|
||
def extract_tasks(self, content: str) -> List[Task]:
|
||
tasks = []
|
||
task_table_match = re.search(
|
||
r"\| 任务 \| 优先级 \| 状态 \| 产出 \|(.+?)\n---", content, re.DOTALL
|
||
)
|
||
if task_table_match:
|
||
table_rows = re.findall(
|
||
r"\|\s*(.+?)\s*\|\s*(.+?)\s*\|\s*(.+?)\s*\|\s*(.+?)\s*\|",
|
||
task_table_match.group(1),
|
||
)
|
||
for row in table_rows:
|
||
name = row[0].strip()
|
||
priority_str = row[1].strip()
|
||
status_str = row[2].strip()
|
||
output_path = row[3].strip()
|
||
|
||
priority = (
|
||
Priority.P1
|
||
if "P1" in priority_str
|
||
else (Priority.P2 if "P2" in priority_str else Priority.P0)
|
||
)
|
||
status = (
|
||
TaskStatus.COMPLETED
|
||
if "✅" in status_str
|
||
else (
|
||
TaskStatus.IN_PROGRESS
|
||
if "⏳" in status_str
|
||
else (
|
||
TaskStatus.PENDING
|
||
if "⏸️" in status_str
|
||
else TaskStatus.BLOCKED
|
||
)
|
||
)
|
||
)
|
||
|
||
tasks.append(
|
||
Task(
|
||
name=name,
|
||
priority=priority,
|
||
status=status,
|
||
output_path=output_path,
|
||
)
|
||
)
|
||
return tasks
|
||
|
||
def extract_institutions(self, content: str) -> List[InstitutionProfile]:
|
||
institutions = []
|
||
inst_table_match = re.search(
|
||
r"\| 机构 \| 类型 \| 质量分 \| 核心数据 \|(.+?)\n\*\*", content, re.DOTALL
|
||
)
|
||
if inst_table_match:
|
||
table_rows = re.findall(
|
||
r"\|\s*\*\*(.+?)\*\*\s*\|\s*(.+?)\s*\|\s*(.+?)\s*\|\s*(.+?)\s*\|",
|
||
inst_table_match.group(1),
|
||
)
|
||
for row in table_rows:
|
||
name = row[0].strip()
|
||
inst_type = row[1].strip()
|
||
quality_str = row[2].strip()
|
||
key_data = row[3].strip()
|
||
|
||
quality_score = (
|
||
float(re.search(r"(\d+)", quality_str).group(1))
|
||
if re.search(r"(\d+)", quality_str)
|
||
else 0.0
|
||
)
|
||
|
||
institutions.append(
|
||
InstitutionProfile(
|
||
name=name,
|
||
type=inst_type,
|
||
quality_score=quality_score,
|
||
key_data=key_data,
|
||
)
|
||
)
|
||
return institutions
|
||
|
||
def extract_metrics(self, content: str) -> List[Metric]:
|
||
metrics = []
|
||
metric_patterns = [
|
||
(r"AI教育广义.*?\|\s*\$?([\d.,]+).*?万", "AI教育广义市场规模", "亿美元"),
|
||
(r"纯EdTech融资.*?\|\s*\$?([\d.,]+).*?万", "纯EdTech融资总额", "亿美元"),
|
||
(r"AI辅导.*?\|\s*\*\*(\$[\d.,]+)\*\*", "AI辅导赛道融资", "亿美元"),
|
||
(r"内容生成.*?\|\s*\$?([\d.,]+).*?万", "内容生成赛道融资", "亿美元"),
|
||
(r"评估评分.*?\|\s*\$?([\d.,]+).*?万", "评估评分赛道融资", "亿美元"),
|
||
]
|
||
|
||
for pattern, metric_name, unit in metric_patterns:
|
||
matches = re.findall(pattern, content)
|
||
for match in matches:
|
||
value = match.replace(",", "")
|
||
metrics.append(Metric(name=metric_name, value=value, unit=unit))
|
||
|
||
return metrics
|
||
|
||
def extract_discoveries(self, content: str) -> List[Discovery]:
|
||
discoveries = []
|
||
discovery_sections = re.findall(
|
||
r"### \d+\.\d+\s+最关键洞察.*?\n(.+?)(?:\n###|\n---|$)", content, re.DOTALL
|
||
)
|
||
|
||
for section in discovery_sections:
|
||
title_match = re.search(r"最关键洞察：(.+?)(?:\n|>)", section)
|
||
if title_match:
|
||
title = title_match.group(1).strip()
|
||
content_text = section.replace(title_match.group(0), "").strip()
|
||
discoveries.append(
|
||
Discovery(
|
||
title=title,
|
||
content=content_text[:500],
|
||
category="洞察",
|
||
impact_level="高",
|
||
)
|
||
)
|
||
|
||
return discoveries
|
||
|
||
def extract_completion_rate(self, content: str) -> Optional[str]:
|
||
match = re.search(r"完成率.*?：(.+?)(?:\n|$)", content)
|
||
return match.group(1).strip() if match else None
|
||
|
||
def parse_report(self, report_file: Path) -> Optional[WeeklyReportData]:
|
||
try:
|
||
content = report_file.read_text(encoding="utf-8")
|
||
week_number = self.extract_week_number(report_file.name)
|
||
if not week_number:
|
||
return None
|
||
|
||
frontmatter = self.extract_frontmatter(content)
|
||
report_period = self.extract_report_period(content)
|
||
theme = self.extract_theme(content)
|
||
author = (
|
||
frontmatter.get("author", ["狗剩"])[0]
|
||
if frontmatter.get("author")
|
||
else "狗剩"
|
||
)
|
||
completion_date = frontmatter.get(
|
||
"created", datetime.date.today().isoformat()
|
||
)
|
||
|
||
tasks = self.extract_tasks(content)
|
||
institutions = self.extract_institutions(content)
|
||
metrics = self.extract_metrics(content)
|
||
discoveries = self.extract_discoveries(content)
|
||
completion_rate = self.extract_completion_rate(content)
|
||
|
||
return WeeklyReportData(
|
||
week_number=week_number,
|
||
report_period=report_period or "",
|
||
theme=theme or "",
|
||
author=author,
|
||
completion_date=completion_date,
|
||
tasks=tasks,
|
||
institutions=institutions,
|
||
metrics=metrics,
|
||
discoveries=discoveries,
|
||
completion_rate=completion_rate or "",
|
||
)
|
||
|
||
except Exception as e:
|
||
print(f"Error parsing {report_file.name}: {str(e)}")
|
||
return None
|
||
|
||
def extract_all_reports(self) -> List[WeeklyReportData]:
|
||
report_files = list(self.reports_dir.glob("*.md"))
|
||
report_files.sort(key=lambda f: f.stat().st_mtime, reverse=True)
|
||
|
||
print(f"Found {len(report_files)} weekly reports")
|
||
print(f"Parsing reports from {len(report_files)} files...")
|
||
print()
|
||
|
||
for report_file in report_files:
|
||
print(f"Parsing: {report_file.name}")
|
||
report_data = self.parse_report(report_file)
|
||
if report_data:
|
||
self.reports_data.append(report_data)
|
||
print(f" ✅ Parsed successfully - Week {report_data.week_number}")
|
||
print(f" Tasks: {len(report_data.tasks)}")
|
||
print(f" Institutions: {len(report_data.institutions)}")
|
||
print(f" Metrics: {len(report_data.metrics)}")
|
||
print(f" Discoveries: {len(report_data.discoveries)}")
|
||
else:
|
||
print(f" ❌ Failed to parse")
|
||
print()
|
||
|
||
return self.reports_data
|
||
|
||
def report_to_dict(self, report: WeeklyReportData) -> Dict[str, Any]:
|
||
def serialize_enum(obj):
|
||
if isinstance(obj, Enum):
|
||
return obj.value
|
||
elif isinstance(obj, list):
|
||
return [serialize_enum(item) for item in obj]
|
||
elif isinstance(obj, dict):
|
||
return {key: serialize_enum(value) for key, value in obj.items()}
|
||
elif hasattr(obj, "__dict__"):
|
||
return {
|
||
key: serialize_enum(value) for key, value in asdict(obj).items()
|
||
}
|
||
return obj
|
||
|
||
return serialize_enum(report)
|
||
|
||
def save_to_json(self, output_path: Path):
|
||
data = {
|
||
"extraction_timestamp": datetime.datetime.now().isoformat(),
|
||
"total_reports": len(self.reports_data),
|
||
"reports": [self.report_to_dict(report) for report in self.reports_data],
|
||
}
|
||
|
||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||
with open(output_path, "w", encoding="utf-8") as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
|
||
print(f"✅ Data saved to: {output_path}")
|
||
|
||
def generate_summary(self) -> Dict[str, Any]:
|
||
if not self.reports_data:
|
||
return {}
|
||
|
||
total_tasks = sum(len(report.tasks) for report in self.reports_data)
|
||
total_institutions = sum(
|
||
len(report.institutions) for report in self.reports_data
|
||
)
|
||
total_metrics = sum(len(report.metrics) for report in self.reports_data)
|
||
total_discoveries = sum(len(report.discoveries) for report in self.reports_data)
|
||
|
||
completed_tasks = sum(
|
||
1
|
||
for report in self.reports_data
|
||
for task in report.tasks
|
||
if task.status == TaskStatus.COMPLETED
|
||
)
|
||
|
||
return {
|
||
"total_reports": len(self.reports_data),
|
||
"total_tasks": total_tasks,
|
||
"completed_tasks": completed_tasks,
|
||
"total_institutions": total_institutions,
|
||
"total_metrics": total_metrics,
|
||
"total_discoveries": total_discoveries,
|
||
"task_completion_rate": f"{(completed_tasks / total_tasks * 100):.1f}%"
|
||
if total_tasks > 0
|
||
else "0%",
|
||
}
|
||
|
||
|
||
def main():
|
||
print("=" * 60)
|
||
print("Weekly Reports Data Extraction Tool")
|
||
print("=" * 60)
|
||
print(f"Execution time: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print()
|
||
|
||
reports_dir = Path(__file__).parent.parent / "每周报告"
|
||
output_file = (
|
||
Path(__file__).parent.parent
|
||
/ "outputs"
|
||
/ f"weekly_reports_extraction_{datetime.date.today()}.json"
|
||
)
|
||
|
||
extractor = WeeklyReportExtractor(reports_dir)
|
||
reports_data = extractor.extract_all_reports()
|
||
|
||
if reports_data:
|
||
extractor.save_to_json(output_file)
|
||
|
||
summary = extractor.generate_summary()
|
||
print()
|
||
print("=" * 60)
|
||
print("Extraction Summary")
|
||
print("=" * 60)
|
||
print(f"Total reports processed: {summary['total_reports']}")
|
||
print(f"Total tasks extracted: {summary['total_tasks']}")
|
||
print(f"Completed tasks: {summary['completed_tasks']}")
|
||
print(f"Task completion rate: {summary['task_completion_rate']}")
|
||
print(f"Total institutions: {summary['total_institutions']}")
|
||
print(f"Total metrics: {summary['total_metrics']}")
|
||
print(f"Total discoveries: {summary['total_discoveries']}")
|
||
print("=" * 60)
|
||
else:
|
||
print("❌ No reports could be parsed")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|