git_ew ¤

async def create_message(  # noqa: PLR0917
    self,
    message_id: str,
    thread_id: int,
    from_email: str,
    from_name: str,
    subject: str,
    date: datetime,
    body: str,
    *,
    in_reply_to: str | None = None,
    is_patch: bool = False,
    patch_content: str | None = None,
    raw_email: str | None = None,
) -> Message:
    """Create a new message.

    Args:
        message_id: Unique message ID.
        thread_id: Thread ID this message belongs to.
        from_email: Sender email.
        from_name: Sender name.
        subject: Message subject.
        date: Message date.
        body: Message body.
        in_reply_to: ID of message this replies to.
        is_patch: Whether this message contains a patch.
        patch_content: Patch content if applicable.
        raw_email: Raw email content.

    Returns:
        Created message.
    """
    message = Message(
        message_id=message_id,
        thread_id=thread_id,
        from_email=from_email,
        from_name=from_name,
        subject=subject,
        date=date,
        body=body,
        in_reply_to=in_reply_to,
        is_patch=is_patch,
        patch_content=patch_content,
        raw_email=raw_email,
    )
    async with self.session() as session:
        session.add(message)

        # Update thread's updated_at
        thread = await session.get(Thread, thread_id)
        if thread:
            thread.updated_at = datetime.now(UTC)

        await session.flush()
        await session.refresh(message)
        return message

async def create_thread(
    self,
    subject: str,
    first_message_id: str,
    *,
    is_patch: bool = False,
) -> Thread:
    """Create a new thread.

    Args:
        subject: Thread subject.
        first_message_id: ID of the first message.
        is_patch: Whether this thread contains patches.

    Returns:
        Created thread.
    """
    now = datetime.now(UTC)
    thread = Thread(
        subject=subject,
        first_message_id=first_message_id,
        created_at=now,
        updated_at=now,
        is_patch=is_patch,
        status="open",
    )
    async with self.session() as session:
        session.add(thread)
        await session.flush()
        await session.refresh(thread)
        return thread

async def get_config(self, key: str, default: Any = None) -> Any:
    """Get a configuration value.

    Args:
        key: Configuration key.
        default: Default value if key not found.

    Returns:
        Configuration value or default.
    """
    async with self.session() as session:
        query = select(Configuration).where(Configuration.key == key)
        result = await session.execute(query)
        config = result.scalar_one_or_none()
        if config:
            return json.loads(config.value)
        return default

async def get_email_sources(self) -> list[EmailSource]:
    """Get all email sources.

    Returns:
        List of email sources.
    """
    async with self.session() as session:
        query = select(EmailSource)
        result = await session.execute(query)
        return list(result.scalars().all())

async def get_message_by_id(self, message_id: str) -> Message | None:
    """Get a message by its message ID.

    Args:
        message_id: Message ID.

    Returns:
        Message or None if not found.
    """
    async with self.session() as session:
        query = select(Message).where(Message.message_id == message_id)
        result = await session.execute(query)
        return result.scalar_one_or_none()

async def get_thread(self, thread_id: int) -> Thread | None:
    """Get a thread by ID with all messages.

    Args:
        thread_id: Thread ID.

    Returns:
        Thread or None if not found.
    """
    async with self.session() as session:
        query = select(Thread).where(Thread.id == thread_id).options(selectinload(Thread.messages))
        result = await session.execute(query)
        return result.scalar_one_or_none()

async def get_thread_by_message_id(self, message_id: str) -> Thread | None:
    """Get a thread by first message ID.

    Args:
        message_id: Message ID.

    Returns:
        Thread or None if not found.
    """
    async with self.session() as session:
        query = select(Thread).where(Thread.first_message_id == message_id)
        result = await session.execute(query)
        return result.scalar_one_or_none()

async def get_threads(
    self,
    status: str | None = None,
    limit: int | None = None,
    offset: int = 0,
) -> list[Thread]:
    """Get threads with optional filtering.

    Args:
        status: Filter by status (open/closed).
        limit: Maximum number of threads to return.
        offset: Number of threads to skip.

    Returns:
        List of threads.
    """
    async with self.session() as session:
        query = select(Thread).order_by(Thread.updated_at.desc())
        if status:
            query = query.where(Thread.status == status)
        if limit is not None:
            query = query.limit(limit)
        query = query.offset(offset)
        result = await session.execute(query)
        return list(result.scalars().all())

async def init_db(self) -> None:
    """Initialize database tables."""
    async with self.engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

