ghpsolutions/quadro
Quadro: team boards with roles, history and an MCP endpoint for Claude and ChatGPT
A remix template · 0 remixes · Remix ghpsolutions/quadro
app/boards/application/use_cases.py
"""Use cases: every action a person can take, written once.
The web page and the MCP tools both call this class, so a rule (a viewer cannot edit, a stranger
cannot see a board) holds the same way whether the request comes from a browser or from Claude.
Boards a person cannot see are reported as "not found", so their existence never leaks.
"""
from __future__ import annotations
import hashlib
import math
import re
import uuid
from datetime import date, datetime, timedelta, timezone
from typing import Callable, Iterable, Optional
from boards.application.ports import Assistant, BoardRepository, NumberTaken
from boards.domain.model import (
DEFAULT_LISTS, Board, BoardList, Card, Conflict, Forbidden, Invalid, NotFound, Person, Role,
board_key, changes_between, check_blocker, clean_due, clean_email, clean_labels, clean_text,
clean_title, with_changes)
__all__ = ["Boards", "NotFound", "Forbidden", "Invalid", "Conflict"]
REF = re.compile(r"^\s*(?:(?P<key>[A-Za-z0-9]+)-)?#?(?P<num>\d+)\s*$")
def _now() -> str:
return datetime.now(timezone.utc).isoformat(timespec="microseconds")
def _new_id() -> str:
return uuid.uuid4().hex[:12]
def _words(text: str) -> set:
return {w for w in re.findall(r"[a-z0-9]{3,}", text.lower())}
def _cosine(a: list, b: list) -> float:
dot = sum(x * y for x, y in zip(a, b))
na, nb = math.sqrt(sum(x * x for x in a)), math.sqrt(sum(y * y for y in b))
return dot / (na * nb) if na and nb else 0.0
class Boards:
def __init__(self, repo: BoardRepository, assistant: Optional[Assistant] = None,
today: Callable[[], date] = lambda: datetime.now(timezone.utc).date()):
self.repo, self.assistant, self.today = repo, assistant, today
# ------------------------------------------------------------------ lookups
def _visible(self, person: Person) -> list:
return self.repo.boards_for(person.email)
def _board(self, person: Person, ref: str) -> Board:
"""A board by id, key or title, among the boards this person can see."""
ref = (ref or "").strip()
if not ref:
raise Invalid("which board? pass its id, key or title")
board = self.repo.get_board(ref)
if board and board.can_view(person):
return board
low = ref.lower().removesuffix(" board").strip()
matches = [b for b in self._visible(person)
if low in (b.key.lower(), b.title.lower(), b.title.lower().removesuffix(" board"))]
if len(matches) == 1:
return matches[0]
if len(matches) > 1:
raise Invalid(f"more than one board matches '{ref}': "
+ ", ".join(f"{b.title} ({b.id})" for b in matches))
raise NotFound(f"no board '{ref}' that you can see")
def _card(self, person: Person, ref: str, board_ref: Optional[str] = None) -> tuple:
"""(card, board) by card id, 'KEY-12' or '12' among the cards this person can see."""
ref = str(ref or "").strip()
card = self.repo.get_card(ref)
if card:
board = self.repo.get_board(card.board_id)
if board and board.can_view(person):
return card, board
m = REF.match(ref)
if not m:
raise NotFound(f"no card '{ref}' that you can see")
boards = [self._board(person, board_ref)] if board_ref else self._visible(person)
if m.group("key"):
boards = [b for b in boards if b.key.lower() == m.group("key").lower()]
number = int(m.group("num"))
hits = []
for board in boards:
c = self.repo.get_card(f"{board.id}-{number}")
if c:
hits.append((c, board))
if len(hits) == 1:
return hits[0]
if len(hits) > 1:
raise Invalid(f"card {number} exists on several boards; say which: "
+ ", ".join(f"{b.key}-{number} ({b.title})" for _, b in hits))
raise NotFound(f"no card '{ref}' that you can see")
def _lists(self, board: Board) -> list:
return sorted(self.repo.lists_on(board.id), key=lambda l: l.position)
def _list(self, board: Board, ref: Optional[str]) -> BoardList:
lists = self._lists(board)
if not ref:
return lists[0]
low = ref.strip().lower()
for lst in lists:
if lst.id == ref or lst.name.lower() == low:
return lst
raise Invalid(f"'{ref}' is not a list on {board.title}; lists are: "
+ ", ".join(l.name for l in lists))
# ------------------------------------------------------------------ views
def _card_view(self, card: Card, board: Board, lists: dict, counts: Optional[dict] = None,
cards: Optional[dict] = None) -> dict:
done_ids = [l.id for l in lists.values() if l.done]
lst = lists.get(card.list_id)
done, total = card.checklist()
blockers = []
for bid in card.blocked_by:
other = (cards or {}).get(bid) or self.repo.get_card(bid)
if other:
blockers.append({"id": other.id, "ref": f"{board.key}-{other.number}",
"title": other.title, "open": other.list_id not in done_ids})
return {
"id": card.id, "ref": f"{board.key}-{card.number}", "number": card.number,
"board_id": board.id, "board": board.title, "list_id": card.list_id,
"list": lst.name if lst else None, "title": card.title,
"description": card.description, "assignee": card.assignee,
"due_date": card.due_date, "labels": list(card.labels),
"overdue": card.is_overdue(self.today(), done_ids),
"done": card.list_id in done_ids,
"checklist": {"done": done, "total": total}, "blocked_by": blockers,
"blocked": any(b["open"] for b in blockers),
"comments": (counts or {}).get(card.id, 0), "position": card.position,
"version": card.version, "updated_at": card.updated_at,
}
def my_boards(self, person: Person) -> list:
out = []
for b in self._visible(person):
out.append({"id": b.id, "key": b.key, "title": b.title,
"role": b.role_of(person).value, "owner": b.owner})
return sorted(out, key=lambda b: b["title"].lower())
def show_board(self, person: Person, board_ref: str) -> dict:
board = self._board(person, board_ref)
lists = self._lists(board)
by_id = {l.id: l for l in lists}
cards = self.repo.cards_on([board.id])
index = {c.id: c for c in cards}
counts = self.repo.comment_counts(board.id)
views = [self._card_view(c, board, by_id, counts, index) for c in cards]
return {
"id": board.id, "key": board.key, "title": board.title,
"role": board.role_of(person).value, "members": board.people(),
"lists": [{"id": l.id, "name": l.name, "done": l.done,
"cards": sorted([v for v in views if v["list_id"] == l.id],
key=lambda v: v["position"])} for l in lists],
}
def card_detail(self, person: Person, card_ref: str, board_ref: Optional[str] = None) -> dict:
card, board = self._card(person, card_ref, board_ref)
lists = {l.id: l for l in self._lists(board)}
view = self._card_view(card, board, lists, self.repo.comment_counts(board.id))
view["comment_list"] = self.repo.comments_for(card.id)
view["history"] = self.repo.activity(board.id, card.id, limit=100)
view["role"] = board.role_of(person).value
return view
def board_activity(self, person: Person, board_ref: str, limit: int = 50) -> list:
board = self._board(person, board_ref)
return self.repo.activity(board.id, None, limit=max(1, min(limit, 200)))
# ------------------------------------------------------------------ history
def _log(self, person: Person, board: Board, action: str, card: Optional[Card] = None,
changes: Optional[list] = None, note: str = "") -> None:
self.repo.record({
"id": _new_id(), "board_id": board.id, "card_id": card.id if card else None,
"card_ref": f"{board.key}-{card.number}" if card else None,
"actor": person.email, "action": action, "changes": changes or [],
"note": note, "happened_at": _now()})
# ------------------------------------------------------------------ boards and people
def create_board(self, person: Person, title: str) -> dict:
title = clean_title(title)
board = Board(id=_new_id(), key=board_key(title), title=title, owner=person.email,
created_at=_now())
self.repo.save_board(board)
self.repo.save_member(board.id, person.email, Role.OWNER)
for i, (name, done) in enumerate(DEFAULT_LISTS):
self.repo.save_list(BoardList(_new_id(), board.id, name, float(i + 1), done))
self._log(person, board, "created the board")
return self.show_board(person, board.id)
def invite_member(self, person: Person, board_ref: str, email: str, role: str = "member") -> dict:
board = self._board(person, board_ref)
email, new_role = clean_email(email), Role.parse(role)
board.check_new_role(person, email, new_role)
before = board.members.get(email)
self.repo.save_member(board.id, email, new_role)
if before is None:
self._log(person, board, "invited a member", note=f"{email} as {new_role.value}")
elif before is not new_role:
self._log(person, board, "changed a role",
note=f"{email}: {before.value} → {new_role.value}")
return {"board_id": board.id, "email": email, "role": new_role.value}
change_role = invite_member
def remove_member(self, person: Person, board_ref: str, email: str) -> dict:
board = self._board(person, board_ref)
email = clean_email(email)
board.check_removal(person, email)
self.repo.remove_member(board.id, email)
self._log(person, board, "removed a member", note=email)
return {"board_id": board.id, "removed": email}
# ------------------------------------------------------------------ lists
def add_list(self, person: Person, board_ref: str, name: str) -> dict:
board = self._board(person, board_ref)
board.require_edit(person)
name = clean_title(name)
lists = self._lists(board)
if any(l.name.lower() == name.lower() for l in lists):
raise Invalid(f"there is already a list called '{name}'")
lst = BoardList(_new_id(), board.id, name, (lists[-1].position if lists else 0) + 1.0,
name.lower() == "done")
self.repo.save_list(lst)
self._log(person, board, "added a list", note=name)
return {"id": lst.id, "name": lst.name, "done": lst.done}
def update_list(self, person: Person, board_ref: str, list_ref: str,
name: Optional[str] = None, done: Optional[bool] = None,
index: Optional[int] = None) -> dict:
board = self._board(person, board_ref)
board.require_edit(person)
lst = self._list(board, list_ref)
notes = []
if name is not None and clean_title(name) != lst.name:
name = clean_title(name)
if any(l.name.lower() == name.lower() and l.id != lst.id for l in self._lists(board)):
raise Invalid(f"there is already a list called '{name}'")
notes.append(f"renamed '{lst.name}' to '{name}'")
lst.name = name
if done is not None and bool(done) != lst.done:
lst.done = bool(done)
notes.append(f"'{lst.name}' {'now counts' if lst.done else 'no longer counts'} as done")
if index is not None:
others = [l for l in self._lists(board) if l.id != lst.id]
index = max(0, min(int(index), len(others)))
others.insert(index, lst)
for i, l in enumerate(others):
if l.position != float(i + 1):
l.position = float(i + 1)
if l.id != lst.id:
self.repo.save_list(l)
notes.append(f"moved '{lst.name}' to position {index + 1}")
self.repo.save_list(lst)
if notes:
self._log(person, board, "changed a list", note="; ".join(notes))
return {"id": lst.id, "name": lst.name, "done": lst.done}
def delete_list(self, person: Person, board_ref: str, list_ref: str) -> dict:
board = self._board(person, board_ref)
board.require_edit(person)
lst = self._list(board, list_ref)
if any(c.list_id == lst.id for c in self.repo.cards_on([board.id])):
raise Invalid(f"'{lst.name}' still has cards; move them first")
if len(self._lists(board)) == 1:
raise Invalid("a board needs at least one list")
self.repo.delete_list(lst.id)
self._log(person, board, "deleted a list", note=lst.name)
return {"deleted": lst.name}
# ------------------------------------------------------------------ cards
def _position_at(self, board: Board, list_id: str, index: Optional[int],
moving: Optional[str] = None) -> float:
"""A position that puts a card at `index` in a list (None = at the end)."""
siblings = sorted((c for c in self.repo.cards_on([board.id])
if c.list_id == list_id and c.id != moving), key=lambda c: c.position)
if index is None or index >= len(siblings):
return (siblings[-1].position if siblings else 0.0) + 1024.0
index = max(0, int(index))
after = siblings[index].position
before = siblings[index - 1].position if index > 0 else after - 2048.0
if after - before < 1e-6: # worn out gaps: spread the list again
for i, c in enumerate(siblings):
self.repo.update_card(with_changes(c, position=(i + 1) * 1024.0), c.version)
return index * 1024.0 + 512.0
return (before + after) / 2
def add_card(self, person: Person, board_ref: str, title: str, list_name: Optional[str] = None,
description: str = "", assignee: Optional[str] = None,
due_date: Optional[str] = None, labels: Iterable[str] = ()) -> dict:
board = self._board(person, board_ref)
board.require_edit(person)
lst = self._list(board, list_name)
assignee = self._assignee(board, assignee)
now = _now()
base = Card(id="", board_id=board.id, number=0, list_id=lst.id, title=clean_title(title),
description=clean_text(description), assignee=assignee,
due_date=clean_due(due_date), labels=clean_labels(labels),
position=self._position_at(board, lst.id, None), created_by=person.email,
created_at=now, updated_at=now)
number = self.repo.next_number(board.id)
for _ in range(20): # concurrent adds race for a number; the database decides
card = with_changes(base, id=f"{board.id}-{number}", number=number)
try:
self.repo.insert_card(card)
break
except NumberTaken:
number += 1
else:
raise Conflict("could not claim a card number, try again")
self._log(person, board, "added the card", card,
[{"field": "list", "from": None, "to": lst.name}], note=f"in {lst.name}")
return self._card_view(card, board, {l.id: l for l in self._lists(board)})
def _assignee(self, board: Board, value: Optional[str]) -> Optional[str]:
if value in (None, ""):
return None
email = clean_email(value)
role = board.role_of(Person(email))
if role is None:
raise Invalid(f"{email} is not on this board; invite them first")
return email
def _write(self, person: Person, board: Board, card: Card, updated: Card, action: str,
expected_version: Optional[int]) -> dict:
if expected_version is not None and int(expected_version) != card.version:
raise Conflict(f"{board.key}-{card.number} changed since you opened it; reload and retry")
changes = changes_between(card, updated)
lists = {l.id: l for l in self._lists(board)}
if not changes and updated.position == card.position:
return self._card_view(card, board, lists)
updated = with_changes(updated, version=card.version + (1 if changes else 0),
updated_at=_now())
if not self.repo.update_card(updated, card.version):
raise Conflict(f"{board.key}-{card.number} changed since you opened it; reload and retry")
for ch in changes: # readable names in the log, not ids
if ch["field"] == "list_id":
ch["field"] = "list"
ch["from"] = lists[ch["from"]].name if ch["from"] in lists else ch["from"]
ch["to"] = lists[ch["to"]].name if ch["to"] in lists else ch["to"]
if ch["field"] == "blocked_by":
ch["from"] = [self._ref(board, i) for i in ch["from"]]
ch["to"] = [self._ref(board, i) for i in ch["to"]]
if changes:
self._log(person, board, action, updated, changes)
return self._card_view(updated, board, lists, self.repo.comment_counts(board.id))
def _ref(self, board: Board, card_id: str) -> str:
return f"{board.key}-{card_id.rsplit('-', 1)[-1]}"
def update_card(self, person: Person, card_ref: str, expected_version: Optional[int] = None,
board_ref: Optional[str] = None, **fields) -> dict:
card, board = self._card(person, card_ref, board_ref)
board.require_edit(person)
allowed = {"title", "description", "assignee", "due_date", "labels"}
unknown = set(fields) - allowed
if unknown:
raise Invalid("cannot change: " + ", ".join(sorted(unknown)))
clean = {}
if "title" in fields:
clean["title"] = clean_title(fields["title"])
if "description" in fields:
clean["description"] = clean_text(fields["description"])
if "assignee" in fields:
clean["assignee"] = self._assignee(board, fields["assignee"])
if "due_date" in fields:
clean["due_date"] = clean_due(fields["due_date"])
if "labels" in fields:
clean["labels"] = clean_labels(fields["labels"])
return self._write(person, board, card, with_changes(card, **clean), "edited the card",
expected_version)
def move_card(self, person: Person, card_ref: str, list_name: str,
index: Optional[int] = None, expected_version: Optional[int] = None,
board_ref: Optional[str] = None) -> dict:
card, board = self._card(person, card_ref, board_ref)
board.require_edit(person)
lst = self._list(board, list_name)
position = self._position_at(board, lst.id, index, moving=card.id)
card = self.repo.get_card(card.id) or card # a respread may have touched it
action = "moved the card" if lst.id != card.list_id else "reordered the card"
return self._write(person, board, card, with_changes(card, list_id=lst.id, position=position),
action, expected_version)
def delete_card(self, person: Person, card_ref: str, board_ref: Optional[str] = None) -> dict:
card, board = self._card(person, card_ref, board_ref)
board.require_edit(person)
for other in self.repo.cards_on([board.id]):
if card.id in other.blocked_by:
self.repo.update_card(with_changes(other, blocked_by=tuple(
b for b in other.blocked_by if b != card.id)), other.version)
self.repo.delete_card(card.id)
self._log(person, board, "deleted the card", card, note=card.title)
return {"deleted": f"{board.key}-{card.number}", "title": card.title}
def comment(self, person: Person, card_ref: str, body: str,
board_ref: Optional[str] = None) -> dict:
card, board = self._card(person, card_ref, board_ref)
board.require_edit(person)
body = clean_text(body)
if not body:
raise Invalid("a comment needs some text")
comment = {"id": _new_id(), "card_id": card.id, "board_id": board.id,
"author": person.email, "body": body, "created_at": _now()}
self.repo.add_comment(comment)
self._log(person, board, "commented", card, note=body[:140])
return comment
# ------------------------------------------------------------------ dependencies
def block(self, person: Person, card_ref: str, blocker_ref: str,
board_ref: Optional[str] = None) -> dict:
card, board = self._card(person, card_ref, board_ref)
board.require_edit(person)
blocker, _ = self._card(person, blocker_ref, board.id)
graph = {c.id: c.blocked_by for c in self.repo.cards_on([board.id])}
check_blocker(card, blocker, graph)
if blocker.id in card.blocked_by:
return self.card_detail(person, card.id)
return self._write(person, board, card,
with_changes(card, blocked_by=card.blocked_by + (blocker.id,)),
"added a blocker", None)
def unblock(self, person: Person, card_ref: str, blocker_ref: str,
board_ref: Optional[str] = None) -> dict:
card, board = self._card(person, card_ref, board_ref)
board.require_edit(person)
blocker, _ = self._card(person, blocker_ref, board.id)
return self._write(person, board, card, with_changes(
card, blocked_by=tuple(b for b in card.blocked_by if b != blocker.id)),
"removed a blocker", None)
def what_blocks(self, person: Person, card_ref: str, board_ref: Optional[str] = None) -> dict:
"""Everything still open that this card waits on, directly or further down the chain."""
card, board = self._card(person, card_ref, board_ref)
lists = {l.id: l for l in self._lists(board)}
done_ids = {l.id for l in lists.values() if l.done}
cards = {c.id: c for c in self.repo.cards_on([board.id])}
chain, seen = [], set()
def walk(c: Card, depth: int, via: Optional[str]):
for bid in c.blocked_by:
if bid in seen or bid not in cards:
continue
seen.add(bid)
b = cards[bid]
if b.list_id not in done_ids:
chain.append({"ref": f"{board.key}-{b.number}", "title": b.title,
"list": lists[b.list_id].name if b.list_id in lists else None,
"assignee": b.assignee, "due_date": b.due_date,
"depth": depth, "blocks": via})
walk(b, depth + 1, f"{board.key}-{b.number}")
walk(card, 1, f"{board.key}-{card.number}")
return {"card": f"{board.key}-{card.number}", "title": card.title,
"clear": not chain, "open_blockers": chain}
# ------------------------------------------------------------------ questions across boards
def _open_cards(self, person: Person, board_ref: Optional[str] = None) -> list:
boards = [self._board(person, board_ref)] if board_ref else self._visible(person)
if not boards:
return []
by_board = {b.id: b for b in boards}
lists = {}
for b in boards:
lists[b.id] = {l.id: l for l in self._lists(b)}
out = []
for c in self.repo.cards_on(list(by_board)):
board_lists = lists[c.board_id]
if c.list_id in board_lists and board_lists[c.list_id].done:
continue
out.append((c, by_board[c.board_id], board_lists))
return out
def my_plate(self, person: Person, days: int = 7) -> dict:
"""Open cards assigned to me: overdue, due in the next `days`, then the rest."""
today = self.today()
until = today + timedelta(days=max(1, min(int(days), 60)))
overdue, soon, later = [], [], []
for c, b, lists in self._open_cards(person):
if c.assignee != person.email:
continue
view = self._card_view(c, b, lists)
d = c.due()
if d and d < today:
overdue.append(view)
elif d and d <= until:
soon.append(view)
else:
later.append(view)
key = lambda v: (v["due_date"] or "9999", v["ref"])
return {"for": person.email, "today": today.isoformat(), "until": until.isoformat(),
"overdue": sorted(overdue, key=key), "due_soon": sorted(soon, key=key),
"no_date_or_later": sorted(later, key=key)}
def overdue(self, person: Person, board_ref: Optional[str] = None) -> list:
today = self.today()
views = [self._card_view(c, b, l) for c, b, l in self._open_cards(person, board_ref)
if c.due() and c.due() < today]
return sorted(views, key=lambda v: (v["due_date"], v["ref"]))
def search(self, person: Person, query: str, board_ref: Optional[str] = None) -> list:
terms = [t for t in (query or "").lower().split() if t]
if not terms:
raise Invalid("search for something")
boards = [self._board(person, board_ref)] if board_ref else self._visible(person)
out = []
for b in boards:
lists = {l.id: l for l in self._lists(b)}
for c in self.repo.cards_on([b.id]):
hay = " ".join([c.title, c.description, " ".join(c.labels), c.assignee or "",
f"{b.key}-{c.number}"]).lower()
if all(t in hay for t in terms):
out.append(self._card_view(c, b, lists))
return out[:50]
# ------------------------------------------------------------------ Dodil Models features
def suggest_checklist(self, person: Person, card_ref: str,
board_ref: Optional[str] = None) -> dict:
"""Suggested steps for a card. Nothing is saved until someone applies them."""
card, board = self._card(person, card_ref, board_ref)
if self.assistant is None:
raise Invalid("suggestions are not configured on this server")
steps = self.assistant.checklist(card.title, card.description)
return {"card": f"{board.key}-{card.number}", "title": card.title, "steps": steps}
def apply_checklist(self, person: Person, card_ref: str, steps: Iterable[str],
board_ref: Optional[str] = None) -> dict:
"""Append steps to the card description as a checklist ('- [ ] step')."""
card, board = self._card(person, card_ref, board_ref)
board.require_edit(person)
lines = [" ".join(str(s).split()) for s in steps if str(s).strip()][:20]
if not lines:
raise Invalid("no steps to add")
text = (card.description.rstrip() + "\n\n" if card.description.strip() else "")
text += "\n".join(f"- [ ] {s}" for s in lines)
return self._write(person, board, card, with_changes(card, description=clean_text(text)),
"added a checklist", None)
def _board_facts(self, board: Board) -> dict:
today = self.today()
lists = self._lists(board)
done_ids = {l.id for l in lists if l.done}
cards = self.repo.cards_on([board.id])
open_ids = {c.id for c in cards if c.list_id not in done_ids}
since = (datetime.now(timezone.utc) - timedelta(hours=24)).isoformat(timespec="microseconds")
recent = [e for e in self.repo.activity(board.id, None, limit=200)
if e["happened_at"] >= since]
ref = lambda c: f"{board.key}-{c.number}"
return {
"board": board.title, "date": today.isoformat(),
"lists": {l.name: sum(1 for c in cards if c.list_id == l.id) for l in lists},
"overdue": [f"{ref(c)} {c.title} (due {c.due_date}, {c.assignee or 'unassigned'})"
for c in cards if c.is_overdue(today, done_ids)],
"due_next_3_days": [f"{ref(c)} {c.title} (due {c.due_date})" for c in cards
if c.due() and today <= c.due() <= today + timedelta(days=3)
and c.list_id not in done_ids],
"blocked": [f"{ref(c)} {c.title}" for c in cards if c.list_id not in done_ids
and any(b in open_ids for b in c.blocked_by)],
"unassigned_open": sum(1 for c in cards if not c.assignee and c.list_id not in done_ids),
"last_24h": [f"{e['actor']} {e['action']} {e['card_ref'] or ''} {e['note'] or ''}".strip()
for e in recent[:40]],
}
def board_summary(self, person: Person, board_ref: str) -> dict:
board = self._board(person, board_ref)
facts = self._board_facts(board)
text, source = None, "facts"
if self.assistant is not None:
try:
text = self.assistant.summary(_facts_text(facts)).strip() or None
source = "dodil-models" if text else "facts"
except Exception: # the summary still has the facts when the model is slow or down
text = None
return {"board": board.title, "date": facts["date"], "summary": text or _facts_text(facts),
"source": source, "facts": facts}
def similar_cards(self, person: Person, card_ref: str, limit: int = 5,
board_ref: Optional[str] = None) -> dict:
"""Cards on the same board that read most like this one."""
card, board = self._card(person, card_ref, board_ref)
lists = {l.id: l for l in self._lists(board)}
cards = [c for c in self.repo.cards_on([board.id])]
others = [c for c in cards if c.id != card.id]
if not others:
return {"card": f"{board.key}-{card.number}", "method": "none", "similar": []}
text = lambda c: f"{c.title}\n{c.description}\n{' '.join(c.labels)}".strip()
scores, method = {}, "keywords"
if self.assistant is not None:
try:
stored = self.repo.embeddings_on(board.id)
missing = [c for c in cards
if stored.get(c.id, (None,))[0] != hashlib.sha1(text(c).encode()).hexdigest()]
for i in range(0, len(missing), 32):
batch = missing[i:i + 32]
for c, vec in zip(batch, self.assistant.embed([text(c) for c in batch])):
digest = hashlib.sha1(text(c).encode()).hexdigest()
self.repo.save_embedding(c.id, board.id, digest, vec)
stored[c.id] = (digest, vec)
target = stored[card.id][1]
scores = {c.id: _cosine(target, stored[c.id][1]) for c in others if c.id in stored}
method = "embeddings"
except Exception:
scores = {}
if not scores:
mine = _words(text(card))
scores = {c.id: (len(mine & _words(text(c))) / len(mine | _words(text(c))))
if mine | _words(text(c)) else 0.0 for c in others}
ranked = sorted(others, key=lambda c: scores.get(c.id, 0), reverse=True)
limit = max(1, min(int(limit), 20))
return {"card": f"{board.key}-{card.number}", "method": method, "similar": [
{**self._card_view(c, board, lists), "score": round(scores.get(c.id, 0.0), 3)}
for c in ranked[:limit] if scores.get(c.id, 0) > 0]}
def _facts_text(f: dict) -> str:
lines = [f"Board: {f['board']} — {f['date']}",
"Cards per list: " + ", ".join(f"{k} {v}" for k, v in f["lists"].items())]
for title, key in (("Overdue", "overdue"), ("Due in the next 3 days", "due_next_3_days"),
("Blocked", "blocked"), ("Last 24 hours", "last_24h")):
items = f[key]
lines.append(f"{title}: " + ("none" if not items else "; ".join(items)))
lines.append(f"Open cards with nobody assigned: {f['unassigned_open']}")
return "\n".join(lines)