"""年度报告 (Helix Wrapped) — a per-person year-in-review. A Spotify-Wrapped-style recap of one employee's year: how long they have been at the company, how often they showed up, what they carried, what they earned, and how that compares to last year and to the people around them. Composed from year-scoped queries over existing tables; self-scoped (the ``person`` arg is the authenticated caller → no target param, no IDOR). The backend stays language-agnostic: every scene carries ``headline_key`` / ``caption_key`` + params, and the *selection* (which tier of headline applies, which scenes exist at all) is decided here so the view never branches on thresholds. Time zones ---------- "The year" is the person's LOCAL calendar year, and — since 2026-08-09 — so is every day bucket inside it. The previous implementation windowed the year in local time but bucketed activity by **UTC date**, so a UTC+8 employee's actions before 08:00 on Jan 1 landed on the previous year's date, were dropped from the heatmap, and yet still counted toward "active days". The heatmap had fewer lit cells than the sentence above it claimed, and the user could count them. Activity is now aggregated per local *hour* (≤8784 buckets for a year, one query) and folded into local days in Python using the real ``ZoneInfo`` — exact across DST rather than approximated with a fixed offset, and it yields the busiest-day and favourite-hour readings for free. """ from __future__ import annotations from collections import Counter from datetime import UTC, date, datetime, timedelta, tzinfo from typing import Any from pydantic import BaseModel from sqlalchemy import func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.core.tenure import ( completed_years, ordinal_year, tenure_reference_date, ) from app.identity.person_models import IdentityPerson class WrappedResponse(BaseModel): person: dict[str, Any] year: int is_current_year: bool tenure: dict[str, Any] security: dict[str, Any] devices_received: int tickets_opened: int # ``[{"code": ..., "unlocked_at": ...}]`` — the code alone rendered as # eight identical 🏅 with no name and no date, which communicated only a # count. The frontend localises via ``badge..name``. badges: list[dict[str, Any]] activity: dict[str, Any] # Year-over-year deltas for the headline numbers, or ``None`` when there is # no prior year to compare against (first year at the company). comparison: dict[str, Any] | None persona: dict[str, Any] scenes: list[dict[str, Any]] generated_at: str # Accent hues per scene — translucent-friendly brand tones that read on the # dark gradient cards (display only; the frontend may override). _ACCENT = { "opening": "#7c3aed", "logins": "#4f46e5", "actions": "#2563eb", "devices": "#0891b2", "security": "#16a34a", "badges": "#f59e0b", "social": "#d4537e", "support": "#ea580c", } # Audit actions that are "you logged in", not "you did something". Anchored # with a prefix + explicit membership rather than ``like("auth.%login%")`` so a # future ``auth.impersonate_login`` can't silently inflate someone's recap. _LOGIN_ACTIONS = ("auth.login", "auth.login_password", "auth.feishu_login") def is_wrapped_meaningful(w: WrappedResponse) -> bool: """Whether a wrapped payload carries any *real* year-in-review signal. The opening tenure scene, a ``persona`` (falls back to ``"steady"``) and the poster are ALWAYS present — even for a brand-new hire whose year is all zeros — so none of those can gate the year-end push. Meaningful == at least one genuine content dimension the recap actually brags about. Everything else (tenure alone) is a degenerate card that would deep-link to an all-zeros ``/me/wrapped/`` while claiming "你留下了不少足迹". """ act = w.activity or {} return bool( w.devices_received or w.security.get("answered") or w.badges or w.tickets_opened or act.get("logins") or act.get("platform_actions") or act.get("posts") or act.get("comments") or act.get("likes_given") ) def _pct_delta(now: int, before: int) -> int | None: """Whole-percent change, or ``None`` when there is no honest baseline. A jump from 0 is not "+∞%" and not "+100%" — it has no percentage, so the frontend is given ``None`` and shows the absolute number instead. """ if before <= 0: return None return round((now - before) * 100 / before) async def build_my_wrapped( db: AsyncSession, person: IdentityPerson, year: int, *, headcount: int | None = None, ) -> WrappedResponse: """``headcount`` lets a batch caller compute the company-wide count ONCE (#2193-3). It is person-independent, but the year-end push calls this in a loop over every employee — so it was a full ``identity_persons`` COUNT per person. Passing it in keeps the single-person path unchanged (None → computed here) while turning N table scans into one for the push. """ from app.audit.models import AuditLog from app.core.working_hours import resolve_timezone from app.domains.asset.models import AssetHistory, AssetOperation from app.domains.feed.models import FeedComment, FeedLike, FeedPost from app.domains.gamification.models import GamificationUnlock from app.domains.security_quiz.models import SecQuizAnswer from app.domains.security_quiz.service import _local_today, compute_stats from app.domains.support.models import SupportTicket # Guard the year at the boundary that actually does the unsafe arithmetic # (``datetime(year + 1, ...)`` below raises ValueError → unhandled 500 for # an out-of-range value like 0 / 99999). Clamping here protects EVERY caller # (HTTP route + MCP tool + any future one) from one place. year = max(2000, min(year, 2100)) tz = resolve_timezone(person) today_local = _local_today(person) jan1 = date(year, 1, 1) dec31 = date(year, 12, 31) dt_start = datetime(year, 1, 1, tzinfo=tz).astimezone(UTC) dt_end = datetime(year + 1, 1, 1, tzinfo=tz).astimezone(UTC) is_current = today_local.year == year # ── Tenure. Two DIFFERENT numbers, separately named: 「满 N 年」 (cardinal) # and 「第 N 年」 (ordinal). One value rendered as both was the bug. tenure_as_of = tenure_reference_date(year, today_local) joined_this_year = bool(person.hire_date and person.hire_date.year == year) tenure_completed: int | None = None tenure_ordinal: int | None = None if person.hire_date is not None: tenure_completed = completed_years(person.hire_date, tenure_as_of) tenure_ordinal = ordinal_year(person.hire_date, tenure_as_of) # ── Where the caller sits among colleagues. # # ``join_rank`` is historical truth (counts leavers), matching the footprint # page. ``earlier_than_current`` is the number of people STILL HERE who # joined after the caller — computed by its own query rather than # subtracting one measure from the other. # # The old code did ``current_headcount - join_rank`` on the frontend, mixing # a leaver-inclusive rank with a leaver-exclusive headcount. Once cumulative # hires exceeded current headcount (i.e. always, for any company with # attrition) that went negative and clamped to 0, so every recent joiner # read 「第 850 位加入 · 早于 0 位同事」. join_rank: int | None = None earlier_than_current = 0 if person.hire_date is not None: # Tie-break same-day joiners by created_at via a scalar subquery rather # than reading ``person.created_at`` in Python — that attribute is a # server-default column that lazy-loads, which trips MissingGreenlet # when touched mid-statement-build. my_created = ( select(IdentityPerson.created_at) .where(IdentityPerson.id == person.id) .scalar_subquery() ) joined_before_me = or_( IdentityPerson.hire_date < person.hire_date, (IdentityPerson.hire_date == person.hire_date) & (IdentityPerson.created_at < my_created), ) earlier = await db.scalar( select(func.count()).select_from(IdentityPerson).where( IdentityPerson.deleted_at.is_(None), IdentityPerson.hire_date.is_not(None), joined_before_me, ) ) join_rank = int(earlier or 0) + 1 earlier_than_current = int(await db.scalar( select(func.count()).select_from(IdentityPerson).where( IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status != "left", IdentityPerson.hire_date.is_not(None), IdentityPerson.id != person.id, ~joined_before_me, ) ) or 0) current_headcount = ( headcount if headcount is not None else int(await db.scalar( select(func.count()).select_from(IdentityPerson).where( IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status != "left", ) ) or 0) ) # ── Security-quiz effort within the year (quiz_date is a DATE). sec_row = (await db.execute( select( func.count().label("answered"), func.sum(func.if_(SecQuizAnswer.is_correct.is_(True), 1, 0)).label("correct"), ).where( SecQuizAnswer.person_id == person.id, SecQuizAnswer.quiz_date >= jan1, SecQuizAnswer.quiz_date <= dec31, ) )).one() answered = int(sec_row.answered or 0) correct = int(sec_row.correct or 0) # Longest quiz streak WITHIN THE YEAR. ``compute_stats`` returns a lifetime # figure, which the recap rendered under a 「连续 N 天」 chip identical to # the one on the activity slide — two different windows behind one UI. quiz_dates = sorted(set((await db.execute( select(SecQuizAnswer.quiz_date).where( SecQuizAnswer.person_id == person.id, SecQuizAnswer.quiz_date >= jan1, SecQuizAnswer.quiz_date <= dec31, ) )).scalars().all())) quiz_streak = _longest_run(quiz_dates) lifetime_stats = await compute_stats(db, person.id, today_local) # ── Devices that came into the caller's hands during the year. dev_ids: set = set((await db.execute( select(AssetOperation.asset_id).where( AssetOperation.to_user_id == person.id, AssetOperation.status == "confirmed", AssetOperation.operation_type.in_(("transfer", "claim")), AssetOperation.created_at >= dt_start, AssetOperation.created_at < dt_end, ) )).scalars().all()) dev_ids |= set((await db.execute( select(AssetHistory.asset_id).where( AssetHistory.to_user == person.id, AssetHistory.action.in_(("assigned", "transferred")), AssetHistory.created_at >= dt_start, AssetHistory.created_at < dt_end, ) )).scalars().all()) devices_received = len(dev_ids) tickets_opened = int((await db.execute( select(func.count()).select_from(SupportTicket).where( SupportTicket.requester_id == person.id, SupportTicket.created_at >= dt_start, SupportTicket.created_at < dt_end, ) )).scalar_one() or 0) badge_rows = (await db.execute( select(GamificationUnlock.badge_code, GamificationUnlock.unlocked_at) .where( GamificationUnlock.person_id == person.id, GamificationUnlock.unlocked_at >= dt_start, GamificationUnlock.unlocked_at < dt_end, ) .order_by(GamificationUnlock.unlocked_at) )).all() badges = [ {"code": code, "unlocked_at": _local_date_iso(unlocked_at, tz)} for code, unlocked_at in badge_rows ] # ── Activity. pid_str = str(person.id) async def _count(stmt) -> int: return int((await db.execute(stmt)).scalar_one() or 0) logins = await _count( select(func.count()).select_from(AuditLog).where( AuditLog.actor_user_id == person.id, AuditLog.action.in_(_LOGIN_ACTIONS), AuditLog.created_at >= dt_start, AuditLog.created_at < dt_end, ) ) # Everything the person DID, excluding the act of arriving. Logins were # previously counted twice — once on their own slide, then again inside a # bare ``COUNT(*)`` over the whole audit trail — so 「800 次操作」 silently # contained the 「200 次登录」 shown one slide earlier, and the per-day # average inherited the inflation. platform_actions = await _count( select(func.count()).select_from(AuditLog).where( AuditLog.actor_user_id == person.id, AuditLog.created_at >= dt_start, AuditLog.created_at < dt_end, ~AuditLog.action.in_(_LOGIN_ACTIONS), ) ) posts = await _count( select(func.count()).select_from(FeedPost).where( FeedPost.author_person_id == pid_str, FeedPost.post_type == "user", FeedPost.deleted_at.is_(None), FeedPost.created_at >= dt_start, FeedPost.created_at < dt_end, ) ) comments = await _count( select(func.count()).select_from(FeedComment).where( FeedComment.author_person_id == pid_str, FeedComment.deleted_at.is_(None), FeedComment.created_at >= dt_start, FeedComment.created_at < dt_end, ) ) likes_given = await _count( select(func.count()).select_from(FeedLike).where( FeedLike.person_id == pid_str, FeedLike.created_at >= dt_start, FeedLike.created_at < dt_end, ) ) # Per-UTC-quarter-hour counts, folded into LOCAL days below (see # ``_parse_utc_bucket`` for why quarter-hours and not hours). ≤35,136 rows # for a full year, and bounded above by the number of audit rows, so in # practice two orders of magnitude fewer. hour_rows = (await db.execute( select( func.date_format(AuditLog.created_at, "%Y-%m-%d %H").label("h"), func.floor(func.minute(AuditLog.created_at) / 15).label("q"), func.count().label("n"), ).where( AuditLog.actor_user_id == person.id, AuditLog.created_at >= dt_start, AuditLog.created_at < dt_end, ).group_by("h", "q") )).all() per_day: Counter[date] = Counter() per_local_hour: Counter[int] = Counter() for bucket, quarter, n in hour_rows: parsed = _parse_utc_bucket(bucket, quarter) if parsed is None: continue local = parsed.astimezone(tz) # The local-year window is enforced by the SQL bounds, but a DST shift # can push a boundary hour one day outside it; keep the recap inside # its own year rather than lighting a cell in another one. if local.year != year: continue per_day[local.date()] += int(n or 0) per_local_hour[local.hour] += int(n or 0) active_dates = sorted(per_day) active_days = len(active_dates) active_streak = _longest_run(active_dates) # 1-based day-of-year offsets for the heatmap grid. active_doys = [(d - jan1).days + 1 for d in active_dates] busiest = max(per_day.items(), key=lambda kv: (kv[1], kv[0]), default=None) peak_hour = ( max(per_local_hour.items(), key=lambda kv: (kv[1], -kv[0]))[0] if per_local_hour else None ) # Which weekday Jan 1 falls on (Mon=0) — lets the frontend align the # heatmap to real weeks instead of chunking days into arbitrary rows of 7. jan1_weekday = jan1.weekday() activity = { "logins": logins, "platform_actions": platform_actions, "active_days": active_days, "active_streak": active_streak, "active_doys": active_doys, "days_in_year": (date(year + 1, 1, 1) - jan1).days, "jan1_weekday": jan1_weekday, "busiest_day": busiest[0].isoformat() if busiest else None, "busiest_day_actions": busiest[1] if busiest else 0, "peak_hour": peak_hour, "posts": posts, "comments": comments, "likes_given": likes_given, } # ── Last year, for the "比去年" reading. Two cheap counts; skipped entirely # when the caller had not joined yet (a comparison against a year you # weren't here for is noise, not insight). comparison: dict[str, Any] | None = None prev_year = year - 1 hired_before_prev_year = ( person.hire_date is not None and person.hire_date.year <= prev_year ) if hired_before_prev_year: prev_start = datetime(prev_year, 1, 1, tzinfo=tz).astimezone(UTC) prev_end = datetime(year, 1, 1, tzinfo=tz).astimezone(UTC) prev_actions = await _count( select(func.count()).select_from(AuditLog).where( AuditLog.actor_user_id == person.id, AuditLog.created_at >= prev_start, AuditLog.created_at < prev_end, ~AuditLog.action.in_(_LOGIN_ACTIONS), ) ) # Folded to LOCAL days exactly like the current year's count. A # ``COUNT(DISTINCT DATE(created_at))`` here would be UTC days, and # comparing a local-day count against a UTC-day count is the very # mistake this rewrite exists to fix: for a UTC+8 employee any action # before 08:00 local lands on the previous UTC date, so a day with # activity both early and late yields TWO UTC dates. The prior year # would read systematically higher and the delta would show a decline # for someone whose behaviour never changed. prev_hour_rows = (await db.execute( select( func.date_format(AuditLog.created_at, "%Y-%m-%d %H").label("h"), func.floor(func.minute(AuditLog.created_at) / 15).label("q"), ).where( AuditLog.actor_user_id == person.id, AuditLog.created_at >= prev_start, AuditLog.created_at < prev_end, ).group_by("h", "q") )).all() prev_local_days: set[date] = set() for bucket, quarter in prev_hour_rows: parsed = _parse_utc_bucket(bucket, quarter) if parsed is None: continue local = parsed.astimezone(tz) if local.year == prev_year: prev_local_days.add(local.date()) prev_active_days = len(prev_local_days) comparison = { "year": prev_year, "platform_actions": prev_actions, "platform_actions_delta_pct": _pct_delta(platform_actions, prev_actions), "active_days": prev_active_days, "active_days_delta_pct": _pct_delta(active_days, prev_active_days), } # ── Year keyword (persona). persona_scores = { "builder": platform_actions / 800.0, "guardian": answered / 200.0, "social": (posts * 5 + comments * 2 + likes_given) / 100.0, "collector": (devices_received + len(badges)) / 8.0, } top_code = max(persona_scores, key=lambda k: persona_scores[k]) top_score = persona_scores[top_code] persona_code = top_code if top_score >= 0.15 else "steady" persona = { "code": persona_code, "score": round(top_score, 3), "dimensions": {k: round(v, 3) for k, v in persona_scores.items()}, } # ── Scenes. scenes: list[dict[str, Any]] = [] def _scene(scene_id: str, emoji: str, value: Any, headline_key: str, caption_key: str, params: dict[str, Any]) -> None: scenes.append({ "id": scene_id, "emoji": emoji, "value": value, "headline_key": headline_key, "caption_key": caption_key, "params": params, "accent": _ACCENT.get(scene_id, "#4f46e5"), }) if joined_this_year: _scene("opening", "🎉", year, "wrapped.scene.joined.headline", "wrapped.scene.joined.caption", {"year": year}) elif not tenure_completed: # Hired in a PREVIOUS calendar year but not yet a full year ago. # Routing these readers into the "you joined this year" slide — which # is what an earlier `or not tenure_completed` guard did — told someone # hired 2025-12-31 that their story began in 2026, and `shareBody` # would have published that sentence to the company feed. Printing # 「一路同行 0 年」 is no better. So: no number at all, and the poster # cell prints 「—」 for the same reason its neighbour already does. _scene("opening", "🧭", None, "wrapped.scene.tenure.firstYear", "wrapped.scene.tenure.caption", {}) else: # COMPLETED years, not the ordinal — the poster's 「{n} 年 · 一路同行」 # reads from ``completed_years``, and the two differ by exactly one. # Splitting the two numbers was right (the previous version rendered a # single value as both 「第 3 年」 and 「3 年」, and only one of those # could be true); showing BOTH of them two slides apart in the same # story was not. A reader sees 5 then 4 and concludes one is a bug. # ``ordinal_year`` stays on the API for callers that want the milestone # framing — it just no longer appears beside its own sibling. _scene("opening", "🧭", tenure_completed, "wrapped.scene.tenure.headline", "wrapped.scene.tenure.caption", {"years": tenure_completed}) if logins: if active_days >= 200: login_h = "wrapped.scene.logins.headline.daily" elif active_days >= 100: login_h = "wrapped.scene.logins.headline.often" else: login_h = "wrapped.scene.logins.headline.some" _scene("logins", "📅", logins, login_h, "wrapped.scene.logins.caption", {"logins": logins, "activeDays": active_days}) if platform_actions: per_day_avg = ( round(platform_actions / active_days, 1) if active_days else platform_actions ) if platform_actions >= 1000: act_h = "wrapped.scene.actions.headline.high" elif platform_actions >= 100: act_h = "wrapped.scene.actions.headline.mid" else: act_h = "wrapped.scene.actions.headline.low" _scene("actions", "⚡", platform_actions, act_h, "wrapped.scene.actions.caption", {"actions": platform_actions, "perDay": per_day_avg}) if devices_received: _scene("devices", "💻", devices_received, "wrapped.scene.devices.headline", "wrapped.scene.devices.caption", {"n": devices_received}) if answered: rate = round(correct * 100 / answered) sec_h = ("wrapped.scene.security.headline.sharp" if rate >= 80 else "wrapped.scene.security.headline.steady") _scene("security", "🛡️", correct, sec_h, "wrapped.scene.security.caption", {"answered": answered, "correct": correct, "rate": rate, "streak": quiz_streak}) if badges: _scene("badges", "🏅", len(badges), "wrapped.scene.badges.headline", "wrapped.scene.badges.caption", {"n": len(badges)}) if tickets_opened: _scene("support", "🛎️", tickets_opened, "wrapped.scene.support.headline", "wrapped.scene.support.caption", {"n": tickets_opened}) # Social. ``value`` is the dimension the person actually used: someone who # only ever liked things used to get a giant "0" over 「你发了 0 条动态」. if posts or comments or likes_given: if posts: social_value, social_key = posts, "wrapped.scene.social.headline" elif comments: social_value, social_key = comments, "wrapped.scene.social.headline.comments" else: social_value, social_key = likes_given, "wrapped.scene.social.headline.likes" _scene("social", "💬", social_value, social_key, "wrapped.scene.social.caption", {"posts": posts, "comments": comments, "likes": likes_given}) return WrappedResponse( person={ "id": str(person.id), "display_name": person.display_name, "employee_no": person.employee_no, "avatar_url": person.avatar_url, }, year=year, is_current_year=is_current, tenure={ "hire_date": person.hire_date.isoformat() if person.hire_date else None, "joined_this_year": joined_this_year, # 「满 N 年」 — use for "N 年 · 一路同行". "completed_years": tenure_completed, # 「第 N 年」 — use for "这是你在 PingCAP 的第 N 年". # Exposed, but NOT rendered anywhere. Every user-facing tenure # surface (this recap's opening slide + poster, footprint, cohort, # the celebration post) reads ``completed_years``, deliberately: # showing 「第 5 年」 beside 「4 年」 reads as a bug even though # both are correct. If you reach for this field for a new surface, # make sure nothing on the same screen shows the cardinal. "ordinal_year": tenure_ordinal, "join_rank": join_rank, "earlier_than_current": earlier_than_current, "current_headcount": current_headcount, }, security={ "answered": answered, "correct": correct, "year_streak": quiz_streak, "lifetime_streak": int(lifetime_stats["longest_streak"]), }, devices_received=devices_received, tickets_opened=tickets_opened, activity=activity, badges=badges, comparison=comparison, persona=persona, scenes=scenes, generated_at=datetime.now(UTC).isoformat(), ) def _local_date_iso(value: datetime | None, tz: tzinfo) -> str | None: """A stored timestamp as an ISO date in ``tz``. MySQL ``DATETIME`` columns come back NAIVE, and ``naive.astimezone(tz)`` does not mean "this is UTC, convert it" — Python reads a naive datetime as the *host's* local wall clock. On any deployment whose container clock is not UTC that shifted every badge by a day. Stamped UTC explicitly, and aware values (some drivers, and every value constructed in tests) are passed through rather than re-stamped. """ if value is None: return None aware = value if value.tzinfo is not None else value.replace(tzinfo=UTC) return aware.astimezone(tz).date().isoformat() def _parse_utc_bucket(value: Any, quarter: Any) -> datetime | None: """``('%Y-%m-%d %H', quarter)`` → aware UTC datetime at the bucket start. Buckets are FIFTEEN minutes, not an hour, because the fold to local time happens in Python and a whole-hour bucket cannot be folded exactly for the zones that sit on a fractional offset — +05:30 (India), +05:45 (Nepal), +09:30 (Australia/Adelaide), +03:30 (Iran). A UTC 03:00–04:00 bucket is 08:30–09:30 in Kolkata: half of it belongs to local hour 8 and half to hour 9, and a bucket straddling local midnight puts its last 30 minutes on the wrong calendar day. Attributing the whole bucket to its start skewed 「最活跃时段」 systematically early and mis-assigned the tail of every local day for those employees. Every real-world UTC offset is a multiple of 15 minutes, so a 15-minute bucket folds exactly for all of them. """ if isinstance(value, datetime): base = value.replace(minute=0, second=0, microsecond=0, tzinfo=UTC) elif isinstance(value, str): try: base = datetime.strptime(value.strip(), "%Y-%m-%d %H").replace(tzinfo=UTC) except ValueError: return None else: return None try: q = int(quarter) except (TypeError, ValueError): q = 0 return base + timedelta(minutes=15 * max(0, min(3, q))) def _longest_run(days: list[date]) -> int: """Longest run of consecutive calendar days in a sorted, de-duplicated list.""" longest = 0 run = 0 prev: date | None = None for d in days: run = run + 1 if (prev is not None and d - prev == timedelta(days=1)) else 1 prev = d longest = max(longest, run) return longest async def available_wrapped_years( db: AsyncSession, person: IdentityPerson, ) -> list[int]: """Years this person has a recap for, newest first. From their hire year to the current local year. Without this the only way to reach last year's recap was to hand-edit the URL — the entry point hard- coded ``new Date().getFullYear()``, so on January 2nd the button opened an almost-empty current-year report. """ from app.domains.security_quiz.service import _local_today this_year = _local_today(person).year first = person.hire_date.year if person.hire_date else this_year first = max(2000, min(first, this_year)) return list(range(this_year, first - 1, -1)) __all__ = [ "available_wrapped_years", "build_my_wrapped", "is_wrapped_meaningful", "WrappedResponse", ]