Merge pull request 'production' (#10) from production into main

build.ipynb

@@ -756,6 +756,35 @@
     "### fastapi prod service"
    ]
   },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "721bee4c",
+   "metadata": {
+    "vscode": {
+     "languageId": "shellscript"
+    }
+   },
+   "outputs": [],
+   "source": [
+    "%%bash\n",
+    "ssh kultmost 'tee /app/production/start-uvicorn.sh >/dev/null' <<'EOF'\n",
+    "#!/usr/bin/bash\n",
+    "set -euo pipefail\n",
+    "\n",
+    "workers=\"$(nproc)\"\n",
+    "\n",
+    "exec /app/production/.venv/bin/uvicorn main:app \\\n",
+    "  --host 127.0.0.1 \\\n",
+    "  --port 30081 \\\n",
+    "  --workers \"$workers\"\n",
+    "EOF\n",
+    "\n",
+    "ssh kultmost 'chmod +x /app/production/start-uvicorn.sh'\n",
+    "ssh kultmost 'ls -ln /app/production/start-uvicorn.sh'"
+   ],
+   "display_name": ".venv"
+  },
   {
    "cell_type": "code",
    "execution_count": null,
@@ -775,16 +804,16 @@
     "\n",
     "[Service]\n",
     "WorkingDirectory=/app/production\n",
-    "ExecStart=/app/production/.venv/bin/uvicorn main:app --host 127.0.0.1 --port 30081 --workers 3\n",
+    "ExecStart=/app/production/start-uvicorn.sh\n",
     "Restart=always\n",
     "EnvironmentFile=/app/production/env\n",
     "CPUWeight=60\n",
     "\n",
     "[Install]\n",
-    "WantedBy=default.target\n",
+    "WantedBy=multi-user.target\n",
     "EOF\n",
     "\n",
-    "ssh kultmost 'systemctl daemon-reload && systemctl enable --now production.service && systemctl status --no-pager -l production.service'\n"
+    "ssh kultmost 'systemctl daemon-reload && systemctl enable --now production.service && systemctl status --no-pager -l production.service'"
    ],
    "display_name": ".venv"
   },

main.ipynb

@@ -173,6 +173,33 @@
    ],
    "display_name": ".venv"
   },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "e7845a8a",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "def get_delivery_aggregate(_filter: dict = {}, _status: str = None):\n",
