231 lines
11 KiB
Python
231 lines
11 KiB
Python
"""岗位 (position) → resource subscription **sync engine**.
|
|
|
|
A user's position decides which agents / skills / scheduled tasks they are
|
|
subscribed to, and which recommended questions they can see. This service turns
|
|
a position's *tag bindings* into concrete, materialized rows in the existing
|
|
"我的" tables (idempotent, re-runnable).
|
|
|
|
Resolution rule (per resource type):
|
|
|
|
* The user's position is ``users.position_id``; if unset, the ``is_default``
|
|
position is used; if there is none, every type falls back to its default set.
|
|
* For a resource type **with** bound tags → the union of resources carrying any
|
|
of those tags.
|
|
* For a resource type **without** bound tags → the *default set*:
|
|
- ``agent`` -> built-ins first, then filtered out of
|
|
position-granted favorites
|
|
- ``scheduled_task`` → empty (no built-in tasks)
|
|
- ``skill`` → empty (the 广场 still shows every visible skill;
|
|
a position only *adds* its tagged skills to the member's "我的技能")
|
|
- ``recommended_question`` → *untagged-only* (handled at query time, never
|
|
materialized — see :meth:`resolve_visible`). The default set is the
|
|
questions carrying **no** 标签; tagged questions surface only for a position
|
|
that binds their tag. (The recommended-questions router interprets a
|
|
``None`` from :meth:`resolve_visible` as "show only untagged".)
|
|
|
|
Materialization (``sync_user``) only ever touches rows it created itself
|
|
(``origin='position'``), so a user's own favorites / subscriptions are never
|
|
clobbered. ``agent`` keeps the original position-granted behavior for ordinary
|
|
tagged agents, but filters out built-ins and business-chain seats; those only
|
|
enter "mine" after an explicit user favorite. ``skill`` is still materialized
|
|
into ``skill_favorites`` so tagged skills are added to the user's "mine"
|
|
collection while the browse-all square stays complete (it is **not**
|
|
restricted).
|
|
Only recommended-question visibility is computed live by :meth:`resolve_visible`
|
|
and is *not* materialized. (The admin-configured prompts shown to users ARE the
|
|
recommended questions — there is no separate prompt-template store.)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import uuid
|
|
from datetime import UTC, datetime
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
|
|
|
from deerflow.persistence.agents.model import AgentFavoriteRow, AgentRow
|
|
from deerflow.persistence.positions.base import PositionStore
|
|
from deerflow.persistence.roundtable_chains.model import RoundtableChainRow
|
|
from deerflow.persistence.scheduled_tasks.model import ScheduledTaskSubscriptionRow
|
|
from deerflow.persistence.skills.model import SkillFavoriteRow
|
|
from deerflow.persistence.tags.base import TagStore
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Reconciled by sync_user. ``agent`` preserves ordinary position-granted rows,
|
|
# but filters built-ins and business-chain seats.
|
|
_MATERIALIZED_TYPES = ("agent", "skill", "scheduled_task")
|
|
# Resolved live at query time (visibility filters), never materialized.
|
|
_VISIBILITY_TYPES = ("recommended_question",)
|
|
|
|
|
|
class PositionSyncService:
|
|
def __init__(self, session_factory: async_sessionmaker[AsyncSession], position_store: PositionStore, tag_store: TagStore) -> None:
|
|
self._sf = session_factory
|
|
self._positions = position_store
|
|
self._tags = tag_store
|
|
|
|
# ----------------------------------------------------------- resolution
|
|
|
|
async def _effective_position_id(self, user_id: str) -> str | None:
|
|
pid = await self._positions.get_user_position_id(user_id)
|
|
if pid:
|
|
return pid
|
|
return await self._positions.get_default_position_id()
|
|
|
|
async def resolve_visible(self, user_id: str, resource_type: str) -> set[str] | None:
|
|
"""Return the set of visible ``resource_type`` ids, or ``None`` = no tag binding.
|
|
|
|
Used by the recommended-questions filter (the only remaining query-time
|
|
visibility type; skills are now *materialized*, not filtered). A set
|
|
means "restrict to exactly these ids". ``None`` means the effective
|
|
position binds no tag for this type — the recommended-questions router
|
|
treats that as "默认岗位 → show only untagged questions" (see
|
|
``app/gateway/routers/recommended_questions.py``).
|
|
"""
|
|
pid = await self._effective_position_id(user_id)
|
|
if not pid:
|
|
return None
|
|
bindings = await self._positions.get_tag_bindings(pid)
|
|
tag_ids = bindings.get(resource_type) or []
|
|
if not tag_ids:
|
|
return None
|
|
return set(await self._tags.list_targets_for_tags(resource_type, tag_ids))
|
|
|
|
async def _resolve_materialized(self, user_id: str, resource_type: str, session: AsyncSession) -> set[str]:
|
|
pid = await self._effective_position_id(user_id)
|
|
bindings = await self._positions.get_tag_bindings(pid) if pid else {}
|
|
tag_ids = bindings.get(resource_type) or []
|
|
if tag_ids:
|
|
target = set(await self._tags.list_targets_for_tags(resource_type, tag_ids))
|
|
else:
|
|
target = await self._default_set(resource_type, session)
|
|
if resource_type == "agent":
|
|
return await self._filter_materializable_agents(target, session)
|
|
return target
|
|
|
|
async def _default_set(self, resource_type: str, session: AsyncSession) -> set[str]:
|
|
if resource_type == "agent":
|
|
rows = (await session.execute(select(AgentRow.id).where(AgentRow.user_id.is_(None)))).all()
|
|
return {r[0] for r in rows}
|
|
# scheduled_task (and any other materialized type) defaults to empty.
|
|
return set()
|
|
|
|
async def _filter_materializable_agents(self, agent_ids: set[str], session: AsyncSession) -> set[str]:
|
|
if not agent_ids:
|
|
return set()
|
|
rows = (
|
|
await session.execute(
|
|
select(AgentRow.id).where(
|
|
AgentRow.id.in_(agent_ids),
|
|
AgentRow.user_id.is_not(None),
|
|
)
|
|
)
|
|
).all()
|
|
candidate_ids = {r[0] for r in rows}
|
|
return candidate_ids - await self._business_chain_agent_ids(session)
|
|
|
|
async def _business_chain_agent_ids(self, session: AsyncSession) -> set[str]:
|
|
rows = (await session.execute(select(RoundtableChainRow.seats))).scalars()
|
|
agent_ids: set[str] = set()
|
|
for raw in rows:
|
|
if not raw:
|
|
continue
|
|
try:
|
|
seats = json.loads(raw) if isinstance(raw, str) else raw
|
|
except Exception:
|
|
continue
|
|
if not isinstance(seats, list):
|
|
continue
|
|
for seat in seats:
|
|
if isinstance(seat, dict) and isinstance(seat.get("agent_id"), str):
|
|
agent_ids.add(seat["agent_id"])
|
|
return agent_ids
|
|
|
|
# --------------------------------------------------------- materialize
|
|
|
|
async def sync_user(self, user_id: str) -> None:
|
|
"""Reconcile all materialized subscriptions for one user (idempotent)."""
|
|
if self._sf is None:
|
|
return
|
|
pid = await self._effective_position_id(user_id)
|
|
async with self._sf() as session:
|
|
await self._sync_agents(session, user_id, pid)
|
|
await self._sync_skills(session, user_id, pid)
|
|
await self._sync_tasks(session, user_id, pid)
|
|
await session.commit()
|
|
|
|
async def _sync_agents(self, session: AsyncSession, user_id: str, position_id: str | None) -> None:
|
|
target = await self._resolve_materialized(user_id, "agent", session)
|
|
rows = list((await session.execute(select(AgentFavoriteRow).where(AgentFavoriteRow.user_id == user_id))).scalars())
|
|
owned_any = {r.agent_id for r in rows}
|
|
position_rows = {r.agent_id: r for r in rows if r.origin == "position"}
|
|
now = datetime.now(UTC)
|
|
for agent_id in target - owned_any:
|
|
session.add(AgentFavoriteRow(id=uuid.uuid4().hex, user_id=user_id, agent_id=agent_id, origin="position", position_id=position_id, created_at=now))
|
|
for agent_id, row in position_rows.items():
|
|
if agent_id not in target:
|
|
await session.delete(row)
|
|
|
|
async def _sync_skills(self, session: AsyncSession, user_id: str, position_id: str | None) -> None:
|
|
target = await self._resolve_materialized(user_id, "skill", session)
|
|
rows = list((await session.execute(select(SkillFavoriteRow).where(SkillFavoriteRow.user_id == user_id))).scalars())
|
|
owned_any = {r.skill_name for r in rows}
|
|
position_rows = {r.skill_name: r for r in rows if r.origin == "position"}
|
|
now = datetime.now(UTC)
|
|
for skill_name in target - owned_any:
|
|
session.add(SkillFavoriteRow(id=uuid.uuid4().hex, user_id=user_id, skill_name=skill_name, origin="position", position_id=position_id, created_at=now))
|
|
for skill_name, row in position_rows.items():
|
|
if skill_name not in target:
|
|
await session.delete(row)
|
|
|
|
async def _sync_tasks(self, session: AsyncSession, user_id: str, position_id: str | None) -> None:
|
|
target = await self._resolve_materialized(user_id, "scheduled_task", session)
|
|
rows = list((await session.execute(select(ScheduledTaskSubscriptionRow).where(ScheduledTaskSubscriptionRow.user_id == user_id))).scalars())
|
|
owned_any = {r.task_id for r in rows}
|
|
position_rows = {r.task_id: r for r in rows if r.origin == "position"}
|
|
now = datetime.now(UTC)
|
|
for task_id in target - owned_any:
|
|
session.add(ScheduledTaskSubscriptionRow(id=uuid.uuid4().hex, task_id=task_id, user_id=user_id, notify_element=False, origin="position", position_id=position_id, created_at=now, updated_at=now))
|
|
for task_id, row in position_rows.items():
|
|
if task_id not in target:
|
|
await session.delete(row)
|
|
|
|
# ------------------------------------------------------------- fan-out
|
|
|
|
async def sync_position(self, position_id: str) -> int:
|
|
"""Re-sync every member of a position. Returns the member count."""
|
|
member_ids = await self._positions.list_member_ids(position_id)
|
|
for uid in member_ids:
|
|
try:
|
|
await self.sync_user(uid)
|
|
except Exception:
|
|
logger.exception("position sync failed for user %s (position %s)", uid, position_id)
|
|
return len(member_ids)
|
|
|
|
async def sync_default_members(self) -> int:
|
|
"""Re-sync users with NO explicit position (they inherit the default)."""
|
|
default_id = await self._positions.get_default_position_id()
|
|
if not default_id:
|
|
return 0
|
|
assigned = set(await self._positions.list_assigned_user_ids())
|
|
count = 0
|
|
if self._sf is None:
|
|
return 0
|
|
from deerflow.persistence.user.model import UserRow
|
|
|
|
async with self._sf() as session:
|
|
all_ids = [uid for (uid,) in (await session.execute(select(UserRow.id))).all()]
|
|
for uid in all_ids:
|
|
if uid in assigned:
|
|
continue
|
|
try:
|
|
await self.sync_user(uid)
|
|
count += 1
|
|
except Exception:
|
|
logger.exception("default position sync failed for user %s", uid)
|
|
return count
|