"""Feed domain — HTTP surface. Routes mounted under ``/api/v1/feed`` (employee + admin): * GET ``/posts`` — list (tab + category + cursor) * POST ``/posts`` — create user post * GET ``/posts/{id}`` — detail * PATCH ``/posts/{id}`` — author edits body * DELETE ``/posts/{id}`` — soft-delete * GET ``/posts/{id}/comments`` * POST ``/posts/{id}/comments`` * PATCH ``/comments/{id}`` — author edits body * DELETE ``/comments/{id}`` — author / post-author / admin delete * POST ``/posts/{id}/like`` — like * DELETE ``/posts/{id}/like`` — unlike * GET ``/search`` — LIKE search across visible posts * GET ``/mention-search`` — composer @ picker (auth required, rate-limited) * GET ``/today/celebrations`` — top banners * GET ``/celebrations/{id}`` — celebration detail (incl. blessings) * POST ``/celebrations/{id}/blessings`` * POST ``/celebrations/{id}/blessings/{bid}/ack`` — celebrant thanks one * POST ``/celebrations/{id}/thanks`` — celebrant thanks all * GET ``/announcements`` — public sidebar list * GET ``/admin/announcements`` (admin) * POST ``/admin/announcements`` * PATCH ``/admin/announcements/{id}`` * POST ``/admin/announcements/batch-status`` — batch publish/archive * DELETE ``/admin/announcements/{id}`` Plus on the existing ``/api/v1/me`` surface: * GET / PATCH ``/me/preferences`` — birthday + feed prefs. """ from __future__ import annotations from datetime import UTC, datetime from typing import Any from fastapi import APIRouter, Depends, Query, Request from sqlalchemy import or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.database import get_db from app.core.dependencies import ( check_directory_read_rate_limit, get_current_user, require_active_employee_for_mutations, require_browser_session, ) from app.core.exceptions import AppException, ForbiddenError, NotFoundError from app.core.response import success_response from app.core.safe_logging import get_safe_logger from app.domains.feed import notifications as feed_notifications from app.domains.feed.models import ( FeedAnnouncement, FeedPost, ) from app.domains.feed.schemas import ( FeedAnnouncementBatchStatus, FeedAnnouncementCreate, FeedAnnouncementUpdate, FeedAuthorRevealRequest, FeedBlessingCreate, FeedCelebrationThanks, FeedCommentBodyEdit, FeedCommentCreate, FeedPostBodyEdit, FeedPostCreate, FeedReportCreate, FeedReportResolve, PreferencesUpdate, ) from app.domains.feed.service import FeedService, _is_admin, _iso_utc from app.identity.person_models import IdentityPerson logger = get_safe_logger(__name__) # Router-level Leaver-Blocked gate (mirrors provisioning + notifications # routers): GET/reads pass through so a leaver can still view the feed, # but POST/PATCH/DELETE (post / comment / like / blessing) are rejected # for non-active employees mid-offboarding. Without this, the feed was # the one social surface a leaver_pending user could still mutate — # a Leaver-Blocked invariant hole. Fixed 2026-06-19 (backend review). router = APIRouter( prefix="/api/v1/feed", tags=["feed"], dependencies=[Depends(require_active_employee_for_mutations)], ) # Datetime UTC anchoring uses the shared ``service._iso_utc`` helper — # DateTime(timezone=True) loses tzinfo on MySQL round-trip, so every # response-layer datetime is forced through ``_iso_utc`` (anchor to UTC, # emit ``+00:00`` marker). Maintained in one place so a future TZ- # anchoring change touches one file, not two. def _serialize_announcement( ann: Any, author: IdentityPerson | None, ) -> dict[str, Any]: return { "id": ann.id, "title": ann.title, "body_markdown": ann.body_markdown, "category": ann.category, "color": ann.color, "is_pinned": ann.is_pinned, "publish_at": _iso_utc(ann.publish_at), "expire_at": _iso_utc(ann.expire_at), "status": ann.status, "attachments_json": ann.attachments_json, "author": ( { "id": str(author.id), "display_name": author.display_name, "avatar_url": author.avatar_url, } if author else None ), "created_at": _iso_utc(ann.created_at), "updated_at": _iso_utc(ann.updated_at), } # ────────────────────────────────────────────────────────────────── # Posts # ────────────────────────────────────────────────────────────────── @router.get("/posts") async def list_posts( tab: str = Query("public"), category: str | None = Query(None), cursor: str | None = Query(None), limit: int = Query(20, ge=1, le=50), db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), ) -> dict[str, Any]: svc = FeedService(db) posts, next_cursor = await svc.list_posts( viewer=user, tab=tab, category=category, cursor=cursor, limit=limit, ) items = await svc.hydrate_posts(posts, viewer=user) return success_response({"items": items, "next_cursor": next_cursor}) @router.post("/posts") async def create_post( body: FeedPostCreate, request: Request, db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), # Rate-limit creates like edits — without it a valid token can script # POST /posts at machine speed, and each public post with mentions fans # out Feishu DMs. Same bucket as edit_post / the directory picker. _rate_limit: None = Depends(check_directory_read_rate_limit), ) -> dict[str, Any]: svc = FeedService(db) post, created = await svc.create_user_post( author=user, body=body.body, visibility=body.visibility, category=body.category, mentioned_person_ids=body.mentioned_person_ids, attachments=[a.model_dump() for a in body.attachments], client_token=body.client_token, author_visibility=body.author_visibility, ) if not created: # ``client_token`` replay. Answer 200 with the post that already # exists — a retry after a dropped response must look like the success # it actually was — but write no audit row and, above all, send no # second round of @-mention pushes to everyone named in it. await db.commit() [dto] = await svc.hydrate_posts([post], viewer=user) return success_response({**dto, "replayed": True}) await AuditService(db).log( trace_id=getattr(request.state, "trace_id", "feed.post.create"), actor_type="user", actor_user_id=user.id, action="feed.post.create", status="success", resource_type="feed_post", resource_id=post.id, detail={ "visibility": post.visibility, "category": post.category, "author_visibility": post.author_visibility, }, ) await db.commit() await db.refresh(post) await feed_notifications.notify_mentions_on_post( db, post=post, mentioned_person_ids=body.mentioned_person_ids, ) [dto] = await svc.hydrate_posts([post], viewer=user) return success_response({**dto, "replayed": False}) @router.get("/posts/{post_id}") async def get_post( post_id: str, db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), ) -> dict[str, Any]: post = await db.get(FeedPost, post_id) svc = FeedService(db) if post is None or post.deleted_at is not None: raise NotFoundError("post not found") if not svc._can_view_post(post, user): raise ForbiddenError("cannot view this post") [dto] = await svc.hydrate_posts([post], viewer=user) return success_response(dto) @router.delete("/posts/{post_id}") async def delete_post( post_id: str, request: Request, db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), ) -> dict[str, Any]: svc = FeedService(db) await svc.soft_delete_post(post_id=post_id, viewer=user) await AuditService(db).log( trace_id=getattr(request.state, "trace_id", "feed.post.delete"), actor_type="user", actor_user_id=user.id, action="feed.post.delete", status="success", resource_type="feed_post", resource_id=post_id, ) await db.commit() return success_response({"ok": True}) @router.patch("/posts/{post_id}") async def edit_post( post_id: str, body: FeedPostBodyEdit, request: Request, db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), # Rate-limit edits — without it an attacker with a valid token can # spam PATCH at typing speed, bloating the audit log + DB write # bandwidth. Same bucket as the directory picker — comfortable for # human use (occasional edits) but kills sustained scripted abuse. _rate_limit: None = Depends(check_directory_read_rate_limit), ) -> dict[str, Any]: """Rewrite a user post's body. Author-only — admins can soft-delete via the existing route but cannot rewrite someone else's words. Full before/after lives in the audit log for forensic review. """ svc = FeedService(db) post, prev_body = await svc.edit_post( post_id=post_id, viewer=user, new_body=body.body, ) await AuditService(db).log( trace_id=getattr(request.state, "trace_id", "feed.post.edit"), actor_type="user", actor_user_id=user.id, action="feed.post.edit", status="success", resource_type="feed_post", resource_id=post.id, detail={ "prev_body": prev_body, "new_body": post.body, "edit_count": post.edit_count, }, ) await db.commit() [dto] = await svc.hydrate_posts([post], viewer=user) return success_response(dto) # ────────────────────────────────────────────────────────────────── # Comments # ────────────────────────────────────────────────────────────────── @router.get("/posts/{post_id}/comments") async def list_comments( post_id: str, limit: int = Query(200, ge=1, le=500), db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), ) -> dict[str, Any]: svc = FeedService(db) items = await svc.list_comments(post_id=post_id, viewer=user, limit=limit) return success_response({"items": items}) @router.post("/posts/{post_id}/comments") async def add_comment( post_id: str, body: FeedCommentCreate, db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), # Rate-limit comments like create/edit posts — each comment fans out a # Feishu DM to the post author AND up to MAX_MENTIONS_PER_POST mention DMs; # without this a valid token can script comments at machine speed and flood # the author + mentioned colleagues. Same bucket as the sibling mutations. _rate_limit: None = Depends(check_directory_read_rate_limit), ) -> dict[str, Any]: svc = FeedService(db) comment = await svc.add_comment( post_id=post_id, author=user, body=body.body, mentioned_person_ids=body.mentioned_person_ids, reply_to_comment_id=body.reply_to_comment_id, author_visibility=body.author_visibility, ) await db.commit() # Notifications — comment on post + mentions in comment. await feed_notifications.notify_comment_added(db, comment=comment) await feed_notifications.notify_mentions_on_comment( db, comment=comment, mentioned_person_ids=body.mentioned_person_ids, ) return success_response({ "id": comment.id, "post_id": comment.post_id, # ``service.add_comment`` now sets created_at Python-side, so # this is always populated. ``_iso_utc`` anchors the UTC marker # regardless of MySQL's tz-stripping (see service helper docstring). "created_at": _iso_utc(comment.created_at), # EFFECTIVE reply pointer, not the requested one: ``add_comment`` # drops a pointer that names an unknown / deleted / cross-post # parent, and without echoing it the client would render a reply # the server actually stored as a top-level comment. "reply_to_comment_id": comment.reply_to_comment_id, }) # ────────────────────────────────────────────────────────────────── # Community reporting / moderation # ────────────────────────────────────────────────────────────────── @router.post("/{resource_type}/{resource_id}/reports") async def report_content( resource_type: str, resource_id: str, body: FeedReportCreate, request: Request, db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), ) -> dict[str, Any]: svc = FeedService(db) report = await svc.create_report( resource_type=resource_type, resource_id=resource_id, reporter=user, reason=body.reason, detail=body.detail, ) await AuditService(db).log( trace_id=getattr(request.state, "trace_id", "feed.report.create"), actor_type="user", actor_user_id=user.id, action="feed.report.create", status="success", resource_type=f"feed_{resource_type}", resource_id=resource_id, detail={"report_id": report.id, "reason": body.reason}, ) await db.commit() return success_response({"id": report.id, "status": report.status}) def _require_feed_moderator(user: IdentityPerson) -> None: if not _is_admin(user): raise ForbiddenError("feed moderation requires HR or global admin") @router.get("/admin/reports") async def list_reports( status: str | None = Query(None), db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), ) -> dict[str, Any]: _require_feed_moderator(user) reports = await FeedService(db).list_reports(status=status) return success_response({"items": [{ "id": r.id, "resource_type": r.resource_type, "resource_id": r.resource_id, "reason": r.reason, "detail": r.detail, "status": r.status, "created_at": _iso_utc(r.created_at), "resolved_at": _iso_utc(r.resolved_at), "resolution_note": r.resolution_note, } for r in reports]}) @router.post("/admin/reports/{report_id}/resolve") async def resolve_report( report_id: str, body: FeedReportResolve, request: Request, db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), _browser: None = Depends(require_browser_session), ) -> dict[str, Any]: _require_feed_moderator(user) report = await FeedService(db).resolve_report( report_id=report_id, moderator=user, status=body.status, hide_content=body.hide_content, note=body.note, ) await AuditService(db).log( trace_id=getattr(request.state, "trace_id", "feed.report.resolve"), actor_type="user", actor_user_id=user.id, action="feed.report.resolve", status="success", resource_type="feed_report", resource_id=report.id, detail={"resolution": body.status, "hide_content": body.hide_content, "note": body.note}, ) await db.commit() return success_response({"id": report.id, "status": report.status}) @router.post("/admin/reports/{report_id}/reveal-author") async def reveal_report_author( report_id: str, body: FeedAuthorRevealRequest, request: Request, db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), _browser: None = Depends(require_browser_session), ) -> dict[str, Any]: _require_feed_moderator(user) author = await FeedService(db).reveal_report_author(report_id=report_id) await AuditService(db).log( trace_id=getattr(request.state, "trace_id", "feed.report.reveal_author"), actor_type="user", actor_user_id=user.id, action="feed.report.reveal_author", status="success", resource_type="feed_report", resource_id=report_id, detail={"reason": body.reason, "author_person_id": str(author.id)}, ) await db.commit() return success_response({ "id": str(author.id), "display_name": author.display_name, "work_email": author.work_email, "department": author.primary_department, }) @router.delete("/comments/{comment_id}") async def delete_comment( comment_id: str, request: Request, db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), ) -> dict[str, Any]: """Soft-delete a comment. Authorisation lives in ``FeedService._can_delete_comment`` (author / post author / admin). Audit log captures the pre-delete body so admins can review what was hidden without re-querying the row. """ svc = FeedService(db) comment, prev_body = await svc.soft_delete_comment( comment_id=comment_id, viewer=user, ) await AuditService(db).log( trace_id=getattr(request.state, "trace_id", "feed.comment.delete"), actor_type="user", actor_user_id=user.id, action="feed.comment.delete", status="success", resource_type="feed_comment", resource_id=comment.id, detail={ "post_id": comment.post_id, "prev_body": prev_body, "author_person_id": str(comment.author_person_id), }, ) await db.commit() return success_response({"ok": True}) @router.patch("/comments/{comment_id}") async def edit_comment( comment_id: str, body: FeedCommentBodyEdit, request: Request, db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), _rate_limit: None = Depends(check_directory_read_rate_limit), ) -> dict[str, Any]: """Rewrite a comment's body. Author-only; mirror-blessing comments are immutable (see ``FeedService._can_edit_comment``). Audit logs the before/after for admin forensics. """ svc = FeedService(db) comment, prev_body = await svc.edit_comment( comment_id=comment_id, viewer=user, new_body=body.body, ) await AuditService(db).log( trace_id=getattr(request.state, "trace_id", "feed.comment.edit"), actor_type="user", actor_user_id=user.id, action="feed.comment.edit", status="success", resource_type="feed_comment", resource_id=comment.id, detail={ "post_id": comment.post_id, "prev_body": prev_body, "new_body": comment.body, "edit_count": comment.edit_count, }, ) await db.commit() return success_response({ "id": comment.id, "post_id": comment.post_id, "body": comment.body, "edit_count": comment.edit_count, "last_edited_at": _iso_utc(comment.last_edited_at), }) # ────────────────────────────────────────────────────────────────── # Likes # ────────────────────────────────────────────────────────────────── @router.post("/posts/{post_id}/like") async def like_post( post_id: str, db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), ) -> dict[str, Any]: svc = FeedService(db) n, changed = await svc.toggle_like(post_id=post_id, person=user, like=True) await db.commit() # Only notify on an actual state change — a repeated like # (double-click / retry) must not re-deliver "X liked your post". if changed: await feed_notifications.notify_like_added( db, post_id=post_id, liker=user, ) return success_response({"like_count": n, "liked_by_me": True}) @router.delete("/posts/{post_id}/like") async def unlike_post( post_id: str, db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), ) -> dict[str, Any]: svc = FeedService(db) n, _changed = await svc.toggle_like(post_id=post_id, person=user, like=False) await db.commit() return success_response({"like_count": n, "liked_by_me": False}) # ────────────────────────────────────────────────────────────────── # Search # ────────────────────────────────────────────────────────────────── @router.get("/search") async def search( q: str = Query(min_length=1, max_length=200), tab: str = Query( "public", description=( "Same tab vocabulary as GET /posts (public | mine | department). " "Scopes the search to what the caller is currently looking at." ), ), category: str | None = Query( None, description="Same category vocabulary as GET /posts.", ), limit: int = Query(30, ge=1, le=50), db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), ) -> dict[str, Any]: """Search within the caller's current tab / category. The search box lives in the tab bar, so an unscoped search answered a different question than the UI asked — typing in 工作计划 returned every public post in the company while the tab highlight stayed put. Defaults reproduce the old unscoped public behaviour for callers that don't pass the params. """ svc = FeedService(db) rows = await svc.search_posts( viewer=user, query=q, tab=tab, category=category, limit=limit, ) items = await svc.hydrate_posts(rows, viewer=user) return success_response({"items": items, "next_cursor": None}) @router.get("/mention-search") async def mention_search( q: str = Query(default="", max_length=200), limit: int = Query(20, ge=1, le=30), db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), # Same bucket as /directory/people — keystrokes from the composer's # @ trigger fire at typing speed and could otherwise hammer the DB # if the FE debounce ever regresses or a caller scripts the endpoint. _rate_limit: None = Depends(check_directory_read_rate_limit), ) -> dict[str, Any]: """Lightweight people search powering the feed composer's @ trigger. Distinct from ``/directory/people`` (admin-tier only) because the feed is a普通员工 surface — every active employee can @-mention any other active colleague, same as Feishu / Slack / WeChat. The fields returned (id / display_name / department / avatar_url / work_email) are all already visible in the company-wide internal directory the user has via Feishu, so exposing them via this endpoint is consistent with the PingCAP security baseline and not a new disclosure surface. Limited to ``employment_status in (active, leaver_pending)`` — soft tombstones (``hris_record_state = purged``) and full leavers are filtered out so a @ accidentally pinging an already-gone person can't happen. """ keyword = (q or "").strip() stmt = select(IdentityPerson).where( IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status.in_(["active", "leaver_pending"]), IdentityPerson.hris_record_state != "purged", ) if keyword: # Escape LIKE wildcards, like every other search in this repo # (``search_posts``, ``list_admin_announcements``, # ``support.list_queue``, the KB tokenizer). Unescaped, a typed # ``_`` becomes a single-character wildcard and ``%`` matches # everything — the results stop meaning what the user asked for # (#2185). No escalation here (the endpoint already lists live # colleagues on an empty query); it is a correctness gap. escaped = ( keyword.replace("\\", "\\\\") .replace("%", "\\%") .replace("_", "\\_") ) like = f"%{escaped}%" stmt = stmt.where( or_( IdentityPerson.display_name.ilike(like, escape="\\"), IdentityPerson.work_email.ilike(like, escape="\\"), ) ) stmt = stmt.order_by(IdentityPerson.display_name).limit(limit) rows = (await db.execute(stmt)).scalars().all() items = [ { "id": str(p.id), "display_name": p.display_name, "work_email": p.work_email, "department": p.primary_department, "avatar_url": p.avatar_url, } for p in rows ] return success_response({"items": items}) # ────────────────────────────────────────────────────────────────── # Celebrations # ────────────────────────────────────────────────────────────────── @router.get("/today/celebrations") async def today_celebrations( db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), ) -> dict[str, Any]: svc = FeedService(db) # The cron seeds each row's occasion_date in the employee's own local TZ, # so the display "today" must also be the VIEWER's local date — otherwise a # US viewer (UTC-7/8) sees their birthday banner a calendar day off from a # hardcoded Asia/Shanghai date. Resolve the viewer's TZ the same way the # cron resolves each employee's (falls back to a sane default internally). from app.core.working_hours import resolve_timezone today = datetime.now(resolve_timezone(user)).date() birthdays, anniversaries = await svc.get_today_celebrations(today=today) def to_lite(c: Any) -> dict[str, Any]: return { "person_id": str(c.person_id), "celebration_id": c.id, "display_name": c.display_name_snapshot, "department": c.department_snapshot, "avatar_url": c.avatar_url_snapshot, "years": c.years, "illustration_key": c.illustration_key, } return success_response({ "occasion_date": today.isoformat(), "birthdays": [to_lite(c) for c in birthdays], "anniversaries": [to_lite(c) for c in anniversaries], }) @router.get("/celebrations/{celebration_id}") async def get_celebration_detail( celebration_id: str, db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), ) -> dict[str, Any]: svc = FeedService(db) c = await svc.get_celebration(celebration_id) persons = await svc._hydrate_persons({str(c.person_id)}) target = persons.get(str(c.person_id)) blessings = await svc.list_blessings(celebration_id=c.id) # has-the-viewer-already-blessed already = any( b["sender"] and str(b["sender"]["id"]) == str(user.id) for b in blessings ) is_self = target is not None and str(target.id) == str(user.id) # Reply-loop state for the celebrant's CTA. Resolved server-side (the # button must not be the only thing standing between a second call and # a duplicate 回谢 post) and only for the celebrant — for anyone else # these are inert. thanks_post = ( await svc._find_thanks_post(c.id, str(user.id)) if is_self else None ) unacked_count = sum(1 for b in blessings if not b.get("acked_at")) return success_response({ "id": c.id, "celebration_type": c.celebration_type, "occasion_date": c.occasion_date.isoformat(), "years": c.years, "person": ( { "id": str(target.id), "display_name": ( c.display_name_snapshot or target.display_name ), "department": ( c.department_snapshot or target.primary_department ), "avatar_url": c.avatar_url_snapshot or target.avatar_url, # ``work_email`` deliberately omitted: the celebration card UI # only shows name / department / avatar, and this endpoint # fetches any celebration by id for any authenticated user with # no per-row visibility check, so returning the email turned an # id-enumeration into a roster-email leak. Keep it out of the # payload (see CelebrationDetail FE type, kept in sync). } if target else None ), "illustration_key": c.illustration_key, # ``service.list_blessings`` already serialises ``created_at`` # to UTC ISO via ``_iso_utc``; pass the dicts straight through. "blessings": blessings, "already_blessed": already, "is_self": is_self, # Celebrant-only reply-loop state (always False / None / 0 for # other viewers). "already_thanked": thanks_post is not None, "thanks_post_id": thanks_post.id if thanks_post is not None else None, "unacked_blessing_count": unacked_count if is_self else 0, }) @router.post("/celebrations/{celebration_id}/blessings") async def send_blessing( celebration_id: str, body: FeedBlessingCreate, db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), ) -> dict[str, Any]: svc = FeedService(db) blessing = await svc.send_blessing( celebration_id=celebration_id, sender=user, sticker_key=body.sticker_key, message=body.message, ) await db.commit() await feed_notifications.notify_blessing_received(db, blessing_id=blessing.id) return success_response({ "id": blessing.id, "celebration_id": celebration_id, "created_at": _iso_utc(blessing.created_at), }) @router.post("/celebrations/{celebration_id}/blessings/{blessing_id}/ack") async def ack_blessing( celebration_id: str, blessing_id: str, request: Request, db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), ) -> dict[str, Any]: """Celebrant acknowledges ONE blessing ("谢谢" on a single card). Celebrant-only (403 for anyone else, checked before the blessing row is read so ids stay unenumerable) and idempotent: re-acking returns 200 with ``changed=false`` rather than a 409, because this sits at the same interaction weight as a like and a double tap must not read as an error. Only a real state change writes an audit row. """ svc = FeedService(db) blessing, changed = await svc.ack_blessing( celebration_id=celebration_id, blessing_id=blessing_id, viewer=user, ) if changed: await AuditService(db).log( trace_id=getattr(request.state, "trace_id", "feed.blessing.ack"), actor_type="user", actor_user_id=user.id, action="feed.blessing.ack", status="success", resource_type="feed_blessing", resource_id=blessing.id, detail={"celebration_id": celebration_id}, ) await db.commit() return success_response({ "id": blessing.id, "celebration_id": celebration_id, "acked_at": _iso_utc(blessing.acked_at), "changed": changed, }) @router.post("/celebrations/{celebration_id}/thanks") async def thank_all_blessings( celebration_id: str, body: FeedCelebrationThanks, request: Request, db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), # Same bucket as the sibling post/comment mutations — this one publishes # a post AND fans out up to MAX_MENTIONS_PER_POST Feishu DMs. _rate_limit: None = Depends(check_directory_read_rate_limit), ) -> dict[str, Any]: """Celebrant thanks everyone at once. Marks every blessing on the celebration as acknowledged and publishes one public 回谢 post that @-mentions each well-wisher, so the thank-you is delivered through the existing mention-notification chain instead of a bespoke channel. Celebrant-only. Once per celebration — a replay raises ``CELEBRATION_ALREADY_THANKED`` (409) off the stored post, not off a disabled button; ``CELEBRATION_NO_BLESSINGS`` (409) when there is nothing to thank yet. """ svc = FeedService(db) post, acked_count, mentioned_ids = await svc.thank_all_blessings( celebration_id=celebration_id, viewer=user, body=body.body, ) await AuditService(db).log( trace_id=getattr(request.state, "trace_id", "feed.celebration.thanks"), actor_type="user", actor_user_id=user.id, action="feed.celebration.thanks", status="success", resource_type="feed_celebration", resource_id=celebration_id, detail={ "post_id": post.id, "acked_count": acked_count, "mentioned_count": len(mentioned_ids), }, ) await db.commit() await db.refresh(post) # Reuse the ordinary mention fan-out — no second notification mechanism. await feed_notifications.notify_mentions_on_post( db, post=post, mentioned_person_ids=mentioned_ids, ) [dto] = await svc.hydrate_posts([post], viewer=user) return success_response({ "post": dto, "acked_count": acked_count, "thanked_person_ids": mentioned_ids, }) # ────────────────────────────────────────────────────────────────── # Announcements (employee read) # ────────────────────────────────────────────────────────────────── @router.get("/announcements") async def list_announcements( db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), ) -> dict[str, Any]: svc = FeedService(db) now = datetime.now(UTC) rows = await svc.list_active_announcements(now=now, limit=8) author_ids = {str(a.author_person_id) for a in rows} persons = await svc._hydrate_persons(author_ids) items = [ _serialize_announcement( a, persons.get(str(a.author_person_id)), ) for a in rows ] return success_response({"items": items}) # ────────────────────────────────────────────────────────────────── # Announcement admin (HR + IT) # ────────────────────────────────────────────────────────────────── @router.get("/admin/announcements") async def admin_list_announcements( include_archived: bool = Query(False), q: str | None = Query( None, max_length=100, description=( "Case-insensitive substring match on title / category / " "body_markdown. LIKE metacharacters match literally." ), ), limit: int = Query(200, ge=1, le=500), db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), ) -> dict[str, Any]: if not _is_admin(user): raise ForbiddenError("not authorized") svc = FeedService(db) rows = await svc.list_admin_announcements( include_archived=include_archived, search=q, limit=limit, ) author_ids = {str(a.author_person_id) for a in rows} persons = await svc._hydrate_persons(author_ids) items = [ _serialize_announcement(a, persons.get(str(a.author_person_id))) for a in rows ] return success_response({"items": items}) @router.post("/admin/announcements") async def admin_create_announcement( body: FeedAnnouncementCreate, db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), ) -> dict[str, Any]: svc = FeedService(db) ann = await svc.create_announcement( author=user, title=body.title, body_markdown=body.body_markdown, category=body.category, color=body.color, is_pinned=body.is_pinned, publish_at=body.publish_at, expire_at=body.expire_at, status=body.status, attachments_json=body.attachments_json, ) await db.commit() return success_response(_serialize_announcement(ann, user)) @router.patch("/admin/announcements/{ann_id}") async def admin_update_announcement( ann_id: str, body: FeedAnnouncementUpdate, db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), ) -> dict[str, Any]: svc = FeedService(db) ann = await svc.update_announcement( actor=user, announcement_id=ann_id, patch=body.model_dump(exclude_unset=True), ) await db.commit() author_persons = await svc._hydrate_persons({str(ann.author_person_id)}) return success_response( _serialize_announcement( ann, author_persons.get(str(ann.author_person_id)), ) ) @router.post("/admin/announcements/batch-status") async def admin_batch_announcement_status( request: Request, body: FeedAnnouncementBatchStatus, db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), ) -> dict[str, Any]: """Batch publish / archive announcements (admin). Per-row semantics (platform batch convention): * each mutation runs inside its own SAVEPOINT — one bad row never poisons the rest of the batch; * rows already in the target status count as ``skipped`` (idempotent replays are safe); * missing / soft-deleted ids count as ``failed`` with a per-row error message; * one aggregate audit row (``feed.announcements.batch_status``) is written for the whole call — not one per row. RBAC matches the single-row PATCH exactly: the same ``_is_admin`` gate that ``FeedService.update_announcement`` enforces, checked up-front so a non-admin gets a clean 403 before any row is read. """ if not _is_admin(user): raise ForbiddenError("not authorized") svc = FeedService(db) results: list[dict[str, Any]] = [] succeeded = skipped = failed = 0 for ann_id in body.ids: ann = await db.get(FeedAnnouncement, ann_id) if ann is None or ann.deleted_at is not None: failed += 1 results.append({ "id": ann_id, "status": "failed", "error": "announcement not found", }) continue if ann.status == body.status: skipped += 1 results.append({"id": ann_id, "status": "skipped"}) continue try: async with db.begin_nested(): await svc.update_announcement( actor=user, announcement_id=ann_id, patch={"status": body.status}, ) succeeded += 1 results.append({"id": ann_id, "status": "ok"}) except (AppException, ValueError) as exc: failed += 1 error_code = exc.code if isinstance(exc, AppException) else "VALIDATION_ERROR" logger.warning( "feed announcement batch row failed announcement_id=%s trace_id=%s " "error_type=%s error_code=%s", ann_id, getattr(request.state, "trace_id", None), type(exc).__name__, error_code, ) results.append({ "id": ann_id, "status": "failed", "error": error_code, }) await AuditService(db).log( trace_id=getattr( request.state, "trace_id", "feed.announcements.batch_status", ), actor_type="user", actor_user_id=user.id, action="feed.announcements.batch_status", status="success", resource_type="feed_announcement", resource_id=f"batch:{len(body.ids)}", detail={ "target_status": body.status, "total": len(body.ids), "succeeded": succeeded, "skipped": skipped, "failed": failed, "ids": body.ids[:50], }, ) await db.commit() return success_response( { "succeeded": succeeded, "skipped": skipped, "failed": failed, "results": results, }, request=request, ) @router.delete("/admin/announcements/{ann_id}") async def admin_delete_announcement( ann_id: str, db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), ) -> dict[str, Any]: svc = FeedService(db) await svc.soft_delete_announcement( actor=user, announcement_id=ann_id, ) await db.commit() return success_response({"ok": True}) # ────────────────────────────────────────────────────────────────── # Me / Preferences # ────────────────────────────────────────────────────────────────── me_router = APIRouter(prefix="/api/v1/me", tags=["me-preferences"]) @me_router.get("/preferences") async def get_my_preferences( db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), ) -> dict[str, Any]: svc = FeedService(db) prefs = await svc.get_preferences(person=user) # Serialize date for JSON. return success_response({ "birthday": prefs["birthday"].isoformat() if prefs["birthday"] else None, "birthday_visibility": prefs["birthday_visibility"], "notify_comment_feishu": prefs["notify_comment_feishu"], "notify_like_aggregate": prefs["notify_like_aggregate"], "theme_mode": prefs["theme_mode"], "festive_effects_enabled": prefs["festive_effects_enabled"], "wrapped_push_opt_out": prefs["wrapped_push_opt_out"], }) @me_router.patch("/preferences") async def patch_my_preferences( body: PreferencesUpdate, db: AsyncSession = Depends(get_db), user: IdentityPerson = Depends(get_current_user), ) -> dict[str, Any]: svc = FeedService(db) prefs = await svc.update_preferences( person=user, patch=body.model_dump(exclude_unset=True), ) await db.commit() return success_response({ "birthday": prefs["birthday"].isoformat() if prefs["birthday"] else None, "birthday_visibility": prefs["birthday_visibility"], "notify_comment_feishu": prefs["notify_comment_feishu"], "notify_like_aggregate": prefs["notify_like_aggregate"], "theme_mode": prefs["theme_mode"], "festive_effects_enabled": prefs["festive_effects_enabled"], "wrapped_push_opt_out": prefs["wrapped_push_opt_out"], }) __all__ = ["router", "me_router"]