"""Intern attendance monthly-report models. Helix pulls attendance facts from Feishu (考勤 / attendance), builds one **report per intern per cycle**, and drives a two-stage confirmation workflow (intern → leader). This module owns the persistence: * :class:`AttendanceCycle` — one period (e.g. 2026-05-21 .. 2026-06-20). * :class:`AttendanceReport` — one intern's report for one cycle, carrying the immutable, versioned ``snapshot_json`` of what was shown/confirmed plus denormalised summary columns for the list view. * :class:`AttendanceReportEvent` — append-only audit of every state transition (generated / sent / confirmed / returned / undone), each stamped with the request ``trace_id``. Feishu is read-only here: Helix never writes attendance back. 补卡 happens in Feishu; Helix re-pulls on regenerate (see service.generate_cycle). State machine (see ``app/domains/attendance/constants.py`` for the canonical transition guards used by the lifecycle service in PR-B): generated → pending_intern → (returned | pending_leader) pending_leader → (returned | completed | back to pending_intern via undo) """ from __future__ import annotations import uuid from datetime import date, datetime from sqlalchemy import ( JSON, Date, DateTime, Float, ForeignKey, Index, Integer, String, Text, UniqueConstraint, ) from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.sql import func from app.core.database import Base from app.core.sql_types import GUID as Uuid class AttendanceCycle(Base): """A single attendance period (inclusive ``start_date`` .. ``end_date``). The default cadence is the 21st of one month through the 20th of the next (configurable via the ``attendance.cycle_start_day`` platform setting). ``cycle_key`` is the canonical idempotency handle — the ``YYYY-MM`` of the period's ``end_date`` (so 2026-05-21..2026-06-20 → ``"2026-06"``, the "June cycle"). Re-generating a cycle resolves the same row by key rather than creating duplicates. """ __tablename__ = "attendance_cycles" __mapper_args__ = {"eager_defaults": True} __table_args__ = ( UniqueConstraint("cycle_key", name="uq_attendance_cycles_key"), Index("ix_attendance_cycles_end_date", "end_date"), ) id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) cycle_key: Mapped[str] = mapped_column(String(16), nullable=False) start_date: Mapped[date] = mapped_column(Date, nullable=False) end_date: Mapped[date] = mapped_column(Date, nullable=False) # The department-name filter snapshot used to resolve the intern # roster for this cycle — recorded so a later config change doesn't # rewrite history of "who was in scope when this cycle ran". intern_department: Mapped[str | None] = mapped_column(String(128)) # Feishu department members with no matching active Helix person at the # last generation — surfaced to HR as 「待补」 (interns not in HRIS / # not yet onboarded). List of {name, email, open_id}. pending_not_in_helix_json: Mapped[list | None] = mapped_column(JSON, nullable=True) # Plain id columns (no DB FK to identity_persons) — TiDB-FK-safe, per # the migration-324 convention. Person identity is resolved at the # serializer layer, not via an ORM relationship. created_by: Mapped[uuid.UUID | None] = mapped_column(Uuid) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now(), ) reports: Mapped[list[AttendanceReport]] = relationship( back_populates="cycle", cascade="all, delete-orphan", ) class AttendanceReport(Base): """One intern's attendance report for one cycle. ``snapshot_json`` is the versioned source of truth for *what the intern / leader saw and confirmed*. It is re-pulled (and ``version`` bumped) on regenerate / resend, so a confirmation always references a concrete snapshot version. The denormalised count columns mirror ``snapshot_json["summary"]`` for cheap list rendering and filtering. ``leader_person_id`` snapshots the intern's manager at generation time (derived from ``identity_persons.manager_person_id``) so a later org change doesn't silently re-route an in-flight report. """ __tablename__ = "attendance_reports" __mapper_args__ = {"eager_defaults": True} __table_args__ = ( UniqueConstraint( "cycle_id", "person_id", name="uq_attendance_reports_cycle_person", ), Index("ix_attendance_reports_status", "status"), Index("ix_attendance_reports_person", "person_id"), Index("ix_attendance_reports_leader", "leader_person_id"), Index("ix_attendance_reports_cycle_status", "cycle_id", "status"), ) id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) cycle_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("attendance_cycles.id", ondelete="CASCADE"), nullable=False, ) person_id: Mapped[uuid.UUID] = mapped_column(Uuid, nullable=False) leader_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid) status: Mapped[str] = mapped_column( String(24), nullable=False, default="generated", ) version: Mapped[int] = mapped_column(Integer, nullable=False, default=1) # Immutable per-version snapshot of the pulled attendance detail + # summary (see facts.build_snapshot). NULL only transiently before # the first pull completes. snapshot_json: Mapped[dict | None] = mapped_column(JSON, nullable=True) # Denormalised summary (mirrors snapshot_json["summary"]) for list views. # ``work_days`` is duration-weighted (fractional): a partial-attendance # day accrues min(1.0, span/9h), so this is a Float, not a day count. work_days: Mapped[float] = mapped_column(Float, nullable=False, default=0.0) abnormal_days: Mapped[int] = mapped_column(Integer, nullable=False, default=0) lack_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) late_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) early_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) remedied_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) # Data-quality flags: was the person mapped to a Feishu user, did # Feishu authorise reading their attendance, etc. Surfaced to HR so a # blank report reads as "couldn't pull" not "perfect attendance". data_quality_json: Mapped[dict | None] = mapped_column(JSON, nullable=True) # Workflow bookkeeping. return_reason: Mapped[str | None] = mapped_column(Text) returned_by: Mapped[str | None] = mapped_column(String(16)) # 'intern' | 'leader' returned_by_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid) generated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) last_sent_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) # Last time a pending-confirmation reminder card was pushed for this # report — gates the reminder cron so a stale report isn't nagged # every run. last_reminded_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) intern_confirmed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) leader_confirmed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) returned_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now(), ) cycle: Mapped[AttendanceCycle] = relationship(back_populates="reports") events: Mapped[list[AttendanceReportEvent]] = relationship( back_populates="report", cascade="all, delete-orphan", order_by="AttendanceReportEvent.created_at", ) class AttendanceRosterOverride(Base): """HR-maintained overlay on the Feishu-derived intern roster. The base roster is still resolved live from the Feishu 通讯录 department (``attendance.intern_department``) on every generate — this table only **patches** it, so new interns joining the Feishu department keep flowing in with zero maintenance: * ``mode='include'`` — pull this person into the report even though they are NOT in the Feishu department (e.g. an intern onboarded in Helix whom HR never added to the Feishu group, like the 许晴 case). * ``mode='exclude'`` — keep this person OUT of the report even though they ARE in the Feishu department (mis-grouped, transferred, etc.). Effective roster = (Feishu members ∪ includes) − excludes. Global (not per-cycle): overrides apply to every subsequent generate. ``person_id`` is a plain GUID column with no DB FK to ``identity_persons`` — TiDB-FK-safe, per the migration-324 convention the sibling cycle/report tables already follow. A person has at most one override row (``UNIQUE(person_id)``); switching mode upserts in place. """ __tablename__ = "attendance_roster_overrides" __mapper_args__ = {"eager_defaults": True} __table_args__ = ( UniqueConstraint("person_id", name="uq_attendance_roster_overrides_person"), ) id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) person_id: Mapped[uuid.UUID] = mapped_column(Uuid, nullable=False) mode: Mapped[str] = mapped_column(String(16), nullable=False) # 'include' | 'exclude' note: Mapped[str | None] = mapped_column(Text) created_by: Mapped[uuid.UUID | None] = mapped_column(Uuid) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now(), ) class AttendanceReportEvent(Base): """Append-only audit row for one report state transition. Every generate / send / confirm / return / undo writes one row, carrying the request ``trace_id`` so the report's history is fully reconstructable and cross-referenceable with ``audit_logs``. ``actor_person_id`` is NULL for system-driven transitions (e.g. the auto push-to-leader after intern confirmation). """ __tablename__ = "attendance_report_events" __mapper_args__ = {"eager_defaults": True} __table_args__ = ( Index("ix_attendance_report_events_report", "report_id"), Index("ix_attendance_report_events_type", "event_type"), ) id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) report_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("attendance_reports.id", ondelete="CASCADE"), nullable=False, ) event_type: Mapped[str] = mapped_column(String(32), nullable=False) actor_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid) from_status: Mapped[str | None] = mapped_column(String(24)) to_status: Mapped[str | None] = mapped_column(String(24)) reason: Mapped[str | None] = mapped_column(Text) detail_json: Mapped[dict | None] = mapped_column(JSON, nullable=True) trace_id: Mapped[str | None] = mapped_column(String(64)) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) report: Mapped[AttendanceReport] = relationship(back_populates="events")