@asynccontextmanager
async def session(self) -> AsyncIterator[AsyncSession]:
    """Get a database session."""
    async with self.session_maker() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

async def set_config(self, key: str, value: Any) -> None:
    """Set a configuration value.

    Args:
        key: Configuration key.
        value: Configuration value.
    """
    async with self.session() as session:
        query = select(Configuration).where(Configuration.key == key)
        result = await session.execute(query)
        config = result.scalar_one_or_none()

        if config:
            config.value = json.dumps(value)
        else:
            config = Configuration(key=key, value=json.dumps(value))
            session.add(config)

async def update_message_patch(self, message_id: str, patch_content: str) -> None:
    """Add patch content to an existing message."""
    async with self.session() as session:
        query = select(Message).where(Message.message_id == message_id)
        result = await session.execute(query)
        message = result.scalar_one_or_none()
        if message and not message.patch_content:
            message.patch_content = patch_content
            message.is_patch = True
            thread = await session.get(Thread, message.thread_id)
            if thread:
                thread.is_patch = True

async def update_thread_status(self, thread_id: int, status: str) -> None:
    """Update thread status.

    Args:
        thread_id: Thread ID.
        status: New status (open/closed).
    """
    async with self.session() as session:
        thread = await session.get(Thread, thread_id)
        if thread:
            thread.status = status
            thread.updated_at = datetime.now(UTC)

def fetch_emails(self, since: str | None = None) -> AsyncIterator[ParsedEmail]:
    """Fetch emails from the source.

    Args:
        since: Fetch emails since this date (ISO format).

    Yields:
        ParsedEmail objects.
    """
    raise NotImplementedError

def build_email(  # noqa: PLR0917
    self,
    to_email: str,
    subject: str,
    body: str,
    in_reply_to: str | None = None,
    references: list[str] | None = None,
    cc: list[str] | None = None,
) -> EmailMessage:
    """Build an email without sending it."""
    msg = EmailMessage()
    msg.set_content(body)

    # Set headers
    msg["From"] = f"{self.from_name} <{self.from_email}>"
    msg["To"] = to_email
    msg["Subject"] = subject
    msg["Date"] = formatdate(localtime=True)
    msg["Message-ID"] = make_msgid(domain=self.from_email.split("@")[1])

    if cc:
        msg["Cc"] = ", ".join(cc)

    if in_reply_to:
        msg["In-Reply-To"] = f"<{in_reply_to}>"

    if references:
        msg["References"] = " ".join(f"<{ref}>" for ref in references)

    return msg

def build_reply(  # noqa: PLR0917
    self,
    to_email: str,
    subject: str,
    body: str,
    in_reply_to: str,
    references: list[str] | None = None,
    cc: list[str] | None = None,
) -> EmailMessage:
    """Build a reply email without sending it."""
    if not subject.lower().startswith("re:"):
        subject = f"Re: {subject}"

    references = list(references or [])
    if in_reply_to not in references:
        references.append(in_reply_to)

    return self.build_email(
        to_email=to_email,
        subject=subject,
        body=body,
        in_reply_to=in_reply_to,
        references=references,
        cc=cc,
    )

def send_email(  # noqa: PLR0917
    self,
    to_email: str,
    subject: str,
    body: str,
    in_reply_to: str | None = None,
    references: list[str] | None = None,
    cc: list[str] | None = None,
) -> str:
    """Send a plain text email.

    Args:
        to_email: Recipient email address.
        subject: Email subject.
        body: Email body (plain text).
        in_reply_to: Message ID this replies to.
        references: List of message IDs in the thread.
        cc: List of CC recipients.

    Returns:
        Message ID of sent email.
    """
    msg = self.build_email(to_email, subject, body, in_reply_to, references, cc)
    self.send_message(msg)
    return msg["Message-ID"].strip("<>")

def send_message(self, msg: EmailMessage) -> None:
    """Send a prepared email over SMTP."""
    with smtplib.SMTP(self.smtp_host, self.smtp_port) as smtp:
        if self.use_tls:
            smtp.starttls()

        if self.username and self.password:
            smtp.login(self.username, self.password)

        smtp.send_message(msg)

def send_reply(  # noqa: PLR0917
    self,
    to_email: str,
    subject: str,
    body: str,
    in_reply_to: str,
    references: list[str] | None = None,
    cc: list[str] | None = None,
) -> str:
    """Send a reply email.

    Args:
        to_email: Recipient email address.
        subject: Email subject (will be prefixed with "Re:" if not present).
        body: Email body (plain text).
        in_reply_to: Message ID this replies to.
        references: List of message IDs in the thread.
        cc: List of CC recipients.

    Returns:
        Message ID of sent email.
    """
    msg = self.build_reply(to_email, subject, body, in_reply_to, references, cc)
    self.send_message(msg)
    return msg["Message-ID"].strip("<>")