+    "    cursor = delivery_collection.aggregate([\n",
+    "        {\"$match\": {\n",
+    "            \"prod\": PROD, **_filter\n",
+    "        }},\n",
+    "        {\"$sort\": {\"_id\": -1}},\n",
+    "        {\"$group\": {\n",
+    "            \"_id\": \"$post_id\",\n",
+    "            \"doc\": {\"$first\": \"$$ROOT\"}\n",
+    "        }},\n",
+    "        {\"$replaceRoot\": {\"newRoot\": \"$doc\"}},\n",
+    "        {\"$sort\": {\"_id\": -1}},\n",
+    "    ])\n",
+    "    if _status:\n",
+    "        return (doc async for doc in cursor if doc.get('status') == _status)\n",
+    "    else:\n",
+    "        return cursor\n"
+   ],
+   "display_name": ".venv"
+  },
   {
    "cell_type": "code",
    "execution_count": null,
@@ -573,7 +600,6 @@
     "\n",
     "    prod_filter = {\"prod\": PROD}\n",
     "\n",
-    "    # 1) If there's an exhibition that is going to be finished soon -> move deliveries to \"Возврат художнику\"\n",
     "    exhibition_finished = await exhibition_collection.find_one_and_update(\n",
     "        {**prod_filter, \"status\": 1, \"date_finish\": {\"$lt\": limit_dt}},\n",
     "        {\"$set\": {\"status\": 2}},\n",
@@ -581,20 +607,9 @@
     "    )\n",
     "    if exhibition_finished:\n",
     "        exhibition_id_str = str(exhibition_finished[\"_id\"])\n",
-    "        cursor = delivery_collection.aggregate([\n",
-    "            {\"$match\": {\n",
-    "                \"exhibition_id\": exhibition_id_str, **prod_filter\n",
-    "            }},\n",
-    "            {\"$sort\": {\"_id\": -1}},\n",
-    "            {\"$group\": {\n",
-    "                \"_id\": \"$post_id\",\n",
-    "                \"doc\": {\"$first\": \"$$ROOT\"}\n",
-    "            }},\n",
-    "            {\"$replaceRoot\": {\"newRoot\": \"$doc\"}},\n",
-    "            {\"$sort\": {\"_id\": -1}},\n",
-    "        ])\n",
     "\n",
-    "        data = [doc async for doc in cursor if doc.get('status') == 'На выставке']\n",
+    "        data = [doc async for doc in get_delivery_aggregate({\"exhibition_id\": exhibition_id_str}, 'На выставке')]\n",
+    "\n",
     "        for doc in data:\n",
     "            post_id = doc.get(\"post_id\")\n",
     "            exhibition_id = doc.get(\"exhibition_id\")\n",
@@ -619,27 +634,25 @@
     "        return {}\n",
     "    tasks = []\n",
     "    async for exhibition_to_start in exhibition_collection.find(\n",
-    "        {**prod_filter}\n",
+    "        {\"status\": {'$lte':1}, **prod_filter}\n",
     "    ):\n",
     "        exhibition_id_str = str(exhibition_to_start[\"_id\"])\n",
     "        exhibition_title = exhibition_to_start.get(\"title\")\n",
     "\n",
     "        notified_user_id = None\n",
-    "        notified_post_user_id = None\n",
+    "        notified_post_user_ids = {}\n",
     "\n",
-    "        async for doc in delivery_collection.find(\n",
+    "        data = [doc async for doc in get_delivery_aggregate({\"exhibition_id\": exhibition_id_str}, 'Получено')]\n",
-    "            {\"exhibition_id\": exhibition_id_str, \"status\": \"Получено\", **prod_filter},\n",
+    "        \n",
-    "            {\"_id\": 0},\n",
+    "        for doc in data:\n",
-    "        ):\n",
     "            post_id = doc.get(\"post_id\")\n",
     "            exhibition_id = doc.get(\"exhibition_id\")\n",
     "\n",
     "            post = await posts_collection.find_one({\"_id\": ObjectId(post_id)})\n",
     "\n",
-    "            # Use user from post; keep them even if multiple deliveries exist\n",
+    "            if post and post.get(\"user\"):\n",
-    "            if post and post.get(\"user\") and not notified_post_user_id:\n",
+    "                notified_post_user_ids.update({str(post[\"user\"].get(\"_id\")):1})\n",
-    "                notified_post_user_id = str(post[\"user\"].get(\"_id\"))\n",
+    "            \n",
-    "\n",
     "            new_doc = {\n",
     "                **doc,\n",
     "                \"exhibition\": exhibition_to_start,\n",
@@ -659,12 +672,10 @@
     "                {\"$set\": {\"status\": \"На выставке\"}}\n",
     "            ))\n",
     "\n",
-    "        # NOTE: your original code used `post` from inside the loop (could be last item / even undefined).\n",
+    "        for user_id in notified_post_user_ids.keys():\n",
-    "        # This version notifies once using the first found post user.\n",
-    "        if notified_post_user_id:\n",
     "            await notify_in_profile(\n",
     "                NotifyItem(\n",
-    "                    user_id=notified_post_user_id,\n",
+    "                    user_id=user_id,\n",
     "                    type_item=\"exhibtion\",\n",
     "                    title=f\"Выставка {exhibition_title}\",\n",
     "                    link=\"/profile/exhibition\",\n",
@@ -673,7 +684,7 @@
     "            )\n",
     "    if tasks:\n",
     "        asyncio.gather(*tasks)\n",
-    "    # 2) Otherwise find an exhibition that should start soon -> move deliveries to \"На выставке\"\n",
+    "\n",
     "    exhibition_to_start = await exhibition_collection.find_one_and_update(\n",
     "        {**prod_filter, \"status\": 0, \"datetime_start\": {\"$lt\": limit_dt}},\n",
     "        {\"$set\": {\"status\": 1}},\n",
@@ -683,7 +694,6 @@
     "    if not exhibition_to_start:\n",
     "        return {}\n",
     "\n",
-    "    # If vk/tg arrays are empty -> send social invitation\n",
     "    if exhibition_to_start.get(\"vk\") == [] or exhibition_to_start.get(\"tg\") == []:\n",
     "        await service_social_exhibition(bson_clear(exhibition_to_start))\n",
     "    return {}\n"
@@ -1535,7 +1545,8 @@
     "    ]\n",
     "\n",
     "    cursor = competition_request_col.aggregate(pipeline)\n",
-    "    return await cursor.to_list(length=None)"
+    "    data = await cursor.to_list(length=None)\n",
+    "    return await posts_collection.find({\"_id\": {\"$in\":[ ObjectId(doc['_id']) if type(doc['_id']) is str else doc['_id'] for doc in data ]}}).to_list()"
    ],
    "display_name": ".venv"
   },
@@ -1642,22 +1653,8 @@
     "):\n",
     "    exhibition = await exhibition_collection.find_one({'_id': ObjectId(exhibition_id)})\n",
     "    if exhibition:\n",
