#!/usr/bin/env python3 """ Combat Room Knowledge Harvesting & Synchronization Engine. Transforms static meeting minutes into an actionable, cross-department knowledge base: 1. Parses YAML frontmatter and Markdown tables for Decisions (DEC-*) and Action Items (ACT-*). 2. Generates and updates combat_room/_indexes/ACTION_ITEMS_TRACKER.md. 3. Generates and updates combat_room/_indexes/DECISION_REGISTER.md. 4. Generates and updates combat_room/_indexes/TOPIC_TAXONOMY.md. 5. Audits permissions against combat_room/.permissions.json. 6. Optionally pushes core decisions into Mem0 long-term agent memory. """ from __future__ import annotations import argparse import datetime import os import sys from pathlib import Path from typing import Any, Dict, List, Optional, Tuple if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") if hasattr(sys.stderr, "reconfigure"): sys.stderr.reconfigure(encoding="utf-8") PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) try: from tools.knowledge.combat_room_helpers import ( ActionItem, Decision, MeetingDoc, audit_permissions as helper_audit_permissions, extract_action_items_from_table, extract_decisions_from_table, parse_frontmatter, sync_to_mem0_local, ) except ImportError: from combat_room_helpers import ( # type: ignore ActionItem, Decision, MeetingDoc, audit_permissions as helper_audit_permissions, extract_action_items_from_table, extract_decisions_from_table, parse_frontmatter, sync_to_mem0_local, ) COMBAT_ROOM_DIR = PROJECT_ROOT / "combat_room" INDEXES_DIR = COMBAT_ROOM_DIR / "_indexes" PERMISSIONS_FILE = COMBAT_ROOM_DIR / ".permissions.json" def audit_permissions(docs: List[MeetingDoc], permissions_file: Optional[Path] = None) -> List[str]: """Audit meeting permissions against policy file.""" return helper_audit_permissions(docs, permissions_file or PERMISSIONS_FILE) def collect_meeting_docs(root_dir: Path) -> List[MeetingDoc]: """Scan and parse all valid meeting minutes in combat_room/.""" docs: List[MeetingDoc] = [] for path in root_dir.rglob("*.md"): rel = path.relative_to(root_dir).as_posix() if rel.startswith("_") or path.name.lower() in {"readme.md", "permissions.md"}: continue try: content = path.read_text(encoding="utf-8") except Exception as e: print(f"[WARN] Failed to read {path}: {e}") continue fm, body = parse_frontmatter(content) if not fm: continue meeting_id = str(fm.get("meeting_id", path.stem)) title = str(fm.get("title", path.stem)) date = str(fm.get("date", "2026-01-01")) dept = str(fm.get("department", path.parent.name)) tags = fm.get("tags", []) if isinstance(tags, str): tags = [t.strip() for t in tags.split(",") if t.strip()] doc_info_dec = (meeting_id, title, date, rel, dept) doc_info_act = (meeting_id, title, rel, dept) decisions = extract_decisions_from_table(body, doc_info_dec) action_items = extract_action_items_from_table(body, doc_info_act) docs.append(MeetingDoc( path=path, rel_path=rel, frontmatter=fm, body=body, decisions=decisions, action_items=action_items, tags=tags, )) docs.sort(key=lambda d: str(d.frontmatter.get("date", "")), reverse=True) return docs def _render_action_table_rows(items: List[ActionItem], today: datetime.date) -> List[str]: """Render a table section of action items in Markdown.""" if not items: return ["*Không có đầu việc nào.*", ""] t_lines = [ "| Mã Việc | Nhiệm Vụ | Người Phụ Trách | Phòng Ban | Hạn Chót | Trạng Thái | Biên Bản Gốc |", "| :--- | :--- | :--- | :--- | :---: | :---: | :--- |", ] for it in items: link = f"[{it.meeting_id}](file:///c:/Projects/FreeExile/combat_room/{it.meeting_path})" is_overdue = False if it.deadline != "TBD" and it.status in {"OPEN", "IN_PROGRESS"}: try: dl = datetime.datetime.strptime(it.deadline, "%Y-%m-%d").date() if dl < today: is_overdue = True except ValueError: pass dl_str = f"**{it.deadline} (QUÁ HẠN)**" if is_overdue else it.deadline t_lines.append(f"| `{it.action_id}` | {it.task} | `{it.owner}` | `{it.department}` | {dl_str} | `{it.status}` | {link} |") t_lines.append("") return t_lines def generate_action_items_tracker(docs: List[MeetingDoc]) -> str: """Generate Markdown for ACTION_ITEMS_TRACKER.md.""" all_items: List[ActionItem] = [i for d in docs for i in d.action_items] today = datetime.date.today() in_progress = [i for i in all_items if i.status == "IN_PROGRESS"] open_items = [i for i in all_items if i.status == "OPEN"] done_items = [i for i in all_items if i.status in {"DONE", "RESOLVED"}] overdue_count = 0 for i in all_items: if i.status in {"OPEN", "IN_PROGRESS"} and i.deadline != "TBD": try: if datetime.datetime.strptime(i.deadline, "%Y-%m-%d").date() < today: overdue_count += 1 except ValueError: pass lines = [ "# SỔ THEO DÕI ĐẦU VIỆC HÀNH ĐỘNG (COMBAT ROOM ACTION TRACKER)", "", "> **Cơ chế cập nhật**: Tự động sinh từ `python tools/knowledge/combat_room_sync.py`. ", f"> **Thời điểm cập nhật**: `{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}` | **Tổng số đầu việc**: `{len(all_items)}` ", f"> **Trạng thái**: `{len(in_progress)} Đang làm` | `{len(open_items)} Mới mở` | `{len(done_items)} Hoàn thành` | `{overdue_count} Quá hạn` ", "", "---", "", "## 1. CÁC ĐẦU VIỆC ĐANG THỰC HIỆN (IN PROGRESS)", "", ] lines.extend(_render_action_table_rows(in_progress, today)) lines.extend(["---", "", "## 2. CÁC ĐẦU VIỆC MỚI MỞ (OPEN)", ""]) lines.extend(_render_action_table_rows(open_items, today)) lines.extend(["---", "", "## 3. CÁC ĐẦU VIỆC ĐÃ HOÀN THÀNH (DONE)", ""]) lines.extend(_render_action_table_rows(done_items, today)) return "\n".join(lines) def generate_decision_register(docs: List[MeetingDoc]) -> str: """Generate Markdown for DECISION_REGISTER.md.""" all_decisions: List[Decision] = [dec for d in docs for dec in d.decisions] lines = [ "# SỔ CÁI QUYẾT ĐỊNH CÔNG TY (COMBAT ROOM DECISION REGISTER)", "", "> **Mục đích**: Lưu trữ toàn bộ các quyết định chiến lược, kiến trúc và phân công kỹ thuật đã được thông qua. ", f"> **Thời điểm cập nhật**: `{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}` | **Tổng số quyết định**: `{len(all_decisions)}` ", "", "---", "", "| Mã Quyết Định | Ngày | Quyết Định Đã Chốt | Bối Cảnh / Lý Do | Người Duyệt | Phòng Ban | Biên Bản Gốc |", "| :--- | :---: | :--- | :--- | :--- | :--- | :--- |", ] for dec in all_decisions: link = f"[{dec.meeting_id}](file:///c:/Projects/FreeExile/combat_room/{dec.meeting_path})" lines.append( f"| `{dec.decision_id}` | {dec.meeting_date} | **{dec.summary}** | {dec.context} | `{dec.approved_by}` | `{dec.department}` | {link} |" ) lines.append("") return "\n".join(lines) def generate_topic_taxonomy(docs: List[MeetingDoc]) -> str: """Generate Markdown for TOPIC_TAXONOMY.md.""" tag_map: Dict[str, List[MeetingDoc]] = {} for d in docs: for tag in d.tags: tag_clean = tag.strip().lower() if tag_clean: tag_map.setdefault(tag_clean, []).append(d) lines = [ "# BẢN ĐỒ TỪ KHÓA & CHỈ MỤC TRI THỨC (TOPIC TAXONOMY)", "", "> **Mục đích**: Tra cứu biên bản họp nhanh theo chủ đề kỹ thuật, cơ chế game và từ khóa nghiệp vụ. ", f"> **Tổng số chủ đề**: `{len(tag_map)}` ", "", "---", "", "## 1. DANH MỤC TỪ KHÓA CHỦ ĐẠO", "", ] sorted_tags = sorted(tag_map.items(), key=lambda x: len(x[1]), reverse=True) for tag, m_docs in sorted_tags: lines.append(f"### #{tag} (`{len(m_docs)} biên bản`)") for md in m_docs: date_str = str(md.frontmatter.get("date", "2026")) title_str = str(md.frontmatter.get("title", md.rel_path)) link = f"[{title_str}](file:///c:/Projects/FreeExile/combat_room/{md.rel_path})" lines.append(f"- **{date_str}** | {link} (`{md.frontmatter.get('department', 'N/A')}`)") lines.append("") return "\n".join(lines) def main() -> int: """CLI Entry point for Combat Room Synchronization.""" parser = argparse.ArgumentParser(description="Combat Room Knowledge Synchronizer & Auditing Engine") parser.add_argument("--audit-permissions", action="store_true", help="Audit confidentiality levels against policy") parser.add_argument("--sync-mem0", action="store_true", help="Synchronize approved decisions to Mem0 Agent memory") parser.add_argument("--strict", action="store_true", help="Fail if any permission violation is found") args = parser.parse_args() print("===================================================================") print(" FREEEXILE: COMBAT ROOM KNOWLEDGE HARVESTING & SYNC ENGINE ") print("===================================================================") print(f"[*] Scanning directory: {COMBAT_ROOM_DIR}") INDEXES_DIR.mkdir(parents=True, exist_ok=True) docs = collect_meeting_docs(COMBAT_ROOM_DIR) print(f"[*] Found {len(docs)} meeting minutes files.") total_decisions = sum(len(d.decisions) for d in docs) total_actions = sum(len(d.action_items) for d in docs) print(f"[*] Extracted: {total_decisions} Decisions, {total_actions} Action Items.") # 1. Action Items Tracker (INDEXES_DIR / "ACTION_ITEMS_TRACKER.md").write_text(generate_action_items_tracker(docs), encoding="utf-8") print(f"[+] Updated: {(INDEXES_DIR / 'ACTION_ITEMS_TRACKER.md').relative_to(PROJECT_ROOT)}") # 2. Decision Register (INDEXES_DIR / "DECISION_REGISTER.md").write_text(generate_decision_register(docs), encoding="utf-8") print(f"[+] Updated: {(INDEXES_DIR / 'DECISION_REGISTER.md').relative_to(PROJECT_ROOT)}") # 3. Topic Taxonomy (INDEXES_DIR / "TOPIC_TAXONOMY.md").write_text(generate_topic_taxonomy(docs), encoding="utf-8") print(f"[+] Updated: {(INDEXES_DIR / 'TOPIC_TAXONOMY.md').relative_to(PROJECT_ROOT)}") # 4. Audit Permissions violations = audit_permissions(docs) if violations: print("\n[!] PERMISSION AUDIT WARNINGS:") for v in violations: print(f" - {v}") if args.strict: print("\n[FAIL] Strict mode enabled. Aborting due to permission violations.") return 1 else: print("[+] Permission policy audit passed: 100% compliant.") # 5. Mem0 Sync if args.sync_mem0: all_decisions = [dec for d in docs for dec in d.decisions] sync_to_mem0_local(all_decisions) print("\n[SUCCESS] Combat Room Knowledge Base synchronized successfully!") return 0 if __name__ == "__main__": sys.exit(main())