"""Family document router — sorts classified documents by confidence/priority.
Adapts Costco router.py for family document routing:
- Zones → Members
- Items → Documents
- Zone order → Member priority
Sovereign: Zero imports from costco_route.
"""
from typing import Optional
from icarus.core.family_loader import get_family_config
Member priority order (kids first, then parents, then general family)
MEMBER_ORDER = ["sully", "harper", "aundrea", "matt", "family"]
def generate_routing(
classified: dict[str, list[dict]],
learned_overrides: Optional[dict[str, dict]] = None
) -> list[dict]:
"""Generate prioritized routing for family documents.
Args:
classified: {member_id: [document_dicts]} from LLM or inference engine
learned_overrides: Override from ChromaDB memory {pattern: {member_id, ...}}
Returns:
List of member groups in priority order, each with documents.
"""
if learned_overrides:
classified = _apply_overrides(classified, learned_overrides)
family_config = get_family_config()
route = []
for member_id in MEMBER_ORDER:
docs = classified.get(member_id, [])
if not docs:
continue
member = family_config.get_member(member_id) or {"name": member_id.title()}
route.append({
"member_id": member_id,
"member_name": member.get("name", member_id.title()),
"documents": docs,
"priority": _calculate_priority(docs),
"document_count": len(docs)
})
return route
def _apply_overrides(
classified: dict[str, list[dict]],
overrides: dict[str, dict]
) -> dict[str, list[dict]]:
"""Apply learned overrides from memory.
If a document pattern was previously calibrated to a different member,
move it to the learned member.
"""
# Build lookup: pattern_lower → override member
override_map = {}
for pattern, override_info in overrides.items():
override_map[pattern.lower().strip()] = override_info["member_id"]
# Rebuild classified dict with overrides applied
result = {}
for member_id, docs in classified.items():
remaining = []
for doc in docs:
# Check document content for pattern matches
content = doc.get("content", "").lower().strip()
matched = False
for pattern, target_member in override_map.items():
if pattern in content:
result.setdefault(target_member, []).append(doc)
matched = True
break
if not matched:
remaining.append(doc)
if remaining:
result[member_id] = remaining
return result
def _calculate_priority(docs: list[dict]) -> float:
"""Calculate priority score for a member's documents.
Factors:
- Urgency (deadline proximity)
- Document confidence
- Presence of deadlines/dates
"""
if not docs:
return 0.0
scores = []
for doc in docs:
score = 0.5 # Base priority
# Higher confidence = higher priority
confidence = doc.get("confidence", 0.5)
score += confidence * 0.3
# Deadline proximity boost
if doc.get("deadline"):
score += 0.2
# Urgent keywords
content = doc.get("content", "").lower()
if any(word in content for word in ["urgent", "deadline", "required", "mandatory"]):
score += 0.15
scores.append(min(score, 1.0))
return round(sum(scores) / len(scores), 2)
def format_briefing_queue(route: list[dict], family_name: str = "Hoffmann") -> str:
"""Format routing into a Telegram-friendly briefing queue.
Args:
route: Output from generate_routing()
family_name: Display name for the family
Returns:
Formatted string ready for Telegram.
"""
lines = [f"📋 Document Queue — {family_name} Family"]
total_docs = sum(r["document_count"] for r in route)
member_count = len(route)
for member_group in route:
member_id = member_group["member_id"]
docs = member_group["documents"]
priority = member_group["priority"]
# Emoji based on member
emoji = {
"sully": "👦",
"harper": "👧",
"aundrea": "👩",
"matt": "👨",
"family": "👨👩👧👦"
}.get(member_id, "📄")
lines.append(f"\n{emoji} {member_group['member_name']} — "
f"{len(docs)} doc(s) (priority: {priority:.0%})")
for doc in docs:
title = doc.get("title", "Untitled")
confidence = doc.get("confidence", 0.0)
conf_emoji = "🟢" if confidence > 0.9 else "🟡" if confidence > 0.7 else "🔴"
lines.append(f" {conf_emoji} {title}")
lines.append(f"\n✅ {total_docs} documents across {member_count} members")
return "\n".join(lines)
def format_briefing_queue_markdown(route: list[dict], family_name: str = "Hoffmann") -> str:
"""Format routing with markdown (for CLI / non-Telegram output)."""
lines = [f"## 📋 Document Queue — {family_name} Family"]
total_docs = sum(r["document_count"] for r in route)
member_count = len(route)
for member_group in route:
member_id = member_group["member_id"]
docs = member_group["documents"]
priority = member_group["priority"]
lines.append(f"\n### {member_group['member_name']} "
f"({len(docs)} docs, priority: {priority:.0%})")
for doc in docs:
title = doc.get("title", "Untitled")
confidence = doc.get("confidence", 0.0)
lines.append(f"- [{title}](confidence: {confidence:.0%})")
lines.append(f"\n**{total_docs} documents across {member_count} members**")
return "\n".join(lines)