-    "        match_ = {\n",
+    "        posts = [doc async for doc in get_delivery_aggregate({\"exhibition_id\": exhibition_id}, 'На выставке')]\n",
-    "            \"exhibition_id\": exhibition_id,\n",
+    "        \n",
-    "            'prod': PROD,\n",
-    "        }\n",
-    "\n",
-    "        cursor = delivery_collection.aggregate([\n",
-    "            {\"$match\": match_},\n",
-    "            {\"$sort\": {\"_id\": -1}},\n",
-    "            {\"$group\": {\n",
-    "                \"_id\": \"$post_id\",\n",
-    "                \"doc\": {\"$first\": \"$$ROOT\"}\n",
-    "            }},\n",
-    "            {\"$replaceRoot\": {\"newRoot\": \"$doc\"}},\n",
-    "            {\"$sort\": {\"_id\": -1}},\n",
-    "        ])\n",
-    "        posts = [doc async for doc in cursor if doc.get('status') == 'На выставке']\n",
     "        return templates.TemplateResponse(\n",
     "            request=request, name=\"exhibition/index.html\", context={\"id\": 1, \"user\": user, \"exhibition\": exhibition, \"posts\": posts, 'now': datetime.now()}\n",
     "        )\n",
@@ -4342,25 +4339,15 @@
     "            {\"sender._id\": user.get('_id')},\n",
     "            {\"user._id\": user.get('_id')},\n",
     "            {\"receiver._id\": user.get('_id')}\n",
-    "        ],\n",
+    "        ]\n",
-    "        'prod': PROD,\n",
     "    }\n",
     "    if data.get('competition'):\n",
     "        match_.update({'competition_id': data.get('competition').get('_id')})\n",
     "    if data.get('exhibition_id'):\n",
     "        match_.update({'exhibition_id': data.get('exhibition_id')})\n",
-    "    cursor = delivery_collection.aggregate([\n",
+    "    \n",
-    "        {\"$match\": match_},\n",
+    "    posts = [doc async for doc in get_delivery_aggregate(match_)]\n",
-    "        {\"$sort\": {\"_id\": -1}},\n",
-    "        {\"$group\": {\n",
-    "            \"_id\": \"$post_id\",\n",
-    "            \"doc\": {\"$first\": \"$$ROOT\"}\n",
-    "        }},\n",
-    "        {\"$replaceRoot\": {\"newRoot\": \"$doc\"}},\n",
-    "        {\"$sort\": {\"_id\": -1}},\n",
-    "    ])\n",
     "\n",
-    "    posts = [doc async for doc in cursor]\n",
     "    return templates.TemplateResponse(\n",
     "        request=request, name=\"profile/invitation.html\", context={ \"user\": user, \"doc\": data, \"data\": posts}\n",
     "    )"
@@ -4402,7 +4389,7 @@
     "    user: Dict = Security(get_current_user)\n",
     "):\n",
     "    await social_collection.update_one({'_id': ObjectId(id), 'user_id': user.get('_id')},{'$set': {key:check}})\n",
-    "    return {'ok': True}"
+    "    return {'ok': True}\n"
    ],
    "display_name": ".venv"
   },
@@ -4414,22 +4401,8 @@
    "outputs": [],
    "source": [
     "async def get_delivery_post(post_id):\n",
-    "    cursor = delivery_collection.aggregate([\n",
+    "    data = [doc async for doc in get_delivery_aggregate({'post_id': post_id})]\n",
-    "        {\"$match\": {\n",
+    "    return data[0] if data else False\n"
-    "            'post_id': post_id,\n",
-    "            'prod': PROD,\n",
-    "        }},\n",
-    "        {\"$sort\": {\"_id\": -1}},\n",
-    "        {\"$group\": {\n",
-    "            \"_id\": \"$post_id\",\n",
-    "            \"doc\": {\"$first\": \"$$ROOT\"}\n",
-    "        }},\n",
-    "        {\"$replaceRoot\": {\"newRoot\": \"$doc\"}},\n",
-    "        {\"$sort\": {\"_id\": -1}},\n",
-    "    ])\n",
-    "\n",
-    "    data = [doc async for doc in cursor]\n",
-    "    return data[0] if data else False"
    ],
    "display_name": ".venv"
   },
@@ -4715,6 +4688,17 @@
    ],
    "display_name": ".venv"
   },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "5d762f57",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# TODO BEGIN"
