"""Attendance monthly-report service — cycle generation + queries. Generation flow (``generate_cycle``): 1. Resolve config (intern department, cycle start day) and the cycle period for the requested ``YYYY-MM``. 2. Resolve the intern roster from department membership. 3. **Pull from Feishu first** (resolve user_ids → query user_tasks). Any scope/connectivity failure aborts here, before a single DB write, so we never leave a half-built cycle behind. 4. Find-or-create the cycle row, then create-or-refresh one report per intern — each in its own SAVEPOINT so one bad row can't poison the batch. In-flight / completed reports are never clobbered (only ``generated`` / ``returned`` refresh in place); a confirmation always pins a concrete snapshot version. The lifecycle transitions (send / confirm / return / undo) live in PR-B's ``lifecycle.py``; this module only generates and reads. """ from __future__ import annotations import logging from collections import defaultdict from datetime import UTC, datetime from uuid import UUID, uuid4 from sqlalchemy import delete, func, select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import ValidationError from app.domains.attendance import constants as c from app.domains.attendance.config import get_cycle_start_day, get_intern_department from app.domains.attendance.cycle import CyclePeriod, period_ending_in from app.domains.attendance.facts import build_person_snapshot from app.domains.attendance.feishu_attendance import ( ApprovalSpan, AttendanceFeishuError, FeishuAttendanceClient, ) from app.domains.attendance.models import ( AttendanceCycle, AttendanceReport, AttendanceReportEvent, AttendanceRosterOverride, ) from app.domains.attendance.roster import resolve_intern_roster from app.identity.person_models import IdentityPerson logger = logging.getLogger(__name__) class AttendanceService: def __init__(self, db: AsyncSession) -> None: self.db = db # ── Generation ────────────────────────────────────────────────── async def generate_cycle( self, *, year: int, month: int, actor_user_id: UUID, trace_id: str, ) -> dict: """Generate (or refresh) all intern reports for the cycle whose end date falls in ``year``-``month``. Idempotent.""" start_day = await get_cycle_start_day(self.db) department = await get_intern_department(self.db) period = period_ending_in(year, month, start_day) # ── Resolve roster + pull from Feishu BEFORE any DB write ── # Interns come from the Feishu 通讯录 department (not Helix's # HRIS org tree). Members with no matching active Helix person are # surfaced as 「待补」 rather than dropped. client = await FeishuAttendanceClient.from_config(session=self.db) roster = await resolve_intern_roster(self.db, department, client) if not roster.department_found: raise AttendanceFeishuError( f"在飞书通讯录未找到部门「{department}」。请在 平台设置 " "attendance.intern_department 配置正确的部门名,并确认该部门在" "应用可见范围内。" ) emails = [p.work_email for p in roster.mapped if p.work_email] email_to_uid = await client.batch_resolve_user_ids(emails) uids = sorted(set(email_to_uid.values())) pulled = await client.query_user_tasks(uids, period.start_date, period.end_date) results_by_uid: dict[str, list[dict]] = defaultdict(list) for res in pulled.results: uid = res.get("user_id") if uid: results_by_uid[uid].append(res) unauthorized = set(pulled.unauthorized_user_ids) invalid = set(pulled.invalid_user_ids) # Authoritative leave / trip / outing overlay (假勤审批). Best-effort # enrichment — degrades to the打卡-result supplement if unavailable. approvals = await client.query_user_approvals( uids, period.start_date, period.end_date, ) approval_overlay: dict[str, dict[str, dict]] = { uid: self._spans_to_overlay(spans, period) for uid, spans in approvals.items() } # ── Find-or-create the cycle ── cycle = await self._find_or_create_cycle( period, department=department, actor_user_id=actor_user_id, ) cycle.pending_not_in_helix_json = roster.unmapped stats = { "cycle_id": str(cycle.id), "cycle_key": cycle.cycle_key, "period": period.label, "feishu_members": len(roster.mapped) + len(roster.unmapped), "roster_size": len(roster.mapped), "not_in_helix": len(roster.unmapped), "not_in_helix_members": roster.unmapped, "created": 0, "refreshed": 0, "skipped_in_flight": 0, "unmapped": 0, "unauthorized": 0, "with_abnormalities": 0, "failed": 0, "removed_excluded": 0, "excluded_in_flight": 0, } # ── Prune stale rows for manually-excluded people ── # HR adds an exclude AFTER a cycle was generated, then regenerates: # the upsert loop below never visits excluded persons, so their # already-created report rows would linger in the cycle forever. # Only regeneratable (generated / returned) rows are removed — # in-flight / completed reports keep the "regeneration never # clobbers a confirmation" invariant and are surfaced in stats so # HR knows to withdraw them first. removed, kept_in_flight = await self._prune_excluded_reports( cycle=cycle, excluded_ids=roster.excluded_ids, ) stats["removed_excluded"] = removed stats["excluded_in_flight"] = kept_in_flight for person in roster.mapped: # SAVEPOINT per intern: a flush failure on one row rolls back # just that row, leaving the rest of the batch intact. try: async with self.db.begin_nested(): outcome = await self._upsert_report( cycle=cycle, person=person, email_to_uid=email_to_uid, results_by_uid=results_by_uid, approval_overlay=approval_overlay, unauthorized=unauthorized, invalid=invalid, period=period, actor_user_id=actor_user_id, trace_id=trace_id, ) except Exception: # noqa: BLE001 — isolate one bad intern logger.exception( "attendance: failed to upsert report for person %s in cycle %s", person.id, cycle.cycle_key, ) # Surface the failure in the summary — without a counter a # cron reading {"created": 5} could not tell that N interns # errored out (deadlock, transient DB fault), masking a # systemic problem as a clean partial run. stats["failed"] += 1 continue for key in ("created", "refreshed", "skipped_in_flight", "unmapped", "unauthorized", "with_abnormalities"): stats[key] += outcome.get(key, 0) await self.db.flush() return stats async def _find_or_create_cycle( self, period: CyclePeriod, *, department: str, actor_user_id: UUID, ) -> AttendanceCycle: existing = ( await self.db.execute( select(AttendanceCycle).where( AttendanceCycle.cycle_key == period.key ) ) ).scalar_one_or_none() if existing is not None: # ``cycle_key`` names the month in which a cycle ENDS, so a # changed start-day can produce a different window with the same # key. Never silently move an existing cycle (and its reports) # to that new window: it would rewrite payroll history while its # immutable snapshots still described the old dates. if (existing.start_date, existing.end_date) != ( period.start_date, period.end_date, ): raise ValidationError( "该考勤周期已按不同的起止日生成，不能用新的周期配置覆盖历史记录" ) # The department is also a cycle snapshot. A changed setting # takes effect for newly created cycles; existing cycles keep the # source that HR originally reviewed. return existing cycle = AttendanceCycle( cycle_key=period.key, start_date=period.start_date, end_date=period.end_date, intern_department=department, created_by=actor_user_id, ) self.db.add(cycle) await self.db.flush() return cycle async def _prune_excluded_reports( self, *, cycle: AttendanceCycle, excluded_ids: set[UUID], ) -> tuple[int, int]: """Delete this cycle's report rows for manually-excluded persons. Returns ``(removed, kept_in_flight)``. Only regeneratable rows (``generated`` / ``returned``) are deleted; in-flight or completed reports are kept (a confirmation is never destroyed by a regeneration) and counted so the caller can surface them to HR. Event rows are deleted explicitly — the ORM cascade would need a lazy load per report (MissingGreenlet in async), and being explicit beats relying on the DB-level ON DELETE CASCADE. """ if not excluded_ids: return 0, 0 rows = ( await self.db.execute( select(AttendanceReport).where( AttendanceReport.cycle_id == cycle.id, AttendanceReport.person_id.in_(excluded_ids), ) ) ).scalars().all() if not rows: return 0, 0 removable = [r for r in rows if r.status in c.REGENERATABLE_STATUSES] kept = len(rows) - len(removable) if kept: logger.warning( "attendance: %d excluded person(s) in cycle %s have in-flight/" "completed reports — not removed; HR must withdraw first: %s", kept, cycle.cycle_key, [str(r.person_id) for r in rows if r.status not in c.REGENERATABLE_STATUSES], ) if removable: ids = [r.id for r in removable] await self.db.execute( delete(AttendanceReportEvent).where( AttendanceReportEvent.report_id.in_(ids) ) ) await self.db.execute( delete(AttendanceReport) .where(AttendanceReport.id.in_(ids)) .execution_options(synchronize_session="fetch") ) logger.info( "attendance: pruned %d excluded report(s) from cycle %s", len(ids), cycle.cycle_key, ) return len(removable), kept async def _upsert_report( self, *, cycle: AttendanceCycle, person: IdentityPerson, email_to_uid: dict[str, str], results_by_uid: dict[str, list[dict]], approval_overlay: dict[str, dict[str, dict]], unauthorized: set[str], invalid: set[str], period: CyclePeriod, actor_user_id: UUID, trace_id: str, ) -> dict: uid = email_to_uid.get((person.work_email or "").lower()) raw = results_by_uid.get(uid, []) if uid else [] approval_dates = approval_overlay.get(uid) if uid else None snapshot = build_person_snapshot( raw, feishu_user_id=uid, period=period, approval_dates=approval_dates, ) summary = snapshot["summary"] data_quality = { "feishu_mapped": bool(uid), "unauthorized": bool(uid and uid in unauthorized), "invalid_user": bool(uid and uid in invalid), "no_attendance_rows": bool(uid and not raw), } has_abnormal = summary["abnormal_days"] > 0 existing = ( await self.db.execute( select(AttendanceReport).where( AttendanceReport.cycle_id == cycle.id, AttendanceReport.person_id == person.id, ) ) ).scalar_one_or_none() now = datetime.now(UTC) outcome = { "created": 0, "refreshed": 0, "skipped_in_flight": 0, "unmapped": 0 if uid else 1, "unauthorized": 1 if data_quality["unauthorized"] else 0, "with_abnormalities": 1 if has_abnormal else 0, } if existing is None: # Assign the id explicitly so the event row can reference it # without flushing the report first — and, crucially, never # touch the lazy ``report.events`` collection (appending to it # would trigger a sync-context lazy-load during the next # person's autoflush → MissingGreenlet). report = AttendanceReport( id=uuid4(), cycle_id=cycle.id, person_id=person.id, leader_person_id=person.manager_person_id, status=c.STATUS_GENERATED, version=1, snapshot_json=snapshot, data_quality_json=data_quality, generated_at=now, ) self._apply_summary(report, summary) self.db.add(report) self.db.add(AttendanceReportEvent( report_id=report.id, event_type=c.EVENT_GENERATED, actor_person_id=actor_user_id, to_status=c.STATUS_GENERATED, detail_json={"version": 1, "data_quality": data_quality}, trace_id=trace_id, )) await self.db.flush() outcome["created"] = 1 return outcome if existing.status not in c.REGENERATABLE_STATUSES: outcome["skipped_in_flight"] = 1 return outcome # Refresh in place — re-pull reflects 补卡 done in Feishu. existing.snapshot_json = snapshot existing.data_quality_json = data_quality existing.version += 1 existing.leader_person_id = person.manager_person_id existing.generated_at = now # Returning a report to GENERATED on refresh clears any prior # return metadata so the next send starts clean. if existing.status == c.STATUS_RETURNED: existing.status = c.STATUS_GENERATED existing.return_reason = None existing.returned_by = None existing.returned_by_person_id = None self._apply_summary(existing, summary) self.db.add(AttendanceReportEvent( report_id=existing.id, event_type=c.EVENT_REGENERATED, actor_person_id=actor_user_id, to_status=existing.status, detail_json={"version": existing.version, "data_quality": data_quality}, trace_id=trace_id, )) await self.db.flush() outcome["refreshed"] = 1 return outcome async def repull_report( self, report: AttendanceReport, *, actor_user_id: UUID, trace_id: str, ) -> None: """Re-pull ONE intern's latest Feishu attendance into ``report`` and bump the version. Used by HR resend after the intern has done 补卡 in Feishu — the resend always reflects the current data.""" cycle = await self.db.get(AttendanceCycle, report.cycle_id) person = await self.db.get(IdentityPerson, report.person_id) if cycle is None or person is None: return period = CyclePeriod(start_date=cycle.start_date, end_date=cycle.end_date) client = await FeishuAttendanceClient.from_config(session=self.db) emails = [person.work_email] if person.work_email else [] email_to_uid = await client.batch_resolve_user_ids(emails) uid = email_to_uid.get((person.work_email or "").lower()) pulled = await client.query_user_tasks( [uid] if uid else [], period.start_date, period.end_date, ) raw = [r for r in pulled.results if r.get("user_id") == uid] if uid else [] # Re-pull the authoritative 假勤审批 overlay too — without it the resend # would drop approved leave/trip/out classification and regress days # back to absent/lack/late (the打卡-result supplement is only a fallback). approval_dates: dict | None = None if uid: approvals = await client.query_user_approvals( [uid], period.start_date, period.end_date, ) approval_dates = self._spans_to_overlay(approvals.get(uid, []), period) snapshot = build_person_snapshot( raw, feishu_user_id=uid, period=period, approval_dates=approval_dates, ) report.snapshot_json = snapshot report.data_quality_json = { "feishu_mapped": bool(uid), "unauthorized": bool(uid and uid in set(pulled.unauthorized_user_ids)), "invalid_user": bool(uid and uid in set(pulled.invalid_user_ids)), "no_attendance_rows": bool(uid and not raw), } report.version += 1 # A resend after HR corrects the reporting line must route the new # confirmation round to the current leader, not the stale snapshot. report.leader_person_id = person.manager_person_id report.generated_at = datetime.now(UTC) self._apply_summary(report, snapshot["summary"]) self.db.add(AttendanceReportEvent( report_id=report.id, event_type=c.EVENT_REGENERATED, actor_person_id=actor_user_id, to_status=report.status, detail_json={"version": report.version, "resend_repull": True}, trace_id=trace_id, )) await self.db.flush() @staticmethod def _spans_to_overlay( spans: list[ApprovalSpan], period: CyclePeriod, ) -> dict[str, dict]: """Expand approved leave/trip/out spans into a per-date overlay clamped to the cycle. Leave wins over trip/out on overlapping days (most attendance-relevant).""" from datetime import date as _date from datetime import timedelta as _td overlay: dict[str, dict] = {} for sp in spans: try: sd = _date.fromisoformat(sp.start_date) ed = _date.fromisoformat(sp.end_date) except (ValueError, TypeError): continue if ed < sd: continue cur = max(sd, period.start_date) last = min(ed, period.end_date) while cur <= last: iso = cur.isoformat() if overlay.get(iso) is None or sp.kind == "leave": previous = overlay.get(iso) or {} overlay[iso] = {"kind": sp.kind, "name": sp.name} if sp.kind == "leave": intervals = previous.get("leave_intervals", []) interval = { "start_ts": sp.start_ts, "end_ts": sp.end_ts, # Never apply a multi-day approval's TOTAL to # every date, nor divide it equally across dates. "seconds": sp.leave_seconds if sd == ed else None, } if interval not in intervals: intervals = [*intervals, interval] overlay[iso]["leave_intervals"] = intervals cur += _td(days=1) return overlay @staticmethod def _apply_summary(report: AttendanceReport, summary: dict) -> None: report.work_days = summary["work_days"] report.abnormal_days = summary["abnormal_days"] report.lack_count = summary["lack_count"] report.late_count = summary["late_count"] report.early_count = summary["early_count"] report.remedied_count = summary["remedied_count"] # ── Queries ───────────────────────────────────────────────────── async def list_cycles(self, *, limit: int = 36) -> list[AttendanceCycle]: rows = ( await self.db.execute( select(AttendanceCycle) .order_by(AttendanceCycle.end_date.desc()) .limit(limit) ) ).scalars().all() return list(rows) async def get_cycle(self, cycle_id: UUID) -> AttendanceCycle | None: return await self.db.get(AttendanceCycle, cycle_id) async def get_cycles( self, cycle_ids: set[UUID] | list[UUID], ) -> dict[UUID, AttendanceCycle]: """Batch-load cycles by id (one round-trip) for serializing a page of reports — avoids the per-distinct-cycle N+1.""" ids = list(cycle_ids) if not ids: return {} rows = ( await self.db.execute( select(AttendanceCycle).where(AttendanceCycle.id.in_(ids)) ) ).scalars().all() return {cyc.id: cyc for cyc in rows} async def report_counts( self, cycle_ids: list[UUID], ) -> dict[UUID, int]: if not cycle_ids: return {} rows = ( await self.db.execute( select( AttendanceReport.cycle_id, func.count() ) .where(AttendanceReport.cycle_id.in_(cycle_ids)) .group_by(AttendanceReport.cycle_id) ) ).all() return {cid: int(n) for cid, n in rows} async def list_reports( self, *, cycle_id: UUID | None = None, status: str | None = None, person_id: UUID | None = None, leader_person_id: UUID | None = None, page: int = 1, page_size: int = 50, ) -> tuple[list[AttendanceReport], int]: conds = [] if cycle_id is not None: conds.append(AttendanceReport.cycle_id == cycle_id) if status is not None: conds.append(AttendanceReport.status == status) if person_id is not None: conds.append(AttendanceReport.person_id == person_id) if leader_person_id is not None: conds.append(AttendanceReport.leader_person_id == leader_person_id) total = ( await self.db.execute( select(func.count()).select_from(AttendanceReport).where(*conds) ) ).scalar_one() rows = ( await self.db.execute( select(AttendanceReport) .where(*conds) .order_by(AttendanceReport.created_at.desc(), AttendanceReport.id) .offset((page - 1) * page_size) .limit(page_size) ) ).scalars().all() return list(rows), int(total) async def get_report(self, report_id: UUID) -> AttendanceReport | None: return await self.db.get(AttendanceReport, report_id) # ── Roster overlay (HR-maintained include / exclude) ───────────── async def list_roster_overrides(self) -> list[AttendanceRosterOverride]: rows = ( await self.db.execute( select(AttendanceRosterOverride) .order_by(AttendanceRosterOverride.created_at.desc()) ) ).scalars().all() return list(rows) async def set_roster_override( self, *, person_id: UUID, mode: str, note: str | None, actor_user_id: UUID, ) -> AttendanceRosterOverride: """Upsert one person's roster override. Switching mode (include ↔ exclude) updates the existing row in place — one override per person. Idempotent.""" existing = await self._get_roster_override(person_id) if existing is not None: existing.mode = mode existing.note = note await self.db.flush() return existing row = AttendanceRosterOverride( person_id=person_id, mode=mode, note=note, created_by=actor_user_id, ) try: # ``add`` + ``flush`` MUST happen inside the savepoint: the # INSERT collision is then confined to the SAVEPOINT and the # outer txn survives. Adding before ``begin_nested`` would let # its pre-flush INSERT the row before the savepoint exists, # poisoning the whole transaction on collision. async with self.db.begin_nested(): self.db.add(row) await self.db.flush() except IntegrityError: # UNIQUE(person_id) collision from a concurrent PUT (HR double- # click / client retry) that inserted between our SELECT above # and this flush. The constraint kept it fail-closed (no dup); # turn the loser into the same in-place upsert the found-branch # would have done, not an unhandled 500. # The savepoint rollback already detached the pending ``row``. # Locking re-select: a plain consistent read would keep this # txn's REPEATABLE READ snapshot and miss the concurrent winner # (same class as the ensure_run fix). with_for_update reads the # latest committed row. winner = await self._get_roster_override(person_id, lock=True) if winner is None: raise winner.mode = mode winner.note = note await self.db.flush() return winner return row async def _get_roster_override( self, person_id: UUID, *, lock: bool = False, ) -> AttendanceRosterOverride | None: stmt = select(AttendanceRosterOverride).where( AttendanceRosterOverride.person_id == person_id ) if lock: stmt = stmt.with_for_update().execution_options( populate_existing=True, ) return (await self.db.execute(stmt)).scalar_one_or_none() async def remove_roster_override(self, person_id: UUID) -> bool: existing = await self._get_roster_override(person_id) if existing is None: return False await self.db.delete(existing) await self.db.flush() return True async def get_report_events( self, report_id: UUID, ) -> list[AttendanceReportEvent]: rows = ( await self.db.execute( select(AttendanceReportEvent) .where(AttendanceReportEvent.report_id == report_id) .order_by(AttendanceReportEvent.created_at) ) ).scalars().all() return list(rows)