{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "da5a2e63",
   "metadata": {},
   "outputs": [],
   "source": [
    "#ignore\n",
    "import os\n",
    "\n",
    "def load_env(file_path):\n",
    "    \"\"\"Загружает переменные окружения из файла .env с поддержкой булевых значений.\"\"\"\n",
    "    with open(file_path) as f:\n",
    "        for line in f:\n",
    "            line = line.strip()\n",
    "            if not line or line.startswith('#'):\n",
    "                continue\n",
    "            \n",
    "            try:\n",
    "                key, value = line.split('=', 1)\n",
    "                key = key.strip()\n",
    "                value = value.strip('\"\\' ')\n",
    "                # Преобразование строковых значений в булевые\n",
    "                if value.lower() in ['true', '1', 'yes']:\n",
    "                    value = True\n",
    "                elif value.lower() in ['false', '0', 'no']:\n",
    "                    value = False\n",
    "                \n",
    "                # Проверка, что ключ не пустой\n",
    "                if key:\n",
    "                    os.environ[key] = str(value)  # Храним как строку\n",
    "            except ValueError:\n",
    "                print(f\"Ошибка: неверный формат в строке '{line}'\")\n",
    "\n",
    "try:\n",
    "    load_env('env')\n",
    "except:\n",
    "    load_env('env_example')"
   ],
   "display_name": ".venv"
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "afd6bc25",
   "metadata": {},
   "outputs": [],
   "source": [
    "import smtplib\n",
    "import os\n",
    "from email.mime.application import MIMEApplication\n",
    "from email.mime.multipart import MIMEMultipart\n",
    "from email.mime.text import MIMEText\n",
    "from email.utils import formatdate, make_msgid\n",
    "from typing import Optional\n",
    "import gnupg\n",
    "from celery import Celery\n",
    "from config import *\n",
    "from db import *\n",
    "from templates import templates\n",
    "import logging\n",
    "import asyncio \n"
   ],
   "display_name": ".venv"
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ea1066a5",
   "metadata": {},
   "outputs": [],
   "source": [
    "def build_unsubscribe_url(email: str, token: str, base_url: str) -> str:\n",
    "    return f\"{base_url}/api/v1/email/unsubscribe?token={token}&email={email}\"\n"
   ],
   "display_name": ".venv"
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "05fadc2d",
   "metadata": {},
   "outputs": [],
   "source": [
    "GPG_HOME = os.getenv(\"GPG_HOME\", \"/var/lib/app/.gnupg\")\n",
    "GPG_BINARY = os.getenv(\"GPG_BINARY\", \"gpg\")\n",
    "\n",
    "gpg = gnupg.GPG(\n",
    "    gnupghome=GPG_HOME,\n",
    "    gpgbinary=GPG_BINARY,\n",
    ")\n",
    "\n",
    "\n",
    "def build_unsubscribe_url(\n",
    "    to_email: str,\n",
    "    user_id: Optional[int],\n",
    "    domain: str,\n",
    ") -> str:\n",
    "    \"\"\"\n",
    "    Replace this implementation with your real unsubscribe URL builder.\n",
    "    Avoid putting a raw email address in the URL in production.\n",
    "    \"\"\"\n",
    "    from urllib.parse import quote\n",
    "\n",
    "    return f\"https://{domain}/unsubscribe?email={quote(to_email)}\"\n",
    "\n",
    "\n",
    "def import_public_key(public_key: str) -> str:\n",
    "    \"\"\"\n",
    "    Import an ASCII-armored public key and return its fingerprint.\n",
    "\n",
    "    The database value should contain the complete public key, for example:\n",
    "\n",
    "    -----BEGIN PGP PUBLIC KEY BLOCK-----\n",
    "    ...\n",
    "    -----END PGP PUBLIC KEY BLOCK-----\n",
    "    \"\"\"\n",
    "    result = gpg.import_keys(public_key)\n",
    "\n",
    "    if not result.fingerprints:\n",
    "        raise RuntimeError(\n",
    "            f\"Unable to import the recipient's public key: {result}\"\n",
    "        )\n",
    "\n",
    "    # Use the imported fingerprint rather than the email address. The email\n",
    "    # address in the key's UID may not exactly match the recipient address.\n",
    "    return result.fingerprints[0]\n",
    "\n",
    "\n",
    "def encrypt_mime_message(\n",
    "    message,\n",
    "    recipient_fingerprint: str,\n",
    "):\n",
    "    \"\"\"\n",
    "    Encrypt an already-built MIME message and return a PGP/MIME message.\n",
    "    \"\"\"\n",
    "\n",
    "    # Serialize the original MIME message before encryption.\n",
    "    plaintext = message.as_bytes()\n",
    "\n",
    "    # Binary OpenPGP data is appropriate for application/octet-stream.\n",
    "    encrypted = gpg.encrypt(\n",
    "        plaintext,\n",
    "        recipients=[recipient_fingerprint],\n",
    "        armor=False,\n",
    "        always_trust=False,\n",
    "    )\n",
    "\n",
    "    if not encrypted.ok:\n",
    "        raise RuntimeError(\n",
    "            f\"GPG encryption failed: {encrypted.status}; \"\n",
    "            f\"stderr: {encrypted.stderr}\"\n",
    "        )\n",
    "\n",
    "    encrypted_bytes = encrypted.data\n",
    "\n",
    "    if not encrypted_bytes:\n",
    "        raise RuntimeError(\"GPG returned an empty encrypted message\")\n",
    "\n",
    "    encrypted_message = MIMEMultipart(\n",
    "        \"encrypted\",\n",
    "        protocol=\"application/pgp-encrypted\",\n",
    "    )\n",
    "\n",
    "    # Copy normal mail headers. Content-Type and MIME-Version are generated\n",
    "    # automatically for the new multipart/encrypted message.\n",
    "    for header in (\n",
    "        \"From\",\n",
    "        \"To\",\n",
    "        \"Subject\",\n",
    "        \"Message-ID\",\n",
    "        \"Date\",\n",
    "        \"X-Priority\",\n",
    "        \"Importance\",\n",
    "        \"X-Auto-Response-Suppress\",\n",
    "        \"List-Unsubscribe\",\n",
    "        \"List-Unsubscribe-Post\",\n",
    "    ):\n",
    "        value = message.get(header)\n",
    "        if value is not None:\n",
    "            encrypted_message[header] = value\n",
    "\n",
    "    # Required PGP/MIME version part.\n",
    "    version_part = MIMEApplication(\n",
    "        b\"Version: 1\\r\\n\",\n",
    "        _subtype=\"pgp-encrypted\",\n",
    "    )\n",
    "    version_part.add_header(\n",
    "        \"Content-Description\",\n",
    "        \"PGP/MIME version identification\",\n",
    "    )\n",
    "    encrypted_message.attach(version_part)\n",
    "\n",
    "    # Encrypted MIME payload.\n",
    "    encrypted_part = MIMEApplication(\n",
    "        encrypted_bytes,\n",
    "        _subtype=\"octet-stream\",\n",
    "    )\n",
    "    encrypted_part.add_header(\n",
    "        \"Content-Description\",\n",
    "        \"OpenPGP encrypted message\",\n",
    "    )\n",
    "    encrypted_part.add_header(\n",
    "        \"Content-Disposition\",\n",
    "        \"inline\",\n",
    "        filename=\"encrypted.asc\",\n",
    "    )\n",
    "    encrypted_message.attach(encrypted_part)\n",
    "\n",
    "    return encrypted_message\n",
    "\n",
    "\n",
    "async def create_email_message(\n",
    "    to_email: str,\n",
    "    subject: str,\n",
    "    html_content: str,\n",
    "    user_id: Optional[int] = None,\n",
    "):\n",
    "    \"\"\"\n",
    "    Create an HTML email.\n",
    "\n",
    "    If the recipient has a public GPG key in the database, the complete\n",
    "    MIME message is encrypted using PGP/MIME.\n",
    "\n",
    "    Expected database structure:\n",
    "\n",
    "        {\n",
    "            \"email\": \"recipient@example.com\",\n",
    "            \"gnupg\": \"-----BEGIN PGP PUBLIC KEY BLOCK-----\\\\n...\"\n",
    "        }\n",
    "    \"\"\"\n",
    "\n",
    "    msg = MIMEMultipart(\"alternative\")\n",
    "\n",
    "    msg[\"From\"] = f\"{FROM_NAME} <{FROM_EMAIL}>\"\n",
    "    msg[\"To\"] = to_email\n",
    "    msg[\"Subject\"] = subject\n",
    "    msg[\"Message-ID\"] = make_msgid(\n",
    "        domain=FROM_EMAIL.rsplit(\"@\", 1)[1],\n",
    "    )\n",
    "    msg[\"Date\"] = formatdate(localtime=True)\n",
    "    msg[\"X-Priority\"] = \"3\"\n",
    "    msg[\"Importance\"] = \"normal\"\n",
    "    msg[\"X-Auto-Response-Suppress\"] = \"All\"\n",
    "\n",
    "    # Do not send unsubscribe headers to yourself.\n",
    "    if to_email.lower() != FROM_EMAIL.lower():\n",
    "        unsubscribe_url = build_unsubscribe_url(\n",
    "            to_email=to_email,\n",
    "            user_id=user_id,\n",
    "            domain=FULL_DOMAIN,\n",
    "        )\n",
    "\n",
    "        msg[\"List-Unsubscribe\"] = f\"<{unsubscribe_url}>\"\n",
    "        msg[\"List-Unsubscribe-Post\"] = \"List-Unsubscribe=One-Click\"\n",
    "\n",
    "    msg.attach(\n",
    "        MIMEText(\n",
    "            html_content,\n",
    "            \"html\",\n",
    "            \"utf-8\",\n",
    "        )\n",
    "    )\n",
    "\n",
    "    # Fetch the recipient's public key.\n",
    "    user = await users_collection.find_one(\n",
    "        {\"email\": to_email},\n",
    "        {\"gnupg\": 1},\n",
    "    )\n",
    "\n",
    "    public_key = user.get(\"gnupg\") if user else None\n",
    "\n",
    "    if not public_key:\n",
    "        # Return a normal unencrypted email when no key is available.\n",
    "        return msg\n",
    "\n",
    "    if not isinstance(public_key, str):\n",
    "        raise TypeError(\"The gnupg database field must be a string\")\n",
    "\n",
    "    try:\n",
    "        fingerprint = import_public_key(public_key)\n",
    "        return encrypt_mime_message(msg, fingerprint)\n",
    "\n",
    "    except Exception:\n",
    "        # Encryption is mandatory whenever a key exists.\n",
    "        # Do not silently send the message unencrypted.\n",
    "        raise\n"
   ],
   "display_name": ".venv"
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6687f2ab",
   "metadata": {},
   "outputs": [],
   "source": [
    "def connect(email_server, email_port):\n",
    "    server = smtplib.SMTP(email_server, email_port)\n",
    "    server.starttls(context=ctx)\n",
    "    return server\n",
    "\n",
    "\n",
    "def send_email(\n",
    "    to_email: str,\n",
    "    html_content: str,\n",
    "    subject: str,\n",
    "    user_id: Optional[str] = None,\n",
    ") -> bool:\n",
    "    msg =asyncio.run(create_email_message(\n",
    "        to_email=to_email,\n",
    "        subject=subject,\n",
    "        html_content=html_content,\n",
    "        user_id=user_id,\n",
    "    ))\n",
    "\n",
    "    try:\n",
    "        with connect(EMAIL_SMTP_SERVER, EMAIL_SMTP_PORT) as server:\n",
    "            server.login(FROM_EMAIL, EMAIL_PASS)\n",
    "            server.send_message(msg)\n",
    "            # server.close()\n",
    "\n",
    "        logging.info(f\"Email sent successfully to: {to_email} | Subject: {subject}\")\n",
    "        return True\n",
    "\n",
    "    except Exception as e:\n",
    "        logging.error(\n",
    "            f\"Error sending email to: {to_email}, \"\n",
    "            f\"Subject: {subject}, Error: {str(e)}\"\n",
    "        )\n",
    "        return False\n"
   ],
   "display_name": ".venv"
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "81e35a88",
   "metadata": {},
   "outputs": [],
   "source": [
    "celery_app_email = Celery('emails', broker=f'{redis_uri}2', backend=f'{redis_uri}2', task_track_started=True)\n"
   ],
   "display_name": ".venv"
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "30ca10ab",
   "metadata": {},
   "outputs": [],
   "source": [
    "@celery_app_email.task(\n",
    "    bind=True,\n",
    "    rate_limit='30/m',\n",
    "    name='email_task.send_email',\n",
    "    retry_backoff=True,\n",
    "    retry_backoff_max=600,\n",
    "    retry_jitter=True,\n",
    "    max_retries=3,\n",
    "    soft_time_limit=30,\n",
    "    time_limit=45,\n",
    ")\n",
    "def notify(self, to_email: str, data: dict, subject: str = EMAIL_SUBJECT, template: str = 'main.html'):\n",
    "    logging.info(f\"Отправка email на {to_email} (попытка {self.request.retries + 1})\")\n",
    "    try:\n",
    "        html_content = templates.get_template(f\"email/{template}\").render({\n",
    "            \"subject\": subject, \n",
    "            \"data\": data\n",
    "        })\n",
    "        \n",
    "        ok = send_email(to_email, html_content, subject=subject)\n",
    "        if ok:\n",
    "            return ok\n",
    "        else:\n",
    "            raise self.retry(exc=exc)\n",
    "        \n",
    "    except Exception as exc:\n",
    "        logging.warning(\n",
    "            f\"Ошибка при отправке email на {to_email} (попытка {self.request.retries + 1}): {exc}\"\n",
    "        )\n",
    "        raise self.retry(exc=exc)\n"
   ],
   "display_name": ".venv"
  },
  {
   "cell_type": "markdown",
   "id": "1837d7c2",
   "metadata": {},
   "source": [
    "## Run"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3aeae7e8",
   "metadata": {},
   "outputs": [],
   "source": [
    "!mkdir -p /var/lib/app/.gnupg\n",
    "!chmod 700 /var/lib/app/.gnupg"
   ],
   "display_name": ".venv"
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "daf4fe38",
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "import os\n",
    "\n",
    "IGNORE = ['build.ipynb', 'data.ipynb']\n",
    "\n",
    "for _filename in [i for i in os.listdir() if 'ipynb' in i and i not in IGNORE]:\n",
    "    filename_without_extension = os.path.splitext(os.path.basename(_filename))[0]\n",
    "\n",
    "    with open(f'{filename_without_extension}.ipynb', 'r') as f:\n",
    "        data = f.read()\n",
    "\n",
    "    main = json.loads(data)\n",
    "    app = []\n",
    "    for cell in main['cells']:\n",
    "        if '## Run' in cell['source']:\n",
    "            break\n",
    "        if '# ignore\\n' in cell['source']:\n",
    "            continue\n",
    "        app.append(''.join(cell['source']))\n",
    "\n",
    "    with open(f'{filename_without_extension}.py', 'w') as f:\n",
    "        f.write('\\n\\n'.join(app))\n"
   ],
   "display_name": ".venv"
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f5194611",
   "metadata": {},
   "outputs": [],
   "source": [
    "!celery -A email_task.celery_app_email worker --loglevel=info"
   ],
   "display_name": ".venv"
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.13.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}