+   ],
+   "display_name": ".venv"
+  },
   {
    "cell_type": "code",
    "execution_count": null,
@@ -4724,26 +4708,16 @@
    "source": [
     "# @app.get('/profile/delivery/{type_d}/{item_id}')\n",
     "# async def profile_delivery_by_old(request: Request, type_d:str, item_id:str, user: Dict = Security(get_current_user)):\n",
-    "#     cursor = delivery_collection.aggregate([\n",
+    "#     _match = {\n",
-    "#         {\"$match\": {\n",
     "#             \"$or\": [\n",
     "#                 {\"sender._id\": user.get('_id')},\n",
     "#                 {\"user._id\": user.get('_id')},\n",
     "#                 {\"receiver._id\": user.get('_id')}\n",
     "#             ],\n",
     "#             f'{type_d}_id': ObjectId(item_id),\n",
-    "#             'prod': PROD,\n",
+    "#         }\n",
-    "#         }},\n",
-    "#         {\"$sort\": {\"_id\": -1}},\n",
-    "#         {\"$group\": {\n",
-    "#             \"_id\": \"$post_id\",\n",
-    "#             \"doc\": {\"$first\": \"$$ROOT\"}\n",
-    "#         }},\n",
-    "#         {\"$replaceRoot\": {\"newRoot\": \"$doc\"}},\n",
-    "#         {\"$sort\": {\"_id\": -1}},\n",
-    "#     ])\n",
     "\n",
-    "#     data = [doc async for doc in cursor]\n",
+    "#     data = [doc async for doc in get_delivery_aggregate(_match)]\n",
     "#     data.sort(key=lambda doc: doc[\"_id\"].generation_time, reverse=True)\n",
     "\n",
     "#     return templates.TemplateResponse(\n",
@@ -4766,25 +4740,15 @@
    "source": [
     "@app.get('/profile/delivery/competition/{competition_id}')\n",
     "async def profile_delivery_by_competition(request: Request, competition_id:str, user: Dict = Security(get_current_user)):\n",
-    "    cursor = delivery_collection.aggregate([\n",
+    "    _match = {\n",
-    "        {\"$match\": {\n",
     "            \"$or\": [\n",
     "                {\"sender._id\": user.get('_id')},\n",
     "                {\"user._id\": user.get('_id')},\n",
     "                {\"receiver._id\": user.get('_id')}\n",
-    "            ],\n",
+    "            ]\n",
-    "            'prod': PROD,\n",
+    "        }\n",
-    "        }},\n",
-    "        {\"$sort\": {\"_id\": -1}},\n",
-    "        {\"$group\": {\n",
-    "            \"_id\": \"$post_id\",\n",
-    "            \"doc\": {\"$first\": \"$$ROOT\"}\n",
-    "        }},\n",
-    "        {\"$replaceRoot\": {\"newRoot\": \"$doc\"}},\n",
-    "        {\"$sort\": {\"_id\": -1}},\n",
-    "    ])\n",
     "\n",
-    "    data = [doc async for doc in cursor]\n",
+    "    data = [doc async for doc in get_delivery_aggregate(_match)]\n",
     "    data.sort(key=lambda doc: doc[\"_id\"].generation_time, reverse=True)\n",
     "\n",
     "    return templates.TemplateResponse(\n",
@@ -4807,26 +4771,16 @@
    "source": [
     "@app.get('/profile/delivery/exhibition/{exhibition_id}')\n",
     "async def profile_delivery_by_exhibition(request: Request, exhibition_id:str, user: Dict = Security(get_current_user)):\n",
-    "    cursor = delivery_collection.aggregate([\n",
+    "    _match = {\n",
-    "        {\"$match\": {\n",
     "            \"$or\": [\n",
     "                {\"sender._id\": user.get('_id')},\n",
     "                {\"user._id\": user.get('_id')},\n",
     "                {\"receiver._id\": user.get('_id')}\n",
     "            ],\n",
     "            'exhibition_id': exhibition_id,\n",
-    "            'prod': PROD,\n",
+    "        }\n",
-    "        }},\n",
-    "        {\"$sort\": {\"_id\": -1}},\n",
-    "        {\"$group\": {\n",
-    "            \"_id\": \"$post_id\",\n",
-    "            \"doc\": {\"$first\": \"$$ROOT\"}\n",
-    "        }},\n",
-    "        {\"$replaceRoot\": {\"newRoot\": \"$doc\"}},\n",
-    "        {\"$sort\": {\"_id\": -1}},\n",
-    "    ])\n",
     "\n",
-    "    data = [doc async for doc in cursor]\n",
+    "    data = [doc async for doc in get_delivery_aggregate(_match)]\n",
     "    data.sort(key=lambda doc: doc[\"_id\"].generation_time, reverse=True)\n",
     "\n",
     "    return templates.TemplateResponse(\n",
@@ -4840,6 +4794,17 @@
    ],
    "display_name": ".venv"
   },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "e26f23d4",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# TODO END"
+   ],
+   "display_name": ".venv"
+  },
   {
    "cell_type": "code",
    "execution_count": null,
@@ -4876,26 +4841,16 @@
     "    Returns:\n",
     "        HTMLResponse с шаблоном profile/delivery.html\n",
     "    \"\"\"\n",
-    "    # Получаем все доставки пользователя (отправитель ИЛИ получатель)\n",
+    "    _match = {\n",
-    "    cursor = delivery_collection.aggregate([\n",
-    "        {\"$match\": {\n",
     "            \"$or\": [\n",
     "                {\"sender._id\": user.get('_id')},\n",
     "                {\"user._id\": user.get('_id')},\n",
     "                {\"receiver._id\": user.get('_id')}\n",
-    "            ],\n",
+    "            ]\n",
-    "            'prod': PROD,\n",
+    "        }\n",
-    "        }},\n",
+    "    \n",
-    "        {\"$sort\": {\"_id\": -1}},\n",
+    "    data = [doc async for doc in get_delivery_aggregate(_match)]\n",
-    "        {\"$group\": {\n",
-    "            \"_id\": \"$post_id\",\n",
-    "            \"doc\": {\"$first\": \"$$ROOT\"}\n",
-    "        }},\n",
-    "        {\"$replaceRoot\": {\"newRoot\": \"$doc\"}},\n",
-    "        {\"$sort\": {\"_id\": -1}},\n",
-    "    ])\n",
     "\n",
-    "    data = [doc async for doc in cursor]\n",
     "    exhibitions = await exhibition_collection.find({'_id': {'$in': [ObjectId(i.get('exhibition_id')) for i in data if i.get('exhibition_id')]}}).to_list()\n",
     "    competitions = await competitions_col.find({'_id': {'$in': [i.get('competition_id') for i in data if i.get('competition_id')]}}).to_list()\n",
     "    return templates.TemplateResponse(\n",

main.py

@@ -81,6 +81,25 @@ def transliterate(text: str) -> str:
     return ''.join(RUS_TO_LAT.get(ch, '' if ch not in string.ascii_letters else ch) for ch in text)
 
 
+def get_delivery_aggregate(_filter: dict = {}, _status: str = None):
+    cursor = delivery_collection.aggregate([
+        {"$match": {
+            "prod": PROD, **_filter
+        }},
+        {"$sort": {"_id": -1}},
+        {"$group": {
+            "_id": "$post_id",
+            "doc": {"$first": "$$ROOT"}
+        }},
+        {"$replaceRoot": {"newRoot": "$doc"}},
+        {"$sort": {"_id": -1}},
+    ])
+    if _status:
+        return (doc async for doc in cursor if doc.get('status') == _status)
+    else:
+        return cursor
+
+
 async def get_latest_news() -> List[dict]:
     cursor = news_collection.find({'publish_on_platform': True}).sort("_id", -1).limit(30)
     docs: List[dict] = []
@@ -353,7 +372,6 @@ async def cron_every_five_min(token: str = Security(get_token)):
 
     prod_filter = {"prod": PROD}
 
-    # 1) If there's an exhibition that is going to be finished soon -> move deliveries to "Возврат художнику"
     exhibition_finished = await exhibition_collection.find_one_and_update(
         {**prod_filter, "status": 1, "date_finish": {"$lt": limit_dt}},
         {"$set": {"status": 2}},
@@ -361,20 +379,9 @@ async def cron_every_five_min(token: str = Security(get_token)):
     )
     if exhibition_finished:
         exhibition_id_str = str(exhibition_finished["_id"])
-        cursor = delivery_collection.aggregate([
-            {"$match": {
-                "exhibition_id": exhibition_id_str, **prod_filter
-            }},
-            {"$sort": {"_id": -1}},
-            {"$group": {
-                "_id": "$post_id",
-                "doc": {"$first": "$$ROOT"}
-            }},
-            {"$replaceRoot": {"newRoot": "$doc"}},
-            {"$sort": {"_id": -1}},
-        ])
 
-        data = [doc async for doc in cursor if doc.get('status') == 'На выставке']
+        data = [doc async for doc in get_delivery_aggregate({"exhibition_id": exhibition_id_str}, 'На выставке')]
+
         for doc in data:
             post_id = doc.get("post_id")
             exhibition_id = doc.get("exhibition_id")
@@ -399,27 +406,25 @@ async def cron_every_five_min(token: str = Security(get_token)):
         return {}
     tasks = []
     async for exhibition_to_start in exhibition_collection.find(
-        {**prod_filter}
+        {"status": {'$lte':1}, **prod_filter}
     ):
         exhibition_id_str = str(exhibition_to_start["_id"])
         exhibition_title = exhibition_to_start.get("title")
 
         notified_user_id = None
-        notified_post_user_id = None
+        notified_post_user_ids = {}
 
-        async for doc in delivery_collection.find(
+        data = [doc async for doc in get_delivery_aggregate({"exhibition_id": exhibition_id_str}, 'Получено')]
-            {"exhibition_id": exhibition_id_str, "status": "Получено", **prod_filter},
+        
-            {"_id": 0},
+        for doc in data:
-        ):
             post_id = doc.get("post_id")
             exhibition_id = doc.get("exhibition_id")
 
             post = await posts_collection.find_one({"_id": ObjectId(post_id)})
 
-            # Use user from post; keep them even if multiple deliveries exist
+            if post and post.get("user"):
-            if post and post.get("user") and not notified_post_user_id:
+                notified_post_user_ids.update({str(post["user"].get("_id")):1})
-                notified_post_user_id = str(post["user"].get("_id"))
+            
-
             new_doc = {
                 **doc,
                 "exhibition": exhibition_to_start,
@@ -439,12 +444,10 @@ async def cron_every_five_min(token: str = Security(get_token)):
                 {"$set": {"status": "На выставке"}}
             ))
 
