"""我的装备墙 — an RPG character-sheet view of everything the caller "owns" on the platform: equipped gear (current hardware, rarity-tiered), attributes (a 5-axis spread of assets / accounts / permissions / cloud / security), inventory counts, an achievements preview, and a security rank chip. This is the dark-mode / game-y face of the personal IT overview — composed almost entirely from existing, battle-tested self-scoped builders so there is no new aggregation logic to drift: * equipped gear + rarity ← ``device_gallery.build_my_device_gallery`` * tenure / seats / permissions / milestones ← ``footprint.build_my_footprint`` * security streak / correct ← ``security_quiz.service.compute_stats`` Only two extra read counts are added (both on indexed self-scoped columns): external-identity accounts and active cloud grants — the two attribute axes the footprint doesn't already carry. Read-only, self-scoped (the ``person`` arg is always the authenticated caller → no target param, no IDOR). """ from __future__ import annotations from datetime import UTC, datetime from typing import Any from pydantic import BaseModel from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.identity.person_models import IdentityExternalIdentity, IdentityPerson # Category → equipment-slot key. Anything unmatched falls back to "gear". _SLOT_BY_CATEGORY: dict[str, str] = { "laptop": "weapon", "notebook": "weapon", "desktop": "weapon", "workstation": "weapon", "monitor": "offhand", "display": "offhand", "phone": "comm", "mobile": "comm", "tablet": "comm", "keyboard": "gear", "mouse": "gear", "headset": "gear", "accessory": "gear", "dock": "gear", } def _slot_for(category: str | None) -> str: if not category: return "gear" return _SLOT_BY_CATEGORY.get(category.strip().lower(), "gear") def _rank_tier(total_correct: int) -> str: """Security rank tier for the loadout chip. Delegates to the canonical ladder (`security_quiz.ranks`) so the chip and the 段位擂台 never diverge.""" from app.domains.security_quiz.ranks import rank_for return rank_for(total_correct) class EquippedItem(BaseModel): id: str slot: str # weapon | offhand | comm | gear name: str | None category: str | None brand: str | None model: str | None service_days: int rarity: str # rookie | veteran | elder | legend class LoadoutResponse(BaseModel): person: dict[str, Any] tenure: dict[str, Any] level: int power: int rank: dict[str, Any] equipped: list[EquippedItem] attributes: dict[str, int] # raw axis counts; frontend scales the radar inventory: dict[str, Any] achievements_preview: list[dict[str, Any]] generated_at: str #: Badges shown on the loadout card. The wall itself is the full view; this is #: a teaser strip, so it stays short. _BADGE_PREVIEW_LIMIT = 6 async def _badge_preview( db: AsyncSession, person: IdentityPerson, fallback: list[dict[str, Any]], ) -> list[dict[str, Any]]: """The caller's most recently unlocked badges, newest first. This used to serve the footprint's milestone timeline as a stand-in because the badge system had not shipped yet — which meant the "成就" panel on the loadout card showed join/first-device events rather than any actual achievement. It reads the real unlock table now. Read-only: the wall's own endpoint owns lazy evaluation, so building a character card never writes — which means someone who has never opened the wall (and whom the nightly scan has not reached yet) has no unlock rows at all. Rather than render an empty 成就 panel for them, fall back to the footprint milestones this section used to show. An empty panel on a page whose whole job is "look at everything you have" is the one outcome not worth having. """ from app.domains.gamification.catalog import BADGES_BY_CODE from app.domains.gamification.models import GamificationUnlock rows = (await db.execute( select(GamificationUnlock) .where(GamificationUnlock.person_id == person.id) # ``badge_code`` breaks the tie: the first nightly scan stamps a whole # batch with one server_default timestamp, so ordering by time alone # made which six appear change between page loads. .order_by(GamificationUnlock.unlocked_at.desc(), GamificationUnlock.badge_code) .limit(_BADGE_PREVIEW_LIMIT) )).scalars().all() preview: list[dict[str, Any]] = [] for row in rows: badge = BADGES_BY_CODE.get(row.badge_code) if badge is None: # A retired code still in someone's unlock history — skip rather # than render an untranslatable key. continue preview.append({ "kind": "badge", "code": badge.code, "icon": badge.icon, "tier": badge.tier, "title_key": badge.name_key, "params": {}, "date": row.unlocked_at.date().isoformat() if row.unlocked_at else None, }) return preview or fallback async def build_my_loadout( db: AsyncSession, person: IdentityPerson, ) -> LoadoutResponse: # Local imports keep the domain-boundary lint happy and avoid a heavy # import graph at module load (mirrors the other me_universe builders). from app.domains.asset.models import AssetItem from app.domains.cmdb.models import CmdbCloudAccessGrant from app.domains.me_universe.device_gallery import build_my_device_gallery from app.domains.me_universe.footprint import build_my_footprint from app.domains.security_quiz.service import _local_today, compute_stats now = datetime.now(UTC) footprint = (await build_my_footprint(db, person=person)).model_dump() gallery = (await build_my_device_gallery(db, person=person)).model_dump() sec = await compute_stats(db, person.id, _local_today(person)) # ── Equipped gear: source from the LIVE assignment (assets currently # assigned to the caller), NOT the gallery's ownership-trail cards. # A device assigned directly (admin assign writes only history, and # some seed/import paths write neither operation nor the assigned # history action) is genuinely held now but may carry no trail event # — sourcing equipped from the trail would show empty slots for a # user who clearly holds hardware. We enrich service_days / rarity # from the matching gallery card when present, else fall back. gallery_by_id = {c["id"]: c for c in gallery["cards"]} live_assets = list((await db.execute( select(AssetItem).where( AssetItem.assigned_to == person.id, AssetItem.deleted_at.is_(None), ) )).scalars().all()) equipped: list[EquippedItem] = [] for a in live_assets: card = gallery_by_id.get(str(a.id)) equipped.append(EquippedItem( id=str(a.id), slot=_slot_for(a.category), name=a.asset_name, category=a.category, brand=a.brand, model=a.model, service_days=int(card["service_days"]) if card else 0, rarity=card["rarity"] if card else "rookie", )) # Longest-serving first so the hero gear tops the loadout; cap so the # sheet stays a "loadout" not a warehouse. equipped.sort(key=lambda e: -e.service_days) equipped = equipped[:8] # ── Extra attribute axes the footprint doesn't carry. account_rows = (await db.execute( select(IdentityExternalIdentity.provider).where( IdentityExternalIdentity.person_id == person.id, ) )).scalars().all() account_providers = sorted({p for p in account_rows if p}) accounts_count = len(account_providers) # ``CmdbCloudAccessGrant.person_id`` is a plain String(36), not the GUID # type — comparing it against a ``uuid.UUID`` matches zero rows silently, # which is why this card showed 0 cloud grants for everyone. Every other # call site in the cmdb domain wraps the id in ``str()``; so does this one. cloud_grants = (await db.execute( select(func.count()) .select_from(CmdbCloudAccessGrant) .where( CmdbCloudAccessGrant.person_id == str(person.id), CmdbCloudAccessGrant.status == "active", ) )).scalar_one() or 0 devices_current = int(footprint["devices"]["current_count"]) seats_current = int(footprint["seats"]["current"]) permissions_active = int(footprint["permissions"]["active"]) tenure_days = int(footprint["tenure"]["days"] or 0) total_correct = int(sec["total_correct"]) attributes = { "assets": devices_current, "accounts": accounts_count, "permissions": permissions_active, "cloud": int(cloud_grants), # 0–100 security score: lifetime correct, soft-capped at 100. "security": min(100, total_correct), } # ── Playful derived stats. Level grows ~1 per 40 days of tenure (cap 50); # power is a weighted footprint of everything you hold. Both are # cosmetic — rounded ints, never used for any gate. level = max(1, min(50, 1 + tenure_days // 40)) power = ( devices_current * 100 + accounts_count * 60 + permissions_active * 30 + int(cloud_grants) * 120 + total_correct * 20 + tenure_days ) rank = { "tier": _rank_tier(total_correct), "total_correct": total_correct, "total_answered": int(sec["total_answered"]), "current_streak": int(sec["current_streak"]), "longest_streak": int(sec["longest_streak"]), } inventory = { # total_ever from the footprint (live + history aware) — the gallery # summary is ownership-trail-only and undercounts directly-assigned # devices that never wrote a transfer/claim event. "devices_total": max( int(footprint["devices"].get("total_ever", 0) or 0), devices_current, ), "devices_current": devices_current, "seats": seats_current, "permissions": permissions_active, "accounts": accounts_count, "account_providers": account_providers, "cloud_grants": int(cloud_grants), } return LoadoutResponse( person=footprint["person"], tenure=footprint["tenure"], level=level, power=power, rank=rank, equipped=equipped, attributes=attributes, inventory=inventory, achievements_preview=await _badge_preview( db, person, fallback=footprint["milestones"], ), generated_at=now.isoformat(), ) __all__ = ["build_my_loadout", "LoadoutResponse", "EquippedItem"]