"""Handbook live-fact builders. Each function returns plain dicts (JSON-ready) describing one structural facet of the platform. They read the in-process registries (scheduled-job handlers, integration drivers) and the DB for enable-state only. """ from __future__ import annotations from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.integrations.models import Integration from app.integrations.registry import list_drivers from app.scheduled_jobs.models import ScheduledJob from app.scheduled_jobs.registry import get_handler, list_handlers # ── Domain labels ─────────────────────────────────────────────────── # The prefix before the first dot of a job key / a driver kind maps to a # human domain label. Kept here (not the DB) because it is documentation # taxonomy, not runtime state. _JOB_DOMAIN_LABELS: dict[str, str] = { "ai_spending": "AI 花费", "apikey": "API Key", "asset": "资产", "attendance": "考勤", "auditor": "安全审计", "auth": "认证", "bailian": "百炼 AI", "cmdb": "云与网络", "compliance": "合规治理", "cron": "平台运维", "feed": "动态广场", "feishu": "飞书", "finance": "财务", "gamification": "成就", "google": "Google Workspace", "hr": "HR 生命周期", "hris": "HRIS 同步", "identity": "身份治理", "identity_sync": "身份出站同步", "insights": "洞察", "jamf": "Jamf MDM", "jira": "Jira", "leaver": "离职", "notification": "通知", "offboarding": "离职", "onboarding": "入职", "procurement": "采购", "provisioning": "开通编排", "relationships": "关系图谱", "role_change": "角色变更", "secquiz": "安全答题", "slack": "Slack", "software_billing": "软件账单", "software_seat": "软件坐席", "sso": "SSO", "wrapped": "年度回顾", "zoom": "Zoom", "data_access": "数据仓库", } _KIND_LABELS: dict[str, str] = { "cloud": "云厂商", "software_provisioning": "软件坐席", "hris": "HRIS", "chat_approval": "审批 / 通讯", "helpdesk": "服务台", "email": "邮件发送", "data_access": "数据仓库", "identity_sync_endpoint": "身份出站", "identity_sync_feishu": "身份出站", "identity_sync_huilianyi": "身份出站", "identity_sync_netsuite": "身份出站", } def _job_domain(key: str) -> str: prefix = key.split(".", 1)[0] return _JOB_DOMAIN_LABELS.get(prefix, prefix) def _schedule_str(job: ScheduledJob) -> str: if job.trigger_type == "interval" and job.interval_seconds: secs = job.interval_seconds if secs % 3600 == 0: return f"每 {secs // 3600} 小时" if secs % 60 == 0: return f"每 {secs // 60} 分钟" return f"每 {secs} 秒" if job.cron_expr: return f"cron: {job.cron_expr} ({job.timezone})" return job.trigger_type or "—" async def list_scheduled_jobs(db: AsyncSession) -> list[dict]: """Every scheduled job the platform can run — the union of the code registry (authoritative catalogue of handlers) and the ``scheduled_jobs`` DB rows (which carry schedule + enabled). A handler with no DB row shows as ``未调度`` / disabled; a DB row with no handler is still listed (so a stale row is visible, not hidden). No run history (kept out on purpose).""" db_rows = ( await db.execute(select(ScheduledJob)) ).scalars().all() by_key: dict[str, ScheduledJob] = {r.handler_key: r for r in db_rows} keys = set(by_key) | {h.key for h in list_handlers()} out: list[dict] = [] for key in sorted(keys): job = by_key.get(key) handler = get_handler(key) name = ( (job.display_name if job and job.display_name else None) or (handler.display_name if handler else None) or key ) description = ( (job.description if job and job.description else None) or (handler.description if handler else "") or "" ) out.append( { "key": key, "name": name, "domain": _job_domain(key), "schedule": _schedule_str(job) if job else "未调度", "enabled": bool(job.enabled) if job else False, "description": description, } ) return out async def list_connectors(db: AsyncSession) -> list[dict]: """Every integration driver this build knows about, with capabilities and whether at least one instance is enabled.""" # provider+kind → any enabled instance? enabled_rows = ( await db.execute( select(Integration.kind, Integration.provider, Integration.enabled) ) ).all() enabled_map: dict[tuple[str, str], bool] = {} for kind, provider, enabled in enabled_rows: cur = enabled_map.get((kind, provider), False) enabled_map[(kind, provider)] = cur or bool(enabled) out: list[dict] = [] for meta in list_drivers(): out.append( { "kind": meta.kind, "kind_label": _KIND_LABELS.get(meta.kind, meta.kind), "provider": meta.provider, "name": meta.display_name or meta.provider, "capabilities": list(meta.capabilities or []), "configured": (meta.kind, meta.provider) in enabled_map, "enabled": enabled_map.get((meta.kind, meta.provider), False), } ) return out # ── RBAC roles ────────────────────────────────────────────────────── # The 9 canonical roles + manager-derived scope. Curated (stable) rather # than reflected, because the scope prose is documentation, not a runtime # value. Mirrors CLAUDE.md's RBAC model + app/core/rbac.py. _ROLES: list[dict] = [ { "slug": "global_admin", "name": "全局管理员", "scope": "全平台读写;不可逆注销 / 连接器凭证仅此角色", "derived": False, }, { "slug": "employee", "name": "员工", "scope": "本人范围:查看自己的资产 / 账号 / 权限,提交申请,读选定云资源", "derived": False, }, { "slug": "auditor", "name": "审计员", "scope": "审计 / 安全治理只读 + 明确的复核决策", "derived": False, }, { "slug": "asset_admin", "name": "资产管理员", "scope": "资产域管理,按办公区派生并区域限权", "derived": True, }, { "slug": "asset_global_admin", "name": "全局资产管理员", "scope": "资产域全办公区写权限;域外无权;联动派生 asset_admin", "derived": False, }, { "slug": "cloud_admin", "name": "云管理员", "scope": "云 / 网络域写权限;不可逆注销与凭证仍留 global_admin", "derived": False, }, { "slug": "hr", "name": "HR", "scope": "HRIS / 通讯录 / HR 工具面;可由 HR 部门自动派生", "derived": False, }, { "slug": "finance", "name": "财务", "scope": "账单相邻只读:采购 / 软件成本 / 云成本 / AI 花费;可由财务 BU 自动派生", "derived": False, }, { "slug": "management", "name": "管理层", "scope": "全管理面只读的领导视角;排除在管理员广播通知之外", "derived": False, }, { "slug": "manager", "name": "经理(派生)", "scope": "由 identity_persons.manager_person_id 派生:可见下属的资产 / 账号 / 权限。非存储角色", "derived": True, }, ] async def list_rbac_roles(_db: AsyncSession) -> list[dict]: return _ROLES