-        # NOTE: your original code used `post` from inside the loop (could be last item / even undefined).
+        for user_id in notified_post_user_ids.keys():
-        # This version notifies once using the first found post user.
-        if notified_post_user_id:
             await notify_in_profile(
                 NotifyItem(
-                    user_id=notified_post_user_id,
+                    user_id=user_id,
                     type_item="exhibtion",
                     title=f"Выставка {exhibition_title}",
                     link="/profile/exhibition",
@@ -453,7 +456,7 @@ async def cron_every_five_min(token: str = Security(get_token)):
             )
     if tasks:
         asyncio.gather(*tasks)
-    # 2) Otherwise find an exhibition that should start soon -> move deliveries to "На выставке"
+
     exhibition_to_start = await exhibition_collection.find_one_and_update(
         {**prod_filter, "status": 0, "datetime_start": {"$lt": limit_dt}},
         {"$set": {"status": 1}},
@@ -463,7 +466,6 @@ async def cron_every_five_min(token: str = Security(get_token)):
     if not exhibition_to_start:
         return {}
 
-    # If vk/tg arrays are empty -> send social invitation
     if exhibition_to_start.get("vk") == [] or exhibition_to_start.get("tg") == []:
         await service_social_exhibition(bson_clear(exhibition_to_start))
     return {}
@@ -1186,7 +1188,8 @@ async def get_unique_posts_by_user(
     ]
 
     cursor = competition_request_col.aggregate(pipeline)
-    return await cursor.to_list(length=None)
+    data = await cursor.to_list(length=None)
+    return await posts_collection.find({"_id": {"$in":[ ObjectId(doc['_id']) if type(doc['_id']) is str else doc['_id'] for doc in data ]}}).to_list()
 
 async def get_unique_posts_с(competition_request_col: AsyncIOMotorCollection, competition_id: str, finish: bool = False) -> List[Any]:
     match_criteria = {
@@ -1221,22 +1224,8 @@ async def exhibition_members(
 ):
     exhibition = await exhibition_collection.find_one({'_id': ObjectId(exhibition_id)})
     if exhibition:
-        match_ = {
+        posts = [doc async for doc in get_delivery_aggregate({"exhibition_id": exhibition_id}, 'На выставке')]
-            "exhibition_id": exhibition_id,
+        
-            'prod': PROD,
-        }
-
-        cursor = delivery_collection.aggregate([
-            {"$match": match_},
-            {"$sort": {"_id": -1}},
-            {"$group": {
-                "_id": "$post_id",
-                "doc": {"$first": "$$ROOT"}
-            }},
-            {"$replaceRoot": {"newRoot": "$doc"}},
-            {"$sort": {"_id": -1}},
-        ])
-        posts = [doc async for doc in cursor if doc.get('status') == 'На выставке']
         return templates.TemplateResponse(
             request=request, name="exhibition/index.html", context={"id": 1, "user": user, "exhibition": exhibition, "posts": posts, 'now': datetime.now()}
         )
@@ -3339,25 +3328,15 @@ async def profile_invitation(
             {"sender._id": user.get('_id')},
             {"user._id": user.get('_id')},
             {"receiver._id": user.get('_id')}
-        ],
+        ]
-        'prod': PROD,
     }
     if data.get('competition'):
         match_.update({'competition_id': data.get('competition').get('_id')})
     if data.get('exhibition_id'):
         match_.update({'exhibition_id': data.get('exhibition_id')})
-    cursor = delivery_collection.aggregate([
+    
-        {"$match": match_},
+    posts = [doc async for doc in get_delivery_aggregate(match_)]
-        {"$sort": {"_id": -1}},
-        {"$group": {
-            "_id": "$post_id",
-            "doc": {"$first": "$$ROOT"}
-        }},
-        {"$replaceRoot": {"newRoot": "$doc"}},
-        {"$sort": {"_id": -1}},
-    ])
 
-    posts = [doc async for doc in cursor]
     return templates.TemplateResponse(
         request=request, name="profile/invitation.html", context={ "user": user, "doc": data, "data": posts}
     )
@@ -3383,24 +3362,12 @@ async def social_setting(
     await social_collection.update_one({'_id': ObjectId(id), 'user_id': user.get('_id')},{'$set': {key:check}})
     return {'ok': True}
 
-async def get_delivery_post(post_id):
-    cursor = delivery_collection.aggregate([
-        {"$match": {
-            'post_id': post_id,
-            'prod': PROD,
-        }},
-        {"$sort": {"_id": -1}},
-        {"$group": {
-            "_id": "$post_id",
-            "doc": {"$first": "$$ROOT"}
-        }},
-        {"$replaceRoot": {"newRoot": "$doc"}},
-        {"$sort": {"_id": -1}},
-    ])
 
-    data = [doc async for doc in cursor]
+async def get_delivery_post(post_id):
+    data = [doc async for doc in get_delivery_aggregate({'post_id': post_id})]
     return data[0] if data else False
 
+
 @app.post('/api/v1/views')
 async def get_views_count(request: Request):
     url = (await request.json())['url']
@@ -3608,28 +3575,20 @@ async def profile_delivery_history_post(request: Request, post_id:str, user: Dic
     )
 
 
+# TODO BEGIN
+
 # @app.get('/profile/delivery/{type_d}/{item_id}')
 # async def profile_delivery_by_old(request: Request, type_d:str, item_id:str, user: Dict = Security(get_current_user)):
-#     cursor = delivery_collection.aggregate([
+#     _match = {
-#         {"$match": {
 #             "$or": [
 #                 {"sender._id": user.get('_id')},
 #                 {"user._id": user.get('_id')},
 #                 {"receiver._id": user.get('_id')}
 #             ],
 #             f'{type_d}_id': ObjectId(item_id),
-#             'prod': PROD,
+#         }
-#         }},
-#         {"$sort": {"_id": -1}},
-#         {"$group": {
-#             "_id": "$post_id",
-#             "doc": {"$first": "$$ROOT"}
-#         }},
-#         {"$replaceRoot": {"newRoot": "$doc"}},
-#         {"$sort": {"_id": -1}},
-#     ])
 
-#     data = [doc async for doc in cursor]
+#     data = [doc async for doc in get_delivery_aggregate(_match)]
 #     data.sort(key=lambda doc: doc["_id"].generation_time, reverse=True)
 
 #     return templates.TemplateResponse(
@@ -3644,25 +3603,15 @@ async def profile_delivery_history_post(request: Request, post_id:str, user: Dic
 
 @app.get('/profile/delivery/competition/{competition_id}')
 async def profile_delivery_by_competition(request: Request, competition_id:str, user: Dict = Security(get_current_user)):
-    cursor = delivery_collection.aggregate([
+    _match = {
-        {"$match": {
             "$or": [
                 {"sender._id": user.get('_id')},
                 {"user._id": user.get('_id')},
                 {"receiver._id": user.get('_id')}
-            ],
+            ]
-            'prod': PROD,
+        }
-        }},
-        {"$sort": {"_id": -1}},
-        {"$group": {
-            "_id": "$post_id",
-            "doc": {"$first": "$$ROOT"}
-        }},
-        {"$replaceRoot": {"newRoot": "$doc"}},
-        {"$sort": {"_id": -1}},
-    ])
 
-    data = [doc async for doc in cursor]
+    data = [doc async for doc in get_delivery_aggregate(_match)]
     data.sort(key=lambda doc: doc["_id"].generation_time, reverse=True)
 
     return templates.TemplateResponse(
@@ -3677,26 +3626,16 @@ async def profile_delivery_by_competition(request: Request, competition_id:str,
 
 @app.get('/profile/delivery/exhibition/{exhibition_id}')
 async def profile_delivery_by_exhibition(request: Request, exhibition_id:str, user: Dict = Security(get_current_user)):
-    cursor = delivery_collection.aggregate([
+    _match = {
-        {"$match": {
             "$or": [
                 {"sender._id": user.get('_id')},
                 {"user._id": user.get('_id')},
                 {"receiver._id": user.get('_id')}
             ],
             'exhibition_id': exhibition_id,
-            'prod': PROD,
+        }
-        }},
-        {"$sort": {"_id": -1}},
-        {"$group": {
-            "_id": "$post_id",
-            "doc": {"$first": "$$ROOT"}
-        }},
-        {"$replaceRoot": {"newRoot": "$doc"}},
-        {"$sort": {"_id": -1}},
-    ])
 
-    data = [doc async for doc in cursor]
+    data = [doc async for doc in get_delivery_aggregate(_match)]
     data.sort(key=lambda doc: doc["_id"].generation_time, reverse=True)
 
     return templates.TemplateResponse(
@@ -3708,6 +3647,8 @@ async def profile_delivery_by_exhibition(request: Request, exhibition_id:str, us
         }
     )
 
+# TODO END
+
 @app.get(
     "/profile/delivery",
     response_class=HTMLResponse,
@@ -3737,26 +3678,16 @@ async def profile_delivery(
     Returns:
         HTMLResponse с шаблоном profile/delivery.html
     """
-    # Получаем все доставки пользователя (отправитель ИЛИ получатель)
+    _match = {
-    cursor = delivery_collection.aggregate([
-        {"$match": {
             "$or": [
                 {"sender._id": user.get('_id')},
                 {"user._id": user.get('_id')},
                 {"receiver._id": user.get('_id')}
-            ],
+            ]
-            'prod': PROD,
+        }
-        }},
+    
-        {"$sort": {"_id": -1}},
+    data = [doc async for doc in get_delivery_aggregate(_match)]
-        {"$group": {
-            "_id": "$post_id",
-            "doc": {"$first": "$$ROOT"}
-        }},
-        {"$replaceRoot": {"newRoot": "$doc"}},
-        {"$sort": {"_id": -1}},
-    ])
 
-    data = [doc async for doc in cursor]
     exhibitions = await exhibition_collection.find({'_id': {'$in': [ObjectId(i.get('exhibition_id')) for i in data if i.get('exhibition_id')]}}).to_list()
     competitions = await competitions_col.find({'_id': {'$in': [i.get('competition_id') for i in data if i.get('competition_id')]}}).to_list()
     return templates.TemplateResponse(

tasks.ipynb

@@ -69,7 +69,7 @@
    "outputs": [],
    "source": [
     "async def notify_in_profile(item: NotifyItem):\n",
-    "    return await notify_collection.update_one(\n",
+    "    result = await notify_collection.update_one(\n",
     "        {\n",
     "            \"user_id\": item.user_id,\n",
     "            \"type\": item.type_item,\n",
@@ -89,7 +89,43 @@
     "            }\n",
     "        },\n",
     "        upsert=True,\n",
-    "    )\n"
+    "    )\n",
+    "\n",
+    "    if result.upserted_id is None:\n",
+    "        return result  # Existing notification was updated; no email\n",
+    "\n",
+    "    user = await users_collection.find_one(\n",
+    "        {\"_id\": ObjectId(str(item.user_id))},\n",
+    "        {\"email\": 1},\n",
+    "    )\n",
+    "\n",
+    "    if not user:\n",
+    "        return result\n",
+    "\n",
+    "    token = await token_create(\n",
+    "        str(user[\"_id\"]),\n",
+    "        url=\"/profile\",\n",
+    "    )\n",
+    "    login_link = f\"{FULL_DOMAIN}/login/{token}\"\n",
+    "\n",
+    "    if PROD:\n",
+    "        email_task.notify.delay(\n",
+    "            user[\"email\"],\n",
+    "            {\n",
+    "                \"text\": item.text,\n",
+    "                \"list\": [[]],\n",
+    "                \"primary\": {\n",
+    "                    \"href\": login_link,\n",
+    "                    \"text\": \"Войти и посмотреть\",\n",
+    "                },\n",
+    "                \"title\": item.title,\n",
+    "                \"login_link\": login_link,\n",
+    "            },\n",
+    "            \"Уведомление\",\n",
+    "            \"main.html\",\n",
+    "        )\n",
+    "\n",
+    "    return result\n"
    ],
    "display_name": ".venv"
   },

tasks.py

@@ -12,7 +12,7 @@ import asyncio
 
 
 async def notify_in_profile(item: NotifyItem):
-    return await notify_collection.update_one(
+    result = await notify_collection.update_one(
         {
             "user_id": item.user_id,
             "type": item.type_item,
@@ -34,6 +34,42 @@ async def notify_in_profile(item: NotifyItem):
         upsert=True,
     )
 
+    if result.upserted_id is None:
+        return result  # Existing notification was updated; no email
+
+    user = await users_collection.find_one(
+        {"_id": ObjectId(str(item.user_id))},
+        {"email": 1},
+    )
+
+    if not user:
+        return result
+
+    token = await token_create(
+        str(user["_id"]),
+        url="/profile",
+    )
+    login_link = f"{FULL_DOMAIN}/login/{token}"
+
+    if PROD:
+        email_task.notify.delay(
+            user["email"],
+            {
+                "text": item.text,
+                "list": [[]],
+                "primary": {
+                    "href": login_link,
+                    "text": "Войти и посмотреть",
+                },
+                "title": item.title,
+                "login_link": login_link,
+            },
+            "Уведомление",
+            "main.html",
+        )
+
+    return result
+
 
 async def email_new_post(new_post: Dict):
     async for user in users_collection.find({'roles':'partner'}):

vk.py

@@ -188,6 +188,7 @@ async def send_vk(data:dict, request: Request):
     return {"ok": True}
 
 
+
 # TODO: function call like send_tg from tg.py (without router)
 @router.post("/edit")
 async def edit_vk(data:dict, request: Request):
@@ -257,6 +258,7 @@ async def edit_vk(data:dict, request: Request):
     return {"ok": True}
 
 
+
 # TODO: function call like send_tg from tg.py (without router)
 @router.post("/delete")
 async def wall_delete(data: dict, request: Request):