"""ORM models for the bounty domain (5 tables). See ``server/alembic/versions/533_bounty_ops.py`` for the canonical schema and ``constants.py`` for the value enumerations. Money discipline: every amount is USD and stored as ``Numeric(12, 2)``. There is deliberately no multi-currency column — rewards are quoted and paid in USD, and a nullable ``currency`` that is always ``USD`` would only invite an inconsistent second code path. If a non-USD payout ever happens, convert at the finance FX rate and note the native figure in ``payment_note`` (a schema change is cheaper than a wrong sum). """ from __future__ import annotations import uuid from datetime import date, datetime from decimal import Decimal from typing import Any from sqlalchemy import ( JSON, BigInteger, Boolean, Date, DateTime, ForeignKey, Index, Integer, Numeric, String, Text, UniqueConstraint, ) from sqlalchemy.orm import Mapped, mapped_column from sqlalchemy.sql import func from app.core.database import Base from app.core.sql_types import GUID as Uuid from app.domains.bounty import constants as const # ════════════════════════════════════════════════════════════════════ # 1. Researchers (the external white-hats) # ════════════════════════════════════════════════════════════════════ class BountyResearcher(Base): """One external submitter, keyed by the email they write from. Deliberately thin: name, handle, country, notes. **No banking data, no address, no identity documents** — payout details stay in the mail thread and in finance's own system (decision 2026-08-24). The value of this table is continuity: "this is the same person who sent us four reports, three of them valid" is what makes the reward decision and the researcher leaderboard possible. """ __tablename__ = "bounty_researchers" __mapper_args__ = {"eager_defaults": True} id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) # Lower-cased on write by the service — the natural key. 320 = RFC # max (64 local + @ + 255 domain). email: Mapped[str] = mapped_column(String(320), nullable=False) display_name: Mapped[str | None] = mapped_column(String(255)) # Public alias the researcher asks to be credited by, when different # from their legal/display name. handle: Mapped[str | None] = mapped_column(String(128)) country: Mapped[str | None] = mapped_column(String(64)) #: Canonical researcher this address is an alias of. Confirmed in #: production: two addresses sharing the ``suhaspandya`` prefix are one #: person, and every per-researcher figure was wrong for them. #: #: Reports keep pointing at the address that sent them — a reply has to #: go back to the address the mail arrived on, and "which address did #: this come in on" is a real question in a dispute. Aggregates resolve #: through this pointer instead of anything being rewritten. #: #: One hop only: the service refuses to make an alias an alias target, #: so no recursive resolution and no cycle can hang a query. merged_into_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("bounty_researchers.id", ondelete="SET NULL"), nullable=True, index=True, ) notes: Mapped[str | None] = mapped_column(Text) # Noisy / abusive submitters: the intake form warns on a blocked # email but never refuses it — dropping a report because of a flag # is how a real finding gets lost. is_blocked: Mapped[bool] = mapped_column( Boolean, nullable=False, default=False, server_default="0", ) first_report_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) last_report_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) created_by: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), ) 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(), ) __table_args__ = ( UniqueConstraint("email", name="uq_bounty_researchers_email"), Index("ix_bounty_researchers_last_report", "last_report_at"), ) def __repr__(self) -> str: return f"" # ════════════════════════════════════════════════════════════════════ # 2. Reports # ════════════════════════════════════════════════════════════════════ class BountyReport(Base): """One vulnerability report from an external researcher. Three independent axes live side by side — see ``constants.py``: * ``status`` — triage decision * ``remediation_status`` — is the hole plugged * ``payment_status`` — did the researcher get paid Attribution snapshots (``assignee_department``, ``assignee_name``) are stored rather than joined at read time on purpose: the "which team ships the most findings" chart must not silently re-attribute two-year-old reports when someone changes department, and it must keep working after the person is soft-deleted. """ __tablename__ = "bounty_reports" __mapper_args__ = {"eager_defaults": True} id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) # Human handle used in mail subjects and finance references: # ``BB-2026-001``. Unique and never reused, including after a soft # delete — the researcher may quote it back at us years later. code: Mapped[str] = mapped_column(String(32), nullable=False) title: Mapped[str] = mapped_column(String(const.MAX_TITLE_CHARS), nullable=False) researcher_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("bounty_researchers.id"), nullable=False, ) source: Mapped[str] = mapped_column( String(16), nullable=False, default=const.SOURCE_EMAIL, server_default=const.SOURCE_EMAIL, ) # When the researcher's mail arrived — NOT when it was keyed in. The # "how fast do we answer" metric depends on this being the real date. reported_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, ) # Verbatim paste of the researcher's mail. The evidentiary record — # never rewritten by AI parsing, which only fills the fields below. raw_report: Mapped[str | None] = mapped_column(Text) # Operator/AI-written internal summary (Chinese) — what the business # owner reads on the confirmation page instead of a raw English mail. summary: Mapped[str | None] = mapped_column(Text) # Steps to reproduce, split out so the owner page can show it in a # code block; optional because plenty of reports have none. reproduction: Mapped[str | None] = mapped_column(Text) affected_system: Mapped[str | None] = mapped_column(String(255)) #: Somebody looked and there is no internal system to link this to — #: an open-source product finding, a CI workflow. Distinct from #: ``system_id IS NULL``, which means nobody has looked yet. Without #: the distinction the data-completeness read counts 44 reports that #: can never be linked, and a number that cannot reach zero teaches #: people to ignore it. system_not_applicable: Mapped[bool] = mapped_column( Boolean, nullable=False, default=False, server_default="0", ) # Optional link into the internal system registry (``registry_systems``). # Referenced by table name, not by importing the registry's model — # domain modules do not import each other (CLAUDE.md §3); the # registry's data is read through # ``core.coordinators.system_registry_coordinator``. # # ``affected_system`` stays the free-text fallback and is NOT # derived from this column: a researcher can name a host we have # never registered, and rewriting what they said would lose evidence. system_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("registry_systems.id", ondelete="SET NULL"), ) # Which report this one duplicates. Two fields on purpose: the # original is often NOT another Helix row (an internal issue, a # commit, or an email predating this domain), and forcing an FK # would push operators to leave it blank — which is the state this # replaces. See migration 544. duplicate_of_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("bounty_reports.id", ondelete="SET NULL"), ) duplicate_note: Mapped[str | None] = mapped_column(String(500)) affected_urls: Mapped[list[str] | None] = mapped_column(JSON) vuln_type: Mapped[str | None] = mapped_column(String(32)) cve_id: Mapped[str | None] = mapped_column(String(64)) # Final severity (a ``bounty_severity_tiers.key``). NULL until the # security team rules on it — never defaulted, because a defaulted # severity is indistinguishable from a decided one in the charts. severity: Mapped[str | None] = mapped_column(String(24)) # What the AI / the business owner proposed. Kept separate so # "how often is the owner's guess right" stays answerable. suggested_severity: Mapped[str | None] = mapped_column(String(24)) #: True while ``severity`` is the level a report ARRIVED with rather #: than one a person chose. Without it the console cannot tell "somebody #: looked and said informational" from "nobody has looked" — and the #: reply composer would offer a critical finding the $0 informational #: reward letter. Cleared by any explicit ``set_severity``, including #: one that re-picks informational: that is a person deciding. severity_is_default: Mapped[bool] = mapped_column( Boolean, nullable=False, default=False, server_default="0", ) #: The model's structured reading of the report — verdict, reasoning, #: exploitability, impact, what to check, what to ask the researcher. #: Advisory only: nothing downstream reads it, and it never sets the #: status, the grade or the reply. See migration 551. ai_assessment: Mapped[dict | None] = mapped_column(JSON) ai_assessed_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), ) status: Mapped[str] = mapped_column( String(24), nullable=False, default=const.STATUS_NEW, server_default=const.STATUS_NEW, ) remediation_status: Mapped[str] = mapped_column( String(24), nullable=False, default=const.REMEDIATION_NOT_APPLICABLE, server_default=const.REMEDIATION_NOT_APPLICABLE, ) payment_status: Mapped[str] = mapped_column( String(24), nullable=False, default=const.PAYMENT_NOT_APPLICABLE, server_default=const.PAYMENT_NOT_APPLICABLE, ) # ── Business owner (assignee) + attribution snapshot ── assignee_person_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), ) assignee_name: Mapped[str | None] = mapped_column(String(255)) assignee_department: Mapped[str | None] = mapped_column(String(255)) assigned_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) assigned_by: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), ) # Last time the operator pushed a reminder — throttles the nudge # button so an owner cannot be carded five times in a minute. last_nudged_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) owner_decision: Mapped[str | None] = mapped_column(String(24)) owner_note: Mapped[str | None] = mapped_column(Text) owner_responded_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) # ── Remediation ── fix_eta: Mapped[date | None] = mapped_column(Date) # Severity SLA deadline, stamped when severity is set. Denormalised # so "overdue" is a plain indexed comparison instead of a per-row # join against the tier table (whose SLA may since have changed — # the report keeps the deadline it was actually given). fix_due_date: Mapped[date | None] = mapped_column(Date) fixed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) # ── Money ── # Which of the two funding pots this reward is charged to. An explicit # column rather than a lookup through the affected system: the system # registry is not the authority on who pays, and a report can be # re-tracked at triage without rewriting anything else. budget_track: Mapped[str] = mapped_column( String(16), nullable=False, default=const.DEFAULT_BUDGET_TRACK, server_default=const.DEFAULT_BUDGET_TRACK, ) reward_amount_usd: Mapped[Decimal | None] = mapped_column(Numeric(12, 2)) # True when the operator deviated from the tier default, so the # budget report can explain a figure that does not match the ladder. reward_is_override: Mapped[bool] = mapped_column( Boolean, nullable=False, default=False, server_default="0", ) reward_decided_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) # Fiscal year the reward is charged to — stamped when the reward is # decided, from the platform's FY start month. Stored, not derived, # so a reward decided in FY26 and paid in FY27 stays charged to FY26. fiscal_year: Mapped[int | None] = mapped_column(Integer) payment_requested_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) paid_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) # Finance's own handle for the payment (PO / expense id / invoice # number). Free text — Helix does not own the payment system. payment_reference: Mapped[str | None] = mapped_column(String(128)) payment_note: Mapped[str | None] = mapped_column(Text) # ── Correspondence bookkeeping ── last_replied_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) last_reply_template_key: Mapped[str | None] = mapped_column(String(64)) reply_count: Mapped[int] = mapped_column( Integer, nullable=False, default=0, server_default="0", ) closed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) created_by: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), ) 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(), ) # Soft delete only — a mis-keyed report is hidden, never erased, and # its ``code`` is not recycled. deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) __table_args__ = ( UniqueConstraint("code", name="uq_bounty_reports_code"), Index("ix_bounty_reports_status", "status", "deleted_at"), Index("ix_bounty_reports_assignee", "assignee_person_id", "status"), Index("ix_bounty_reports_researcher", "researcher_id", "reported_at"), Index("ix_bounty_reports_budget", "fiscal_year", "payment_status"), Index( "ix_bounty_reports_track_budget", "budget_track", "fiscal_year", "payment_status", ), Index("ix_bounty_reports_severity", "severity", "status"), Index("ix_bounty_reports_reported", "reported_at"), Index("ix_bounty_reports_remediation", "remediation_status", "fix_due_date"), Index("ix_bounty_reports_system", "system_id", "status"), Index("ix_bounty_reports_duplicate_of", "duplicate_of_id"), ) def __repr__(self) -> str: return f"" # ════════════════════════════════════════════════════════════════════ # 3. Report events (timeline + comments + correspondence log) # ════════════════════════════════════════════════════════════════════ class BountyReportEvent(Base): """Append-only log: state transitions, human comments, and the verbatim text of every reply we sent the researcher. One table rather than three because the detail page wants a single chronological stream, and because "what happened to this report" is the same question whether the answer is a status flip or a sentence someone typed. ``visibility`` is load-bearing, not cosmetic: ``internal`` events (reward amounts, payment references, the researcher's identity) are filtered out of the business owner's view at the query level, not in the frontend. """ __tablename__ = "bounty_report_events" __mapper_args__ = {"eager_defaults": True} id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) report_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("bounty_reports.id", ondelete="CASCADE"), nullable=False, ) event_type: Mapped[str] = mapped_column(String(32), nullable=False) actor_kind: Mapped[str] = mapped_column( String(16), nullable=False, default=const.ACTOR_USER, server_default=const.ACTOR_USER, ) actor_person_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), ) # Snapshot so the timeline still reads correctly after the actor is # soft-deleted or leaves the company. actor_name: Mapped[str | None] = mapped_column(String(255)) visibility: Mapped[str] = mapped_column( String(16), nullable=False, default=const.VISIBILITY_INTERNAL, server_default=const.VISIBILITY_INTERNAL, ) # Comment text, reply body, or the note attached to a transition. body: Mapped[str | None] = mapped_column(Text) # Structured payload: {"from": ..., "to": ..., "template_key": ..., # "amount_usd": ..., "subject": ...} meta: Mapped[dict[str, Any] | None] = mapped_column(JSON) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) __table_args__ = ( Index("ix_bounty_events_report", "report_id", "created_at"), Index("ix_bounty_events_type", "event_type", "created_at"), ) def __repr__(self) -> str: return f"" # ════════════════════════════════════════════════════════════════════ # 4. Severity tiers (the reward ladder) # ════════════════════════════════════════════════════════════════════ class BountySeverityTier(Base): """A configurable severity level: label, default reward, override ceiling, and the remediation SLA it implies. Seeded from ``catalog.SEVERITY_TIERS``. Builtin rows can be edited (amounts change year to year) and disabled, but not deleted — a deleted tier would orphan the ``severity`` value on historical reports and silently drop them out of every severity chart. """ __tablename__ = "bounty_severity_tiers" __mapper_args__ = {"eager_defaults": True} id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) key: Mapped[str] = mapped_column(String(24), nullable=False) label_zh: Mapped[str] = mapped_column(String(64), nullable=False) label_en: Mapped[str] = mapped_column(String(64), nullable=False) label_ja: Mapped[str] = mapped_column(String(64), nullable=False) default_reward_usd: Mapped[Decimal] = mapped_column( Numeric(12, 2), nullable=False, default=Decimal("0"), server_default="0", ) # Guardrail on the per-report override. NULL = no ceiling. max_reward_usd: Mapped[Decimal | None] = mapped_column(Numeric(12, 2)) fix_sla_days: Mapped[int | None] = mapped_column(Integer) color: Mapped[str | None] = mapped_column(String(16)) sort_order: Mapped[int] = mapped_column( Integer, nullable=False, default=100, server_default="100", ) enabled: Mapped[bool] = mapped_column( Boolean, nullable=False, default=True, server_default="1", ) is_builtin: Mapped[bool] = mapped_column( Boolean, nullable=False, default=False, server_default="0", ) updated_by: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), ) 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(), ) __table_args__ = ( UniqueConstraint("key", name="uq_bounty_severity_tiers_key"), Index("ix_bounty_severity_tiers_order", "sort_order"), ) def __repr__(self) -> str: return f"" # ════════════════════════════════════════════════════════════════════ # 5. Reply templates # ════════════════════════════════════════════════════════════════════ class BountyReplyTemplate(Base): """An English reply the operator copies into their mail client. Bodies are English-only by design (the audience is an external researcher); the *names* are tri-lingual because they appear in the Helix picker. Seeded from ``catalog.REPLY_TEMPLATES``; builtin rows are editable and restorable but not deletable. """ __tablename__ = "bounty_reply_templates" __mapper_args__ = {"eager_defaults": True} id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) key: Mapped[str] = mapped_column(String(64), nullable=False) name_zh: Mapped[str] = mapped_column(String(128), nullable=False) name_en: Mapped[str] = mapped_column(String(128), nullable=False) name_ja: Mapped[str] = mapped_column(String(128), nullable=False) category: Mapped[str] = mapped_column(String(24), nullable=False) subject: Mapped[str] = mapped_column(String(500), nullable=False) body: Mapped[str] = mapped_column(Text, nullable=False) # Declared placeholders, for the "variables used" hint in the editor. # Advisory: rendering substitutes whatever tokens it recognises. variables: Mapped[list[str] | None] = mapped_column(JSON) sort_order: Mapped[int] = mapped_column( Integer, nullable=False, default=100, server_default="100", ) enabled: Mapped[bool] = mapped_column( Boolean, nullable=False, default=True, server_default="1", ) is_builtin: Mapped[bool] = mapped_column( Boolean, nullable=False, default=False, server_default="0", ) # How many times a reply was rendered from this template — tells the # operator which templates actually earn their place. use_count: Mapped[int] = mapped_column( Integer, nullable=False, default=0, server_default="0", ) updated_by: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), ) 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(), ) __table_args__ = ( UniqueConstraint("key", name="uq_bounty_reply_templates_key"), Index("ix_bounty_reply_templates_order", "category", "sort_order"), ) def __repr__(self) -> str: return f"" class BountyMailIntake(Base): """A mail that arrived on ``security@`` and is waiting for a person. Deliberately not a report. ``security@`` receives vendor pitches, internal mail and automated scan output as well as researchers, and auto-creating reports from all of it would pollute the ledger, the budget attribution and every statistic built on them. The precedent is concrete: an HRIS predicate slightly too loose created 26 accounts for system users because nothing asked a person first. The body is NOT stored — only a snippet. Researchers send tax forms and bank details unprompted, and the FY27 import already established that their correspondence stays in the mailbox. """ __tablename__ = "bounty_mail_intake" id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) #: RFC822 Message-ID, unique: one mail is offered once however often #: the poller runs or however far its watermark rewinds. message_id: Mapped[str] = mapped_column(String(512), nullable=False) #: Used to ATTACH follow-ups to the report the thread already created, #: instead of creating a second report for the same conversation. thread_id: Mapped[str | None] = mapped_column(String(128)) from_email: Mapped[str] = mapped_column(String(320), nullable=False) from_name: Mapped[str | None] = mapped_column(String(255)) subject: Mapped[str | None] = mapped_column(String(500)) snippet: Mapped[str | None] = mapped_column(Text) received_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, ) status: Mapped[str] = mapped_column( String(24), nullable=False, default="pending", server_default="pending", ) #: What the AI read out of the subject and snippet. Advisory only. parsed: Mapped[dict[str, Any] | None] = mapped_column(JSON) report_id: Mapped[uuid.UUID | None] = mapped_column(Uuid) decided_by: Mapped[uuid.UUID | None] = mapped_column(Uuid) decided_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) note: Mapped[str | None] = mapped_column(String(1000)) 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 BountyReportAttachment(Base): """A file the researcher sent with their report. The write-up import read text out of the MIME tree and walked past every attachment part, so 319 files across 54 reports — 266 screenshots, 25 HTTP logs, 19 PDFs and 9 PoC screen recordings — existed only in the mailbox. 「有些白帽子给了附件 这个才是重要的」: for a good share of these reports the screenshot or the recording IS the evidence, and the prose is a caption for it. Bytes live in S3 through ``core.storage`` (the same path the auditor domain's evidence files take); this table holds the metadata and the key. ``gmail_message_id`` + ``gmail_attachment_id`` keep the provenance so a file can be re-fetched from the thread it came from without guessing which message it hung off. Two columns exist to keep the UI honest rather than to store data: * ``is_inline`` — a signature logo and a proof-of-concept screenshot are both ``image/png`` parts of a mail. One report carries 90 files for 6 MiB, which is what a page looks like when nobody separates them. Inline parts are kept (they are occasionally the actual screenshot, pasted rather than attached) but collapse by default. * ``sha256`` — researchers resend threads, and the same PNG arrives three times. Deduped per report, so a re-run of the importer is a no-op instead of tripling the gallery. """ __tablename__ = "bounty_report_attachments" __mapper_args__ = {"eager_defaults": True} id: Mapped[uuid.UUID] = mapped_column( Uuid, primary_key=True, default=uuid.uuid4, ) report_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("bounty_reports.id", ondelete="CASCADE"), nullable=False, ) #: FULL S3 object key including whatever prefix was active at upload #: time — never a URL. A presigned GET is minted per download. object_key: Mapped[str] = mapped_column(String(512), nullable=False) filename: Mapped[str] = mapped_column(String(255), nullable=False) content_type: Mapped[str] = mapped_column( String(128), nullable=False, server_default="application/octet-stream", ) size_bytes: Mapped[int | None] = mapped_column(BigInteger) #: Hex sha256 of the bytes. Dedupe key together with ``report_id``. sha256: Mapped[str | None] = mapped_column(String(64)) is_inline: Mapped[bool] = mapped_column( Boolean, nullable=False, default=False, server_default="0", ) #: ``researcher_mail`` today. Named rather than assumed so an operator #: upload later is distinguishable from what the researcher sent — #: which matters when the question is 「这是他给的证据吗」. source: Mapped[str] = mapped_column( String(24), nullable=False, server_default="researcher_mail", ) gmail_message_id: Mapped[str | None] = mapped_column(String(128)) #: Gmail attachment ids are long opaque strings, well past String(255). gmail_attachment_id: Mapped[str | None] = mapped_column(Text) imported_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) __table_args__ = ( Index("idx_bounty_attach_report", "report_id"), # Dedupe is an index, not a UNIQUE constraint: sha256 is nullable # (a zero-byte or unreadable part still deserves a row saying it # was there), and MySQL treats NULLs as distinct anyway, so the # constraint would not hold the invariant it appears to. Index("idx_bounty_attach_dedupe", "report_id", "sha256"), ) def __repr__(self) -> str: return ( f"" )