async def fetch_emails(self, since: str | None = None) -> AsyncIterator[ParsedEmail]:
    """Fetch and parse messages from configured IMAP folders."""
    raw_messages = await asyncio.to_thread(self._fetch_messages, since)
    for raw_message in raw_messages:
        try:
            yield parse_email(raw_message)
        except Exception as e:  # noqa: BLE001
            _logger.debug("Skipping malformed IMAP message: %s", e)
        await asyncio.sleep(0)

def handle_data(self, data: str) -> None:
    """Extract date information from pre-formatted text."""
    if not self._in_pre:
        return

    self._current_line += data

    # Look for date patterns in the line (DD-MMM-YYYY format)
    # Example: "12-Jun-1995"
    if len(self._current_line) > 50:  # Approximate line length  # noqa: PLR2004
        parts = self._current_line.split()
        for i, part in enumerate(parts):
            # Try to parse date from parts
            if "-" in part and len(parts) > i + 1:
                try:
                    # Try common date formats
                    date_str = part
                    # Convert French month names to English
                    date_str = date_str.replace("janv.", "Jan")
                    date_str = date_str.replace("févr.", "Feb")
                    date_str = date_str.replace("mars", "Mar")
                    date_str = date_str.replace("avril", "Apr")
                    date_str = date_str.replace("mai", "May")
                    date_str = date_str.replace("juin", "Jun")
                    date_str = date_str.replace("juil.", "Jul")
                    date_str = date_str.replace("août", "Aug")
                    date_str = date_str.replace("sept.", "Sep")
                    date_str = date_str.replace("oct.", "Oct")
                    date_str = date_str.replace("nov.", "Nov")
                    date_str = date_str.replace("déc.", "Dec")

                    parsed_time = strptime(date_str, "%d-%b-%Y")
                    parsed = date(parsed_time.tm_year, parsed_time.tm_mon, parsed_time.tm_mday)
                    # Associate this date with the last archived filename found
                    if self.archives:
                        last_filename = list(self.archives.keys())[-1]
                        if self.archives[last_filename] is None:
                            self.archives[last_filename] = parsed
                except (ValueError, IndexError):
                    _logger.debug("Could not parse archive date from %r", part, exc_info=True)

        self._current_line = ""

def handle_endtag(self, tag: str) -> None:
    """Track end of pre tag."""
    if tag == "pre":
        self._in_pre = False

def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
    """Track pre tags and extract href attributes."""
    if tag == "pre":
        self._in_pre = True
    elif tag == "a" and self._in_pre:
        for attr, value in attrs:
            if attr == "href" and value and value.endswith(".tgz"):
                filename = value.split("/")[-1]
                # Date will be extracted from text content after the link
                self.archives[filename] = None

async def fetch_emails(self, since: str | None = None) -> AsyncIterator[ParsedEmail]:  # noqa: ARG002
    """Fetch emails from Maildir.

    Args:
        since: Fetch emails since this date (ISO format).

    Yields:
        ParsedEmail objects.
    """
    if not self.maildir_path.exists():
        return

    # Use mailbox.Maildir to read the maildir
    mbox = mailbox.Maildir(str(self.maildir_path))
    try:
        for message in mbox:
            try:
                raw = message.as_bytes()
                parsed = parse_email(raw)
                yield parsed
            except Exception as e:  # noqa: BLE001
                # Skip malformed emails
                _logger.debug("Skipping malformed Maildir message: %s", e)
                continue

            # Allow event loop to process
            await asyncio.sleep(0)
    finally:
        mbox.close()

async def fetch_emails(self, since: str | None = None) -> AsyncIterator[ParsedEmail]:  # noqa: ARG002
    """Fetch emails from mbox.

    Args:
        since: Fetch emails since this date (ISO format).

    Yields:
        ParsedEmail objects.
    """
    if not self.mbox_path.exists():
        return

    mbox = mailbox.mbox(str(self.mbox_path))
    try:
        for message in mbox:
            try:
                raw = message.as_bytes()
                parsed = parse_email(raw)
                yield parsed
            except Exception as e:  # noqa: BLE001
                # Skip malformed emails
                _logger.debug("Skipping malformed mbox message: %s", e)
                continue

            # Allow event loop to process
            await asyncio.sleep(0)
    finally:
        mbox.close()